Unable to send the logs to a dynamically generated topic based on the field topic_key

[stefannmih]

Bugs should be filed for issues encountered whilst operating logging-operator.
You should first attempt to resolve your issues through the community support
channels, e.g. Slack, in order to rule out individual configuration errors. #logging-operator
Please provide as much detail as possible.

Describe the bug:
I want to send logs from fluentd to a kafka cluster. The logs are coming from multiple namespaces and I want to send them in a specific topic named after the namespace. E.g logs from namespace cattle-monitoring-system should go to the topic cattle-monitoring-system-logs

I have used the following manifest for ClusterFlow to extract a namespace_name key.

apiVersion: logging.banzaicloud.io/v1beta1
kind: ClusterFlow
metadata:
  name: kafka-systemlogs-clusterflow
  namespace: cattle-logging-system
  annotations:
    field.cattle.io/description: used for system logs only
    argocd.argoproj.io/sync-wave: "2"
  labels:
    workload_type: systemlogs
spec:
  filters:
    - tag_normaliser:
        format: ${namespace_name}
    - record_transformer:
        enable_ruby: true
        records: 
          - namespace_name: "${record['kubernetes']['namespace_name']}-logs"
    - stdout:
        output_type: json
    - dedot:
        de_dot_separator: "-"
        de_dot_nested: true
  globalOutputRefs:
    - kafka-systemlogs-clusteroutput
  match:
    - select:
        namespaces: [] 

I have used this ClusterOutput:

apiVersion: logging.banzaicloud.io/v1beta1
kind: ClusterOutput
metadata:
  annotations:
    field.cattle.io/description: used for system logs only
    argocd.argoproj.io/sync-wave: "2"
  labels:
    workload_type: systemlogs
  name: kafka-systemlogs-clusteroutput
  namespace: cattle-logging-system
spec:
  kafka:
    brokers: "kafka-cluster.kafka:9094"
    topic_key: namespace_name
    default_topic: "topic1"
    username:
      valueFrom:
        secretKeyRef:
          key: username
          name: fluentd-kafka-secret
    password:
      valueFrom:
        secretKeyRef:
          key: password
          name: fluentd-kafka-secret
    scram_mechanism: "sha512"
    sasl_over_ssl: false
    ssl_verify_hostname: false
    format:
      type: json
    default_partition_key: k8s-clusterName-${tag}
    buffer:
      timekey: 10m
      timekey_wait: 1m
      timekey_use_utc: true
      chunk_limit_size: "512k"
      # total_limit_size: "2G"    
      flush_interval: 10s      
      flush_thread_count: 4     
      retry_forever: true       

This is a part of the generated fluentd.conf:

  <match **>
    @type kafka2
    @id clusterflow:cattle-logging-system:kafka-systemlogs-clusterflow:clusteroutput:cattle-logging-system:kafka-systemlogs-clusteroutput
    brokers kafka-cluster.kafka:9094
    default_partition_key k8s-clusterName-${tag}
    default_topic topic1
    password .....
    sasl_over_ssl false
    scram_mechanism sha512
    ssl_verify_hostname false
    topic_key namespace_name

Expected behaviour:
I would expect fluentd to send the logs to a dynamically generated topic based on the field topic_key so I can consume the logs from that specific topic.

Additional context:
If I set the default_topic, I can consume the messages from all the namespaces, but if I try to consume the messages from another topic, let's say cattle-monitoring-system-logs, I get no messages. I also created all the topics in advance.

Environment details:

• Kubernetes version : 1.26
• Cloud-provider/provisioner (e.g. AKS, GKE, EKS, PKE etc): Rancher v2.7.9
• logging-operator version (e.g. 2.1.1): Rancher chart version 103.0.1+up3.17.10
• Install method (e.g. helm or static manifests): Operator installed with helm and ClusterOutput/ClusterFlow with manifests
• Logs from the misbehaving component (and any other relevant logs):

2025-02-03 18:47:58.878572656 +0000 cattle-monitoring-system: {"log":"ts=2025-02-03T18:47:58.878Z caller=main.go:167 level=info scrape_id=01084bd4-0885-4107-8428-699dac7c675b msg=\"Retrieved scrape response\"\n","stream":"stderr","time":"2025-02-03T18:47:58.878572656Z","kubernetes":{"pod_name":"pushprox-kube-controller-manager-client-96cgz","namespace_name":"cattle-monitoring-system","pod_id":"138b53a5-4554-4c8f-a6c7-d102f4894ab2","labels":{"component":"kube-controller-manager","controller-revision-hash":"f69cbd454","k8s-app":"pushprox-kube-controller-manager-client","pod-template-generation":"4","provider":"kubernetes","release":"rancher-monitoring"},"host":"....","container_name":"pushprox-client","docker_id":"ab3c3f305ca7c78b3e4b48f5dbce466842587938ac21e2a12fbe5595c3d12f49","container_hash":"artifactory..../rancher/pushprox-client@sha256:e5784cfff95cc8590cf64bdf0a0c8e173e4a49ffc81457b2543edc9d7c8adbd6","container_image":"artifactory.../rancher/pushprox-client:v0.1.3-rancher2-client"},"namespace_name":"cattle-monitoring-system-logs"}

• Resource definition (possibly in YAML format) that caused the issue, without sensitive data:

/kind bug