feat: setup support for multi process workers

Initial support for multiproccess workers

Signed-off-by: jcriadomarco <jcriadomarco@vmware.com>

Author: javiercri

base-image/Makefile

@@ -20,7 +20,6 @@ run: build-image
 	  -v `pwd`/failsafe.conf:/fluentd/etc/fluent.conf \
 	  -v /var/log:/var/log \
 	  -v /var/run:/var/run \
-	  -e FLUENTD_OPT="--no-supervisor" \
 	  $(IMAGE):$(TAG)
 
 test: build-image
@@ -29,9 +28,9 @@ test: build-image
 	  -v `pwd`:/workspace \
 	  -v `pwd/plugins`:/fluentd/plugins \
 	  -v `pwd`/test/local.conf:/fluentd/etc/fluent.conf \
-	  -e FLUENTD_OPT="--no-supervisor" \
-		-e PAPERTRAIL_PORT=$$PAPERTRAIL_PORT \
-		-e PAPERTRAIL_HOST=$$PAPERTRAIL_HOST \
+	  -p 0.0.0.0:24231:24231 \
+	  -e PAPERTRAIL_PORT=$$PAPERTRAIL_PORT \
+	  -e PAPERTRAIL_HOST=$$PAPERTRAIL_HOST \
 	  $(IMAGE):$(TAG)
 
 list-gems:

config-reloader/config/config.go

@@ -56,6 +56,7 @@ type Config struct {
 	ParsedMetaValues    map[string]string
 	ParsedLabelSelector labels.Set
 	ExecTimeoutSeconds  int
+	SplitPattern        string
 }
 
 var defaultConfig = &Config{
@@ -78,6 +79,7 @@ var defaultConfig = &Config{
 	MetricsPort:          9000,
 	AdminNamespace:       "kube-system",
 	ExecTimeoutSeconds:   30,
+	SplitPattern:         "infra-system",
 }
 
 var reValidID = regexp.MustCompile("([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]")
@@ -250,6 +252,8 @@ func (cfg *Config) ParseFlags(args []string) error {
 
 	app.Flag("admin-namespace", "Configurations defined in this namespace are copied as is, without further processing. Virtual plugins can also be defined in this namespace").Default(defaultConfig.AdminNamespace).StringVar(&cfg.AdminNamespace)
 
+	app.Flag("split-pattern", "Configurations defined special pattern for different workers").Default(defaultConfig.SplitPattern).StringVar(&cfg.SplitPattern)
+
 	app.Flag("exec-timeout", "Timeout duration (in seconds) for exec command during validation").Default(strconv.Itoa(defaultConfig.ExecTimeoutSeconds)).IntVar(&cfg.ExecTimeoutSeconds)
 	_, err := app.Parse(args)
 

config-reloader/fluentd/reloader.go

@@ -33,7 +33,7 @@ func (r *Reloader) ReloadConfiguration() {
 
 	logrus.Infof("Reloading fluentd configuration gracefully via POST to /api/config.gracefulReload")
 
-	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/api/config.gracefulReload", r.port))
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/api/config.reload", r.port))
 	if err != nil {
 		logrus.Errorf("fluentd config.gracefulReload request failed: %+v", err)
 	} else if resp.StatusCode != 200 {

config-reloader/fluentd/reloader_test.go

@@ -25,7 +25,7 @@ func TestReloaderCalls(t *testing.T) {
 
 	handler := func(w http.ResponseWriter, r *http.Request) {
 		fmt.Printf("req %+v", r)
-		if r.Method == "GET" && r.RequestURI == "/api/config.gracefulReload" {
+		if r.Method == "GET" && r.RequestURI == "/api/config.reload" {
 			counter++
 		}
 	}

config-reloader/generator/generator.go

@@ -337,9 +337,11 @@ func (g *generatorInstance) renderIncludableFile(templateFile string, dest strin
 	model := struct {
 		ID                string
 		PrometheusEnabled bool
+		SplitPattern      string
 	}{
 		ID:                util.MakeFluentdSafeName(g.cfg.ID),
 		PrometheusEnabled: g.cfg.PrometheusEnabled,
+		SplitPattern:      g.cfg.SplitPattern,
 	}
 
 	err = util.TemplateAndWriteFile(tmpl, model, dest)

config-reloader/templates/fluent.conf

@@ -5,6 +5,7 @@
 
   # needed to enable /api/config.reload
   rpc_endpoint 127.0.0.1:24444
+  workers 3
 </system>
 
 # you can turn this on for debug

config-reloader/templates/kubernetes-postprocess.conf

@@ -36,12 +36,14 @@
 # Remove the unnecessary field as the information is already available on other fields.
 <filter kube.*.*.*>
   @type record_modifier
+  @id k8s_remove_unused_field
   remove_keys $.kubernetes.pod_id, $.kubernetes.master_url, $.kubernetes.container_image_id, $.kubernetes.namespace_id, kubernetes_namespace_container_name, $.kubernetes.labels.pod-template-generation, $.kubernetes.labels.controller-revision-hash, $.kubernetes.labels.pod-template-hash
 </filter>
 
 # Parse logs in the kube-system namespace using the kubernetes formatter.
 <filter kube.kube-system.**>
   @type parser
+  @id k8s_parser_kube_system
   reserve_data true
   key_name log
   emit_invalid_record_to_error false