fluentbit forwarder: refactor before introducing new output

The forward input/output is a more solid data exchange protocol between
fluent bit instances. We will abandon the HTTP output in favor of this.
The first step is to refactor some code to support the cutoff.

Change-Id: Idec1f76724bb96782b7fe87d6bbc0440a73edf7f

api/v1/softwarefactory_types.go

@@ -43,6 +43,11 @@ type FluentBitForwarderSpec struct {
 	// +optional
 	// Run fluent bit sidecars in debug mode. This will output forwarded payloads and additional info in the sidecar's logs. Defaults to false.
 	Debug *bool `json:"debug,omitempty"`
+	// The Host for the Fluent Bit Forward Input to forward logs to.
+	ForwardInputHost string `json:"forwardInputHost,omitempty"`
+	// The (optional) port of the forward input, defaults to 24224.
+	// +kubebuilder:default:=24224
+	ForwardInputPort int32 `json:"forwardInputPort,omitempty"`
 }
 
 type StorageSpec struct {

config/crd/bases/sf.softwarefactory-project.io_softwarefactories.yaml

@@ -62,6 +62,16 @@ spec:
                       output forwarded payloads and additional info in the sidecar's
                       logs. Defaults to false.
                     type: boolean
+                  forwardInputHost:
+                    description: The Host for the Fluent Bit Forward Input to forward
+                      logs to.
+                    type: string
+                  forwardInputPort:
+                    default: 24224
+                    description: The (optional) port of the forward input, defaults
+                      to 24224.
+                    format: int32
+                    type: integer
                   httpInputHost:
                     description: The Host for the Fluent Bit HTTP Input to forward
                       logs to.

controllers/libs/logging/logging.go

@@ -23,10 +23,17 @@ type FluentBitLabel struct {
 	Value string
 }
 
-type PythonTemplateLoggingParams struct {
-	LogLevel    string
-	ForwardLogs bool
-	BaseURL     string
+type TemplateInputParams struct {
+	InUse bool
+	Host  string
+	Port  string
+}
+
+type TemplateLoggingParams struct {
+	Tag                string
+	LogLevel           string
+	HTTPInputConfig    TemplateInputParams
+	ForwardInputConfig TemplateInputParams
 }
 
 func CreateForwarderEnvVars(name string, extraLabels []FluentBitLabel) []apiv1.EnvVar {
@@ -45,9 +52,46 @@ func CreateForwarderEnvVars(name string, extraLabels []FluentBitLabel) []apiv1.E
 	return forwarderEnvVars
 }
 
+func CreateForwarderConfigTemplateParams(tag string, forwarderSpec *v1.FluentBitForwarderSpec) TemplateLoggingParams {
+	var httpInputParams = TemplateInputParams{
+		InUse: false,
+		Host:  "",
+		Port:  "",
+	}
+	var forwardInputParams = TemplateInputParams{
+		InUse: false,
+		Host:  "",
+		Port:  "",
+	}
+	var loggingParams = TemplateLoggingParams{
+		Tag:                tag,
+		LogLevel:           "info",
+		HTTPInputConfig:    httpInputParams,
+		ForwardInputConfig: forwardInputParams,
+	}
+	if forwarderSpec != nil {
+		if forwarderSpec.HTTPInputHost != "" {
+			loggingParams.HTTPInputConfig.InUse = true
+			loggingParams.HTTPInputConfig.Host = forwarderSpec.HTTPInputHost
+			loggingParams.HTTPInputConfig.Port = strconv.Itoa(int(forwarderSpec.HTTPInputPort))
+		}
+		if forwarderSpec.ForwardInputHost != "" {
+			loggingParams.ForwardInputConfig.InUse = true
+			loggingParams.ForwardInputConfig.Host = forwarderSpec.ForwardInputHost
+			loggingParams.ForwardInputConfig.Port = strconv.Itoa(int(forwarderSpec.ForwardInputPort))
+		}
+
+		if forwarderSpec.Debug != nil && *forwarderSpec.Debug {
+			loggingParams.LogLevel = "debug"
+		}
+	}
+	return loggingParams
+}
+
 func SetupLogForwarding(serviceName string, forwarderSpec *v1.FluentBitForwarderSpec, extraLabels []FluentBitLabel, annotations map[string]string) []apiv1.EnvVar {
 	if forwarderSpec != nil {
 		annotations["log-forwarding"] = forwarderSpec.HTTPInputHost + ":" + strconv.Itoa(int(forwarderSpec.HTTPInputPort))
+		annotations["log-forwarding"] += forwarderSpec.ForwardInputHost + ":" + strconv.Itoa(int(forwarderSpec.ForwardInputPort))
 		return CreateForwarderEnvVars(serviceName, extraLabels)
 	} else {
 		annotations["log-forwarding"] = "disabled"

controllers/mariadb.go

@@ -47,18 +47,14 @@ type ZuulDBOpts struct {
 func createLogForwarderSidecar(r *SFController, annotations map[string]string) (apiv1.Volume, apiv1.Container) {
 
 	fbForwarderConfig := make(map[string]string)
-	var fbLogLevel = "info"
-	if r.cr.Spec.FluentBitLogForwarding.Debug != nil && *r.cr.Spec.FluentBitLogForwarding.Debug {
-		fbLogLevel = "debug"
-	}
+	var loggingParams = logging.CreateForwarderConfigTemplateParams("mariadb", r.cr.Spec.FluentBitLogForwarding)
+
 	fbForwarderConfig["fluent-bit.conf"], _ = utils.ParseString(
 		mariadbFluentBitForwarderConfig,
 		struct {
-			ExtraKeys              []logging.FluentBitLabel
-			FluentBitHTTPInputHost string
-			FluentBitHTTPInputPort string
-			LogLevel               string
-		}{[]logging.FluentBitLabel{}, r.cr.Spec.FluentBitLogForwarding.HTTPInputHost, strconv.Itoa(int(r.cr.Spec.FluentBitLogForwarding.HTTPInputPort)), fbLogLevel})
+			ExtraKeys     []logging.FluentBitLabel
+			LoggingParams logging.TemplateLoggingParams
+		}{[]logging.FluentBitLabel{}, loggingParams})
 	r.EnsureConfigMap("fluentbit-mariadb-cfg", fbForwarderConfig)
 
 	volume := base.MkVolumeCM("mariadb-log-forwarder-config",

controllers/nodepool.go

@@ -94,19 +94,15 @@ var nodepoolFluentBitLabels = []logging.FluentBitLabel{
 }
 
 func createImageBuildLogForwarderSidecar(r *SFController, annotations map[string]string) (apiv1.Volume, apiv1.Container) {
-	var fbLogLevel = "info"
-	if r.cr.Spec.FluentBitLogForwarding.Debug != nil && *r.cr.Spec.FluentBitLogForwarding.Debug {
-		fbLogLevel = "debug"
-	}
 	fbForwarderConfig := make(map[string]string)
+	var loggingParams = logging.CreateForwarderConfigTemplateParams("diskimage-builder", r.cr.Spec.FluentBitLogForwarding)
+
 	fbForwarderConfig["fluent-bit.conf"], _ = utils.ParseString(
 		fluentBitForwarderConfig,
 		struct {
-			ExtraKeys              []logging.FluentBitLabel
-			FluentBitHTTPInputHost string
-			FluentBitHTTPInputPort string
-			LogLevel               string
-		}{[]logging.FluentBitLabel{}, r.cr.Spec.FluentBitLogForwarding.HTTPInputHost, strconv.Itoa(int(r.cr.Spec.FluentBitLogForwarding.HTTPInputPort)), fbLogLevel})
+			ExtraKeys     []logging.FluentBitLabel
+			LoggingParams logging.TemplateLoggingParams
+		}{[]logging.FluentBitLabel{}, loggingParams})
 	fbForwarderConfig["parsers.conf"] = fluentBitForwarderParsersConfig
 	r.EnsureConfigMap("fluentbit-dib-cfg", fbForwarderConfig)
 
@@ -197,20 +193,16 @@ func (r *SFController) mkLoggingTemplate(serviceName string) (string, error) {
 		selectedLogLevel = logLevel
 	}
 
-	var forwardLogs = false
-	var inputBaseURL = ""
-	if r.cr.Spec.FluentBitLogForwarding != nil {
-		forwardLogs = true
-		inputBaseURL = "http://" + r.cr.Spec.FluentBitLogForwarding.HTTPInputHost + ":" + strconv.Itoa(int(r.cr.Spec.FluentBitLogForwarding.HTTPInputPort))
-	}
+	var loggingParams = logging.CreateForwarderConfigTemplateParams("nodepool."+serviceName, r.cr.Spec.FluentBitLogForwarding)
+	// Change logLevel to what we actually want
+	loggingParams.LogLevel = string(selectedLogLevel)
 
 	loggingConfig, err := utils.ParseString(
 		loggingConfigTemplate,
-		logging.PythonTemplateLoggingParams{
-			LogLevel:    string(selectedLogLevel),
-			ForwardLogs: forwardLogs,
-			BaseURL:     inputBaseURL,
-		})
+		struct {
+			ExtraKeys     []logging.FluentBitLabel
+			LoggingParams logging.TemplateLoggingParams
+		}{[]logging.FluentBitLabel{}, loggingParams})
 
 	return loggingConfig, err
 }

controllers/static/mariadb/fluentbit/fluent-bit.conf.tmpl

@@ -1,10 +1,10 @@
 [SERVICE]
     http_server On
     http_port   2020
-    log_level   {{ .LogLevel }}
+    log_level   {{ .LoggingParams.LogLevel }}
 [INPUT]
     name                tail
-    tag                 ${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.mariadb
+    tag                 mariadb
     path                /watch/*.log
     path_key            full_path
     refresh_interval    5
@@ -24,14 +24,18 @@
 {{- range .ExtraKeys }}
     add     {{ .Key }} ${K8S_{{ .Value -}}}
 {{- end }}
+{{- if eq .LoggingParams.LogLevel "debug" }}
 [OUTPUT]
     name    stdout
     match   *
     format  json_lines
+{{ end }}
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
 [OUTPUT]
     name    http
     match   *
-    uri     /${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.mariadb
+    uri     /{{ .LoggingParams.Tag }}
     format  json
-    host    {{ .FluentBitHTTPInputHost }}
-    port    {{ .FluentBitHTTPInputPort }}
+    host    {{ .LoggingParams.HTTPInputConfig.Host }}
+    port    {{ .LoggingParams.HTTPInputConfig.Port }}
+{{ end }}

controllers/static/nodepool/fluentbit/fluent-bit.conf.tmpl

@@ -1,11 +1,11 @@
 [SERVICE]
     http_server On
     http_port   2020
-    log_level   {{ .LogLevel }}
+    log_level   {{ .LoggingParams.LogLevel }}
     parsers_file    parsers.conf
 [INPUT]
     name                tail
-    tag                 ${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.dib
+    tag                 diskimage-builder
     path                /watch/*/*.log
     path_key            full_path
     refresh_interval    5
@@ -27,21 +27,21 @@
 {{- end }}
 [FILTER]
     Name parser
-    Match *dib
+    Match *
     Key_Name full_path
     Parser filename_shortener
     Preserve_Key On
     Reserve_Data On
 [FILTER]
     Name parser
-    Match *dib
+    Match *
     Key_Name file
     Parser image_name
     Preserve_Key On
     Reserve_Data On
 [FILTER]
     Name parser
-    Match *dib
+    Match *
     Key_Name log
     Parser timestamp_callback
     Preserve_Key On
@@ -50,14 +50,18 @@
     Name modify
     Match *dib
     Remove full_path
+{{- if eq .LoggingParams.LogLevel "debug" }}
 [OUTPUT]
     name    stdout
     match   *
     format  json_lines
+{{ end }}
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
 [OUTPUT]
     name    http
     match   *
-    uri     /${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.dib
+    uri     /{{ .LoggingParams.Tag }}
     format  json
-    host    {{ .FluentBitHTTPInputHost }}
-    port    {{ .FluentBitHTTPInputPort }}
+    host    {{ .LoggingParams.HTTPInputConfig.Host }}
+    port    {{ .LoggingParams.HTTPInputConfig.Port }}
+{{ end }}

controllers/static/nodepool/logging.yaml.tmpl

@@ -3,37 +3,37 @@ formatters:
   console:
     class: 'nodepool.logconfig.MultiLineFormatter'
     format: "%(asctime)s %(levelname)s %(name)s: %(message)s"
-{{- if .ForwardLogs }}
-  logforward:
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
+  fluenthttp:
     format: '%(asctime)s %(levelname)s %(name)s: %(message)s'
 {{ end }}
 handlers:
   console:
     class: logging.StreamHandler
     formatter: console
-    level: {{.LogLevel}}
+    level: {{ .LoggingParams.LogLevel}}
     stream: ext://sys.stdout
-{{- if .ForwardLogs }}
-  logforward:
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
+  fluenthttp:
     class: sfExtras.SimpleFluentBitHTTPInputHandler
-    level: {{ .LogLevel }}
-    formatter:
-    url: '{{ .BaseURL }}/nodepool'
+    level: {{ .LoggingParams.LogLevel }}
+    formatter: fluenthttp
+    url: 'http://{{ .LoggingParams.HTTPInputConfig.Host }}:{{ .LoggingParams.HTTPInputConfig.Port }}/{{ .LoggingParams.Tag }}'
     env_prefix: K8S_
 {{ end }}
 loggers:
   nodepool:
     handlers:
       - console
-{{- if .ForwardLogs }}
-      - logforward
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
+      - fluenthttp
 {{ end }}
-    level: {{.LogLevel}}
+    level: {{.LoggingParams.LogLevel}}
     propagate: 0
 root:
   handlers:
     - console
-{{- if .ForwardLogs }}
-    - logforward
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
+    - fluenthttp
 {{ end }}
-  level: {{.LogLevel}}
+  level: {{.LoggingParams.LogLevel}}

controllers/static/zookeeper/fluent-bit.conf.tmpl

@@ -1,10 +1,10 @@
 [SERVICE]
     http_server On
     http_port   2020
-    log_level   {{ .LogLevel }}
+    log_level   {{ .LoggingParams.LogLevel }}
 [INPUT]
     name                tail
-    tag                 ${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.zookeeper
+    tag                 zookeeper
     path                /watch/*.log
     path_key            full_path
     refresh_interval    5
@@ -24,14 +24,18 @@
 {{- range .ExtraKeys }}
     add     {{ .Key }} ${K8S_{{ .Value -}}}
 {{- end }}
+{{- if eq .LoggingParams.LogLevel "debug" }}
 [OUTPUT]
     name    stdout
     match   *
     format  json_lines
+{{ end }}
+{{- if .LoggingParams.HTTPInputConfig.InUse }}
 [OUTPUT]
     name    http
     match   *
-    uri     /${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.dib
+    uri     /{{ .LoggingParams.Tag }}
     format  json
-    host    {{ .FluentBitHTTPInputHost }}
-    port    {{ .FluentBitHTTPInputPort }}
+    host    {{ .LoggingParams.HTTPInputConfig.Host }}
+    port    {{ .LoggingParams.HTTPInputConfig.Port }}
+{{ end }}