Using kubernetes filter with conditional routing

[mc-prospectsdeveloper]

Hi,

In order to create retention policies in OpenSearch that can provide different retention periods based on logging levels, I have updated the configuration for a FluentBit instance running in Kubernetes to use version 4.2.2 and conditional routing as a test:

      pipeline:
        inputs:
          - name: tail
            path: /var/log/containers/some-service*.log
            tag: app.*
            parser: docker-message
            db: /var/log/flb_kube.db
            mem_buf_limit: 5MB
            skip_long_lines: on
            refresh_interval: 10
            routes:
              logs:
                - name: app_long
                  condition:
                    rules:
                      - field: "$messaqge['log']['level']"
                        op: eq
                        value: "ERROR"
                  to:
                    outputs:
                      - app_long
                - name: app_medium
                  condition:
                    rules:
                      - field: "$messaqge['log']['level']"
                        op: eq
                        value: "WARN"
                  to:
                    outputs:
                      - app_medium
                - name: app_default
                  condition:
                    default: true
                  to:
                    outputs:
                      - app_short

        outputs:
          - name: file
            alias: app_long
            mkdir: True
            path: /fluent-bit/etc/output
            file: app-long.log

          - name: file
            alias: app_medium
            mkdir: True
            path: /fluent-bit/etc/output
            file: app-medium.log

          - name: file
            alias: app_short
            mkdir: True
            path: /fluent-bit/etc/output
            file: app-short.log

That does appear to route log records to the correct outputs.

However, the pipeline also includes the kubernetes filter:

        filters:
          - name: kubernetes
            match: app*
            kube_tag_prefix:  app.var.log.containers.

With the conditional routing rules, the kubernetes filter no longer adds its fields to log entries. This seems to be because conditional routing changes the tag from app.*, or the replacement tag that the tail input plugin generates, to app_default, or whichever output name defined in the routing rules matches. FluentBit's logs include errors such as:
[2026/01/14 10:50:--.0000000000] [error] [filter:kubernetes:kubernetes.0] incoming record tag (app_default) is shorter than the kube_tag_prefix value (app.var.log.containers.), skip filter

As the kubernetes filter relies on the tag generated by the tail input plugin in order to determine the name of the pod the service is running in, it can't work correctly.

Is there an alternative way of combining conditional routing and the kubernetes filter that allows both to work together?

Thanks

[vincent-zurczak]

Hello,

I had almost the same question until few minutes ago.
In fact, my goal was to route some records and discard the other ones. I proceeded slightly differently by using the Kubernetes filter as a processor (which works as the K8s metadata are correctly added). However, it seemed all my records were sent to my (single) output. That was in fact because it seems their need to be an output for any routed record. So, if you have a single output, all your records will land into it. With an additional default output, it works correctly (eventually, I will use the null output to discard these records). BTW, it allows you to route on a metadata added by the Kubernetes filter/process (e.g. the namespace).

Given your sample, could you try...

pipeline:
  inputs:
  - name: tail
    path: /var/log/containers/some-service*.log
    tag: app.*
    parser: docker-message
    db: /var/log/flb_kube.db
    mem_buf_limit: 5MB
    skip_long_lines: on
    refresh_interval: 10
    
    processors:
      logs:
      - name: kubernetes
        match: app*
        kube_tag_prefix:  app.var.log.containers.
    
    routes:
      logs:
      - name: app_long
        condition:
          rules:
          - field: "$messaqge['log']['level']"
            op: eq
            value: "ERROR"
        to:
          outputs:
          - app_long
      - name: app_medium
        condition:
          rules:
          - field: "$messaqge['log']['level']"
            op: eq
            value: "WARN"
        to:
          outputs:
          - app_medium
      - name: app_default
        condition:
          default: true
        to:
          outputs:
          - app_short

  outputs:
  - name: file
    alias: app_long
    mkdir: True
    path: /fluent-bit/etc/output
    file: app-long.log

  - name: file
    alias: app_medium
    mkdir: True
    path: /fluent-bit/etc/output
    file: app-medium.log

  - name: file
    alias: app_short
    mkdir: True
    path: /fluent-bit/etc/output
    file: app-short.log

[mc-prospectsdeveloper]

Thanks Vincent, that worked perfectly.