help request: Can anyone explain how the clusterfilter works on Fluentd with tags?

[Crazyigor1987]

Describe the issue

Hi!

Can anyone explain how the clusterfilter works on Fluentd with tags?

I have a fluentbit -> fluentd forward setup. The logs in fluentbit are correctly tagged and passed to fluentd. In my case, i want to create a clusterfilter on fluentd side, which matches the tag and performs the filter in ordered manner. I quite don't understand, why the tag is only set per filter. I expect that all filters are performed matched per one tag.

My setup looks like this:

apiVersion: fluentd.fluent.io/v1alpha1
kind: ClusterFilter
metadata:
  name: 001-istio-proxy-parser
  labels:
    fluentd.default.filter: "true"
spec:
  filters:
    - tag: "istio-proxy.*"
    - parser:
        keyName: log
        parse:
          type: json
        reserveData: true 
        removeKeyNameField: false
        injectKeyPrefix: "parsed."
        emitInvalidRecordToError: false
    - recordTransformer:
        enableRuby: true
        records:
          - key: istio
            value: '${record["container_name"] == "istio-proxy" && record["parsed"] ? record["parsed"] : nil}'
          - key: log
            value: '${record["container_name"] == "istio-proxy" && record["parsed"] ? nil : record["log"]}'
        removeKeys: "parsed"

The result looks like this:
Got <filter "istio-proxy.*>" and the following filters "<filter **>".

<source>
  @type  forward
  bind  0.0.0.0
  port  24224
</source>
<match **>
  @id  main
  @type  label_router
  <route>
    @label  @db681e4cb763ca5b7cdbf9ab76f67bbe
    <match>
    </match>
  </route>
</match>
<label @db681e4cb763ca5b7cdbf9ab76f67bbe>
  <filter istio-proxy.*>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusterfilter::001-istio-proxy-parser-0
  </filter>
  <filter **>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusterfilter::001-istio-proxy-parser-1
    @type  parser
    emit_invalid_record_to_error  false
    inject_key_prefix  parsed.
    key_name  log
    remove_key_name_field  false
    reserve_data  true
    <parse>
      @type  json
    </parse>
  </filter>
  <filter **>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusterfilter::001-istio-proxy-parser-2
    @type  record_transformer
    enable_ruby  true
    remove_keys  parsed
    <record>
      istio  ${record["container_name"] == "istio-proxy" && record["parsed"] ? record["parsed"] : nil}
      log  ${record["container_name"] == "istio-proxy" && record["parsed"] ? nil : record["log"]}
    </record>
  </filter>
  <match **>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusteroutput::fluentd-output-stdout-0
    @type  stdout
  </match>
</label>

How did you install fluent operator?

No response

Additional context

No response

[cw-Guo]

have you tired to specify the tag for each filters?

https://github.com/fluent/fluent-operator/blob/master/docs/plugins/fluentd/filter/types.md

[Crazyigor1987]

Hi @cw-Guo .
Thank you for this Information. The documention lacks the implementation of the commonfilters. Even in the CRDs there are not described. How should that work? Do you have an example for me?

Best regards

[cw-Guo]

you can run kubectl explain clusterfilters.fluentd.fluent.io.spec.filters to check the definitions too.

But a simple example is as follows:

apiVersion: fluentd.fluent.io/v1alpha1
kind: ClusterOutput
metadata:
  name: fd-generic-output
spec:
  outputs:
  - elasticsearch:
      SSLVerify: false
      hosts: example
    tag: 'kube.*'

[Crazyigor1987]

Thanks a lot @cw-Guo !

So grouped filters doesn't work with tags. I have separated my filters and provided the tag for each filter. Like:

apiVersion: fluentd.fluent.io/v1alpha1
kind: ClusterFilter
metadata:
  name: 001-istio-proxy-0-parsing
  labels:
    fluentd.eso-default.filter: "true"
spec:
  filters:
    - parser:
        keyName: log
        parse:
          type: json
        reserveData: true
        removeKeyNameField: false
        hashValueField: "istio"
        emitInvalidRecordToError: false
      tag: "istio.*"
---
apiVersion: fluentd.fluent.io/v1alpha1
kind: ClusterFilter
metadata:
  name: 001-istio-proxy-1-transforming
  labels:
    fluentd.eso-default.filter: "true"
spec:
  filters:
    - recordTransformer:
        enableRuby: true
        removeKeys: "parsed"
        records:
          - key: log
            value: "${record['kubernetes'] && record['kubernetes']['container_name'] == 'istio-proxy' && record['istio'] && !record['istio'].empty? ? nil : record['log']}"
      tag: "istio.*"

This results to my desired filter names.
Config:

<label @db681e4cb763ca5b7cdbf9ab76f67bbe>
  <filter istio.*>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusterfilter::001-istio-proxy-0-parsing-0
    @type  parser
    emit_invalid_record_to_error  false
    hash_value_field  istio
    key_name  log
    remove_key_name_field  false
    reserve_data  true
    <parse>
      @type  json
    </parse>
  </filter>
  <filter istio.*>
    @id  ClusterFluentdConfig-cluster-fluentd-config::cluster::clusterfilter::001-istio-proxy-1-transforming-0
    @type  record_transformer
    enable_ruby  true
    remove_keys  parsed
    <record>
      log  ${record['kubernetes'] && record['kubernetes']['container_name'] == 'istio-proxy' && record['istio'] && !record['istio'].empty? ? nil : record['log']}
    </record>
  </filter>

[cw-Guo]

grouped filters doesn't work with tags.

@Crazyigor1987 it should work.
you can try the following:

apiVersion: fluentd.fluent.io/v1alpha1
kind: ClusterOutput
metadata:
  name: fd-generic-output
spec:
  outputs:
    - parser:
        keyName: log
        parse:
          type: json
        reserveData: true
        removeKeyNameField: false
        hashValueField: "istio"
        emitInvalidRecordToError: false
      tag: "istio.*"
    - recordTransformer:
        enableRuby: true
        removeKeys: "parsed"
        records:
          - key: log
            value: "${record['kubernetes'] && record['kubernetes']['container_name'] == 'istio-proxy' && record['istio'] && !record['istio'].empty? ? nil : record['log']}"
      tag: "istio.*"