# Use ingest pipelines with output.kafka specified

**URL:** <https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [April 13, 2020, 10:57am UTC](https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762 "2020-04-13T10:57:32Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![romanfurst](https://avatars.discourse-cdn.com/v4/letter/r/3d9bf3/32.png) [@romanfurst](https://discuss.elastic.co/u/romanfurst)\
**Post date:** [April 13, 2020, 10:57am UTC](https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762/1 "2020-04-13T10:57:32Z")

</div>

Hello there, is it possible use ingest pipeline in filebeat without direct connection to logstash (or elastic) ? I mean in our scenario we have specified ouput.kafka (filebeat -\> kafka -\> logstash -\> elastic), however we need make some changes in logs fileds and structure before log is send from filebeat to kafka topic. Also I think it's good to be mention that we use autodiscover setup for our kubernetes environment. So we made custom module and ingest pipeline but it seems it doesnt work. As long as I read documentation I understand that this is not possible to filebate use ingest pipelines to tranfrom logs. Am I right or Im missing something ? Is there any option how make advanced logs transformation in filebaet ? Thank you

---

<div class="post-metadata">

**Author:** ![rugenl](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rugenl/32/12887_2.png) [@rugenl](https://discuss.elastic.co/u/rugenl)\
**Post date:** [April 13, 2020, 2:20pm UTC](https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762/2 "2020-04-13T14:20:26Z")

</div>

You can use conditional in your ultimate logstash output to specify ingest pipeline. It the sample below, I have a tag and must have a module name that gets passed to the "pipeline" option.

```
else if "use_ingest" in [tags] and [agent][module] {
    elasticsearch {
      hosts => [{{ ES_http }}]
      cacert => "/etc/logstash/certs/https_interm.cer"
      user => "{{ elastic.user }}"
      password => "{{elastic.pass }}"
      sniffing => false
      manage_template => false
      pipeline => "%{[@metadata][beat]}-%{[@metadata][version]}-%{[agent][module]}-%{[fileset][name]}-pipeline"
      ilm_enabled => true
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{[fields][app_id]}-%{[fields][campus]}"
    }
  }
```

---

<div class="post-metadata">

**Author:** ![romanfurst](https://avatars.discourse-cdn.com/v4/letter/r/3d9bf3/32.png) [@romanfurst](https://discuss.elastic.co/u/romanfurst)\
**Post date:** [April 13, 2020, 4:30pm UTC](https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762/3 "2020-04-13T16:30:38Z")

</div>

Well thanks for your reply I appreciate it. But Im not sure this is what Im looking for (or maybe I just missinterpret your answer). We dont want to make any changes in logstash or elastic. Those components (logstash,elastic) are shared for many teams in our corporation so we dont want to doing any unnecessary changes just becauce we decided to use filebeat. While those components are maintained by another squad, every change request is long running corporate process. Our idea was just use filebeat to read logs from docker container, transfrom theirs structure or fileds as we want, and send it to kafka topic. Thats it. We thought use filebat custom filebeat modules and ingest piplines is right way for that purpose. But now it seems we have to find out another solution

---

<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:** [May 11, 2020, 4:30pm UTC](https://discuss.elastic.co/t/use-ingest-pipelines-with-output-kafka-specified/227762/4 "2020-05-11T16:30:42Z")

</div>

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