OpenTelemetry Setup

package io.kubemq.example.observability;

import io.kubemq.sdk.client.KubeMQClient;
import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;
import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.data.MetricData;
import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
import java.util.Collection;

/**
 * OpenTelemetry Setup Example
 *
 * Demonstrates how the KubeMQ Java SDK integrates with OpenTelemetry metrics.
 *
 * The SDK auto-detects the OTel API on the classpath and, when present,
 * records metrics under the instrumentation scope "io.kubemq.sdk" using the
 * OTel messaging semantic conventions (messaging.client.*), reading from
 * whatever MeterProvider is registered as GlobalOpenTelemetry. There is no
 * KubeMQClient builder method to inject a custom MeterProvider directly, so
 * this example registers one globally *before* creating the client — the
 * same effect the -javaagent: agent has in production, just done in code
 * with an in-memory reader so the example is self-contained.
 *
 * In production, skip the manual OpenTelemetrySdk.builder() step below and
 * instead attach the OpenTelemetry Java Agent on the JVM command line:
 *   java -javaagent:opentelemetry-javaagent.jar \
 *        -Dotel.service.name=my-kubemq-service \
 *        -Dotel.metrics.exporter=otlp \
 *        -jar myapp.jar
 */
public class OpenTelemetrySetupExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-observability-otel-setup-client";

    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== OpenTelemetry Setup Example ===\n");

        // Register an in-memory OTel meter provider as the global instance
        // BEFORE creating the KubeMQClient -- the SDK reads GlobalOpenTelemetry
        // at client-construction time.
        InMemoryMetricReader reader = InMemoryMetricReader.create();
        SdkMeterProvider meterProvider =
                SdkMeterProvider.builder().registerMetricReader(reader).build();
        OpenTelemetrySdk.builder().setMeterProvider(meterProvider).buildAndRegisterGlobal();

        System.out.println("Registered a global OTel MeterProvider with an in-memory reader.");
        System.out.println("The KubeMQ SDK will record metrics under scope \"io.kubemq.sdk\",");
        System.out.println("using messaging.client.* instrument names (OTel messaging semconv),");
        System.out.println("not a custom \"kubemq.*\" namespace.\n");

        // Create a client (OTel auto-detected because the API is on the classpath
        // and a global MeterProvider is now registered)
        try (PubSubClient client = PubSubClient.builder()
                .address(ADDRESS)
                .clientId(CLIENT_ID)
                .logLevel(KubeMQClient.Level.INFO)
                .build()) {

            ServerInfo info = client.ping();
            System.out.println("Connected to: " + info.getHost() + " v" + info.getVersion());

            // Create channel for send/receive (generates OTel spans + metrics)
            String channel = "java-observability.otel-setup";
            client.createEventsStoreChannel(channel);

            // Send messages (increments messaging.client.sent.messages)
            System.out.println("\nSending messages...");
            for (int i = 1; i <= 3; i++) {
                EventStoreMessage message = EventStoreMessage.builder()
                        .id("otel-" + i)
                        .channel(channel)
                        .body(("Traced message #" + i).getBytes())
                        .metadata("otel-example")
                        .build();

                EventSendResult result = client.publishEventStore(message);
                System.out.println("  Sent #" + i + " (sent=" + result.isSent() + ")");
            }

            // Subscribe and receive (increments messaging.client.consumed.messages)
            System.out.println("\nSubscribing...");
            EventsStoreSubscription sub = EventsStoreSubscription.builder()
                    .channel(channel)
                    .eventsStoreType(EventsStoreType.StartFromFirst)
                    .onReceiveEventCallback(event ->
                        System.out.println("  Received: " + new String(event.getBody())))
                    .onErrorCallback(err ->
                        System.err.println("  Error: " + err.getMessage()))
                    .build();

            client.subscribeToEventsStore(sub);
            Thread.sleep(2000);

            // Clean up resources
            sub.cancel();
            client.deleteEventsStoreChannel(channel);

        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }

        // Read the metrics back from the in-memory reader to confirm the real
        // instrument names -- this is the reproducible proof that the SDK
        // emits messaging.client.* (and NOT a kubemq.* meter namespace).
        System.out.println("\nRecorded metrics (instrumentation scope \"io.kubemq.sdk\"):");
        Collection<MetricData> metrics = reader.collectAllMetrics();
        for (MetricData metric : metrics) {
            System.out.println("  " + metric.getName() + " (" + metric.getType() + ")");
        }

        meterProvider.close();
        GlobalOpenTelemetry.resetForTest();
        System.out.println("\nOpenTelemetry setup example completed.");
    }
}