Fix Kafka input for filebeat #30277

rdner · 2022-02-08T14:05:06Z

What does this PR do?

The Kafka input was broken and had 2 issues:

A serialization error on filebeat output:

Looks like the ack function was put in the reader.Message.Meta map by mistake in
20d6038
Meta is a MapStr type that does not support function values, therefore fails to
serialize itself when requested later on output.

ack was not called for incoming messages:

The ack function was never used because it was supposed
to be a part of the beat.Event.Private and it was not put in there.

Why is it important?

We have reports from customers that the Kafka input is completely broken starting with v7.16.2

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~~
I have added tests that prove my fix is effective or that my feature works
I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

Start kafka and zookeeper locally in their standard configurations.
Create a test topic:

kafka-topics --create --topic cyberark --bootstrap-server localhost:9092

Run filebeat with the following configuration

filebeat.inputs:
  - type: kafka
    hosts:
      - localhost:9092
    topics: ["cyberark"]
    group_id: "filebeat"

output.file:
  path: "/Path/To/Your/LogDir"
logging.level: debug

mind the "/Path/To/Your/LogDir" path that you need to adjust

Publish a message (or a few) to the topic:

kafka-console-producer --topic cyberark --bootstrap-server localhost:9092

You can see the logs saying that the message passed through:

{"log.level":"debug","@timestamp":"2022-02-08T12:16:15.724+0100","log.logger":"processors","log.origin":{"file.name":"processing/processors.go","file.line":209},"message":"Publish event: {\n  \"@timestamp\": \"2022-02-08T10:10:54.878Z\",\n  \"@metadata\": {\n    \"beat\": 
\"filebeat\",\n    \"type\": \"_doc\",\n    \"version\": \"8.2.0\"\n  },\n  \"message\": \"this should definitely not be json\",\n  \"input\": {\n    \"type\": \"kafka\"\n  },\n  \"ecs\": {\n    \"version\": \"8.0.0\"\n  },\n  \"host\": {\n    \"name\": \"MacBook-Pro.local
domain\"\n  },\n  \"agent\": {\n    \"ephemeral_id\": \"25448a31-5031-45bf-a051-5689b7cb4dd3\",\n    \"id\": \"4fd6b173-0f13-4792-888a-b54dcd85496c\",\n    \"name\": \"MacBook-Pro.localdomain\",\n    \"type\": \"filebeat\",\n    \"version\": \"8.2.0\"\n  },\n  \"kafka\": {
\n    \"topic\": \"cyberark\",\n    \"partition\": 0,\n    \"offset\": 0,\n    \"key\": \"\",\n    \"headers\": []\n  }\n}","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:15.724+0100","log.logger":"processors","log.origin":{"file.name":"processing/processors.go","file.line":209},"message":"Publish event: {\n  \"@timestamp\": \"2022-02-08T10:11:00.656Z\",\n  \"@metadata\": {\n    \"beat\": 
\"filebeat\",\n    \"type\": \"_doc\",\n    \"version\": \"8.2.0\"\n  },\n  \"input\": {\n    \"type\": \"kafka\"\n  },\n  \"ecs\": {\n    \"version\": \"8.0.0\"\n  },\n  \"host\": {\n    \"name\": \"MacBook-Pro.localdomain\"\n  },\n  \"agent\": {\n    \"ephemeral_id\": \"
25448a31-5031-45bf-a051-5689b7cb4dd3\",\n    \"id\": \"4fd6b173-0f13-4792-888a-b54dcd85496c\",\n    \"name\": \"MacBook-Pro.localdomain\",\n    \"type\": \"filebeat\",\n    \"version\": \"8.2.0\"\n  },\n  \"kafka\": {\n    \"headers\": [],\n    \"topic\": \"cyberark\",\n  
  \"partition\": 0,\n    \"offset\": 1,\n    \"key\": \"\"\n  },\n  \"message\": \"why is this failing?\"\n}","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.725+0100","log.logger":"rotator","log.origin":{"file.name":"file/rotator.go","file.line":440},"message":"Rotating file","service.name":"filebeat","rotator":{"filename":"/Users/rdner/Projects/tests/kfk.log/filebeat-2022
0208-2.ndjson","reason":"initializing","ecs.version":"1.6.0"}}                                                                           
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.727+0100","log.logger":"publisher","log.origin":{"file.name":"memqueue/ackloop.go","file.line":160},"message":"ackloop: receive ack [0: 0, 2]","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.727+0100","log.logger":"publisher","log.origin":{"file.name":"memqueue/eventloop.go","file.line":535},"message":"broker ACK events: count=2, start-seq=1, end-seq=2\n","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.727+0100","log.logger":"acker","log.origin":{"file.name":"beater/acker.go","file.line":64},"message":"stateless ack","service.name":"filebeat","count":2,"ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.727+0100","log.logger":"publisher","log.origin":{"file.name":"memqueue/ackloop.go","file.line":128},"message":"ackloop: return ack to broker loop:2","service.name":"filebeat","ecs.version":"1.6.0"}
{"log.level":"debug","@timestamp":"2022-02-08T12:16:16.727+0100","log.logger":"publisher","log.origin":{"file.name":"memqueue/ackloop.go","file.line":131},"message":"ackloop:  done send ack","service.name":"filebeat","ecs.version":"1.6.0"}

Before this change this was failing with:

{"log.level":"error","@timestamp":"2022-02-08T11:11:02.675+0100","log.logger":"file","log.origin":{"file.name":"fileout/file.go","file.line":128},"message":"Failed to serialize the event: unsupported","service.name":"filebeat","ecs.version":"1.6.0"}

{"log.level":"debug","@timestamp":"2022-02-08T11:11:02.675+0100","log.logger":"file","log.origin":{"file.name":"fileout/file.go","file.line":132},"message":"Failed event: &{{2022-02-08 11:11:00.656 +0100 CET Not valid json: json: unsupported type: func() {\"agent\":{\"ephe
meral_id\":\"8a333eee-02aa-45d7-9dde-af4d95571e5c\",\"id\":\"4fd6b173-0f13-4792-888a-b54dcd85496c\",\"name\":\"MacBook-Pro.localdomain\",\"type\":\"filebeat\",\"version\":\"8.2.0\"},\"ecs\":{\"version\":\"8.0.0\"},\"host\":\"xxxxx\",\"input\":{\"type\":\"kafka\"},\"kafka\"
:{\"headers\":[],\"key\":\"\",\"offset\":1,\"partition\":0,\"topic\":\"cyberark\"},\"message\":\"why is this failing?\"} <nil> false} 1 {map[]}}","service.name":"filebeat","ecs.version":"1.6.0"}

Stop filebeat
Verify that a log file with your message was created in "/Path/To/Your/LogDir" and remove all the files in that directory
Restart Filebeat and make sure the same messages don't get logged again (to verify that Kafka messages were acknowledged)

Related issues

Fixes [Filebeat] Kafka Input not working in 7.16.2 #29746

The Kafka input was broken and had 2 issues: A serialization error on filebeat output: Looks like the ack function was put in the `reader.Message.Meta` map by mistake in 20d6038 `Meta` is a `MapStr` type that does not support function values, therefore fails to serialize itself when requested later on output. `ack` was not called for incoming messages: The ack function was never used because it was supposed to be a part of the `beat.Event.Private` and it was not put in there.

elasticmachine · 2022-02-08T14:05:09Z

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

cmacknz · 2022-02-08T15:24:35Z

The 8.0 GA release date is on Feb 10th so it feels too risky to backport this into that branch until after that date.

@jlind23 any thoughts on where we should backport this change?

jlind23 · 2022-02-08T15:25:43Z

@cmacknz we should backport it all the way down to 7.17..
There won't be any more 7.16 release so backporting it to 7.16 is not needed.

filebeat/input/kafka/kafka_integration_test.go

cmacknz · 2022-02-08T18:05:34Z

LGTM, as mentioned previously don't merge the 8.0 backport until after February 10th when the release happens.

I'd also wait for @faec to take a look before merging :)

elasticmachine · 2022-02-08T18:06:02Z

💔 Tests Failed

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

Start Time: 2022-02-09T10:00:30.422+0000
Duration: 242 min 31 sec

Test stats 🧪

Test	Results
Failed	541
Passed	5178
Skipped	520
Total	6239

Test errors