convert groovy smoke tests to java

Author: zeitlinger

smoke-tests/src/test/java/io/opentelemetry/smoketest/JavaTelemetryRetriever.java

@@ -10,13 +10,21 @@
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.testing.internal.armeria.client.WebClient;
-import java.nio.charset.StandardCharsets;
+import io.opentelemetry.testing.internal.jackson.core.JsonProcessingException;
+import io.opentelemetry.testing.internal.jackson.databind.ObjectMapper;
+import io.opentelemetry.testing.internal.proto.collector.trace.v1.ExportTraceServiceRequest;
+import io.opentelemetry.testing.internal.protobuf.GeneratedMessage;
+import io.opentelemetry.testing.internal.protobuf.InvalidProtocolBufferException;
+import io.opentelemetry.testing.internal.protobuf.util.JsonFormat;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 public class JavaTelemetryRetriever {
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
   private final WebClient client;
 
   public JavaTelemetryRetriever(int backendPort) {
@@ -28,49 +36,76 @@ public void clearTelemetry() {
   }
 
   public List<SpanData> waitForTraces() {
-    try {
-      return TelemetryConverter.getSpanData(
-          Collections.singletonList(waitForContent("get-traces")));
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
+    Collection<ExportTraceServiceRequest> requests =
+        waitForTelemetry("get-traces", () -> ExportTraceServiceRequest.newBuilder());
+    return TelemetryConverter.getSpanData(
+        requests.stream()
+            .flatMap(r -> r.getResourceSpansList().stream())
+            .collect(Collectors.toList()));
   }
 
   public Collection<MetricData> waitForMetrics() {
-    try {
-      return TelemetryConverter.getMetricsData(
-          Collections.singletonList(waitForContent("get-metrics")));
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
+    //    try {
+    //      return TelemetryConverter.getMetricsData(singletonList(waitForContent("get-metrics")));
+    //    } catch (InterruptedException e) {
+    //      throw new RuntimeException(e);
+    //    }
+    // todo
+    return Collections.emptyList();
   }
 
   public Collection<LogRecordData> waitForLogs() {
+    //    try {
+    //      return TelemetryConverter.getLogRecordData(singletonList(waitForContent("get-logs")));
+    //    } catch (InterruptedException e) {
+    //      throw new RuntimeException(e);
+    //    }
+    // todo
+    return Collections.emptyList();
+  }
+
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  private <T extends GeneratedMessage, B extends GeneratedMessage.Builder>
+      Collection<T> waitForTelemetry(String path, Supplier<B> builderConstructor) {
     try {
-      return TelemetryConverter.getLogRecordData(
-          Collections.singletonList(waitForContent("get-logs")));
-    } catch (InterruptedException e) {
+      return OBJECT_MAPPER
+          .readTree(waitForContent(path))
+          .valueStream()
+          .map(
+              it -> {
+                B builder = builderConstructor.get();
+                // TODO: Register parser into object mapper to avoid de -> re -> deserialize.
+                try {
+                  JsonFormat.parser().merge(OBJECT_MAPPER.writeValueAsString(it), builder);
+                  return (T) builder.build();
+                } catch (InvalidProtocolBufferException | JsonProcessingException e) {
+                  throw new RuntimeException(e);
+                }
+              })
+          .collect(Collectors.toList());
+    } catch (InterruptedException
+        | io.opentelemetry.testing.internal.jackson.core.JsonProcessingException e) {
       throw new RuntimeException(e);
     }
   }
 
-  private byte[] waitForContent(String path) throws InterruptedException {
+  private String waitForContent(String path) throws InterruptedException {
     long previousSize = 0;
     long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30);
-    byte[] content = "[]".getBytes(StandardCharsets.UTF_8);
+    String content = "[]";
     while (System.currentTimeMillis() < deadline) {
-      content = client.get(path).aggregate().join().content().array();
-      if (content.length > 2 && content.length == previousSize) {
+      content = client.get(path).aggregate().join().contentUtf8();
+      if (content.length() > 2 && content.length() == previousSize) {
         break;
       }
 
-      previousSize = content.length;
+      previousSize = content.length();
       System.out.println("Current content size " + previousSize);
       TimeUnit.MILLISECONDS.sleep(500);
     }
 
     // todo remove debug
-    System.out.println(new String(content, StandardCharsets.UTF_8));
+    //    System.out.println(new String(content, StandardCharsets.UTF_8));
 
     return content;
   }

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/internal/TelemetryConverter.java

@@ -46,7 +46,6 @@
 import io.opentelemetry.sdk.trace.data.StatusData;
 import io.opentelemetry.testing.internal.proto.collector.logs.v1.ExportLogsServiceRequest;
 import io.opentelemetry.testing.internal.proto.collector.metrics.v1.ExportMetricsServiceRequest;
-import io.opentelemetry.testing.internal.proto.collector.trace.v1.ExportTraceServiceRequest;
 import io.opentelemetry.testing.internal.proto.common.v1.AnyValue;
 import io.opentelemetry.testing.internal.proto.common.v1.ArrayValue;
 import io.opentelemetry.testing.internal.proto.common.v1.InstrumentationScope;
@@ -94,19 +93,7 @@ public class TelemetryConverter {
   private static final boolean hasExtendedAttributes =
       classAvailable("io.opentelemetry.api.incubator.common.ExtendedAttributes");
 
-  public static List<SpanData> getSpanData(List<byte[]> exportRequests) {
-    List<ResourceSpans> allResourceSpans =
-        exportRequests.stream()
-            .map(
-                serialized -> {
-                  try {
-                    return ExportTraceServiceRequest.parseFrom(serialized);
-                  } catch (InvalidProtocolBufferException e) {
-                    throw new AssertionError(e);
-                  }
-                })
-            .flatMap(request -> request.getResourceSpansList().stream())
-            .collect(toList());
+  public static List<SpanData> getSpanData(Collection<ResourceSpans> allResourceSpans) {
     List<SpanData> spans = new ArrayList<>();
     for (ResourceSpans resourceSpans : allResourceSpans) {
       Resource resource = resourceSpans.getResource();

testing-common/src/main/java/io/opentelemetry/javaagent/testing/common/AgentTestingExporterAccess.java

@@ -5,10 +5,15 @@
 
 package io.opentelemetry.javaagent.testing.common;
 
+import static java.util.stream.Collectors.toList;
+
 import io.opentelemetry.instrumentation.testing.internal.TelemetryConverter;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.testing.internal.proto.collector.trace.v1.ExportTraceServiceRequest;
+import io.opentelemetry.testing.internal.proto.trace.v1.ResourceSpans;
+import io.opentelemetry.testing.internal.protobuf.InvalidProtocolBufferException;
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
@@ -75,7 +80,21 @@ public static boolean forceFlushCalled() {
   @SuppressWarnings("unchecked")
   public static List<SpanData> getExportedSpans() {
     try {
-      return TelemetryConverter.getSpanData((List<byte[]>) getSpanExportRequests.invokeExact());
+      List<byte[]> bytes = (List<byte[]>) getSpanExportRequests.invokeExact();
+      List<ResourceSpans> allResourceSpans =
+          bytes.stream()
+              .map(
+                  serialized -> {
+                    try {
+                      return ExportTraceServiceRequest.parseFrom(serialized);
+                    } catch (InvalidProtocolBufferException e) {
+                      throw new AssertionError(e);
+                    }
+                  })
+              .flatMap(request -> request.getResourceSpansList().stream())
+              .collect(toList());
+
+      return TelemetryConverter.getSpanData(allResourceSpans);
     } catch (Throwable t) {
       throw new AssertionError("Could not invoke getSpanExportRequests", t);
     }

testing/proto-shaded-for-testing/build.gradle.kts

@@ -5,6 +5,7 @@ plugins {
 
 dependencies {
   implementation("io.opentelemetry.proto:opentelemetry-proto")
+  implementation("com.google.protobuf:protobuf-java-util:4.32.1")
 }
 
 tasks {