---
title: "Telemetry Ingestion via Kafka"
slug: "telemetry-ingestion-via-kafka"
updated: 2026-11-04T18:48:24Z
published: 2026-11-04T18:48:24Z
canonical: "docs.crossworkassurance.cisco.com/telemetry-ingestion-via-kafka"
---

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

# Telemetry Ingestion via Kafka

This article describes how to configure the Telemetry Collector to **consume** metrics from an Apache Kafka topic and forward them through the Sensor Collector to Crosswork Assurance. This guide covers Kafka as a data *input* source.

For exporting data *from* Crosswork Assurance to an external Kafka system, see [Export to an External Kafka](/docs/exporting-to-an-external-kafka).

#### Out-of-the-box vs. Custom Telemetry

Crosswork Assurance provides default Telemetry Collector configurations for IOS-XE and IOS-XR routers. This guide covers **Kafka as a custom data source** that requires a custom Telegraf configuration (TOML). For custom telemetry ingestion engagements, contact your Cisco CX or Partner representative.

The Telemetry Collector embeds [Telegraf](https://github.com/influxdata/telegraf), which includes the `kafka_consumer` input plugin. By customizing the Telemetry Collector's Telegraf configuration (TOML), you can configure it to consume messages from one or more Kafka topics and convert them to OpenMetrics format for ingestion into Crosswork Assurance.

## Prerequisites

- A deployed **Gateway Sensor Collector** with the OpenMetrics Gateway connector enabled.
- Network connectivity from the Telemetry Collector to the Kafka broker(s) on the configured port (typically 9092).
- Knowledge of the Kafka topic name(s), message format (JSON, etc.), and broker addresses.
- Access to the Crosswork Assurance UI with permissions to create Telemetry Collector configurations.

---

## Step 1: Design the Telegraf Configuration (TOML)

The Telemetry Collector's runtime behavior is controlled by a Telegraf TOML configuration file. When you create a Telemetry Collector with a **Custom** Telegraf configuration in the UI, you upload a TOML file that defines:

- **Inputs** — which Kafka topics to consume and how to parse the messages
- **Processors** — how to build session identity tags for Crosswork Assurance

**Important:** Do **not** include an `[[outputs.*]]` section in the TOML file. The product automatically appends its own output configuration for forwarding data to the Sensor Collector. Including an output section causes a "violation of Telegraf's configuration" error.

### Key Design Principles

- **One `[[inputs.kafka_consumer]]` block per topic.** Different topics typically carry different payload structures that require different parsers. A single input subscribing to multiple topics with different schemas applies only one parser and drops fields from non-matching topics.
- **One `consumer_group` per input.** Use a unique consumer group name for each input to avoid conflicts across the consumers.
- **The `objectType` tag is required.** The Telemetry Collector agent requires an `objectType` tag whose value matches the name of the ingestion dictionary that defines the metrics and dimensions being ingested for the objects. You can set `objectType` as a constant tag on the input when a single Kafka topic carries only one object type, or derive it from the Kafka message itself using a `[[processors.template]]` block (similar to how `sessionId` and `sessionName` are derived) when a topic may carry multiple object types. Use [Telemetry Ingestion Staging](/latest/docs/telemetry-ingestion-staging-feature) to create and manage your ingestion dictionaries.
- **Session identity processors are required.** The Telemetry Collector opens one reporting stream per `sessionId`. Include `[[processors.template]]` blocks that build `sessionId` and `sessionName` from your parsed tags.

### Example TOML Configuration

The following example shows a `kafka_consumer` input consuming JSON metrics from a Kafka topic, with `json_v2` parsing and session identity. In this example, a single topic carries one object type, so `objectType` is set as a constant tag. If your topic carries multiple object types, derive `objectType` from the message using a `[[processors.template]]` block instead.

```
###############################################################################
# Kafka Consumer Input — one block per topic
###############################################################################
[[inputs.kafka_consumer]]
  brokers = ["kafka-broker:9092"]
  topics = ["my-telemetry-topic"]
  consumer_group = "crosswork-assurance-my-topic"
  offset = "newest"
  topic_tag = "topic"
  max_message_len = 1000000
  data_format = "json_v2"

  # Optional: SASL authentication for secured Kafka clusters
  # security_protocol = "SASL_SSL"
  # sasl_mechanism = "SCRAM-SHA-512"
  # sasl_username = "${KAFKA_USERNAME}"
  # sasl_password = "${KAFKA_PASSWORD}"

  # objectType must match the ingestion dictionary name.
  # Use a constant tag (shown here) when a topic carries one object type,
  # or derive it from the message via [[processors.template]] for mixed topics.
  [inputs.kafka_consumer.tags]
    objectType = "my_metrics"

  [[inputs.kafka_consumer.json_v2]]
    measurement_name = "my_metrics"
    timestamp_path = "timestamp"
    timestamp_format = "2006-01-02T15:04:05Z07:00"

    # Dimensions (tags) — fields used to identify the source
    [[inputs.kafka_consumer.json_v2.tag]]
      path = "device_name"
    [[inputs.kafka_consumer.json_v2.tag]]
      path = "interface"

    # Metrics (fields) — numeric values to ingest
    [[inputs.kafka_consumer.json_v2.field]]
      path = "metrics.in_octets"
      rename = "in_octets"
      type = "int"
    [[inputs.kafka_consumer.json_v2.field]]
      path = "metrics.out_octets"
      rename = "out_octets"
      type = "int"

###############################################################################
# Session Identity (required)
#
# Build a unique sessionId and sessionName per reporting stream.
# The Telemetry Collector opens one stream per sessionId.
###############################################################################
[[processors.template]]
  namepass = ["my_metrics*"]
  tag = "sessionId"
  template = '{{ .Tag "device_name" }}_{{ .Tag "interface" }}_metrics'

[[processors.template]]
  namepass = ["my_metrics*"]
  tag = "sessionName"
  template = '{{ .Tag "device_name" }}_{{ .Tag "interface" }}_metrics'
```

**Broker address:** The broker address depends on your deployment type. For Docker deployments where the Kafka broker runs on the same Docker network, use the container name (e.g., `kafka:9092`). For Kubernetes deployments, use the internal service DNS name (e.g., `kafka.my-namespace.svc.cluster.local:19092`).

### Kafka Consumer Plugin Reference

The `kafka_consumer` input plugin supports many additional options beyond those shown in the example above, including TLS configuration, SASL authentication (PLAIN, SCRAM, GSSAPI, OAUTHBEARER, AWS-MSK-IAM), compression codecs, consumer group partition strategies, and message processing tuning.

For the complete list of configuration options, see the [Telegraf Kafka Consumer plugin documentation](https://github.com/influxdata/telegraf/blob/master/plugins/inputs/kafka_consumer/README.md).

### Supported Data Formats

The `kafka_consumer` plugin can parse incoming messages in any format supported by Telegraf's input data parsers, including JSON, JSON v2, Avro, CSV, InfluxDB line protocol, Prometheus, OpenMetrics, and Protocol Buffers (via the XPath parser). For JSON payloads, the `json_v2` parser is recommended because it provides explicit field path mapping.

For the full list of supported parsers and their configuration options, see the [Telegraf Input Data Formats documentation](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).

---

## Step 2: Create the Telemetry Collector in the UI

For general instructions on creating and managing Telemetry Collector configurations, see [Configure Telemetry Collector](/latest/docs/configure-telemetry-collector). The steps below are specific to uploading a custom Kafka TOML configuration.

1. In the Crosswork Assurance UI, navigate to **Sensors > Collectors**.
2. Click the **+** button to add a new collector.
3. In the **New collector** dialog, select **Telemetry** and click **Configure collector**.
4. Enter a **Name** for the Telemetry Collector (e.g., `My_Kafka_Telemetry_Collector`).
5. Set the **Telegraf Configuration** dropdown to **Custom**.
6. In the **Custom Telegraf.conf** section, click the upload icon and select your TOML file.
7. Enter the **Sensor Collector Data Address** — the hostname or IP of your Gateway Sensor Collector.
8. Enter the **Sensor Collector Data Port** — the data port of your Gateway Sensor Collector (e.g., `39888`).
9. Configure SSL options as needed:
  - **Use SSL communication** — enable for encrypted connections to the Sensor Collector
  - **Allow self-signed certificates** — enable if the Sensor Collector uses self-signed certificates
  - **Skip SSL hostname check** — enable if the hostname does not match the certificate
10. Click the **checkmark** (save) button.

### Generate the Auth Token

After saving the Telemetry Collector configuration:

1. On the **Summary** tab, click **Generate auth token**.
2. Copy the **Agent secret (YAML)** content. This contains the `agentId` and `authenticationToken` that the Telemetry Collector container needs to authenticate with Crosswork Assurance.
3. Save this content as `secrets.yaml` in your deployment directory.

> **Important:** The auth token is displayed only once. Copy and save it immediately.

### Alternative: API Method (Base64)

You can also create the Telemetry Collector configuration via API by Base64-encoding the TOML file and submitting it as the `customTransformationConfig` field.

1. Base64-encode your TOML file:

```
base64 -w 0 my-config.toml > config_base64.txt
```
2. Use the Agent Configuration API to create the Telemetry Collector. For detailed API instructions, see [Telemetry Collector Configuration - MDT Dial-Out (alternate API method)](/docs/ios-xe-and-xr-via-mdt-dial-out-alternate-api-method).

---

## Step 3: Deploy the Telemetry Collector

Deploy the Telemetry Collector container using the standard procedure for your environment:

- **Docker:** See [Deploy Telemetry Collector in Docker](/latest/docs/deploying-telemetry-collector-in-docker)
- **Kubernetes:** See [Deploy Telemetry Collector in Kubernetes](/latest/docs/deploying-telemetry-collector-in-kubernetes)

The Telemetry Collector connects to the Sensor Collector, receives its custom TOML configuration, and begins consuming from the specified Kafka topic(s).

> **Note:** Ensure the Telemetry Collector container has network access to the Kafka broker(s). For Docker deployments, both containers must be on the same Docker network. For Kubernetes deployments, verify that the Kafka service is accessible from the Telemetry Collector pod's namespace.

---

## Step 4: Verify

1. Check the Telemetry Collector logs for successful startup:

```
Config watcher started
Loading config: /etc/Telemetry/Telemetry.conf
Connected to Kafka broker: kafka-broker:9092
```
2. Confirm that metrics appear in the Crosswork Assurance UI under the expected monitored objects.
3. To enable debug logging for troubleshooting:

```
docker exec -it <container_id> sh
agentstatus telemetry datapathLogging true
exit
docker logs <container_id> -f
```

---

## Troubleshooting

| Symptom | Cause | Resolution |
| --- | --- | --- |
| No data from Kafka topic | Wrong topic name or consumer group | Verify `topics` and `consumer_group` in the TOML match the Kafka broker configuration. |
| Connection refused to Kafka broker | Network or firewall issue | Verify connectivity from the Telemetry Collector host to the Kafka broker(s) on the configured port. For Docker, ensure both containers are on the same network. |
| TLS handshake failure | Missing or incorrect CA certificate | Verify `tls_ca` path and ensure the CA matches the broker's certificate. |
| Authentication failure | SASL credentials incorrect | Verify `sasl_username` and `sasl_password` environment variables are set correctly. |
| "consume: context canceled" | Multiple inputs sharing the same `consumer_group` | Use a unique `consumer_group` for each `[[inputs.kafka_consumer]]` block. |
| "violation of Telegraf's configuration" | Output section defined in the TOML | Remove any `[[outputs.*]]` blocks from the TOML file. The product manages the output automatically. |
| Metrics missing fields | Wrong parser or field paths | Verify `data_format`, `json_v2` field `path` values, and `type` settings match the actual Kafka message structure. |
| Collector status "Unregistered" | Container not deployed or not connected | Deploy the Telemetry Collector container and verify it can reach the Sensor Collector address and port. |

---