> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Monitoring Kafka logs with ClickHouse Observability

> Collect Kafka broker logs with OpenTelemetry and send them to ClickHouse Observability.

Collect Kafka broker logs with the OpenTelemetry `filelog` receiver and send them to ClickHouse Observability. Use the [demo dataset](#demo-dataset) to try the integration without a Kafka deployment.

## Prerequisites

* A ClickHouse Observability service
* Docker and `curl`
* For your own logs, access to Kafka broker log files matching the format below

### Connection details

> [tab] Fully managed
>
> [Beta offering](https://clickhouse.com/docs/reference/settings/beta-and-experimental-features.md#beta-features)
>
> Open ClickHouse Observability and go to **Team Settings → Data → Managed Telemetry**. Copy your OTLP HTTP endpoint and API key. You can use **Copy connection details** to copy the configuration, or **Reveal key** to view the masked key.
>
> ![Managed Telemetry connection details](https://mintcdn.com/private-7c7dfe99/Zk1feNn6fCQBpold/images/clickstack/getting-started/observability-managed-telemetry.png?fit=max\&auto=format\&n=Zk1feNn6fCQBpold\&q=85\&s=e63a2ac2e6ab996d3fc360ebe0f487a5)
>
> For the OTLP exporter examples below, set these variables in the terminal where you will run the collector:
>
> ```bash theme={null}
> export OTEL_EXPORTER_OTLP_ENDPOINT='https://YOUR_OTEL_ENDPOINT:4318'
> export OTEL_EXPORTER_OTLP_AUTHORIZATION='Bearer YOUR_API_KEY'
> ```
>
> Replace the placeholders with the exact endpoint, including its port, and API key from your service. Use the base HTTP endpoint without a `/v1/logs` suffix; the exporter appends the signal path.

> [tab] Self-managed
>
> Use your collector's reachable OTLP HTTP endpoint (for example, `http://YOUR_COLLECTOR_HOST:4318`) and authentication settings. If your collector does not require authentication, omit the exporter's `headers` block from the configurations below. See the [self-managed getting-started guide](https://clickhouse.com/docs/clickstack/self-configured/getting-started/managed.md) for setup guidance.
>
> ```bash theme={null}
> export OTEL_EXPORTER_OTLP_ENDPOINT='http://YOUR_COLLECTOR_HOST:4318'
> export OTEL_EXPORTER_OTLP_AUTHORIZATION='YOUR_INGESTION_KEY'
> ```

The endpoint must be reachable from the collector container. For a destination collector running on this computer, use `http://host.docker.internal:4318`. The Docker commands below include the host mapping.

## Integration with existing Kafka

**1. Verify Kafka logging configuration**

Locate your broker logs. Common locations include `$KAFKA_HOME/logs` and `/var/log/kafka`, depending on your installation and logging configuration:

```bash theme={null}
ls "$KAFKA_HOME/logs/"
ls /var/log/kafka/
```

Common files are `server.log` (broker events), `controller.log` (controller events), and `state-change.log` (partition and replica transitions). Separate files exist only when configured in your logging appenders.

The parser in this guide expects this Log4j text format:

```text theme={null}
[2026-03-09 14:23:45,123] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
```

If your broker logs only to stdout, configure a file appender in the broker's logging configuration and share that directory with the collector. The `filelog` receiver does not read container stdout directly. Adjust the parser if your logging pattern differs.

**2. Configure the collector**

Save this as `kafka-logs-monitoring.yaml`:

```yaml theme={null}
receivers:
  filelog/kafka:
    include:
      - /var/log/kafka/server.log
      - /var/log/kafka/controller.log
      - /var/log/kafka/state-change.log
    start_at: end
    force_flush_period: 1s
    multiline:
      line_start_pattern: '^\[\d{4}-\d{2}-\d{2}'
    operators:
      - type: regex_parser
        regex: '(?s)^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})\] (?P<severity>\w+) (?P<message>.*)$'
        parse_from: body
        parse_to: attributes
        timestamp:
          parse_from: attributes.timestamp
          layout: '%Y-%m-%d %H:%M:%S,%L'
          location: UTC
        severity:
          parse_from: attributes.severity
      - type: move
        from: attributes.message
        to: body
      - type: add
        field: attributes.source
        value: kafka
      - type: add
        field: resource["service.name"]
        value: kafka-service
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 400
    spike_limit_mib: 100
  transform/kafka:
    error_mode: propagate
    log_statements:
      - context: log
        statements:
          - set(log.severity_text, ToLowerCase(log.severity_text))
  batch:
    timeout: 5s
    send_batch_size: 500
    send_batch_max_size: 500
exporters:
  otlp_http/observability:
    endpoint: ${env:OTEL_EXPORTER_OTLP_ENDPOINT}
    headers:
      authorization: ${env:OTEL_EXPORTER_OTLP_AUTHORIZATION}
service:
  pipelines:
    logs/kafka:
      receivers: [filelog/kafka]
      processors: [memory_limiter, transform/kafka, batch]
      exporters: [otlp_http/observability]
```

Set `location` to the timezone used by your broker logs; the example assumes UTC. Multiline entries retain their complete stack traces. The collector reads new entries with `start_at: end` and normalizes severity text to lowercase.

**3. Run the collector**

Set the [connection variables](#connection-details), then run from the configuration directory. Adjust the host log path if needed. For group-readable files, add `--group-add LOG_GROUP_ID` using the numeric group shown by `ls -ln /var/log/kafka`.

```bash theme={null}
docker run --rm --name kafka-logs-collector \
  --add-host=host.docker.internal:host-gateway \
  --memory=512m \
  -e OTEL_EXPORTER_OTLP_ENDPOINT \
  -e OTEL_EXPORTER_OTLP_AUTHORIZATION \
  -v "$(pwd)/kafka-logs-monitoring.yaml:/etc/otelcol-contrib/config.yaml:ro" \
  -v /var/log/kafka:/var/log/kafka:ro \
  otel/opentelemetry-collector-contrib:0.161.0 \
  --config=/etc/otelcol-contrib/config.yaml
```

Leave the terminal open and generate broker activity that writes new log entries. Stop with **Ctrl+C** when finished testing.

**4. Verify logs**

Open **Search**, select **Logs**, filter for `ServiceName:kafka-service`, and select **Last 15 minutes**. You should see broker messages with parsed timestamps and severity.

## Demo dataset

The sample contains 6,376 Kafka broker log entries spanning approximately 24 hours. The collector refreshes their timestamps while preserving messages and stack traces.

**1. Download the sample**

```bash theme={null}
curl -fL -o server.log \
  https://datasets-documentation.s3.eu-west-3.amazonaws.com/clickstack-integrations/kafka/server.log
```

**2. Configure the demo collector**

Save this as `kafka-logs-demo.yaml`:

```yaml theme={null}
receivers:
  filelog/kafka:
    include:
      - /var/log/kafka/server.log
    start_at: beginning
    force_flush_period: 1s
    multiline:
      line_start_pattern: '^\[\d{4}-\d{2}-\d{2}'
    operators:
      - type: regex_parser
        regex: '(?s)^\[(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3})\] (?P<severity>\w+) (?P<message>.*)$'
        parse_from: body
        parse_to: attributes
        timestamp:
          parse_from: attributes.timestamp
          layout: '%Y-%m-%d %H:%M:%S,%L'
          location: UTC
        severity:
          parse_from: attributes.severity
      - type: move
        from: attributes.message
        to: body
      - type: add
        field: attributes.source
        value: kafka-demo
      - type: add
        field: resource["service.name"]
        value: kafka-demo
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 400
    spike_limit_mib: 100
  transform/kafka:
    error_mode: propagate
    log_statements:
      - context: log
        statements:
          - set(log.severity_text, ToLowerCase(log.severity_text))
  transform/demo_timestamps:
    error_mode: propagate
    log_statements:
      - context: log
        statements:
          - set(log.time, Now() - Duration("2m") - (Time("2026-03-09T23:59:58.180Z", "%Y-%m-%dT%H:%M:%S.%LZ") - log.time))
  batch:
    timeout: 5s
    send_batch_size: 500
    send_batch_max_size: 500
exporters:
  otlp_http/observability:
    endpoint: ${env:OTEL_EXPORTER_OTLP_ENDPOINT}
    headers:
      authorization: ${env:OTEL_EXPORTER_OTLP_AUTHORIZATION}
service:
  pipelines:
    logs/kafka:
      receivers: [filelog/kafka]
      processors: [memory_limiter, transform/kafka, transform/demo_timestamps, batch]
      exporters: [otlp_http/observability]
```

**3. Run the demo collector**

Set the [connection variables](#connection-details), then run from the directory containing both files:

```bash theme={null}
docker run --rm --name kafka-logs-demo-collector \
  --add-host=host.docker.internal:host-gateway \
  --memory=512m \
  -e OTEL_EXPORTER_OTLP_ENDPOINT \
  -e OTEL_EXPORTER_OTLP_AUTHORIZATION \
  -v "$(pwd)/kafka-logs-demo.yaml:/etc/otelcol-contrib/config.yaml:ro" \
  -v "$(pwd)/server.log:/var/log/kafka/server.log:ro" \
  otel/opentelemetry-collector-contrib:0.161.0 \
  --config=/etc/otelcol-contrib/config.yaml
```

Leave the terminal open while verifying ingestion. Stop with **Ctrl+C** when finished. Running the demo again re-ingests the sample with refreshed timestamps.

**4. Verify demo logs**

Open **Search**, select **Logs**, filter for `ServiceName:kafka-demo`, and select **Last 2 days**.

![Kafka demo logs with parsed severity in ClickHouse Observability Search](https://mintcdn.com/private-7c7dfe99/dW6bcEakZdn3a5oP/images/clickstack/observability-kafka-demo-logs.png?fit=max\&auto=format\&n=dW6bcEakZdn3a5oP\&q=85\&s=c49e4a1e2e9119875f5ff6c728ac0df7)

## Dashboards and visualization

**1. [Download the dashboard configuration](https://clickhouse-docs-assets.s3.us-east-1.amazonaws.com/examples/kafka-logs-dashboard.json)**

**2. Import the dashboard**

1. Open **Dashboards** and click **Import** in the upper-right corner.
2. Upload `kafka-logs-dashboard.json`.
3. Map the charts to your **Logs** source and click **Finish Import**.

![Import dashboard button](https://mintcdn.com/private-7c7dfe99/dW6bcEakZdn3a5oP/images/clickstack/observability-import-dashboard.png?fit=max\&auto=format\&n=dW6bcEakZdn3a5oP\&q=85\&s=16bcb7e6365ca0f6decac5c693e228eb)

**3. View the dashboard**

For the sample, optionally filter for `ServiceName:kafka-demo` and select **Last 2 days**. For your own logs, use `ServiceName:kafka-service` and a time range covering your records.

![Kafka logs dashboard in ClickHouse Observability showing log volume, severity, replication, and connection events](https://mintcdn.com/private-7c7dfe99/dW6bcEakZdn3a5oP/images/clickstack/observability-kafka-logs-dashboard.png?fit=max\&auto=format\&n=dW6bcEakZdn3a5oP\&q=85\&s=adbd3043742d74777d32803cc8e71784)

## Troubleshooting

### No logs appearing

Check the collector output for file permission, parsing, authentication, or export errors. Confirm your broker writes to the mounted directory and the filenames match the receiver's `include` list. The live configuration reads new entries only.

### Logs not parsing correctly

Compare a broker log entry with the expected format:

```bash theme={null}
tail -n 20 /var/log/kafka/server.log
```

Adjust the regex and multiline start pattern if your logging format differs. Set the timestamp parser's `location` to your broker's logging timezone.

## Next steps

* Set up [alerts](https://clickhouse.com/docs/clickstack/observability/features/alerts.md) for broker failures or replication errors.
* Combine with [Kafka metrics](https://clickhouse.com/docs/clickstack/observability/integration-examples/kafka-metrics.md).
* Create additional [dashboards](https://clickhouse.com/docs/clickstack/observability/features/dashboards/overview.md) for controller events or partition reassignment.
* See [Sending OpenTelemetry data](https://clickhouse.com/docs/clickstack/observability/ingesting-data/opentelemetry.md) for deployment guidance.