# Custom log output to Kafka using Elastic Agent

**URL:** <https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031>\
**Category:** Elastic Agent\
**Created:** [September 4, 2024, 12:00pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031 "2024-09-04T12:00:12Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![tapiojaa](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tapiojaa/32/134828_2.png) [@tapiojaa](https://discuss.elastic.co/u/tapiojaa)\
**Post date:** [September 4, 2024, 12:00pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/1 "2024-09-04T12:00:12Z")

</div>

We have Elasticstack 8.15 in use so that client Filebeat output goes to the Kafka and from there to the Elasticsearch. Filebeat sends different logs to separate Kafka topics. We are using following configuration in "filebeat.yml" (I'll only write lines that are meaningful)

```auto
filebeat.inputs: 
- type: filestream
  id: fs1
  enabled: true
  paths:
    - E:/Logs/folder1/*.log
  fields:
    kafka_topic: topic1

- type: filestream
  id: fs2
  enabled: true
  paths:
    - E:/Logs/folder2/*.log
  fields:
    kafka_topic: topic2

output.kafka:
  enabled: true
  topic: '%{[fields.kafka_topic]}'

```

We'd like to Start using Elastic Agent with Fleet, which is already working, but we have one issue: How to configure Kafka output and dynamic topics? I know that it should be done by adding Custom Logs integration to the Agent policy and the add some code to the "processors"-field. Has someone figured out how this need to be done?

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 4, 2024, 12:05pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/2 "2024-09-04T12:05:15Z")

</div>

> [@tapiojaa](#):
>
> How to configure Kafka output and dynamic topics?

You can't, Elastic Agent does not support dynamic topics.

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 4, 2024, 12:16pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/3 "2024-09-04T12:16:43Z")

</div>

Well, I was wrong, it seems that they added it back.

Support for dynamic topics were removed, but it seems that they reverted this decision and added it back according to this github pull request:

> <https://github.com/elastic/beats/pull/40415>
>
> \## Proposed commit message
> 
> Removed regex validation to allow dynamic topic.
> …
> \## Checklist
> 
> \- \[\] My code follows the style guidelines of this project
> \- \[\] I have commented my code, particularly in hard-to-understand areas
> \- \[\] I have made corresponding changes to the documentation
> \- \[\] I have made corresponding change to the default configuration files
> \- \[x\] I have added tests that prove my fix is effective or that my feature works
> \- \[x\] I have added an entry in \`CHANGELOG.next.asciidoc\` or \`CHANGELOG-developer.next.asciidoc\`.
> 
> \## How to test this PR locally
> 
> \- build beats, copy over agentbeat to the elastic-agent build
> \`\`\` 
> \# build beats agentbeat
> cd ~/beats/x-pack/agentbeat 
> SNAPSHOT=true PLATFORMS="darwin/amd64" mage package
> 
> \# build agent
> cd ~/elastic-agent
> DEV=true EXTERNAL=false SNAPSHOT=true PLATFORMS=darwin/amd64 PACKAGES=tar.gz mage -v package
> 
> \# copy agentbeat
> cp ~/beats/x-pack/agentbeat/build/golang-crossbuild/agentbeat-darwin-amd64 ~/elastic-agent/build/distributions/elastic-agent-8.16.0-SNAPSHOT-darwin-x86\_64/data/elastic-agent-3f22cc/components/agentbeat
> \`\`\`
> 
> To test with a real kafka server, follow this guide: https://hevodata.com/learn/install-kafka-on-mac/#Step\_2\_Download\_Install\_Kafka\_on\_Mac\_Manually\_or\_via\_HomeBrew
> \- downloaded and started zookeeper and kafka server
> \- created a topic \`system.cpu\` and started the script to consume messages
> \`\`\`
> cd ~/Downloads/kafka\_2.13-3.8.0
> ./bin/zookeeper-server-start.sh config/zookeeper.properties
> ./bin/kafka-server-start.sh config/server.properties
> ./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic system.cpu
> ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic system.cpu
> \`\`\`
> \- added kafka output by entering host \`localhost:9092\` and topic \`%{\[data\_stream.dataset\]}\`
> \- after enrolling the agent, see messages showing up in the kafka consumer
> \- in agent logs, expect no errors
> 
> \<img width="818" alt="image" src="https://github.com/user-attachments/assets/c6b1bd48-b173-4440-92a1-817d0f645603"\>
> \<img width="2396" alt="image" src="https://github.com/user-attachments/assets/34766b39-d7ad-4c67-98cb-058393fe8522"\>
> \<img width="811" alt="image" src="https://github.com/user-attachments/assets/e64e47f2-bffb-4cfd-8458-b7bfba42bd37"\>
> 
> 
> 
> \## Related issues
> 
> \<!-- Recommended
> Link related issues below. Insert the issue link or reference after the word "Closes" if merging this should automatically close it.
> 
> \- Closes #123
> \- Relates #123
> \- Requires #123
> \- Superseds #123
> \--\>
> \- Closes https://github.com/elastic/ingest-dev/issues/3618

According to the linked PR you would need to configure it while configuring the output.

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/f/9/f9e4a82a48d878674cf58549d6d7eb2506f5cc63.png)

---

<div class="post-metadata">

**Author:** ![tapiojaa](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tapiojaa/32/134828_2.png) [@tapiojaa](https://discuss.elastic.co/u/tapiojaa)\
**Post date:** [September 4, 2024, 12:30pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/4 "2024-09-04T12:30:53Z")

</div>

Yes, I know that they remove that feature and now it's back.

So, if I configure output that way. How should I configure "custom log" integration?  
 ![image](https://us1.discourse-cdn.com/elastic/original/3X/1/1/11fedf10179749af1f5e1bc0c4212c37faca097d.png)

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 4, 2024, 12:39pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/5 "2024-09-04T12:39:47Z")

</div>

I do not use this integration, but checking the documentation it seems that you would need to use the `add_fields` processor.

Something like this:

```auto
- add_fields:
    target: fields
    fields:
      kafka_topic: test-vm-esagentsda1

```

The main difference is that you would need one Custom Log Integration per input in filebeat, so in your example you have 2 inputs, so you would need to add one Custom Log integration for each one of the inputs.

---

<div class="post-metadata">

**Author:** ![tapiojaa](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tapiojaa/32/134828_2.png) [@tapiojaa](https://discuss.elastic.co/u/tapiojaa)\
**Post date:** [September 4, 2024, 12:46pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/6 "2024-09-04T12:46:43Z")

</div>

PS C:\Program Files\Elastic\Agent\> .\elastic-agent.exe status  
┌─ fleet  
│ └─ status: (STARTING)  
└─ elastic-agent  
├─ status: (DEGRADED) 1 or more components/units in a failed state  
└─ log-b2d12f01-f359-45dd-be6c-e67c43b81cc9  
├─ status: (HEALTHY) Healthy: communicating with pid '4504'  
├─ log-b2d12f01-f359-45dd-be6c-e67c43b81cc9  
│ └─ status: (FAILED) could not start output: failed to reload output: topic '%{[fields.kafka\_topic]}' is invalid, it must match '[a-zA-Z0-9.\_-]' accessing 'kafka'  
└─ log-b2d12f01-f359-45dd-be6c-e67c43b81cc9-logfile-logs-6f3c2805-c5dd-45dc-944a-d8e101173873  
└─ status: (STARTING) Starting

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [September 4, 2024, 1:05pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/7 "2024-09-04T13:05:03Z")

</div>

> [@tapiojaa](#):
>
> status: (FAILED) could not start output: failed to reload output: topic '%{[fields.kafka\_topic]}' is invalid, it must match '[a-zA-Z0-9.\_-]' accessing 'kafka'

Yeah, it seems to not be on 8.15.0 yet.

8.15 was released August 8th, and the revert commit was merged on August 14th.

You will probably need to wait for 8.15.1 or maybe even 8.16.

---

<div class="post-metadata">

**Author:** ![tapiojaa](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/tapiojaa/32/134828_2.png) [@tapiojaa](https://discuss.elastic.co/u/tapiojaa)\
**Post date:** [November 22, 2024, 12:11pm UTC](https://discuss.elastic.co/t/custom-log-output-to-kafka-using-elastic-agent/366031/8 "2024-11-22T12:11:16Z")

</div>

I've installed 8.16 and Dynamic topic is available and working. I put Kafka output like this

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/3/e/3eda35e93affb2219e5893d63f8757d2f3720719.png)

and then in every integration Processors like this (except different topic name on each)

```auto
- add_fields:
    target: fields
    fields:
      integ_topic: test-vm-eagentsda1

```