Use a more efficient serializer and write bytes directly to disk (#2138)

Author: tylerbenson

disk-buffering/build.gradle.kts

@@ -17,6 +17,7 @@ val protos by configurations.creating
 dependencies {
   api("io.opentelemetry:opentelemetry-sdk")
   implementation("io.opentelemetry:opentelemetry-api-incubator")
+  implementation("io.opentelemetry:opentelemetry-exporter-otlp-common")
   compileOnly("com.google.auto.value:auto-value-annotations")
   annotationProcessor("com.google.auto.value:auto-value")
   testImplementation("org.mockito:mockito-inline")
@@ -47,9 +48,10 @@ wire {
   }
 
   root(
-    "opentelemetry.proto.trace.v1.TracesData",
-    "opentelemetry.proto.metrics.v1.MetricsData",
-    "opentelemetry.proto.logs.v1.LogsData",
+    // These are the types used by the Java SDK's OTLP exporters.
+    "opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest",
+    "opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest",
+    "opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest",
   )
 }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/exporter/NoopSerializer.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering.internal.exporter;
+
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Collection;
+
+class NoopSerializer<T> implements SignalSerializer<T> {
+
+  @Override
+  public NoopSerializer<T> initialize(Collection<T> data) {
+    return this;
+  }
+
+  @Override
+  public void writeBinaryTo(OutputStream output) throws IOException {}
+
+  @Override
+  public int getBinarySerializedSize() {
+    return 0;
+  }
+
+  @Override
+  public void reset() {}
+}

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/exporter/ToDiskExporter.java

@@ -38,10 +38,11 @@ public static <T> ToDiskExporterBuilder<T> builder(Storage storage) {
     return new ToDiskExporterBuilder<>(storage);
   }
 
-  public CompletableResultCode export(Collection<EXPORT_DATA> data) {
+  public synchronized CompletableResultCode export(Collection<EXPORT_DATA> data) {
     logger.log("Intercepting exporter batch.", Level.FINER);
     try {
-      if (storage.write(serializer.serialize(data))) {
+      serializer.initialize(data);
+      if (storage.write(serializer)) {
         return CompletableResultCode.ofSuccess();
       }
       logger.log("Could not store batch in disk. Exporting it right away.");
@@ -52,6 +53,8 @@ public CompletableResultCode export(Collection<EXPORT_DATA> data) {
           Level.WARNING,
           e);
       return exportFunction.apply(data);
+    } finally {
+      serializer.reset();
     }
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/exporter/ToDiskExporterBuilder.java

@@ -14,7 +14,7 @@
 
 public final class ToDiskExporterBuilder<T> {
 
-  private SignalSerializer<T> serializer = ts -> new byte[0];
+  private SignalSerializer<T> serializer = new NoopSerializer<T>();
 
   private final Storage storage;
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/LogRecordDataDeserializer.java

@@ -7,7 +7,7 @@
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.logs.ProtoLogsDataMapper;
 import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
-import io.opentelemetry.proto.logs.v1.LogsData;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import java.io.IOException;
 import java.util.List;
@@ -24,7 +24,8 @@ static LogRecordDataDeserializer getInstance() {
   @Override
   public List<LogRecordData> deserialize(byte[] source) throws DeserializationException {
     try {
-      return ProtoLogsDataMapper.getInstance().fromProto(LogsData.ADAPTER.decode(source));
+      return ProtoLogsDataMapper.getInstance()
+          .fromProto(ExportLogsServiceRequest.ADAPTER.decode(source));
     } catch (IOException e) {
       throw new DeserializationException(e);
     }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/MetricDataDeserializer.java

@@ -7,7 +7,7 @@
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.metrics.ProtoMetricsDataMapper;
 import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
-import io.opentelemetry.proto.metrics.v1.MetricsData;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.io.IOException;
 import java.util.List;
@@ -24,7 +24,8 @@ static MetricDataDeserializer getInstance() {
   @Override
   public List<MetricData> deserialize(byte[] source) throws DeserializationException {
     try {
-      return ProtoMetricsDataMapper.getInstance().fromProto(MetricsData.ADAPTER.decode(source));
+      return ProtoMetricsDataMapper.getInstance()
+          .fromProto(ExportMetricsServiceRequest.ADAPTER.decode(source));
     } catch (IOException e) {
       throw new DeserializationException(e);
     }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/SpanDataDeserializer.java

@@ -7,7 +7,7 @@
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.spans.ProtoSpansDataMapper;
 import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
-import io.opentelemetry.proto.trace.v1.TracesData;
+import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.io.IOException;
 import java.util.List;
@@ -24,7 +24,8 @@ static SpanDataDeserializer getInstance() {
   @Override
   public List<SpanData> deserialize(byte[] source) throws DeserializationException {
     try {
-      return ProtoSpansDataMapper.getInstance().fromProto(TracesData.ADAPTER.decode(source));
+      return ProtoSpansDataMapper.getInstance()
+          .fromProto(ExportTraceServiceRequest.ADAPTER.decode(source));
     } catch (IOException e) {
       throw new DeserializationException(e);
     }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/mapping/common/ByteStringMapper.java

@@ -16,10 +16,10 @@ public static ByteStringMapper getInstance() {
   }
 
   public ByteString stringToProto(String source) {
-    return ByteString.encodeUtf8(source);
+    return ByteString.decodeHex(source);
   }
 
   public String protoToString(ByteString source) {
-    return source.utf8();
+    return source.hex();
   }
 }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/mapping/logs/ProtoLogsDataMapper.java

@@ -6,8 +6,8 @@
 package io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.logs;
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.common.BaseProtoSignalsDataMapper;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
 import io.opentelemetry.proto.logs.v1.LogRecord;
-import io.opentelemetry.proto.logs.v1.LogsData;
 import io.opentelemetry.proto.logs.v1.ResourceLogs;
 import io.opentelemetry.proto.logs.v1.ScopeLogs;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
@@ -19,7 +19,7 @@
 
 public final class ProtoLogsDataMapper
     extends BaseProtoSignalsDataMapper<
-        LogRecordData, LogRecord, LogsData, ResourceLogs, ScopeLogs> {
+        LogRecordData, LogRecord, ExportLogsServiceRequest, ResourceLogs, ScopeLogs> {
 
   private static final ProtoLogsDataMapper INSTANCE = new ProtoLogsDataMapper();
 
@@ -39,12 +39,12 @@ protected LogRecordData protoToSignalItem(
   }
 
   @Override
-  protected List<ResourceLogs> getProtoResources(LogsData logsData) {
+  protected List<ResourceLogs> getProtoResources(ExportLogsServiceRequest logsData) {
     return logsData.resource_logs;
   }
 
   @Override
-  protected LogsData createProtoData(
+  protected ExportLogsServiceRequest createProtoData(
       Map<Resource, Map<InstrumentationScopeInfo, List<LogRecord>>> itemsByResource) {
     List<ResourceLogs> items = new ArrayList<>();
     itemsByResource.forEach(
@@ -58,7 +58,7 @@ protected LogsData createProtoData(
           }
           items.add(resourceLogsBuilder.build());
         });
-    return new LogsData.Builder().resource_logs(items).build();
+    return new ExportLogsServiceRequest.Builder().resource_logs(items).build();
   }
 
   private ScopeLogs.Builder createProtoScopeBuilder(InstrumentationScopeInfo scopeInfo) {

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/mapping/metrics/ProtoMetricsDataMapper.java

@@ -6,8 +6,8 @@
 package io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.metrics;
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.common.BaseProtoSignalsDataMapper;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.metrics.v1.Metric;
-import io.opentelemetry.proto.metrics.v1.MetricsData;
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
 import io.opentelemetry.proto.metrics.v1.ScopeMetrics;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
@@ -19,7 +19,7 @@
 
 public final class ProtoMetricsDataMapper
     extends BaseProtoSignalsDataMapper<
-        MetricData, Metric, MetricsData, ResourceMetrics, ScopeMetrics> {
+        MetricData, Metric, ExportMetricsServiceRequest, ResourceMetrics, ScopeMetrics> {
 
   private static final ProtoMetricsDataMapper INSTANCE = new ProtoMetricsDataMapper();
 
@@ -39,12 +39,12 @@ protected MetricData protoToSignalItem(
   }
 
   @Override
-  protected List<ResourceMetrics> getProtoResources(MetricsData protoData) {
+  protected List<ResourceMetrics> getProtoResources(ExportMetricsServiceRequest protoData) {
     return protoData.resource_metrics;
   }
 
   @Override
-  protected MetricsData createProtoData(
+  protected ExportMetricsServiceRequest createProtoData(
       Map<Resource, Map<InstrumentationScopeInfo, List<Metric>>> itemsByResource) {
     List<ResourceMetrics> items = new ArrayList<>();
     itemsByResource.forEach(
@@ -58,7 +58,7 @@ protected MetricsData createProtoData(
           }
           items.add(resourceMetricsBuilder.build());
         });
-    return new MetricsData.Builder().resource_metrics(items).build();
+    return new ExportMetricsServiceRequest.Builder().resource_metrics(items).build();
   }
 
   private ScopeMetrics.Builder createProtoScopeBuilder(InstrumentationScopeInfo scopeInfo) {