Экспорт в Kafka

Профиль kafka использует channel_name и script_name совместно: первый выбирает источник, второй формирует JSON сообщения. Для TCP-, UDP- и IP-сессий используются каналы sessions:tcp, sessions:udp, sessions:ip и скрипты kafka_sessions_tcp.lua, kafka_sessions_udp.lua, kafka_sessions_ip.lua из scripts/tr-kafka/.

Таблица. Параметры kafka

ПараметрЗначенияНазначение и особенности

kafka_bootstrap_servers

host:port,host:port

Список брокеров; обычный порт 9092.

kafka_topic

Строка

Целевой топик.

kafka_security_protocol

plaintext / ssl; plaintext

Без шифрования или TLS. Пустое значение приводит к предупреждению и plaintext.

kafka_ssl_ca_location

Путь PEM

Корневой сертификат для проверки брокера.

kafka_ssl_certificate_location

Путь PEM

Клиентский сертификат для взаимной аутентификации.

kafka_ssl_key_location

Путь PEM

Закрытый ключ клиента.

kafka_ssl_key_password

Строка

Пароль зашифрованного ключа.

kafka_enable_ssl_certificate_verification

true / false; true

Проверка сертификата брокера.

kafka_compression_codec

none / gzip / snappy / lz4 / zstd; none

Сжатие сообщения.

kafka_batch_num_messages

Целое; 10000

Максимальное число сообщений в пакете.

kafka_request_required_acks

Целое; 1

0 — без подтверждения, 1 — подтверждение лидера. Отрицательное значение не применяется.

kafka_log_level

0–7; 7

Уровень журналирования librdkafka.

Поддерживаются plaintext, TLS с проверкой брокера и TLS с клиентским сертификатом. Аутентификация SASL этим профилем не настраивается. Если брокер требует подтверждения клиента, используйте согласованный listener с взаимной TLS-аутентификацией.

{
  "name": "kafka_tcp_sessions",
  "enabled": true,
  "default": false,
  "protocol": "kafka",
  "channel_name": "sessions:tcp",
  "script_name": "kafka_sessions_tcp.lua",
  "kafka_bootstrap_servers": "192.0.2.110:9093",
  "kafka_topic": "nta-tcp-sessions",
  "kafka_security_protocol": "ssl",
  "kafka_ssl_ca_location": "/opt/alertixnta/certs/kafka.ca",
  "kafka_ssl_certificate_location": "/opt/alertixnta/certs/kafka.pem",
  "kafka_ssl_key_location": "/opt/alertixnta/certs/kafka.key",
  "kafka_enable_ssl_certificate_verification": true,
  "kafka_compression_codec": "lz4",
  "kafka_request_required_acks": 1
}

До включения разместите сертификаты, проверьте права чтения ключа и доступность брокеров, которые кластер сообщает клиенту. После reload проверьте не только отсутствие ошибки подключения, но и появление JSON в нужном топике. Выбор acks влияет на подтверждение доставки; не обещает сам по себе отсутствие дублей или сквозную семантику exactly once.