[Kafka output] Add config validation for topic and topics fields #38058

belimawr · 2024-02-19T15:18:24Z

Proposed commit message

This commit adds configuration validation for topic and topics fields from the Kafka output. Validation is performed for both cases: standalone Beat and running under Elastic-Agent. The latter does not support dynamic topics, hence setting topics is rejected by the validation.

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.

~~## Author's Checklist~~

How to test this PR locally

For both test cases first start a Kafka cluster (instructions below)

Standalone Beats (Filebeat):

Start Filebeat using the following configuration file

filebeat.yml

filebeat.inputs:
    - id: filestream-input-id
      type: filestream
      paths:
        - /tmp/flog.log
output:
    kafka:
        hosts:
            - <YOUR LOCAL IP>:9091

queue.mem:
  flush.timeout: 1s

logging:
  level: debug
  selectors:
    - kafka

Validate that

Exiting: error initializing publisher: either 'topic' or 'topics' must be defined accessing 'output.kafka' (source:'filebeat.yml')

is logged

Under Elastic-Agent

Build Filebeat
Download and extract the Elastic-Agent
Replace its filebeat by the one you built
Generate some logs in the file /tmp/flog.log
Start a standalone Elastic-Agent using the configuration below

elastic-agent.yml

outputs:
  default:
    type: kafka
    hosts:
        - <YOUR IP>:9091

inputs:
  - type: filestream
    id: your-input-id
    logging.level: debug
    streams:
      - id: your-filestream-stream-id
        data_stream:
          dataset: generic
        paths:
          - /tmp/flog.log

Validate you see an error log like:

{"log.level":"error","@timestamp":"2024-02-19T15:28:52.127Z","message":"could not start output","component":{"binary":"filebeat","dataset":"elastic_agent.filebeat","id":"filestream-default","type":"filestream"},"log":{"source":"filestream-default"},"log.logger":"centralmgmt.V2-manager","log.origin":{"file.line":631,"file.name":"management/managerV2.go","function":"github.com/elastic/beats/v7/x-pack/libbeat/management.(*BeatV2Manager).reload"},"service.name":"filebeat","error":{"message":"failed to reload output: either 'topic' or 'topics' must be defined accessing 'kafka'"},"ecs.version":"1.6.0","ecs.version":"1.6.0"}

Running a Kafka cluster

Copy the docker-compose.yml provided below
Replace <YOUR LOCAL IP> by your IP address (e.g: 10.0.0.42). Do not use a loopback address
Start the containers docker-compose up
Open Kafdrop (http://localhost:9000/) on your browser. You can use it to manage topics and look at the messages.

docker-compose.yml

version: '3'
services:
  zookeeper:
    image: zookeeper:3.4.9
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: 1
      ZOO_PORT: 2181
      ZOO_SERVERS: server.1=zookeeper:2888:3888
    volumes:
      - ./data/zookeeper/data:/data
      - ./data/zookeeper/datalog:/datalog
  kafka1:
    image: confluentinc/cp-kafka:5.3.0
    hostname: kafka1
    ports:
      - "9091:9091"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19091,LISTENER_DOCKER_EXTERNAL://<YOUR LOCAL IP>:9091
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - ./data/kafka1/data:/var/lib/kafka/data
    depends_on:
      - zookeeper
  kafka2:
    image: confluentinc/cp-kafka:5.3.0
    hostname: kafka2
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka2:19092,LISTENER_DOCKER_EXTERNAL://<YOUR LOCAL IP>:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_BROKER_ID: 2
    volumes:
      - ./data/kafka2/data:/var/lib/kafka/data
    depends_on:
      - zookeeper 
  kafka3:
    image: confluentinc/cp-kafka:5.3.0
    hostname: kafka3
    ports:
      - "9093:9093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka3:19093,LISTENER_DOCKER_EXTERNAL://<YOUR LOCAL IP>:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 3
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    volumes:
      - ./data/kafka3/data:/var/lib/kafka/data
    depends_on:
      - zookeeper
  kafdrop:
    image: obsidiandynamics/kafdrop
    restart: "no"
    ports:
      - "9000:9000"
    environment:
      KAFKA_BROKERCONNECT: "kafka1:19091,kafka2:19092,kafka3:19093"
    depends_on:
      - kafka1
      - kafka2
      - kafka3

Tutorial on running a Kafka cluster with Docker: https://betterprogramming.pub/a-simple-apache-kafka-cluster-with-docker-kafdrop-and-python-cf45ab99e2b9

Related issues

Closes: No data for Kafka output under topic for System integration. elastic-agent#4254

~~## Use cases~~
~~## Screenshots~~
~~## Logs~~

mergify · 2024-02-19T15:21:03Z

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @belimawr? 🙏.
For such, you'll need to label your PR with:

The upcoming major version of the Elastic Stack
The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

backport-v8./d.0 is the label to automatically backport to the 8./d branch. /d is the digit

elasticmachine · 2024-02-19T17:00:25Z

Pinging @elastic/elastic-agent (Team:Elastic-Agent)

elasticmachine · 2024-02-19T17:34:49Z

💚 Build Succeeded

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

Expand to view the summary

Build stats

Duration: 133 min 35 sec

❕ Flaky test report

No test was executed to be analysed.

🤖 GitHub comments

Expand to view the GitHub comments

To re-run your PR in the CI, just comment with:

/test : Re-trigger the build.
/package : Generate the packages and run the E2E tests.
/beats-tester : Run the installation tests with beats-tester.
run elasticsearch-ci/docs : Re-trigger the docs validation. (use unformatted text in the comment!)

libbeat/outputs/kafka/config.go

cmacknz · 2024-02-21T20:38:51Z

libbeat/outputs/kafka/config.go

@@ -169,6 +175,20 @@ func (c *kafkaConfig) Validate() error {
 			return fmt.Errorf("compression_level must be between 0 and 9")
 		}
 	}
+
+	if c.Topic == "" && len(c.Topics) == 0 {


Would it be possible to attempt to evaluate c.Topic as a format string and log a warning if we think it is one?

That would help people figure out why their topic routing is failing. I don't think the %{ syntax is a valid topic name so the only way we'd get that is if someone is trying to use the Beat topic syntax with agent.

The legal characters for a kafka topic is pretty small.

https://github.com/apache/kafka/blob/a126e3a622f2b7142f3543b9dbee54b6412ba9d8/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L33

The bits of code that used to convert the %{foo} notation is deep within some helper from the outputs (outil.BuildSelectorFromConfig) and it will return some errors that might not be what we want.

I could either search for %{ in the topic name and return an error or use a regexp validation with the characters Lee pointed. I prefer the latter and returning an error message stating the valid characters.

Agreed just use the character set Lee linked and link back to that file in our code.

mergify · 2024-02-22T17:15:48Z

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b validation-for-kafka-config upstream/validation-for-kafka-config
git merge upstream/main
git push upstream validation-for-kafka-config

leehinman

LGTM

libbeat/outputs/kafka/config_test.go

mergify · 2024-03-14T19:35:54Z

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b validation-for-kafka-config upstream/validation-for-kafka-config
git merge upstream/main
git push upstream validation-for-kafka-config

This commit adds configuration validation for `topic` and `topics` fields from the Kafka output. Validation is performed for both cases: standalone Beat and running under Elastic-Agent. The latter does not support dynamic topics, hence setting `topics` is rejected by the validation.

This commit adds validation for the topic when running under Elastic-Agent. The topic is validates using a regexpt taken from the Kafka source code: https://github.com/apache/kafka/blob/a126e3a622f2b7142f3543b9dbee54b6412ba9d8/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L33

This commit adds a test case making explicity that topics containing the syntax `%{}` is not supported under Elastic-Agent

elasticmachine · 2024-03-14T19:39:10Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

cc @belimawr

elasticmachine · 2024-03-14T19:43:10Z

💔 Build Failed

Buildkite Build
Commit: e6280db

Failed CI Steps

History

💚 Build #2159 succeeded 50738a1
💚 Build #1822 succeeded 05f96f7
💚 Build #1793 succeeded b140525
💚 Build #1782 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T19:43:16Z

💔 Build Failed

Buildkite Build
Commit: e6280db

Failed CI Steps

History

💔 Build #3500 failed 50738a1
💚 Build #3161 succeeded 05f96f7
💚 Build #3132 succeeded b140525
💚 Build #3121 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T19:53:04Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

History

💚 Build #1863 succeeded 50738a1
💚 Build #1522 succeeded 05f96f7
💚 Build #1489 succeeded b140525
💚 Build #1476 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T19:56:27Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

History

💔 Build #950 failed c24b4c3

cc @belimawr

elasticmachine · 2024-03-14T19:59:17Z

💔 Build Failed

Buildkite Build
Commit: e6280db

Failed CI Steps

History

💔 Build #908 failed c24b4c3

cc @belimawr

elasticmachine · 2024-03-14T20:06:44Z

💔 Build Failed

Buildkite Build
Commit: e6280db

Failed CI Steps

:python: Python Integration Tests

History

💚 Build #3077 succeeded 50738a1
💚 Build #2736 succeeded 05f96f7
💚 Build #2703 succeeded b140525
💚 Build #2690 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T20:09:46Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

History

💚 Build #1870 succeeded 50738a1
💚 Build #1530 succeeded 05f96f7
💚 Build #1497 succeeded b140525
💚 Build #1481 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T20:09:58Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

History

💚 Build #1026 succeeded 50738a1
💚 Build #686 succeeded 05f96f7
💚 Build #653 succeeded b140525
💚 Build #640 succeeded de3d736

cc @belimawr

elasticmachine · 2024-03-14T20:21:48Z

💚 Build Succeeded

Buildkite Build
Commit: e6280db

cc @belimawr

) This commit adds configuration validation for `topic` and `topics` fields from the Kafka output. Validation is performed for both cases: standalone Beat and running under Elastic-Agent. The latter does not support dynamic topics, hence setting `topics` is rejected by the validation. When running under Elastic-Agent the topic is validated using a regexpt taken from the Kafka source code: https://github.com/apache/kafka/blob/a126e3a622f2b7142f3543b9dbee54b6412ba9d8/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L33 (cherry picked from commit 8289144)

) (#38381) This commit adds configuration validation for `topic` and `topics` fields from the Kafka output. Validation is performed for both cases: standalone Beat and running under Elastic-Agent. The latter does not support dynamic topics, hence setting `topics` is rejected by the validation. When running under Elastic-Agent the topic is validated using a regexpt taken from the Kafka source code: https://github.com/apache/kafka/blob/a126e3a622f2b7142f3543b9dbee54b6412ba9d8/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L33 (cherry picked from commit 8289144) Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>

belimawr requested a review from a team as a code owner February 19, 2024 15:18

belimawr requested review from faec and leehinman February 19, 2024 15:18

botelastic bot added the needs_team Indicates that the issue/PR needs a Team:* label label Feb 19, 2024

belimawr force-pushed the validation-for-kafka-config branch from f31c72b to de3d736 Compare February 19, 2024 15:20

mergify bot assigned belimawr Feb 19, 2024

pierrehilbert added the Team:Elastic-Agent Label for the Agent team label Feb 19, 2024

botelastic bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Feb 19, 2024

faec reviewed Feb 20, 2024

View reviewed changes

libbeat/outputs/kafka/config.go Outdated Show resolved Hide resolved

belimawr requested a review from faec February 21, 2024 14:26

cmacknz added the backport-v8.13.0 Automated backport with mergify label Feb 21, 2024

cmacknz reviewed Feb 21, 2024

View reviewed changes

belimawr force-pushed the validation-for-kafka-config branch from 05f96f7 to 50738a1 Compare March 7, 2024 12:19

leehinman approved these changes Mar 8, 2024

View reviewed changes

cmacknz reviewed Mar 8, 2024

View reviewed changes

libbeat/outputs/kafka/config_test.go Show resolved Hide resolved

belimawr added 5 commits March 14, 2024 20:36

Fix lint issues

8848980

Fix type for topics in Kafka config

b8d9aa3

[Kafka output] Add test case for topic with %{} under Elastic-Agent

e6280db

This commit adds a test case making explicity that topics containing the syntax `%{}` is not supported under Elastic-Agent

belimawr force-pushed the validation-for-kafka-config branch from c24b4c3 to e6280db Compare March 14, 2024 19:36

belimawr requested a review from cmacknz March 18, 2024 07:34

cmacknz approved these changes Mar 18, 2024

View reviewed changes

belimawr merged commit 8289144 into elastic:main Mar 18, 2024
106 of 109 checks passed

mergify bot mentioned this pull request Mar 18, 2024

[8.13](backport #38058) [Kafka output] Add config validation for topic and topics fields #38381

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Kafka output] Add config validation for topic and topics fields #38058

[Kafka output] Add config validation for topic and topics fields #38058

belimawr commented Feb 19, 2024 •

edited

mergify bot commented Feb 19, 2024

elasticmachine commented Feb 19, 2024

elasticmachine commented Feb 19, 2024 •

edited

Build stats

cmacknz Feb 21, 2024

leehinman Mar 5, 2024

belimawr Mar 6, 2024

cmacknz Mar 6, 2024

mergify bot commented Feb 22, 2024

leehinman left a comment

mergify bot commented Mar 14, 2024

elasticmachine commented Mar 14, 2024

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024

elasticmachine commented Mar 14, 2024

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024

elasticmachine commented Mar 14, 2024

elasticmachine commented Mar 14, 2024

[Kafka output] Add config validation for topic and topics fields #38058

[Kafka output] Add config validation for topic and topics fields #38058

Conversation

belimawr commented Feb 19, 2024 • edited

Proposed commit message

Checklist

How to test this PR locally

Standalone Beats (Filebeat):

Under Elastic-Agent

Running a Kafka cluster

Related issues

mergify bot commented Feb 19, 2024

elasticmachine commented Feb 19, 2024

elasticmachine commented Feb 19, 2024 • edited

💚 Build Succeeded

Build stats

❕ Flaky test report

🤖 GitHub comments

cmacknz Feb 21, 2024

Choose a reason for hiding this comment

leehinman Mar 5, 2024

Choose a reason for hiding this comment

belimawr Mar 6, 2024

Choose a reason for hiding this comment

cmacknz Mar 6, 2024

Choose a reason for hiding this comment

mergify bot commented Feb 22, 2024

leehinman left a comment

Choose a reason for hiding this comment

mergify bot commented Mar 14, 2024

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

elasticmachine commented Mar 14, 2024 • edited

💔 Build Failed

Failed CI Steps

History

elasticmachine commented Mar 14, 2024 • edited

💔 Build Failed

Failed CI Steps

History

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

History

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

History

elasticmachine commented Mar 14, 2024 • edited

💔 Build Failed

Failed CI Steps

History

elasticmachine commented Mar 14, 2024 • edited

💔 Build Failed

Failed CI Steps

History

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

History

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

History

elasticmachine commented Mar 14, 2024

💚 Build Succeeded

belimawr commented Feb 19, 2024 •

edited

elasticmachine commented Feb 19, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited

elasticmachine commented Mar 14, 2024 •

edited