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.

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, 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 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.

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.

Step 2: Create the Telemetry Collector in the UI

For general instructions on creating and managing Telemetry Collector configurations, see 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).

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
• Kubernetes: See Deploy 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