Overview

Use Observability Pipelines’ Kafka destination to send logs to Kafka topics.

When to use this destination

Common scenarios when you might use this destination:

• To route logs to the following destinations:
  • Clickhouse: An open-source column-oriented database management system used for analyzing large volumes of logs.
  • Snowflake: A data warehouse used for storage and query.
    • Snowflake’s API integration utilizes Kafka as a method to ingest logs into their platform.
  • Databricks: A data lakehouse for analytics and storage.
  • Azure Event Hub: An ingest and processing service in the Microsoft and Azure ecosystem.
• To route data to Kafka and use the Kafka Connect ecosystem.
• To process and normalize your data with Observability Pipelines before routing to Apache Spark with Kafka to analyze data and run machine learning workloads.

Setup

Configure the Kafka destination when you set up a pipeline. You can set up a pipeline in the UI, using the API, or with Terraform. The steps in this section are configured in the UI.

After you select the Kafka destination in the pipeline UI:

1. Enter the identifier for your Kafka bootstrap servers. If you leave it blank, the default is used.
2. Enter the name of the topic you want to send logs to.
3. In the Encoding dropdown menu, select either JSON or Raw message as the output format.

Optional settings

Enable TLS

Toggle the switch to Enable TLS.

• If you are using Secrets Management, enter the identifier for the key pass. See Set secrets for the default used if the field is left blank.
• The following certificate and key files are required:
  • Server Certificate Path: The path to the certificate file that has been signed by your Certificate Authority (CA) root file in DER, PEM, or CRT (X.509).
  • CA Certificate Path: The path to the certificate file that is your Certificate Authority (CA) root file in DER, PEM, or CRT (X.509).
  • Private Key Path: The path to the .key private key file that belongs to your Server Certificate Path in DER, PEM, or CRT (PKCS #8) format.
  • Notes:
    • The configuration data directory /var/lib/observability-pipelines-worker/config/ is automatically appended to the file paths. See Advanced Worker Configurations for more information.
    • The file must be readable by the observability-pipelines-worker group and user.

Enable SASL authentication

1. Toggle the switch to enable SASL Authentication.
2. Enter the identifiers for your Kafka SASL username and password. If you leave them blank, the defaults are used.
3. Select the mechanism (PLAIN, SCHRAM-SHA-256, or SCHRAM-SHA-512) in the dropdown menu.

Enable compression

1. Toggle switch to Enable Compression.
2. In the Compression Algorithm dropdown menu, select a compression algorithm (gzip, zstd, lz4, or snappy).
3. (Optional) Select a Compression Level in the dropdown menu. If the level is not specified, the algorithm’s default level is used.

Buffering

Toggle the switch to enable Buffering Options. Enable a configurable buffer on your destination to ensure intermittent latency or an outage at the destination doesn’t create immediate backpressure, and allow events to continue to be ingested from your source. Disk buffers can also increase pipeline durability by writing data to disk, ensuring buffered data persists through a Worker restart. See Destination buffers for more information.

• If left unconfigured, your destination uses a memory buffer with a capacity of 500 events.
• To configure a buffer on your destination:
  1. Select the buffer type you want to set (Memory or Disk).
  2. Enter the buffer size and select the unit.
     1. Maximum memory buffer size is 128 GB.
     2. Maximum disk buffer size is 500 GB.
  3. In the Behavior on full buffer dropdown menu, select whether you want to block events or drop new events when the buffer is full.

Advanced options

Click Advanced if you want to set any of the following fields:

1. Message Key Field: Specify which log field contains the message key for partitioning, grouping, and ordering.
2. Headers Key: Specify which log field contains your Kafka headers. If left blank, no headers are written.
3. Message Timeout (ms): Local message timeout, in milliseconds. Default is 300,000 ms.
4. Socket Timeout (ms): Default timeout, in milliseconds, for network requests. Default is 60,000 ms.
5. Rate Limit Events: The maximum number of requests the Kafka client can send within the rate limit time window. Default is no rate limit.
6. Rate Limit Time Window (secs): The time window used for the rate limit option.
   • This setting has no effect if the rate limit for events is not set.
   • Default is 1 second if Rate Limit Events is set, but Rate Limit Time Window is not set.
7. To add additional librdkafka options, click Add Option and select an option in the dropdown menu.
   1. Enter a value for that option.
   2. Check your values against the librdkafka documentation to make sure they have the correct type and are within the set range.
   3. Click Add Option to add another librdkafka option.

Secret defaults

These are the defaults used for secret identifiers and environment variables.

• Kafka bootstrap servers identifier:
  • References the bootstrap server that the client uses to connect to the Kafka cluster and discover all the other hosts in the cluster.
  • In your secrets manager, the host and port must be entered in the format of host:port, such as 10.14.22.123:9092. If there is more than one server, use commas to separate them.
  • The default identifier is DESTINATION_KAFKA_BOOTSTRAP_SERVERS.
• Kafka TLS passphrase identifier (when TLS is enabled):
  • The default identifier is DESTINATION_KAFKA_KEY_PASS.
• SASL authentication (when enabled):
  • Kafka SASL username identifier:
    • The default identifier is DESTINATION_KAFKA_SASL_USERNAME.
  • Kafka SASL password identifier:
    • The default identifier is DESTINATION_KAFKA_SASL_PASSWORD.

Kafka bootstrap servers

• The host and port of the Kafka bootstrap servers.
• This is the bootstrap server that the client uses to connect to the Kafka cluster and discover all the other hosts in the cluster. The host and port must be entered in the format of host:port, such as 10.14.22.123:9092. If there is more than one server, use commas to separate them.
• The default environment variable is DD_OP_DESTINATION_KAFKA_BOOTSTRAP_SERVERS.

TLS (when enabled)

• If TLS is enabled, the Kafka TLS passphrase is needed.
• The default environment variable is DD_OP_DESTINATION_KAFKA_KEY_PASS.

SASL (when enabled)

• Kafka SASL username
  • The default environment variable is DD_OP_DESTINATION_KAFKA_SASL_USERNAME.
• Kafka SASL password
  • The default environment variable is DD_OP_DESTINATION_KAFKA_SASL_PASSWORD.

librdkafka options

These are the available librdkafka options:

• client.id
• queue.buffering.max_messages
• transactional.id
• enable.idempotence
• acks

See the librdkafka documentation for more information and to ensure your values have the correct type and are within range.

Metrics

For component metrics and destination buffer metrics emitted by all destinations, see the Pipelines Usage Metrics documentation.

Kafka metrics

• Use the component_id tag to filter or group by individual components.
• The component_type tag is kafka for Kafka destination metrics.

pipelines.kafka_produced_messages_total

Description: The number of messages produced and sent to Kafka brokers.

Metric type: count

pipelines.kafka_produced_messages_bytes_total

Description: The number of message bytes produced and sent to Kafka brokers.

Metric type: count

pipelines.kafka_queue_messages

Description: Current number of messages in the librdkafka producer queue.

Metric type: gauge

pipelines.kafka_queue_messages_bytes

Description: Current total size, in bytes, of messages in the librdkafka producer queue.

Metric type: gauge

pipelines.kafka_requests_total

Description: The number of requests sent to Kafka brokers.

Metric type: count

pipelines.kafka_requests_bytes_total

Description: The number of bytes transmitted to Kafka brokers.

Metric type: count

pipelines.kafka_responses_total

Description: The number of responses received from Kafka brokers.

Metric type: count

pipelines.kafka_responses_bytes_total

Description: The number of bytes received from Kafka brokers.

Metric type: count

Event batching

A batch of events is flushed when one of these parameters is met. See event batching for more information.