feat: Add support for custom multiline filter

docs/user/02-logs.md

@@ -454,7 +454,6 @@ The `unsupportedMode` attribute of a LogPipeline indicates that you are using a
 
 You cannot enable the following plugins, because they potentially harm the stability:
 
-- Multiline Filter
 - Kubernetes Filter
 - Rewrite_Tag Filter
 

internal/fluentbit/config/builder/config_builder.go

@@ -38,6 +38,7 @@ func BuildFluentBitConfig(pipeline *telemetryv1alpha1.LogPipeline, config Builde
 
 	var sb strings.Builder
 	sb.WriteString(createInputSection(pipeline, includePath, excludePath))
+	sb.WriteString(createCustomMultilineFilters(pipeline))
 	sb.WriteString(createRecordModifierFilter(pipeline))
 	sb.WriteString(createKubernetesFilter(pipeline))
 	sb.WriteString(createCustomFilters(pipeline))

internal/fluentbit/config/builder/config_builder_test.go

@@ -82,13 +82,18 @@ func TestMergeSectionsConfig(t *testing.T) {
     db               /data/flb_foo.db
     exclude_path     /var/log/containers/telemetry-fluent-bit-*_kyma-system_fluent-bit-*.log,/var/log/containers/*_*_container1-*.log,/var/log/containers/*_*_container2-*.log
     mem_buf_limit    5MB
-    multiline.parser docker, cri, go, python, java
+    multiline.parser cri
     path             /var/log/containers/*_*_*-*.log
     read_from_head   true
     skip_long_lines  on
     storage.type     filesystem
     tag              foo.*
 
+[FILTER]
+    name             multiline
+    match            foo.*
+    multiline.parser java
+
 [FILTER]
     name   record_modifier
     match  foo.*
@@ -151,6 +156,12 @@ func TestMergeSectionsConfig(t *testing.T) {
 						regex log aa
 					`,
 				},
+				{
+					Custom: `
+						name multiline
+						multiline.parser java
+					`,
+				},
 			},
 			Output: telemetryv1alpha1.Output{
 				HTTP: &telemetryv1alpha1.HTTPOutput{
@@ -181,7 +192,7 @@ func TestMergeSectionsConfigCustomOutput(t *testing.T) {
     alias            foo
     db               /data/flb_foo.db
     mem_buf_limit    5MB
-    multiline.parser docker, cri, go, python, java
+    multiline.parser cri
     path             /var/log/containers/*_*_*-*.log
     read_from_head   true
     skip_long_lines  on

internal/fluentbit/config/builder/custom_filter.go

@@ -11,8 +11,13 @@ func createCustomFilters(pipeline *telemetryv1alpha1.LogPipeline) string {
 	var filters []string
 
 	for _, filter := range pipeline.Spec.Filters {
-		builder := NewFilterSectionBuilder()
 		customFilterParams := parseMultiline(filter.Custom)
+		// skip if the filter is a multiline filter, multiline filter should be first filter in the pipeline chain
+		// see for more details https://docs.fluentbit.io/manual/pipeline/filters/multiline-stacktrace
+		if customFilterParams.ContainsKey("name") && strings.Compare(customFilterParams.GetByKey("name").Value, "multiline") == 0 {
+			continue
+		}
+		builder := NewFilterSectionBuilder()
 		for _, p := range customFilterParams {
 			builder.AddConfigParam(p.Key, p.Value)
 		}
@@ -22,3 +27,21 @@ func createCustomFilters(pipeline *telemetryv1alpha1.LogPipeline) string {
 
 	return strings.Join(filters, "")
 }
+
+func createCustomMultilineFilters(pipeline *telemetryv1alpha1.LogPipeline) string {
+	var filters []string
+
+	for _, filter := range pipeline.Spec.Filters {
+		customFilterParams := parseMultiline(filter.Custom)
+		if customFilterParams.ContainsKey("name") && strings.Compare(customFilterParams.GetByKey("name").Value, "multiline") == 0 {
+			builder := NewFilterSectionBuilder()
+			for _, p := range customFilterParams {
+				builder.AddConfigParam(p.Key, p.Value)
+			}
+			builder.AddConfigParam("match", fmt.Sprintf("%s.*", pipeline.Name))
+			filters = append(filters, builder.Build())
+		}
+	}
+
+	return strings.Join(filters, "")
+}

internal/fluentbit/config/builder/input.go

@@ -14,7 +14,7 @@ func createInputSection(pipeline *telemetryv1alpha1.LogPipeline, includePath, ex
 	inputBuilder.AddConfigParam("alias", pipeline.Name)
 	inputBuilder.AddConfigParam("path", includePath)
 	inputBuilder.AddIfNotEmpty("exclude_path", excludePath)
-	inputBuilder.AddConfigParam("multiline.parser", "docker, cri, go, python, java")
+	inputBuilder.AddConfigParam("multiline.parser", "cri")
 	inputBuilder.AddConfigParam("tag", fmt.Sprintf("%s.*", pipeline.Name))
 	inputBuilder.AddConfigParam("skip_long_lines", "on")
 	inputBuilder.AddConfigParam("db", fmt.Sprintf("/data/flb_%s.db", pipeline.Name))

internal/fluentbit/config/builder/input_test.go

@@ -19,7 +19,7 @@ func TestCreateInput(t *testing.T) {
     db               /data/flb_test-logpipeline.db
     exclude_path     /var/log/containers/telemetry-fluent-bit-*_kyma-system_fluent-bit-*.log
     mem_buf_limit    5MB
-    multiline.parser docker, cri, go, python, java
+    multiline.parser cri
     path             /var/log/containers/*.log
     read_from_head   true
     skip_long_lines  on

main.go

@@ -255,7 +255,7 @@ func main() {
 	flag.StringVar(&metricGatewayDynamicMemoryRequest, "metric-gateway-dynamic-memory-request", "0", "Additional memory request for metrics OpenTelemetry Collector per MetricPipeline")
 	flag.IntVar(&maxMetricPipelines, "metric-gateway-pipelines", 3, "Maximum number of MetricPipelines to be created. If 0, no limit is applied.")
 
-	flag.StringVar(&fluentBitDeniedFilterPlugins, "fluent-bit-denied-filter-plugins", "kubernetes,rewrite_tag,multiline", "Comma separated list of denied filter plugins even if allowUnsupportedPlugins is enabled. If empty, all filter plugins are allowed.")
+	flag.StringVar(&fluentBitDeniedFilterPlugins, "fluent-bit-denied-filter-plugins", "kubernetes,rewrite_tag", "Comma separated list of denied filter plugins even if allowUnsupportedPlugins is enabled. If empty, all filter plugins are allowed.")
 	flag.StringVar(&fluentBitDeniedOutputPlugins, "fluent-bit-denied-output-plugins", "", "Comma separated list of denied output plugins even if allowUnsupportedPlugins is enabled. If empty, all output plugins are allowed.")
 	flag.StringVar(&fluentBitMemoryBufferLimit, "fluent-bit-memory-buffer-limit", "10M", "Fluent Bit memory buffer limit per log pipeline")
 	flag.StringVar(&fluentBitFsBufferLimit, "fluent-bit-filesystem-buffer-limit", "1G", "Fluent Bit filesystem buffer limit per log pipeline")

webhook/dryrun/testdata/expected/pipelines/dynamic/logpipeline-1.conf

@@ -4,7 +4,7 @@
     db               /data/flb_logpipeline-1.db
     exclude_path     /var/log/containers/telemetry-fluent-bit-*_kyma-system_fluent-bit-*.log,/var/log/containers/*_kyma-system_*-*.log,/var/log/containers/*_kube-system_*-*.log,/var/log/containers/*_istio-system_*-*.log,/var/log/containers/*_compass-system_*-*.log
     mem_buf_limit    5MB
-    multiline.parser docker, cri, go, python, java
+    multiline.parser cri
     path             /var/log/containers/*_*_*-*.log
     read_from_head   true
     skip_long_lines  on

webhook/dryrun/testdata/expected/pipelines/fluent-bit.conf

@@ -11,7 +11,7 @@
 [INPUT]
     Name tail
     Path /var/log/containers/*.log
-    multiline.parser docker, cri, go, python, java
+    multiline.parser cri
     Tag tele.*
     Mem_Buf_Limit 5MB
     Skip_Long_Lines On

webhook/dryrun/testdata/given/fluent-bit-configmap.yaml

@@ -21,7 +21,7 @@ data:
     [INPUT]
         Name tail
         Path /var/log/containers/*.log
-        multiline.parser docker, cri, go, python, java
+        multiline.parser cri
         Tag tele.*
         Mem_Buf_Limit 5MB
         Skip_Long_Lines On