Log format compatibility

charts/logging-operator/crds/logging.banzaicloud.io_loggings.yaml

@@ -1063,6 +1063,8 @@ spec:
                       type: string
                     type: array
                 type: object
+              enableDockerParserCompatibilityForCRI:
+                type: boolean
               enableRecreateWorkloadOnImmutableFieldChange:
                 type: boolean
               errorOutputRef:

config/crd/bases/logging.banzaicloud.io_loggings.yaml

@@ -1063,6 +1063,8 @@ spec:
                       type: string
                     type: array
                 type: object
+              enableDockerParserCompatibilityForCRI:
+                type: boolean
               enableRecreateWorkloadOnImmutableFieldChange:
                 type: boolean
               errorOutputRef:

config/samples/containerd-log.yaml

@@ -0,0 +1,41 @@
+# start minikube with containerd
+#     minikube start --driver=docker --container-runtime=containerd
+# install some workload that generates json and non-json logs
+#     helm upgrade --install log-generator kube-logging/log-generator --set app.golang=true
+# verify produced logs in the fluentd container (json logs should be expanded)
+#     kubectl exec -ti containerd-fluentd-0 -- tail -f /fluentd/log/out
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Logging
+metadata:
+  name: containerd
+spec:
+  enableDockerParserCompatibilityForCRI: true
+  fluentd: {}
+  controlNamespace: default
+  fluentbit: {}
+---
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Flow
+metadata:
+  name: all
+spec:
+  filters:
+    # for debugging
+    - stdout: {}
+    # With `enableDockerParserCompatibilityForCRI: true` key and key_name are
+    # set to "log" otherwise these are set to "message" if the runtime is CRI
+    #- concat:
+    #    key: log
+    #- parser:
+    #    key_name: log
+  match:
+    - select: {}
+  localOutputRefs:
+    - "null"
+---
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Output
+metadata:
+  name: "null"
+spec:
+  nullout: {}

config/samples/containerd-merge-log.yaml

@@ -12,12 +12,10 @@ spec:
   fluentd: {}
   controlNamespace: default
   fluentbit:
-    image:
-      tag: 1.9.10-debug
     inputTail:
       Parser: cri-log-key
-    # Parser that populates `log` instead of `message` to enable the kubernetes filter's Merge_Log feature to work
-    # Mind the indentation, otherwise fluentbit will symply parse the whole message into the `log` key
+    # Parser that populates `log` instead of `message` to enable the kubernetes filter's Merge_Log feature to parse json automatically
+    # Indentation is important
     customParsers: |
                     [PARSER]
                         Name cri-log-key

controllers/logging/logging_controller.go

@@ -49,6 +49,7 @@ import (
 	"github.com/kube-logging/logging-operator/pkg/resources/syslogng"
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/render"
 	syslogngconfig "github.com/kube-logging/logging-operator/pkg/sdk/logging/model/syslogng/config"
+	loggingmodeltypes "github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
 
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
 	loggingv1beta1 "github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
@@ -189,7 +190,11 @@ func (r *LoggingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		model.NewValidationReconciler(
 			r.Client,
 			loggingResources,
-			&secretLoaderFactory{Client: r.Client, Path: fluentd.OutputSecretPath},
+			&secretLoaderFactory{
+				Client:  r.Client,
+				Path:    fluentd.OutputSecretPath,
+				Logging: loggingResources.Logging,
+			},
 			log.WithName("validation"),
 		),
 	}
@@ -436,8 +441,9 @@ func (r *LoggingReconciler) clusterConfigurationFluentd(resources model.LoggingR
 	}
 
 	slf := secretLoaderFactory{
-		Client: r.Client,
-		Path:   fluentd.OutputSecretPath,
+		Client:  r.Client,
+		Path:    fluentd.OutputSecretPath,
+		Logging: resources.Logging,
 	}
 
 	fluentConfig, err := model.CreateSystem(resources, &slf, r.Log)
@@ -463,8 +469,9 @@ func (r *LoggingReconciler) clusterConfigurationSyslogNG(resources model.Logging
 	}
 
 	slf := secretLoaderFactory{
-		Client: r.Client,
-		Path:   syslogng.OutputSecretPath,
+		Client:  r.Client,
+		Path:    syslogng.OutputSecretPath,
+		Logging: resources.Logging,
 	}
 
 	_, syslogngSpec := resources.GetSyslogNGSpec()
@@ -487,10 +494,27 @@ func (r *LoggingReconciler) clusterConfigurationSyslogNG(resources model.Logging
 	return b.String(), &slf.Secrets, nil
 }
 
+type SecretLoaderWithLogKeyProvider struct {
+	SecretLoader secret.SecretLoader
+	Logging      loggingv1beta1.Logging
+}
+
+func (s *SecretLoaderWithLogKeyProvider) Load(secret *secret.Secret) (string, error) {
+	return s.SecretLoader.Load(secret)
+}
+
+func (s *SecretLoaderWithLogKeyProvider) GetLogKey() string {
+	if s.Logging.Spec.EnableDockerParserCompatibilityForCRI {
+		return "log"
+	}
+	return loggingmodeltypes.GetLogKey()
+}
+
 type secretLoaderFactory struct {
 	Client  client.Client
 	Secrets secret.MountSecrets
 	Path    string
+	Logging loggingv1beta1.Logging
 }
 
 // Deprecated: use SecretLoaderForNamespace instead
@@ -499,7 +523,10 @@ func (f *secretLoaderFactory) OutputSecretLoaderForNamespace(namespace string) s
 }
 
 func (f *secretLoaderFactory) SecretLoaderForNamespace(namespace string) secret.SecretLoader {
-	return secret.NewSecretLoader(f.Client, namespace, f.Path, &f.Secrets)
+	return &SecretLoaderWithLogKeyProvider{
+		SecretLoader: secret.NewSecretLoader(f.Client, namespace, f.Path, &f.Secrets),
+		Logging:      f.Logging,
+	}
 }
 
 // SetupLoggingWithManager setup logging manager

docs/configuration/crds/v1beta1/logging_types.md

@@ -34,6 +34,11 @@ Namespace for cluster wide configuration resources like ClusterFlow and ClusterO
 Default flow for unmatched logs. This Flow configuration collects all logs that didn't matched any other Flow. 
 
 
+### enableDockerParserCompatibilityForCRI (bool, optional) {#loggingspec-enabledockerparsercompatibilityforcri}
+
+EnableDockerParserCompatibilityForCRI enables a log parser that is compatible with the docker parser. This has the following benefits: - automatic json log parsing using the Merge_Log feature - downstream parsers can use the `log` field instead of `message` as they did with the docker runtime - the `concat` and `parser` filters are automatically set back to use the `log` field 
+
+
 ### enableRecreateWorkloadOnImmutableFieldChange (bool, optional) {#loggingspec-enablerecreateworkloadonimmutablefieldchange}
 
 EnableRecreateWorkloadOnImmutableFieldChange enables the operator to recreate the fluentbit daemonset and the fluentd statefulset (and possibly other resource in the future) in case there is a change in an immutable field that otherwise couldn't be managed with a simple update. 

pkg/resources/fluentbit/config.go

@@ -17,6 +17,7 @@ package fluentbit
 const BaseConfigName = "fluent-bit.conf"
 const UpstreamConfigName = "upstream.conf"
 const CustomParsersConfigName = "custom-parsers.conf"
+const CRIParserConfigName = "cri-log-parser.conf"
 const StockConfigPath = "/fluent-bit/etc"
 const StockBinPath = "/fluent-bit/bin/fluent-bit"
 const OperatorConfigPath = "/fluent-bit/etc-operator"
@@ -224,3 +225,12 @@ var upstreamConfigTemplate = `
     Port {{.Port}}
 {{- end}}
 `
+
+var criParserConfig = `
+[PARSER]
+    Name cri-log-compatibility
+    Format regex
+    Regex ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<log>.*)$
+    Time_Key    time
+    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
+`

pkg/resources/fluentbit/configsecret.go

@@ -239,10 +239,16 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 		switch types.ContainerRuntime {
 		case "docker":
 			r.fluentbitSpec.InputTail.Parser = "docker"
-		case "containerd":
-			r.fluentbitSpec.InputTail.Parser = "cri"
 		default:
-			r.fluentbitSpec.InputTail.Parser = "cri"
+			if r.Logging.Spec.EnableDockerParserCompatibilityForCRI {
+				r.fluentbitSpec.InputTail.Parser = "cri-log-compatibility"
+			} else {
+				r.fluentbitSpec.InputTail.Parser = "cri"
+			}
+		}
+	} else {
+		if r.Logging.Spec.EnableDockerParserCompatibilityForCRI {
+			return nil, nil, errors.New("enableDockerParserCompatibilityForCRI is set, but fluentbit config overrides it with inputTail.parser")
 		}
 	}
 
@@ -429,6 +435,10 @@ func (r *Reconciler) configSecret() (runtime.Object, reconciler.DesiredState, er
 		confs[UpstreamConfigName] = []byte(upstreamConfig)
 	}
 
+	if r.Logging.Spec.EnableDockerParserCompatibilityForCRI {
+		confs[CRIParserConfigName] = []byte(criParserConfig)
+	}
+
 	if r.fluentbitSpec.CustomParsers != "" {
 		confs[CustomParsersConfigName] = []byte(r.fluentbitSpec.CustomParsers)
 	}

pkg/resources/fluentd/fluentd.go

@@ -286,8 +286,11 @@ func (r *Reconciler) statusUpdate(ctx context.Context, patchBase client.Patch, r
 }
 
 func (r *Reconciler) reconcileDrain(ctx context.Context) (*reconcile.Result, error) {
-	if r.fluentdSpec.DisablePvc || !r.fluentdSpec.Scaling.Drain.Enabled {
-		r.Log.Info("fluentd buffer draining is disabled")
+	if !r.fluentdSpec.Scaling.Drain.Enabled {
+		return nil, nil
+	}
+	if r.fluentdSpec.DisablePvc && r.fluentdSpec.Scaling.Drain.Enabled {
+		r.Log.Info("fluentd buffer draining cannot be enabled because PVC for the statefulSet is disabled")
 		return nil, nil
 	}
 

pkg/sdk/logging/api/v1beta1/logging_types.go

@@ -84,6 +84,12 @@ type LoggingSpec struct {
 	// in case there is a change in an immutable field
 	// that otherwise couldn't be managed with a simple update.
 	EnableRecreateWorkloadOnImmutableFieldChange bool `json:"enableRecreateWorkloadOnImmutableFieldChange,omitempty"`
+	// EnableDockerParserCompatibilityForCRI enables a log parser that is compatible with the docker parser.
+	// This has the following benefits:
+	// - automatic json log parsing using the Merge_Log feature
+	// - downstream parsers can use the `log` field instead of `message` as they did with the docker runtime
+	// - the `concat` and `parser` filters are automatically set back to use the `log` field
+	EnableDockerParserCompatibilityForCRI bool `json:"enableDockerParserCompatibilityForCRI,omitempty"`
 }
 
 type ConfigCheckStrategy string

pkg/sdk/logging/model/filter/concat.go

@@ -125,7 +125,11 @@ func (c *Concat) ToDirective(secretLoader secret.SecretLoader, id string) (types
 	}
 	concatConfig := c.DeepCopy()
 	if concatConfig.Key == "" {
-		concatConfig.Key = types.GetLogKey()
+		if logKeyProvider, ok := secretLoader.(types.LogKeyProvider); ok {
+			concatConfig.Key = logKeyProvider.GetLogKey()
+		} else {
+			concatConfig.Key = types.GetLogKey()
+		}
 	}
 	if params, err := types.NewStructToStringMapper(secretLoader).StringsMap(concatConfig); err != nil {
 		return nil, err

pkg/sdk/logging/model/filter/parser.go

@@ -19,6 +19,7 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/cisco-open/operator-tools/pkg/secret"
+
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
 )
 
@@ -341,7 +342,11 @@ func (p *ParserConfig) ToDirective(secretLoader secret.SecretLoader, id string)
 	parserConfig := p.DeepCopy()
 
 	if parserConfig.KeyName == "" {
-		parserConfig.KeyName = types.GetLogKey()
+		if logKeyProvider, ok := secretLoader.(types.LogKeyProvider); ok {
+			parserConfig.KeyName = logKeyProvider.GetLogKey()
+		} else {
+			parserConfig.KeyName = types.GetLogKey()
+		}
 	}
 	if params, err := types.NewStructToStringMapper(secretLoader).StringsMap(parserConfig); err != nil {
 		return nil, err

pkg/sdk/logging/model/types/types.go

@@ -23,6 +23,10 @@ import (
 
 var ContainerRuntime = "containerd"
 
+type LogKeyProvider interface {
+	GetLogKey() string
+}
+
 func GetLogKey() string {
 	switch ContainerRuntime {
 	case "docker":

pkg/sdk/logging/plugins/plugin.go

@@ -19,6 +19,7 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/cisco-open/operator-tools/pkg/secret"
+
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/api/v1beta1"
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
 )