Я пытаюсь заставить filebeat потреблять сообщения от kafka, используя ввод kafka. По какой-то причине я не могу пройти аутентификацию с помощью SASL, и я не уверен, почему это так. Документация как для Kafka, так и для Filebeat немного не хватает при попытке использовать ее с SASL.
Моя конфигурация filebeat следующая:
filebeat.config:
modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
filebeat.inputs:
- type: kafka
hosts: 'the.kafka.server.com:9092'
topics: 'my_topic'
group_id: 'my_group'
ssl.enabled: yes
username: "$ConnectionString"
password: "org.apache.kafka.common.security.plain.PlainLoginModule required username='my_username' password='my_password';"
processors:
- add_cloud_metadata: ~
- add_docker_metadata: ~
output.console:
pretty: true
Вывод показывает
INFO input/input.go:114 Starting input of type: kafka; ID: 14409252276502564738
INFO kafka/log.go:53 kafka message: Initializing new client
INFO kafka/log.go:53 client/metadata fetching metadata for all topics from broker the.kafka.server.com:9092
INFO crawler/crawler.go:106 Loading and starting Inputs completed. Enabled inputs: 1
INFO cfgfile/reload.go:171 Config reloader started
INFO cfgfile/reload.go:226 Loading of config files completed.
INFO kafka/log.go:53 kafka message: Successful SASL handshake. Available mechanisms: %!(EXTRA []string=[PLAIN OAUTHBEARER])
INFO kafka/log.go:53 Failed to read response while authenticating with SASL to broker the.kafka.server.com:9092: EOF
INFO kafka/log.go:53 Closed connection to broker the.kafka.server.com:9092
INFO kafka/log.go:53 client/metadata got error from broker -1 while fetching metadata: EOF
Я не уверен, что здесь происходит. Я также пробовал добавить compression: none
который не помог и подтвердил с помощью openssl, что сертификат сервера может быть проверен. Что я здесь делаю не так? Рассматриваемый сервер kafka является облачным сервером kafka, и я не вижу конфигурацию сервера, мне была предоставлена «строка подключения» из облачного пользовательского интерфейса kafka.
Я нашел проблему, $ConnectionString
синтаксис не работает с конфлюэнтными облачными кластерами кафка. Правильный синтаксис следующий:
filebeat.inputs:
- type: kafka
hosts: 'the.kafka.server.com:9092'
topics: 'my_topic'
group_id: 'my_group'
ssl.enabled: yes
username: <API KEY>
password: <API SECRET>
Этого достаточно, чтобы он подключился и потреблял