Use a more efficient serializer and write bytes directly to disk

Instead of an intermediary byte array.

The OTel Java SDK has a hand-coded and optimized serializer implementation that is well tested and used extensively. This PR switches to using that serializer.

It is also more efficient to serialize directly to the file output stream instead of an intermediary byte array, so I used the same marshaling strategy used by the SDK and passed the marshaler down to the WritableFile.

Author: tylerbenson

disk-buffering/build.gradle.kts

@@ -18,6 +18,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")
   signature("com.toasttab.android:gummy-bears-api-21:0.12.0:coreLib@signature")
@@ -63,9 +64,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) {