Save changes.

Author: kannanjgithub

examples/example-opentelemetry/build.gradle

@@ -33,6 +33,8 @@ dependencies {
     implementation "io.opentelemetry:opentelemetry-sdk:${openTelemetryVersion}"
     implementation "io.opentelemetry:opentelemetry-sdk-metrics:${openTelemetryVersion}"
     implementation "io.opentelemetry:opentelemetry-exporter-logging:${openTelemetryVersion}"
+    implementation "io.opentelemetry:opentelemetry-exporter-otlp:${openTelemetryVersion}"
+    implementation "io.opentelemetry.semconv:opentelemetry-semconv:1.29.0-alpha"
     implementation "io.opentelemetry:opentelemetry-exporter-prometheus:${openTelemetryPrometheusVersion}"
     runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}"
 }

examples/example-opentelemetry/src/main/java/io/grpc/example/opentelemetry/OpenTelemetryClient.java

@@ -26,13 +26,24 @@
 import io.grpc.examples.helloworld.HelloReply;
 import io.grpc.examples.helloworld.HelloRequest;
 import io.grpc.opentelemetry.GrpcOpenTelemetry;
+import io.grpc.opentelemetry.GrpcTraceBinContextPropagator;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import io.opentelemetry.semconv.ResourceAttributes;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
 
 /**
  * A simple gRPC client that requests a greeting from the {@link HelloWorldServer} and
@@ -93,6 +104,35 @@ public static void main(String[] args) throws Exception {
 
     Thread mainThread = Thread.currentThread();
 
+
+    // Adds a PrometheusHttpServer to convert OpenTelemetry metrics to Prometheus format and
+    // expose these via a HttpServer exporter to the SdkMeterProvider.
+    SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder()
+        .registerMetricReader(
+            PrometheusHttpServer.builder().setPort(prometheusPort).build())
+        .build();
+    Resource resource = Resource.create(
+        Attributes.of(ResourceAttributes.SERVICE_NAME, "client-service")
+    );
+
+    // Configure the OTLP Span Exporter with the custom Collector endpoint
+    OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
+        .setEndpoint("http://localhost:7080") // Default OTLP gRPC port
+        .build();
+    SpanProcessor spanProcessor = SimpleSpanProcessor.create(spanExporter);
+
+    // Create the SdkTracerProvider
+    SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
+        .addSpanProcessor(spanProcessor)
+        .setResource(resource)
+        .build();
+    // Initialize OpenTelemetry SDK with MeterProvider configured with Prometeheus.
+    OpenTelemetrySdk openTelemetrySdk =
+        OpenTelemetrySdk.builder().setTracerProvider(sdkTracerProvider).setMeterProvider(sdkMeterProvider)
+            .setPropagators(ContextPropagators.create(TextMapPropagator.composite(
+                GrpcTraceBinContextPropagator.defaultInstance()
+            )))
+            .build();
     Runtime.getRuntime().addShutdownHook(new Thread() {
       @Override
       public void run() {
@@ -105,21 +145,12 @@ public void run() {
         } catch (InterruptedException e) {
           e.printStackTrace(System.err);
         }
+        System.err.println("Closing the otel sdk");
+        openTelemetrySdk.close();
         System.err.println("*** client shut down");
       }
     });
 
-    // Adds a PrometheusHttpServer to convert OpenTelemetry metrics to Prometheus format and
-    // expose these via a HttpServer exporter to the SdkMeterProvider.
-    SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder()
-        .registerMetricReader(
-            PrometheusHttpServer.builder().setPort(prometheusPort).build())
-        .build();
-
-    // Initialize OpenTelemetry SDK with MeterProvider configured with Prometeheus.
-    OpenTelemetrySdk openTelemetrySdk =
-        OpenTelemetrySdk.builder().setMeterProvider(sdkMeterProvider).build();
-
     // Initialize gRPC OpenTelemetry.
     // Following client metrics are enabled by default :
     //     1. grpc.client.attempt.started
@@ -148,6 +179,7 @@ public void run() {
     } finally {
       channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
       // Shut down OpenTelemetry SDK.
+      logger.log(Level.INFO, "Closing the otel sdk");
       openTelemetrySdk.close();
     }
   }

examples/example-opentelemetry/src/main/java/io/grpc/example/opentelemetry/OpenTelemetryServer.java

@@ -23,10 +23,22 @@
 import io.grpc.examples.helloworld.HelloReply;
 import io.grpc.examples.helloworld.HelloRequest;
 import io.grpc.opentelemetry.GrpcOpenTelemetry;
+import io.grpc.opentelemetry.GrpcTraceBinContextPropagator;
 import io.grpc.stub.StreamObserver;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import io.opentelemetry.semconv.ResourceAttributes;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
+
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
@@ -94,8 +106,28 @@ public static void main(String[] args) {
           .build();
 
       // Initialize OpenTelemetry SDK with MeterProvider configured with Prometheus metrics exporter
+      // Define resource attributes (e.g., service name)
+      Resource resource = Resource.create(
+          Attributes.of(ResourceAttributes.SERVICE_NAME, "server-service")
+      );
+
+      // Configure the OTLP Span Exporter with the custom Collector endpoint
+      OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
+          .setEndpoint("http://localhost:7080") // Default OTLP gRPC port
+          .build();
+
+      SpanProcessor spanProcessor = SimpleSpanProcessor.create(spanExporter);
+
+      // Create the SdkTracerProvider
+      SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
+          .addSpanProcessor(spanProcessor)
+          .setResource(resource)
+          .build();
       OpenTelemetrySdk openTelemetrySdk =
-          OpenTelemetrySdk.builder().setMeterProvider(sdkMeterProvider).build();
+          OpenTelemetrySdk.builder().setTracerProvider(sdkTracerProvider)
+              .setPropagators(ContextPropagators.create(TextMapPropagator.composite(
+                  GrpcTraceBinContextPropagator.defaultInstance()
+              ))).setMeterProvider(sdkMeterProvider).build();
 
       // Initialize gRPC OpenTelemetry.
       // Following client metrics are enabled by default :

examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java

@@ -63,7 +63,7 @@ public void greet(String name) {
   public static void main(String[] args) throws Exception {
     String user = "world";
     // Access a service running on the local machine on port 50051
-    String target = "localhost:50051";
+    String target = "localhost:7080";
     // Allow passing in the user and target strings as command line arguments
     if (args.length > 0) {
       if ("--help".equals(args[0])) {

examples/src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java

@@ -35,7 +35,7 @@ public class HelloWorldServer {
   private Server server;
   private void start() throws IOException {
     /* The port on which the server should run */
-    int port = 50051;
+    int port = 7080;
     /*
      * By default gRPC uses a global, shared Executor.newCachedThreadPool() for gRPC callbacks into
      * your application. This is convenient, but can cause an excessive number of threads to be

interop-testing/src/main/java/io/opentelemetry/proto/collector/trace/v1/TraceServiceImpl.java

@@ -1,8 +1,12 @@
 package io.opentelemetry.proto.collector.trace.v1;
 
+import com.google.protobuf.ByteString;
 import io.grpc.ServerServiceDefinition;
 import io.opentelemetry.proto.collector.trace.v1.TraceServiceGrpc.AsyncService;
+import io.opentelemetry.proto.common.v1.KeyValue;
 import io.opentelemetry.proto.trace.v1.ResourceSpans;
+import io.opentelemetry.proto.trace.v1.ScopeSpans;
+import io.opentelemetry.proto.trace.v1.Span;
 
 import java.util.List;
 import java.util.logging.Logger;
@@ -14,19 +18,44 @@ public ServerServiceDefinition bindService() {
       return TraceServiceGrpc.bindService(this);
   }
 
+  private String bytesToHex(ByteString bytes) {
+    // OpenTelemetry trace IDs are 16 bytes (32 hex characters)
+    final byte[] byteArray = bytes.toByteArray();
+    final char[] hexChars = new char[byteArray.length * 2];
+    for (int j = 0; j < byteArray.length; j++) {
+      int v = byteArray[j] & 0xFF;
+      // Convert to lowercase hex as per W3C Trace Context spec
+      hexChars[j * 2] = Character.forDigit((v >>> 4) & 0x0F, 16);
+      hexChars[j * 2 + 1] = Character.forDigit(v & 0x0F, 16);
+    }
+    return new String(hexChars);
+  }
+
   @Override
   public void export(io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest request,
               io.grpc.stub.StreamObserver<io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceResponse> responseObserver) {
     List<ResourceSpans> resourceSpansList = request.getResourceSpansList();
-    logger.info("Received ExportTraceServiceRequest with " + resourceSpansList.size() + " ResourceSpans");
-
-    for (ResourceSpans resourceSpans : resourceSpansList) {
-      // Process the spans here (e.g., log them, store them, forward them)
-      int spanCount = resourceSpans.getScopeSpansList().stream()
-          .mapToInt(ss -> ss.getSpansList().size())
-          .sum();
-      logger.info(String.format("Resource has %d spans. Resource attributes: %s",
-          spanCount, resourceSpans.getResource().getAttributesList()));
+
+    // Iterate over all resource spans in the request
+    for (ResourceSpans resourceSpans : request.getResourceSpansList()) {
+      List<KeyValue> resourceAttrs = resourceSpans.getResource().getAttributesList();
+      for (KeyValue resourceAttr: resourceAttrs) {
+        logger.info("Resource attr key: " + resourceAttr.getKey() + ", value: " + resourceAttr.getValue());
+      }
+      // Iterate over all instrumentation scope spans
+      for (ScopeSpans scopeSpans : resourceSpans.getScopeSpansList()) {
+        // Iterate over individual spans
+        for (Span span : scopeSpans.getSpansList()) {
+          // Get the trace ID as a ByteString
+          ByteString traceIdBytes = span.getTraceId();
+
+          // Convert the byte array to a hex string
+          String traceIdHex = bytesToHex(traceIdBytes);
+
+          logger.info("Extracted Trace ID: " + traceIdHex);
+          logger.info("Span ID: " + bytesToHex(span.getSpanId()));
+        }
+      }
     }
 
     // Create and send the response

interop-testing/src/main/java/io/opentelemetry/proto/collector/trace/v1/TraceServiceServer.java

@@ -6,6 +6,7 @@
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 
+import java.io.IOException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -39,10 +40,10 @@ public void run() {
     server.blockUntilShutdown();
   }
 
-  private void start() {
+  private void start() throws IOException {
     executor = Executors.newSingleThreadScheduledExecutor();
     server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
-        .addService(new TraceServiceImpl()).build();
+        .addService(new TraceServiceImpl()).build().start();
   }
 
   void stop() throws Exception {