How to Integrate OpenTelemetry Tracing with Kourier for Distributed Tracing

Integrate OpenTelemetry tracing with Kourier by adding the amqp-client-opentelemetry module, obtaining a Tracer from your OpenTelemetry SDK, and wrapping your AMQPConnection or AMQPChannel using the withTracing extension functions.

Kourier, the Kotlin-native AMQP client library from guimauvedigital/kourier, provides a dedicated instrumentation module that automatically instruments connection operations, channel management, message publishing, and consumption. This integration propagates the W3C Trace Context through AMQP message headers, enabling end-to-end visibility across distributed microservices.

Understanding Kourier's OpenTelemetry Architecture

The amqp-client-opentelemetry module implements a decorator pattern around Kourier's core client. It intercepts AMQP operations to create OpenTelemetry spans without modifying your business logic.

Core Instrumentation Components

Component	Responsibility	Source Location
OpenTelemetryAMQPConnection	Wraps AMQPConnection to instrument connection-level operations (open, close, heartbeat) and returns traced channels.	OpenTelemetryAMQPConnection.kt
OpenTelemetryAMQPChannel	Wraps AMQPChannel to create spans for basicPublish and basicConsume, and injects trace context into message headers.	OpenTelemetryAMQPChannel.kt
TracingConfig	Controls instrumentation granularity via flags like traceConnectionOperations, traceChannelManagementOperations, and captureMessageBody.	TracingConfig.kt
withTracing Extensions	Public API entry points that wrap existing connections or channels with the instrumented variants.	Extensions.kt

When operations execute, the wrappers use an executeInSpan helper to start a span, make it current in the OpenTelemetry context, execute the underlying AMQP operation, record exceptions if errors occur, and finally end the span.

Step-by-Step Integration Guide

1. Add the OpenTelemetry Module Dependency

Include the Kourier OpenTelemetry instrumentation module and the OpenTelemetry API in your build configuration. Replace 0.4.2 with the latest available version.

// build.gradle.kts
dependencies {
    implementation("dev.kourier:amqp-client-opentelemetry:0.4.2")
    implementation("io.opentelemetry:opentelemetry-api:1.44.1")
}

2. Configure the OpenTelemetry SDK

Initialize the OpenTelemetry SDK with your preferred exporter (OTLP, Jaeger, Zipkin, etc.). The critical requirement is obtaining a io.opentelemetry.api.trace.Tracer instance to pass to Kourier.

import io.opentelemetry.sdk.OpenTelemetrySdk
import io.opentelemetry.sdk.trace.SdkTracerProvider
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter

val exporter = OtlpGrpcSpanExporter.builder()
    .setEndpoint("http://localhost:4317")
    .build()

val tracerProvider = SdkTracerProvider.builder()
    .addSpanProcessor(BatchSpanProcessor.builder(exporter).build())
    .build()

val openTelemetry = OpenTelemetrySdk.builder()
    .setTracerProvider(tracerProvider)
    .build()

val tracer = openTelemetry.getTracer("dev.kourier.amqp")

3. Create and Wrap Your Kourier Connection

Create a standard Kourier connection using createAMQPConnection, then wrap it with tracing capabilities using the withTracing extension function. This returns an OpenTelemetryAMQPConnection that automatically traces all operations.

import dev.kourier.amqp.createAMQPConnection
import dev.kourier.amqp.amqpConfig
import dev.kourier.amqp.opentelemetry.withTracing

val config = amqpConfig {
    server {
        host = "127.0.0.1"
        port = 5672
        user = "guest"
        password = "guest"
    }
}

val connection = createAMQPConnection(this, config)
val tracedConnection = connection.withTracing(tracer)

Alternatively, wrap individual channels if you prefer granular control:

val channel = connection.openChannel()
val tracedChannel = channel.withTracing(tracer)

4. Customize Tracing Behavior

Supply a custom TracingConfig to control which operations generate spans and whether message payloads are captured. Use TracingConfig.debug() for maximum visibility during development, or construct a custom configuration for production.

import dev.kourier.amqp.opentelemetry.TracingConfig

// Debug configuration: traces everything including connection operations
val debugConfig = TracingConfig.debug()

// Production configuration: minimal overhead
val productionConfig = TracingConfig(
    traceConnectionOperations = false,
    traceChannelManagementOperations = false,
    captureMessageBody = false
)

// Custom span naming
val customConfig = TracingConfig(
    publishSpanNameFormatter = { exchange, routingKey ->
        "amqp.publish.$exchange.$routingKey"
    },
    consumeSpanNameFormatter = { queue ->
        "amqp.consume.$queue"
    }
)

val tracedConn = connection.withTracing(tracer, customConfig)

Complete Working Example

The following Kotlin coroutine example demonstrates a complete integration from SDK initialization to message consumption with full tracing enabled:

import dev.kourier.amqp.*
import dev.kourier.amqp.opentelemetry.*
import io.opentelemetry.sdk.OpenTelemetrySdk
import io.opentelemetry.sdk.trace.SdkTracerProvider
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    // 1️⃣ OpenTelemetry SDK setup
    val exporter = OtlpGrpcSpanExporter.builder()
        .setEndpoint("http://localhost:4317")
        .build()
    val tracerProvider = SdkTracerProvider.builder()
        .addSpanProcessor(BatchSpanProcessor.builder(exporter).build())
        .build()
    val otel = OpenTelemetrySdk.builder()
        .setTracerProvider(tracerProvider)
        .build()
    val tracer = otel.getTracer("dev.kourier.amqp")

    // 2️⃣ Kourier configuration
    val amqpCfg = amqpConfig {
        server {
            host = "127.0.0.1"
            port = 5672
            user = "guest"
            password = "guest"
        }
    }

    // 3️⃣ Create and wrap connection with tracing
    val conn = createAMQPConnection(this, amqpCfg)
    val tracedConn = conn.withTracing(tracer, TracingConfig.debug())

    // 4️⃣ Perform AMQP operations (automatically traced)
    val ch = tracedConn.openChannel()
    ch.exchangeDeclare("demo-ex", BuiltinExchangeType.DIRECT)
    ch.queueDeclare("demo-q", durable = true)
    ch.queueBind("demo-q", "demo-ex", "demo-key")
    ch.basicPublish("Hello Traced World!".toByteArray(), "demo-ex", "demo-key")

    val consumer = ch.basicConsume("demo-q")
    for (msg in consumer) {
        println("Received: ${msg.message.body.decodeToString()}")
        ch.basicAck(msg.message)
    }

    // 5️⃣ Cleanup
    ch.close()
    tracedConn.close()
}

Summary

• Add the module: Include amqp-client-opentelemetry alongside your OpenTelemetry SDK dependencies to enable instrumentation.
• Obtain a Tracer: Initialize your OpenTelemetry SDK (OTLP, Jaeger, etc.) and acquire a Tracer instance to pass to Kourier.
• Wrap connections or channels: Use withTracing(tracer) extension functions on AMQPConnection or AMQPChannel to activate tracing.
• Configure granularity: Use TracingConfig to toggle connection-level tracing, channel management tracing, and message body capture.
• Automatic propagation: The integration automatically injects W3C Trace Context headers into AMQP messages, enabling end-to-end distributed tracing across consuming services.

Frequently Asked Questions

How does Kourier propagate trace context across services?

Kourier's OpenTelemetryAMQPChannel automatically injects the W3C Trace Context into AMQP message headers before publishing. When a message is consumed, the channel extracts these headers and resumes the trace context, creating child spans that link the producer and consumer operations. This happens transparently in OpenTelemetryAMQPChannel.kt without requiring manual header manipulation.

Can I use Kourier OpenTelemetry integration with Jaeger?

Yes. The Kourier OpenTelemetry module is exporter-agnostic. Configure your OpenTelemetry SDK with the Jaeger exporter instead of OTLP, obtain the Tracer, and pass it to withTracing(). The instrumentation will generate spans compatible with any OpenTelemetry-compliant backend, including Jaeger, Zipkin, AWS X-Ray, and Grafana Tempo.

What is the performance overhead of enabling tracing?

The overhead depends on your TracingConfig. By default (TracingConfig.default()), only publish and consume operations are traced, minimizing overhead. Connection and channel management operations add minimal latency but more spans. Capturing message bodies (captureMessageBody = true) increases memory usage and payload size. For high-throughput scenarios, use TracingConfig.default() and ensure your span processor is configured with batching (e.g., BatchSpanProcessor).

How do I disable specific operations from being traced?

Use the boolean flags in TracingConfig to selectively disable instrumentation. Set traceConnectionOperations = false to skip spans for connection open/close/heartbeat. Set traceChannelManagementOperations = false to exclude queue and exchange declarations. To disable tracing entirely for a specific channel while keeping it for others, simply do not wrap that channel with withTracing() and use the raw AMQPChannel instead.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}