OTLP Zipkin Collector

README.md

@@ -18,6 +18,17 @@ persists them to a configured collector component.
 |------------------------------------|-----------------------------------------------------------------------------------------|
 | [collector-http](./collector-http) | Implements the [OTLP/HTTP protocol](https://opentelemetry.io/docs/specs/otlp/#otlphttp) |
 
+### Encoders
+
+Encoding is library-specific, as some libraries use `zipkin2.Span` and others
+`brave.handler.MutableSpan`. Both options are available to encode to the
+OTLP format.
+
+| Encoder                                   | Description                                    |
+|-------------------------------------------|------------------------------------------------|
+| [`OtelEncoder.V1`](./encoder-otel-zipkin) | zipkin-reporter `AsyncReporter<Span>`          |
+| [`OtelEncoder`](./encoder-otel-brave)     | zipkin-reporter-brave `AsyncZipkinSpanHandler` |
+
 ## Server integration
 
 If you cannot use our [Docker image](./docker/README.md), you can still integrate

collector-http/pom.xml

@@ -39,6 +39,11 @@
       <artifactId>zipkin-collector</artifactId>
       <version>${zipkin.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-encoder-otel</artifactId>
+      <version>${project.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>${armeria.groupId}</groupId>
@@ -52,11 +57,20 @@
       <version>${zipkin.version}</version>
       <scope>test</scope>
     </dependency>
-
     <dependency>
-      <groupId>${armeria.groupId}</groupId>
-      <artifactId>armeria-junit5</artifactId>
-      <version>${armeria.version}</version>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <version>${awaitility.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-exporter-otlp</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>

collector-http/src/main/java/zipkin2/collector/otel/http/OpenTelemetryHttpCollector.java

@@ -13,20 +13,37 @@
  */
 package zipkin2.collector.otel.http;
 
+import com.google.protobuf.CodedInputStream;
+import com.linecorp.armeria.common.AggregationOptions;
+import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpRequest;
 import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.ResponseHeaders;
+import com.linecorp.armeria.common.encoding.StreamDecoderFactory;
 import com.linecorp.armeria.server.AbstractHttpService;
 import com.linecorp.armeria.server.ServerBuilder;
 import com.linecorp.armeria.server.ServerConfigurator;
 import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.server.encoding.DecodingService;
+import io.netty.buffer.ByteBufAllocator;
+import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import zipkin2.Callback;
+import zipkin2.Span;
 import zipkin2.collector.Collector;
 import zipkin2.collector.CollectorComponent;
 import zipkin2.collector.CollectorMetrics;
 import zipkin2.collector.CollectorSampler;
 import zipkin2.storage.StorageComponent;
+import zipkin2.translation.zipkin.SpanTranslator;
 
 public final class OpenTelemetryHttpCollector extends CollectorComponent
     implements ServerConfigurator {
+
   public static Builder newBuilder() {
     return new Builder();
   }
@@ -36,23 +53,29 @@ public static final class Builder extends CollectorComponent.Builder {
     Collector.Builder delegate = Collector.newBuilder(OpenTelemetryHttpCollector.class);
     CollectorMetrics metrics = CollectorMetrics.NOOP_METRICS;
 
-    @Override public Builder storage(StorageComponent storageComponent) {
+    @Override
+    public Builder storage(StorageComponent storageComponent) {
       delegate.storage(storageComponent);
       return this;
     }
 
-    @Override public Builder metrics(CollectorMetrics metrics) {
-      if (metrics == null) throw new NullPointerException("metrics == null");
+    @Override
+    public Builder metrics(CollectorMetrics metrics) {
+      if (metrics == null) {
+        throw new NullPointerException("metrics == null");
+      }
       delegate.metrics(this.metrics = metrics.forTransport("otel/http"));
       return this;
     }
 
-    @Override public Builder sampler(CollectorSampler sampler) {
+    @Override
+    public Builder sampler(CollectorSampler sampler) {
       delegate.sampler(sampler);
       return this;
     }
 
-    @Override public OpenTelemetryHttpCollector build() {
+    @Override
+    public OpenTelemetryHttpCollector build() {
       return new OpenTelemetryHttpCollector(this);
     }
 
@@ -68,31 +91,77 @@ public static final class Builder extends CollectorComponent.Builder {
     metrics = builder.metrics;
   }
 
-  @Override public OpenTelemetryHttpCollector start() {
+  @Override
+  public OpenTelemetryHttpCollector start() {
     return this;
   }
 
-  @Override public String toString() {
+  @Override
+  public String toString() {
     return "OpenTelemetryHttpCollector{}";
   }
 
   /**
    * Reconfigures the service per https://opentelemetry.io/docs/specs/otlp/#otlphttp-request
    */
-  @Override public void reconfigure(ServerBuilder sb) {
+  @Override
+  public void reconfigure(ServerBuilder sb) {
+    sb.decorator(DecodingService.newDecorator(StreamDecoderFactory.gzip()));
     sb.service("/v1/traces", new HttpService(this));
   }
 
   static final class HttpService extends AbstractHttpService {
+
     final OpenTelemetryHttpCollector collector;
 
     HttpService(OpenTelemetryHttpCollector collector) {
       this.collector = collector;
     }
 
-    @Override protected HttpResponse doPost(ServiceRequestContext ctx, HttpRequest req)
+    @Override
+    protected HttpResponse doPost(ServiceRequestContext ctx, HttpRequest req)
         throws Exception {
-      throw new RuntimeException("Implement me!");
+      CompletableCallback result = new CompletableCallback();
+      req.aggregate(AggregationOptions.usePooledObjects(ByteBufAllocator.DEFAULT, ctx.eventLoop()
+      )).handle((msg, t) -> {
+        if (t != null) {
+          result.onError(t);
+          return null;
+        }
+        try (HttpData content = msg.content()) {
+          if (content.isEmpty()) {
+            result.onSuccess(null);
+            return null;
+          }
+
+          try {
+            ExportTraceServiceRequest request = ExportTraceServiceRequest.parseFrom(CodedInputStream.newInstance(content.byteBuf().nioBuffer()));
+            List<Span> spans = SpanTranslator.translate(request);
+            collector.collector.accept(spans, result);
+          } catch (IOException e) {
+            throw new UncheckedIOException(e);
+          }
+          return null;
+        }
+      });
+      return HttpResponse.of(result);
+    }
+  }
+
+  static final class CompletableCallback extends CompletableFuture<HttpResponse>
+      implements Callback<Void> {
+
+    static final ResponseHeaders ACCEPTED_RESPONSE = ResponseHeaders.of(HttpStatus.ACCEPTED);
+
+    @Override
+    public void onSuccess(Void value) {
+      complete(HttpResponse.of(ACCEPTED_RESPONSE));
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      completeExceptionally(t);
     }
   }
-}
+
+}

collector-http/src/test/java/zipkin2/collector/otel/http/ITOpenTelemetryHttpCollector.java

@@ -13,21 +13,61 @@
  */
 package zipkin2.collector.otel.http;
 
+import static io.opentelemetry.sdk.trace.samplers.Sampler.alwaysOn;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.linecorp.armeria.server.Server;
+import com.linecorp.armeria.server.ServerBuilder;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import zipkin2.collector.CollectorComponent;
 import zipkin2.collector.CollectorSampler;
 import zipkin2.collector.InMemoryCollectorMetrics;
 import zipkin2.storage.InMemoryStorage;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class ITOpenTelemetryHttpCollector {
+
+  private static final Logger log = LoggerFactory.getLogger(ITOpenTelemetryHttpCollector.class);
+
   InMemoryStorage store;
   InMemoryCollectorMetrics metrics;
   CollectorComponent collector;
 
+  SpanExporter spanExporter = OtlpHttpSpanExporter.builder()
+      .setCompression("gzip")
+      .build();
+
+  SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
+      .setSampler(alwaysOn())
+      .addSpanProcessor(BatchSpanProcessor.builder(spanExporter).build())
+      .build();
+
+  OpenTelemetrySdk openTelemetrySdk = OpenTelemetrySdk.builder()
+      .setTracerProvider(sdkTracerProvider)
+      .build();
+
+  Tracer tracer = openTelemetrySdk.getTracerProvider()
+      .get("zipkin2.collector.otel.http");
+
+  Server server;
+
   @BeforeEach public void setup() {
     store = InMemoryStorage.newBuilder().build();
     metrics = new InMemoryCollectorMetrics();
@@ -38,13 +78,42 @@ class ITOpenTelemetryHttpCollector {
         .storage(store)
         .build()
         .start();
+    ServerBuilder serverBuilder = Server.builder().http(4318);
+    ((OpenTelemetryHttpCollector) collector).reconfigure(serverBuilder);
     metrics = metrics.forTransport("otel/http");
+    server = serverBuilder.build();
+    server.start().join();
   }
 
-  @AfterEach void teardown() throws IOException {
+  @AfterEach
+  void teardown() throws IOException {
     store.close();
     collector.close();
+    server.stop().join();
   }
 
-  // TODO: integration test
+  @Test
+  void otelHttpExporterWorksWithZipkinOtelCollector() throws InterruptedException {
+    List<String> traceIds = new ArrayList<>();
+    final int size = 5;
+    for (int i = 0; i < size; i++) {
+      // Given
+      Span span = tracer.spanBuilder("foo " + i)
+          .setAttribute("foo tag", "foo value")
+          .setSpanKind(SpanKind.CONSUMER)
+          .startSpan();
+      String traceId = span.getSpanContext().getTraceId();
+      log.info("Trace Id <" + traceId + ">");
+      Thread.sleep(50);
+      span.addEvent("boom!");
+      Thread.sleep(50);
+
+      // When
+      span.end();
+      traceIds.add(traceId);
+    }
+
+    Awaitility.await().untilAsserted(() -> assertThat(store.acceptedSpanCount()).isEqualTo(5));
+
+  }
 }

collector-http/src/test/java/zipkin2/collector/otel/http/OpenTelemetryHttpCollectorTest.java

@@ -13,11 +13,11 @@
  */
 package zipkin2.collector.otel.http;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import org.junit.jupiter.api.Test;
 import zipkin2.storage.InMemoryStorage;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 class OpenTelemetryHttpCollectorTest {
   OpenTelemetryHttpCollector collector = OpenTelemetryHttpCollector.newBuilder()
       .storage(InMemoryStorage.newBuilder().build())

encoder-otel-zipkin/README.md

@@ -0,0 +1,8 @@
+# encoder-otel-zipkin
+
+This encodes zipkin spans into OTLP proto format.
+
+```java
+// connect the sender to the correct encoding
+reporter = AsyncReporter.newBuilder(sender).build(OtelEncoder.V1);
+```

encoder-otel-zipkin/pom.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright 2024 The OpenZipkin Authors
+
+    Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+    in compliance with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software distributed under the License
+    is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+    or implied. See the License for the specific language governing permissions and limitations under
+    the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <groupId>io.zipkin.contrib.otel</groupId>
+    <artifactId>zipkin-otel-parent</artifactId>
+    <version>0.1.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>zipkin-encoder-otel</artifactId>
+  <name>Zipkin Encoder: OpenTelemetry Trace</name>
+
+  <properties>
+    <main.basedir>${project.basedir}/..</main.basedir>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-translation-otel</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.zipkin.reporter2</groupId>
+      <artifactId>zipkin-reporter</artifactId>
+      <version>${zipkin-reporter.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>${zipkin.groupId}</groupId>
+      <artifactId>zipkin-tests</artifactId>
+      <version>${zipkin.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>