# \[Filebeat\] Define custom Ingest Node for Kafka Output

**URL:** <https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [December 17, 2020, 7:45pm UTC](https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032 "2020-12-17T19:45:23Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![dacamposol](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dacamposol/32/80191_2.png) [@dacamposol](https://discuss.elastic.co/u/dacamposol)\
**Post date:** [December 17, 2020, 7:45pm UTC](https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032/1 "2020-12-17T19:45:23Z")

</div>

Good afternoon guys!

I have a question regarding the configuration of the pipelines and Filebeat, when we're using a Kafka as output:

When we have ElasticSearch as output, we can just define the pipeline we want to use to process the data under `output.elasticsearch.pipeline`, for example:

```auto
output.elasticsearch:
  hosts: ["localhost:9200"]
  pipeline: my_pipeline_id

```

My question is regarding how do I define the Pipeline that I want to use in case that I set up Kafka as output?

As far as [it indicates the documentation](https://www.elastic.co/guide/en/beats/filebeat/master/kafka-output.html), Kafka doesn't have the `pipeline` attribute, but I know that in case that we use **modules** , even using Kafka as output, we can indicate ES that we want it to use the module pipeline for processing the data, so I want to do something similar, but with custom Ingest Nodes.

Also, I'd like to setup a custom pipeline depending of the container image, using the autodiscovery, so I was thinking in something like:

```auto
autodiscover.providers:
    - type: docker
      templates:
        - condition:
            contains:
              docker.container.image: custom_image
          config:
            - pipeline: my_custom_pipeline
              log:
                input:
                  type: container
                  paths:
                    - /var/lib/docker/containers/${data.docker.container.id}/*.log

```

But in case that this is not possible, I could set up different instances of Filebeat and each one using a different custom Ingest Node, so the priority would be to be able to setup this `pipeline` attribute in a Kafka output.

Does anyone know if something like that is possible?

---

<div class="post-metadata">

**Author:** ![mtojek](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/mtojek/32/63863_2.png) [@mtojek](https://discuss.elastic.co/u/mtojek)\
**Post date:** [December 18, 2020, 9:09am UTC](https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032/2 "2020-12-18T09:09:37Z")

</div>

Did you consider adding the logstash between filebeat and Kafka? You could use logstash to process data (run pipelines).

---

<div class="post-metadata">

**Author:** ![dacamposol](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dacamposol/32/80191_2.png) [@dacamposol](https://discuss.elastic.co/u/dacamposol)\
**Post date:** [December 18, 2020, 9:48am UTC](https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032/3 "2020-12-18T09:48:43Z")

</div>

I have LogStash, but it's between Kafka and ElasticSearch.

Basically, I have the following architecture right now in my ELK Stack:

![](https://us1.discourse-cdn.com/elastic/original/3X/7/9/79993db276c0c7c5fc968875f946852b8575d222.png)

* * *

I've though about to add a custom field in the autodiscovery input, taking advantage of the condition of the autodiscovery, and tell LogStash to pass it to ES.

**Filebeat.yml**

```auto
autodiscover.providers:
    - type: docker
      templates:
        - condition:
            contains:
              docker.container.image: custom_image
          config:
            - type: container
              paths:
                 - /var/lib/docker/containers/${data.docker.container.id}/*.log
              processors:
                - add_fields:
                    fields:
                      pipeline: 'my_custom_pipeline'

```

**LogStash Configuration**

```auto
output {
  if [fields][pipeline] {
    elasticsearch {
      pipeline => "%{[fields][pipeline]}"
    }
  }
}

```

Do you think that would work?

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [January 15, 2021, 11:48am UTC](https://discuss.elastic.co/t/filebeat-define-custom-ingest-node-for-kafka-output/259032/4 "2021-01-15T11:48:49Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.