diff --git a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types.go b/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types.go index 893d9db80..29af0f2d2 100644 --- a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types.go +++ b/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) } diff --git a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go b/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go index 9c4606a6d..1e95b271a 100644 --- a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go +++ b/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)) +} diff --git a/controllers/consts.go b/controllers/consts.go index c86a9bb0a..e9bd5337f 100644 --- a/controllers/consts.go +++ b/controllers/consts.go @@ -12,4 +12,6 @@ var ( fluentbitApiGVStr = fluentbitv1alpha2.SchemeGroupVersion.String() fluentdApiGVStr = fluentdv1alpha1.SchemeGroupVersion.String() fluentdAgentMode = "agent" + + configFileFormatYaml = "yaml" ) diff --git a/controllers/fluentbitconfig_controller.go b/controllers/fluentbitconfig_controller.go index 2c23e3227..3df667fa0 100644 --- a/controllers/fluentbitconfig_controller.go +++ b/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 { diff --git a/controllers/fluentbitconfig_controller_test.go b/controllers/fluentbitconfig_controller_test.go new file mode 100644 index 000000000..1aafabb49 --- /dev/null +++ b/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) + } +}