Fluent-bit: Allow Elasticsearch Output Plugin to be able to dynamically add Kubernetes container_name to Logstash_Prefix

[samirspatel]

Is your feature request related to a problem? Please describe.

Elasticsearch is throwing errors because different apps are reporting the same field name with a different datatype. All apps are sending data to the same Elasticsearch index as indicated by the Logstash_Prefix configuration option for the es output plugin

The es output stanza is as follows

  output-elasticsearch.conf: |
    [OUTPUT]
        Name            es
        Match           *
        Host            ${FLUENT_ELASTICSEARCH_HOST}
        Port            ${FLUENT_ELASTICSEARCH_PORT}
        Logstash_Format On
        Logstash_Prefix kubeapps
        Time_Key        @timestamp
        Generate_ID     On
        Include_Tag_Key On
        Replace_Dots    On
        Retry_Limit     False
        Buffer_Size     False
        HTTP_User       ${FLUENT_ELASTICSEARCH_USER}
        HTTP_Passwd     ${FLUENT_ELASTICSEARCH_PASS}
        tls             On
        tls.verify      Off

Describe the solution you'd like

I would like to be able to change this Logstash_Prefix kubeapps to this Logstash_Prefix kube-<container_name> so each application in kubernetes has it's own Logstash_Prefix and hence it's own index in Elasticsearch

Describe alternatives you've considered

The alternative would be to hand configure inputs and create one output per app, which is simply not tenable as the number of services we are adding continues to increase.

Additional context

We are getting errors in Elasticsearch indicating field datatype collisions. ILM is also harder to do because we cannot have a lifecycle policy per app but have to have it on the entire index

[kurosakisora]

Very waiting for this answer, it looks like there isn't have a solution. And I'm also using NSQ queue with Http output Plugin, and I really hope to create topic dynamically, like this : /pub?topic=< container_name >, oh, also in Kafka too

[mr-karan]

+1 This would be really nice to have!

[mtparet]

We have exactly the same issue and so we have to use fluentbit and fluentd to be able to split logs into different indexes...

[ljfuyuan]

+1 , create index dynamically is needed, now , we have to use fluentd instead

[novegit]

You can use option _Logstash_Prefix_Key_ to define a message field, which is used to lookup the elasticsearch index for the message. Additional modifications to the index-name field can be made via lua-filter script (before the output plugin)

[test-account-0]

@novegit could You give an example?

[test-account-0]

I have added something like this to the filter section:

    [FILTER]
        Name    lua
        Match   kube.*
        script  /fluent-bit/etc/fluentbit.lua
        call    set_index

where set_index function looks like this:

function set_index(tag, timestamp, record)
    cluster = "cluster-sandbox-"
    if record["kubernetes"] ~= nil then
        if record["kubernetes"]["namespace_name"] ~= nil then
            if record["kubernetes"]["labels"]["app"] ~= nil then
                record["es_index"] = cluster
                    .. record["kubernetes"]["namespace_name"]
                    .. "-"
                    .. record["kubernetes"]["labels"]["app"]
                return 1, timestamp, record
            end
            record["es_index"] = cluster
                .. record["kubernetes"]["namespace_name"]
            return 1, timestamp, record
        end
    end
    return 1, timestamp, record
end

and it adds es_index field in elastic.

I have added also this thing to to the es output section:

    [OUTPUT]
        Name  es
        Match *
        Host  <somehost>
        Port  <someport>
        Logstash_Format On
        Retry_Limit 0
        Type  flb_type
        Time_Key @timestamp
        Replace_Dots On
        Logstash_Prefix cluster-sandbox
        Logstash_Prefix_Key es_index

But it uses Logstash_Prefix all the time and doesn't resolve Logstash_Prefix_Key.
I'm using version 1.3.6.
Should I be trying to use 1.4-dev for it to work?

[EDIT] It's working - i had unnecessary colon. :)

Thanks @novegit !

Maybe it would be good to add something about it to the docs?
It wasn't clear to me at all after reading them that this is the solution.

[novegit]

@samirspatel
1.3.6 should be ok. Your lua-scipt looks good. Try to debug the record fields, especially your kubernetes fields, by adding an out_stdout output.

[Vfialkin]

Trying to do the same. I think the problem is - "kubernetes.[fieldName]" is not a part of the original log but part of k8s metadata.

[etwillbefine]

Because Logstash_Prefix_Key does not support nested fields I am using a nest Filter to extract Kubernetes Metadata using the following Pipeline (no Script required):

[INPUT]
    Name              tail
    Path              /var/log/containers/*.log
    Parser            docker
    Tag               kube.*
    ....

[FILTER]
    Name                kubernetes
    Match               *
    Kube_Tag_Prefix     kube.var.log.containers.
    ...

[FILTER]
    Name            nest
    Match           *
    Operation       lift
    Wildcard        kubernetes.*
    Nested_under    kubernetes
    Add_prefix      k8s_

[OUTPUT]
    Name                es
    Match               *
    Logstash_Format     On
    Logstash_Prefix_Key k8s_namespace_name
    ....

Still a very limited Option because Logstash_Prefix will be omitted and there is no way of any concatenation or use of multiple variables. In case you want to build the Index from a Label on a Pod for example you would need more nest filters to flatten the data. I'd prefer using "variables" to build the Lagstash Prefix without a custom Script as well.

[Hichem-BenJeddi]

Because Logstash_Prefix_Key does not support nested fields I am using a nest Filter to extract Kubernetes Metadata using the following Pipeline (no Script required):

[INPUT]
    Name              tail
    Path              /var/log/containers/*.log
    Parser            docker
    Tag               kube.*
    ....

[FILTER]
    Name                kubernetes
    Match               *
    Kube_Tag_Prefix     kube.var.log.containers.
    ...

[FILTER]
    Name            nest
    Match           *
    Operation       lift
    Wildcard        kubernetes.*
    Nested_under    kubernetes
    Add_prefix      k8s_

[OUTPUT]
    Name                es
    Match               *
    Logstash_Format     On
    Logstash_Prefix_Key k8s_namespace_name
    ....

Still a very limited Option because Logstash_Prefix will be omitted and there is no way of any concatenation or use of multiple variables. In case you want to build the Index from a Label on a Pod for example you would need more nest filters to flatten the data. I'd prefer using "variables" to build the Lagstash Prefix without a custom Script as well.

Thanks, it working fine when the index name = namespace. it is possible to add logstash_k8s_namespace_name ?
How can i concat Logstash With k8s_namespace_name ?