[Q&A]rewrite_tag_filter plugin not found at fluentd-kubernetes-daemonset:v1.19.0-debian-kafka2-1.1

[86jkuncle]

What is a problem?

I want to place the service logs of k8s into the corresponding topic of the namespace, with coredns logs placed in k8s-coredns-logs and vms-das-ys logs placed in k8s-video-base-logs. Therefore, I used the rewrite_tag_filter plugin. this is my configmap configuration file

configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
  namespace: logging
data:
  fluent.conf: |
    # ================= Source =================
    <source>
      @type tail
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.log.pos
      tag kube.*
      read_from_head true
      <parse>
        @type multi_format
        <pattern>
          format json
          time_key time
          time_type string
          time_format %Y-%m-%dT%H:%M:%S.%NZ
        </pattern>
        <pattern>
          format none
        </pattern>
      </parse>
    </source>

    # ================= Kubernetes Metadata =================
    <filter kube.**>
      @type kubernetes_metadata
    </filter>

    # ================= Rewrite Tags by Namespace =================
    <filter kube.**>
      @type rewrite_tag_filter
      <rule>
        key $.kubernetes.namespace_name
        pattern ^kube-system$
        tag kube.coredns
      </rule>
      <rule>
        key $.kubernetes.namespace_name
        pattern ^video-base$
        tag kube.video-base
      </rule>
      <rule>
        key $.kubernetes.namespace_name
        pattern .*
        tag ignore
      </rule>
    </filter>

    # ================= CoreDNS =================
    <filter kube.coredns>
      @type grep
      <regexp>
        key $.kubernetes.container_name
        pattern ^coredns$
      </regexp>
    </filter>

    <filter kube.coredns>
      @type parser
      key_name log
      reserve_data true
      <parse>
        @type regexp
        expression /^\[(?<level>[A-Z]+)\] (?<client_ip>\S+) - (?<query_id>\d+) "(?<query>.+)" (?<rcode>\S+) (?<flags>.+) (?<size>\d+) (?<duration>\d+\.\d+)s$/
        time_format %Y-%m-%d %H:%M:%S
      </parse>
    </filter>

    # ================= Video-Base Namespace  =================
    <filter kube.video-base>
      @type grep
      <regexp>
        key $.kubernetes.container_name
        pattern ^(vms-das-ys|vms-oss-.*|vms-stdu.*|vms-adm|vms-webuas|vms-das-hk|vms-sss|vms-das-dh|vms-basic-southapi-service|vms-web|vms-sss-scheduler|vms-vps|vms-vds|vms-uas|vms-las|vms-fileserver|vms-ins|vms-das-ys-autoreg|vms-basic-south-gateway|vms-das-jy|vms-das-hk-ehome|vms-das-hk-ar|vms-nas-100000001|vms-basic-taskserver|vms-dis|vms-basic-common-service-server|vms-das-gb|vms-das-hh|vms-bcs|vms-das-dh-autoreg|vms-as-.*|vms-ac-.*|vms-basic-centerdatasyn|vms-basic-mq|vms-basic-centermessagesyn|vms-ars)$
      </regexp>
    </filter>

    # ================= kube mete =================
    <filter kube.**>
      @type record_transformer
      enable_ruby true
      <record>
        service_name ${record["kubernetes"]["container_name"]}
        namespace ${record["kubernetes"]["namespace_name"]}
        pod_name ${record["kubernetes"]["pod_name"]}
      </record>
    </filter>

    # ================= VMS-DAS-YS XML xml =================
    <filter kube.video-base>
      @type parser
      key_name log
      reserve_data true
      <parse>
        @type xml
        force_array false
        xpath_namespaces {}
      </parse>
    </filter>

    <filter kube.video-base>
      @type record_transformer
      enable_ruby true
      <record>
        # 保留原始日志
        raw_message ${record["log"]}
        # 扁平化 M 节点
        ${record["M"] ? record["M"].each { |k,v| record["M_#{k}"]=v } : nil}
        # 扁平化 Param 节点
        ${record.dig("M","Param") ? record["M"]["Param"].each { |k,v| record["Param_#{k}"]=v } : nil}
      </record>
    </filter>

    # ================= no need log =================
    <match ignore>
      @type null
    </match>

    # =================output Kafka =================
    <match kube.coredns>
      @type kafka2
      brokers 10.157.224.251:9092,10.157.224.221:9092,10.157.224.219:9092
      topic k8s-coredns-logs
      required_acks -1
      max_send_retries 5
      use_event_time true
      compression_codec gzip
      max_send_limit_bytes 10485760
      <format>
        @type json
      </format>
      <buffer>
        flush_interval 5s
      </buffer>
    </match>

    <match kube.video-base>
      @type kafka2
      brokers 10.157.224.251:9092,10.157.224.221:9092,10.157.224.219:9092
      topic k8s-video-base-logs
      required_acks -1
      max_send_retries 5
      use_event_time true
      compression_codec gzip
      max_send_limit_bytes 10485760
      <format>
        @type json
      </format>
      <buffer>
        flush_interval 5s
      </buffer>
    </match>

this is my daemonset.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd
  namespace: logging
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd-clusterrole
rules:
- apiGroups: [""]
  resources:
  - pods
  - namespaces
  - nodes
  verbs:
  - get
  - list
  - watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluentd-clusterrolebinding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: fluentd-clusterrole
subjects:
- kind: ServiceAccount
  name: fluentd
  namespace: logging
---
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: logging
  labels:
    app: fluentd
spec:
  selector:
    matchLabels:
      app: fluentd
  template:
    metadata:
      labels:
        app: fluentd
    spec:
      serviceAccountName: fluentd
      containers:
      - name: fluentd
        image: 10.157.224.221:48081/fluent/fluentd-kubernetes-daemonset:v1.19.0-debian-kafka2-1.1
        env:
          - name: K8S_NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName
          - name: FLUENT_UID
            value: "0"
        volumeMounts:
        - name: config-volume
          mountPath: /fluentd/etc/
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
      terminationGracePeriodSeconds: 30
      volumes:
      - name: config-volume
        configMap:
          name: fluentd-config
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers

After using kubectl create -f daemonset.yaml, I saw the following logs

 2025-09-22 03:14:35 +0000 [info]: init supervisor logger path=nil rotate_age=nil rotate_size=nil

 2025-09-22 03:14:35 +0000 [info]: parsing config file is succeeded path="/fluentd/etc/fluent.conf"

 2025-09-22 03:14:35 +0000 [info]: inaccessible include directory was specified path="/etc/fluent/conf.d"

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-avro' version '1.1.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-concat' version '2.6.0'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-detect-exceptions' version '0.0.16'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-grok-parser' version '2.6.2'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-json-in-json-2' version '1.0.2'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-kafka' version '0.19.5'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-kubernetes_metadata_filter' version '3.8.0'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-multi-format-parser' version '1.0.0'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-parser-avro' version '0.3.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-parser-cri' version '0.1.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-prometheus' version '2.2.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-record-modifier' version '2.2.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-rewrite-tag-filter' version '2.4.0'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluent-plugin-systemd' version '1.1.1'

 2025-09-22 03:14:35 +0000 [info]: gem 'fluentd' version '1.19.0'

 2025-09-22 03:14:36 +0000 [error]: config error file="/fluentd/etc/fluent.conf" error_class=Fluent::NotFoundPluginError error="Unknown filter plugin 'rewrite_tag_filter'. Run 'gem search -rd fluent-plugin' to find plugins"

The image I am using is fluent/fluentd-kubernetes-daemonset:v1.19.0-debian-kafka2-1.1,. From the fluentd startup log, it can be seen that the fluent-plugin-rewrite-tag-filter plugin has been successfully installed. Why do I still get an error saying the plugin is not found

Describe the configuration of Fluentd

No response

Describe the logs of Fluentd

No response

Environment

- Fluentd version:
- TD Agent version:
- Fluent Package version:
- Docker image (tag):
- Operating system: centos7.9 
- Kernel version:

[kenhys]

config_include_dir might affects it.
Does it change the behavior if config_include_dir was disabled explicitly?

https://docs.fluentd.org/deployment/system-config#config_include_dir

<system>
 config_include_dir ""
</system>

[86jkuncle]

@kenhys Adding this configuration didn't help at all; it still reports the same error

[86jkuncle]

I found the reason. rewirte_tag_filter is an output plugin and cannot be used in the filter tag. Thank you for your help