fix(fluentbit): render namespaced rewrite_tag filter as YAML when configFileFormat=yaml

apis/fluentbit/v1alpha2/clusterfluentbitconfig_types.go

@@ -351,17 +351,21 @@ func (cfg ClusterFluentBitConfig) RenderMainConfigInYaml(
 	}
 
 	buf.WriteString(inputSections)
-	for _, rtc := range rewriteTagConfigs {
-		buf.WriteString(rtc)
-	}
-	if filterSections == "" && nsFilterSections != nil {
+	// rewriteTagConfigs entries are headerless "filters:" list items (see
+	// generateRewriteTagConfig), so they must share a single "filters:"
+	// header with filterSections/nsFilterSections rather than writing their
+	// own, which would otherwise produce duplicate "filters:" keys.
+	if filterSections == "" && (nsFilterSections != nil || len(rewriteTagConfigs) > 0) {
 		fmt.Fprintf(&buf, "%sfilters:\n", utils.YamlIndent(1))
 	} else {
-		// 1. filterSections == "" && nsFilterSections == nil
-		// 2. filterSections != "" && nsFilterSections != nil
-		// 3. filterSections != "" && nsFilterSections == nil
+		// 1. filterSections == "" && nsFilterSections == nil && rewriteTagConfigs empty
+		// 2. filterSections != "" && (nsFilterSections != nil || rewriteTagConfigs non-empty)
+		// 3. filterSections != "" && nsFilterSections == nil && rewriteTagConfigs empty
 		buf.WriteString(filterSections)
 	}
+	for _, rtc := range rewriteTagConfigs {
+		buf.WriteString(rtc)
+	}
 	for _, filters := range nsFilterSections {
 		buf.WriteString(filters)
 	}

apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go

@@ -2,10 +2,12 @@ package v1alpha2
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	. "github.com/onsi/gomega"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/yaml"
 
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/custom"
@@ -1437,3 +1439,85 @@ func TestClusterFluentBitConfig_RenderMultilineParserConfig(t *testing.T) {
 	g.Expect(err).NotTo(HaveOccurred())
 	g.Expect(config).To(Equal(expectedMultilineParsers))
 }
+
+// TestRenderMainConfigInYaml_RewriteTagConfigMergesIntoSingleFiltersSection
+// is a regression test for
+// https://github.com/fluent/fluent-operator/pull/2019#pullrequestreview-4856326368:
+// a rewrite_tag fragment produced for the "yaml" config format must merge
+// into the single "pipeline.filters" list alongside any cluster/namespaced
+// filters, instead of contributing its own "filters:" key, which would
+// otherwise produce a second, duplicate key under "pipeline:".
+func TestRenderMainConfigInYaml_RewriteTagConfigMergesIntoSingleFiltersSection(t *testing.T) {
+	g := NewGomegaWithT(t)
+	sl := plugins.NewSecretLoader(nil, "testnamespace")
+
+	cfg := ClusterFluentBitConfig{}
+
+	inputs := ClusterInputList{
+		Items: []ClusterInput{
+			{
+				ObjectMeta: metav1.ObjectMeta{Name: "input0"},
+				Spec: InputSpec{
+					Tail: &input.Tail{
+						Tag:  "logs.foo.bar",
+						Path: "/logs/containers/apps0",
+					},
+				},
+			},
+		},
+	}
+
+	filters := ClusterFilterList{
+		Items: []ClusterFilter{
+			{
+				ObjectMeta: metav1.ObjectMeta{Name: "filter0"},
+				Spec: FilterSpec{
+					Match: "logs.foo.bar",
+					FilterItems: []FilterItem{
+						{Modify: &filter.Modify{
+							Rules: []filter.Rule{{Set: map[string]string{"k": "v"}}},
+						}},
+					},
+				},
+			},
+		},
+	}
+
+	outputs := ClusterOutputList{}
+
+	// Build the rewrite_tag fragment the same way
+	// generateRewriteTagConfig does: render a synthetic ClusterFilterList as
+	// YAML and strip its "filters:" header before merging it in.
+	rewriteTagFilters := ClusterFilterList{
+		Items: []ClusterFilter{
+			{
+				Spec: FilterSpec{
+					Match: "kube.*",
+					FilterItems: []FilterItem{
+						{RewriteTag: &filter.RewriteTag{
+							Rules: []string{"$kubernetes['namespace_name'] ^(foobar)$ abc123.$TAG false"},
+						}},
+					},
+				},
+			},
+		},
+	}
+	rendered, err := rewriteTagFilters.LoadAsYaml(sl, 1)
+	g.Expect(err).NotTo(HaveOccurred())
+	rtc := strings.TrimPrefix(rendered, fmt.Sprintf("%sfilters:\n", utils.YamlIndent(1)))
+	g.Expect(rtc).NotTo(ContainSubstring("filters:"))
+
+	config, err := cfg.RenderMainConfigInYaml(sl, inputs, filters, outputs, nil, nil, []string{rtc})
+	g.Expect(err).NotTo(HaveOccurred())
+
+	g.Expect(strings.Count(config, "filters:\n")).To(Equal(1))
+
+	var parsed map[string]interface{}
+	g.Expect(yaml.Unmarshal([]byte(config), &parsed)).To(Succeed())
+
+	pipeline, ok := parsed["pipeline"].(map[string]interface{})
+	g.Expect(ok).To(BeTrue())
+	filterEntries, ok := pipeline["filters"].([]interface{})
+	g.Expect(ok).To(BeTrue())
+	g.Expect(filterEntries).To(HaveLen(2))
+}

controllers/consts.go

@@ -12,4 +12,6 @@ var (
 	fluentbitApiGVStr = fluentbitv1alpha2.SchemeGroupVersion.String()
 	fluentdApiGVStr   = fluentdv1alpha1.SchemeGroupVersion.String()
 	fluentdAgentMode  = "agent"
+
+	configFileFormatYaml = "yaml"
 )

controllers/fluentbitconfig_controller.go

@@ -17,7 +17,6 @@ limitations under the License.
 package controllers
 
 import (
-	"bytes"
 	"context"
 	"crypto/md5"
 	"crypto/sha256"
@@ -26,6 +25,8 @@ import (
 	"sort"
 
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/filter"
+	"github.com/fluent/fluent-operator/v3/pkg/utils"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
@@ -241,7 +242,7 @@ func (r *FluentBitConfigReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			nsFilterLists, nsOutputLists, nsParserLists,
 				nsClusterParserLists, nsMultilineParserLists, nsClusterMultilineParserLists,
 				rewriteTagConfigs, err := r.processNamespacedFluentBitCfgs(
-				ctx, fb, inputs,
+				ctx, fb, inputs, cfg.Spec.ConfigFileFormat,
 			)
 
 			if err != nil {
@@ -287,7 +288,7 @@ func (r *FluentBitConfigReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			}
 
 			configFileName := "fluent-bit.conf"
-			if cfg.Spec.ConfigFileFormat != nil && *cfg.Spec.ConfigFileFormat == "yaml" {
+			if cfg.Spec.ConfigFileFormat != nil && *cfg.Spec.ConfigFileFormat == configFileFormatYaml {
 				configFileName = "fluent-bit.yaml"
 			}
 
@@ -307,6 +308,7 @@ func (r *FluentBitConfigReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 
 func (r *FluentBitConfigReconciler) processNamespacedFluentBitCfgs(
 	ctx context.Context, fb fluentbitv1alpha2.FluentBit, inputs fluentbitv1alpha2.ClusterInputList,
+	configFileFormat *string,
 ) (
 	[]fluentbitv1alpha2.FilterList, []fluentbitv1alpha2.OutputList,
 	[]fluentbitv1alpha2.ParserList, []fluentbitv1alpha2.ClusterParserList,
@@ -359,7 +361,12 @@ func (r *FluentBitConfigReconciler) processNamespacedFluentBitCfgs(
 		clusterMultilineParsers = append(clusterMultilineParsers, clusterMultilineParsersList)
 
 		if _, ok := storeNamespaces[cfg.Namespace]; !ok {
-			rewriteTagConfig := r.generateRewriteTagConfig(cfg, inputs)
+			rewriteTagConfig, err := r.generateRewriteTagConfig(cfg, inputs, configFileFormat)
+			if err != nil {
+				return filters, outputs, parsers,
+					clusterParsers, multilineParsers, clusterMultilineParsers,
+					nil, err
+			}
 			if rewriteTagConfig != "" {
 				rewriteTagConfigs = append(rewriteTagConfigs, rewriteTagConfig)
 				storeNamespaces[cfg.Namespace] = true
@@ -463,8 +470,8 @@ func (r *FluentBitConfigReconciler) ListFluentBitConfigResources(
 }
 
 func (r *FluentBitConfigReconciler) generateRewriteTagConfig(
-	cfg fluentbitv1alpha2.FluentBitConfig, inputs fluentbitv1alpha2.ClusterInputList,
-) string {
+	cfg fluentbitv1alpha2.FluentBitConfig, inputs fluentbitv1alpha2.ClusterInputList, configFileFormat *string,
+) (string, error) {
 	var tag string
 	for _, input := range inputs.Items {
 		if input.Spec.Tail == nil || !strings.Contains(input.Spec.Tail.Path, "/var/log/containers") {
@@ -479,28 +486,48 @@ func (r *FluentBitConfigReconciler) generateRewriteTagConfig(
 		}
 	}
 	if tag == "" {
-		return ""
+		return "", nil
+	}
+
+	rewriteTag := &filter.RewriteTag{
+		Rules: []string{
+			fmt.Sprintf("$kubernetes['namespace_name'] ^(%s)$ %x.$TAG false", cfg.Namespace, md5.Sum([]byte(cfg.Namespace))),
+		},
 	}
-	var buf bytes.Buffer
-	fmt.Fprintln(&buf, "[Filter]")
-	fmt.Fprintln(&buf, "    Name    rewrite_tag")
-	fmt.Fprintf(&buf, "    Match    %s\n", tag)
-	fmt.Fprintf(&buf, "    Rule    $kubernetes['namespace_name'] ^(%s)$ %x.$TAG false\n", cfg.Namespace,
-		md5.Sum([]byte(cfg.Namespace)))
 	if cfg.Spec.Service != nil {
 		if cfg.Spec.Service.EmitterName != "" {
-			fmt.Fprintf(&buf, "    Emitter_Name    %s\n", cfg.Spec.Service.EmitterName)
+			rewriteTag.EmitterName = cfg.Spec.Service.EmitterName
 		} else {
-			fmt.Fprintf(&buf, "    Emitter_Name    re_emitted_%x\n", md5.Sum([]byte(cfg.Namespace)))
+			rewriteTag.EmitterName = fmt.Sprintf("re_emitted_%x", md5.Sum([]byte(cfg.Namespace)))
 		}
-		if cfg.Spec.Service.EmitterStorageType != "" {
-			fmt.Fprintf(&buf, "    Emitter_Storage.type    %s\n", cfg.Spec.Service.EmitterStorageType)
-		}
-		if cfg.Spec.Service.EmitterMemBufLimit != "" {
-			fmt.Fprintf(&buf, "    Emitter_Mem_Buf_Limit    %s\n", cfg.Spec.Service.EmitterMemBufLimit)
+		rewriteTag.EmitterStorageType = cfg.Spec.Service.EmitterStorageType
+		rewriteTag.EmitterMemBufLimit = cfg.Spec.Service.EmitterMemBufLimit
+	}
+
+	filterList := fluentbitv1alpha2.ClusterFilterList{
+		Items: []fluentbitv1alpha2.ClusterFilter{
+			{
+				Spec: fluentbitv1alpha2.FilterSpec{
+					Match:       tag,
+					FilterItems: []fluentbitv1alpha2.FilterItem{{RewriteTag: rewriteTag}},
+				},
+			},
+		},
+	}
+
+	sl := plugins.NewSecretLoader(nil, "")
+	if configFileFormat != nil && *configFileFormat == configFileFormatYaml {
+		rendered, err := filterList.LoadAsYaml(sl, 1)
+		if err != nil {
+			return "", err
 		}
+		// Strip the "filters:" header so callers can merge this into the
+		// single "filters:" section of the main YAML config instead of
+		// emitting a second, duplicate key (see RenderMainConfigInYaml).
+		header := fmt.Sprintf("%sfilters:\n", utils.YamlIndent(1))
+		return strings.TrimPrefix(rendered, header), nil
 	}
-	return buf.String()
+	return filterList.Load(sl)
 }
 
 func (r *FluentBitConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {

controllers/fluentbitconfig_controller_test.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2021.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controllers
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/input"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/yaml"
+
+	fluentbitv1alpha2 "github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2"
+)
+
+// TestGenerateRewriteTagConfigYaml reproduces
+// https://github.com/fluent/fluent-operator/issues/1689: when
+// spec.configFileFormat is "yaml", the auto-generated rewrite_tag filter for
+// namespaced FluentBitConfig resources must be rendered as YAML, not as a
+// classic TOML snippet spliced into the YAML document.
+func TestGenerateRewriteTagConfigYaml(t *testing.T) {
+	r := &FluentBitConfigReconciler{}
+	cfg := fluentbitv1alpha2.FluentBitConfig{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "foobar"},
+	}
+	inputs := fluentbitv1alpha2.ClusterInputList{
+		Items: []fluentbitv1alpha2.ClusterInput{
+			{
+				Spec: fluentbitv1alpha2.InputSpec{
+					Tail: &input.Tail{
+						Tag:  "kube.*",
+						Path: "/var/log/containers/*.log",
+					},
+				},
+			},
+		},
+	}
+
+	yamlFormat := configFileFormatYaml
+	out, err := r.generateRewriteTagConfig(cfg, inputs, &yamlFormat)
+	if err != nil {
+		t.Fatalf("generateRewriteTagConfig returned error: %v", err)
+	}
+
+	if strings.Contains(out, "[Filter]") {
+		t.Fatalf("expected YAML output, got classic TOML snippet:\n%s", out)
+	}
+
+	// The generated snippet is a headerless "filters:" list item, meant to
+	// be merged into the single "pipeline.filters" section alongside
+	// cluster/namespaced filters by RenderMainConfigInYaml. It must not
+	// carry its own "filters:" header, or the merged config would end up
+	// with a duplicate, invalid "filters:" key (see
+	// https://github.com/fluent/fluent-operator/pull/2019#pullrequestreview-4856326368).
+	if strings.Contains(out, "filters:") {
+		t.Fatalf("expected a headerless filter list item, got a \"filters:\" header:\n%s", out)
+	}
+
+	// Wrapping it in a "filters:" key must still parse as valid YAML and
+	// contain the rewrite_tag entry.
+	wrapped := "filters:\n" + out
+	var parsed map[string]interface{}
+	if err := yaml.Unmarshal([]byte(wrapped), &parsed); err != nil {
+		t.Fatalf("generated rewrite_tag config is not valid YAML: %v\n%s", err, wrapped)
+	}
+	if _, ok := parsed["filters"]; !ok {
+		t.Fatalf("expected a top-level \"filters\" key once wrapped, got:\n%s", wrapped)
+	}
+
+	if !strings.Contains(out, "name: rewrite_tag") {
+		t.Fatalf("expected a rewrite_tag filter entry, got:\n%s", out)
+	}
+
+	// classic (default) format must remain unchanged (TOML).
+	classicOut, err := r.generateRewriteTagConfig(cfg, inputs, nil)
+	if err != nil {
+		t.Fatalf("generateRewriteTagConfig returned error: %v", err)
+	}
+	if !strings.Contains(classicOut, "[Filter]") || !strings.Contains(classicOut, "Name    rewrite_tag") {
+		t.Fatalf("expected classic TOML output, got:\n%s", classicOut)
+	}
+}