Overview

OpenTelemetry integration wires the SDK's messaging operations into your tracing and metrics pipeline without hand-instrumenting every call site. In a distributed system where a message might be published by one service, queued, and consumed by three others, per-call logging tells you almost nothing — you need spans that correlate across process boundaries and latency/error metrics broken out by channel and operation. Instrumenting that by hand around every publishEventStore or subscribeToEventsStore call is tedious and easy to get inconsistent; letting the JVM agent do it guarantees uniform coverage with zero SDK code changes.

The SDK relies on the OpenTelemetry Java Agent's auto-instrumentation model: when the agent JAR is attached via -javaagent:, it intercepts the SDK's gRPC calls and creates spans for each send/receive operation automatically, tagged with channel name, message ID, and operation type, and propagates trace context across producer/consumer boundaries. Gotchas: without the agent attached at JVM startup, none of this instrumentation exists — there's no programmatic fallback in application code; the agent must be configured with -Dotel.traces.exporter=... or spans have nowhere to go; and agent-based instrumentation only sees gRPC-level boundaries, so custom business spans still need manual instrumentation in your own coroutine code.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
OpenTelemetry Java agent or SDK on the classpath (optional for the example to compile)

Code

OpenTelemetrySetupExample.kt

package io.kubemq.sdk.examples.observability

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.client.LogLevel
import io.kubemq.sdk.pubsub.StartPosition
import io.kubemq.sdk.pubsub.eventStoreMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-observability-otel-setup"
private const val CHANNEL = "kotlin-observability.otel-setup"

/**
 * OpenTelemetry Setup Example
 *
 * The KubeMQ Kotlin SDK automatically integrates with OpenTelemetry.
 * When the OTel Java agent or SDK is present on the classpath, the SDK will:
 *   1. Create spans for message send/receive operations
 *   2. Record metrics: message counts, latencies, errors
 *   3. Propagate trace context across messages
 *   4. Add structured log context
 *
 * To enable OTel with the Java agent:
 *   java -javaagent:opentelemetry-javaagent.jar \
 *        -Dotel.service.name=my-kubemq-service \
 *        -Dotel.traces.exporter=jaeger \
 *        -jar myapp.jar
 */
fun main() = runBlocking {
    println("=== OpenTelemetry Setup Example ===\n")

    println("The KubeMQ Kotlin SDK automatically integrates with OpenTelemetry.")
    println("When the OTel agent or SDK is present, the SDK will:")
    println("  1. Create spans for message send/receive operations")
    println("  2. Record metrics: message counts, latencies, errors")
    println("  3. Propagate trace context across messages")
    println("  4. Add structured log context\n")

    // Create a client (OTel auto-detected when agent/SDK present)
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
        logLevel = LogLevel.INFO
    }

    client.use {
        try {
            val info = client.ping()
            println("Connected to: ${info.host} v${info.version}")

            client.createEventsStoreChannel(CHANNEL)

            // Send messages (generates send spans and metrics)
            println("\nSending messages (generates OTel spans)...")
            repeat(3) { i ->
                val result = client.publishEventStore(eventStoreMessage {
                    channel = CHANNEL
                    body = "Traced message #${i + 1}".toByteArray()
                    metadata = "otel-example"
                })
                println("  Sent #${i + 1} (sent=${result.sent})")
            }

            // Subscribe and receive (generates receive spans)
            println("\nSubscribing (generates OTel receive spans)...")
            val subJob = launch {
                client.subscribeToEventsStore {
                    channel = CHANNEL
                    startPosition = StartPosition.StartFromFirst
                }.take(3).collect { msg ->
                    println("  Received: ${String(msg.body)}")
                }
            }

            subJob.join()
        } finally {
            try { client.deleteEventsStoreChannel(CHANNEL) } catch (_: Exception) {}
        }
    }

    println("\nOTel integration notes:")
    println("  - Spans appear in your trace backend (Jaeger, Zipkin, etc.)")
    println("  - Metrics available via OTel metrics exporter")
    println("  - Log correlation via trace_id and span_id")
    println("  - No code changes needed -- auto-detection works")
    println("\nOpenTelemetry setup example completed.")
}

How It Works

The SDK automatically integrates with OpenTelemetry when the Java agent or SDK is on the classpath.
No code changes are needed -- OTel is auto-detected.
Send operations create producer spans; receive operations create consumer spans.
Trace context is propagated through message headers for distributed tracing.
To enable, run with the OTel Java agent: java -javaagent:opentelemetry-javaagent.jar -jar myapp.jar

OpenTelemetry Setup

Overview

Prerequisites

Code

How It Works

Related

On this page