Make the disk buffered exporting more efficient

The previous implementation would load the protobuf encoded bytes from disk, deserialize it, then send it through the delegate exporter pipeline which would likely re-serialize it into the exact same format.

This PR reduces the CPU burden by providing an optional alternate path which bypasses the exporter pipeline.
Instead it takes the serialized bytes from disk and sends it straight to an HttpExporter or GrpcExporter, which only takes a `Marshaler` class allowing the serialized bytes to be passed straight through.

Since these exporters like to know the number of serialized elements that are being passed in, I added a function (with the help of Claude) that counts the serialized elements within the protobuf encoded byte array.

A future optimization would be to allow streaming the bytes directly from disk without reading them back into a byte array.

Author: tylerbenson

buildSrc/src/main/kotlin/otel.spotless-conventions.gradle.kts

@@ -5,6 +5,7 @@ plugins {
 spotless {
   java {
     googleJavaFormat()
+    toggleOffOn()
     licenseHeaderFile(rootProject.file("buildscripts/spotless.license.java"), "(package|import|public|// Includes work from:)")
     target("src/**/*.java")
   }

disk-buffering/build.gradle.kts

@@ -22,6 +22,8 @@ dependencies {
   annotationProcessor("com.google.auto.value:auto-value")
   testImplementation("org.mockito:mockito-inline")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+  testImplementation("io.opentelemetry:opentelemetry-exporter-otlp")
+  testImplementation("io.opentelemetry:opentelemetry-exporter-sender-okhttp")
 
   protos("io.opentelemetry.proto:opentelemetry-proto:1.7.0-alpha@jar")
 }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/LogRecordFromDiskExporter.java

@@ -9,26 +9,49 @@
 import io.opentelemetry.contrib.disk.buffering.internal.exporter.FromDiskExporterImpl;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
+import io.opentelemetry.exporter.internal.grpc.GrpcExporter;
+import io.opentelemetry.exporter.internal.http.HttpExporter;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.logs.export.LogRecordExporter;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
 public class LogRecordFromDiskExporter implements FromDiskExporter {
 
-  private final FromDiskExporterImpl<LogRecordData> delegate;
+  private final FromDiskExporterImpl delegate;
 
   public static LogRecordFromDiskExporter create(LogRecordExporter exporter, Storage storage)
       throws IOException {
-    FromDiskExporterImpl<LogRecordData> delegate =
-        FromDiskExporterImpl.<LogRecordData>builder(storage)
-            .setDeserializer(SignalDeserializer.ofLogs())
-            .setExportFunction(exporter::export)
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<LogRecordData>builder(storage, SignalTypes.logs)
+            .setExportFunction(exporter::export, SignalDeserializer.ofLogs())
             .build();
     return new LogRecordFromDiskExporter(delegate);
   }
 
-  private LogRecordFromDiskExporter(FromDiskExporterImpl<LogRecordData> delegate) {
+  public static LogRecordFromDiskExporter create(HttpExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<LogRecordData>builder(storage, SignalTypes.logs)
+            .setExporter(exporter)
+            .build();
+    return new LogRecordFromDiskExporter(delegate);
+  }
+
+  // Private because untested.
+  @SuppressWarnings("unused")
+  private static LogRecordFromDiskExporter create(GrpcExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<LogRecordData>builder(storage, SignalTypes.logs)
+            .setExporter(exporter)
+            .build();
+    return new LogRecordFromDiskExporter(delegate);
+  }
+
+  private LogRecordFromDiskExporter(FromDiskExporterImpl delegate) {
     this.delegate = delegate;
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/MetricFromDiskExporter.java

@@ -9,26 +9,49 @@
 import io.opentelemetry.contrib.disk.buffering.internal.exporter.FromDiskExporterImpl;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
+import io.opentelemetry.exporter.internal.grpc.GrpcExporter;
+import io.opentelemetry.exporter.internal.http.HttpExporter;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
 public class MetricFromDiskExporter implements FromDiskExporter {
 
-  private final FromDiskExporterImpl<MetricData> delegate;
+  private final FromDiskExporterImpl delegate;
 
   public static MetricFromDiskExporter create(MetricExporter exporter, Storage storage)
       throws IOException {
-    FromDiskExporterImpl<MetricData> delegate =
-        FromDiskExporterImpl.<MetricData>builder(storage)
-            .setDeserializer(SignalDeserializer.ofMetrics())
-            .setExportFunction(exporter::export)
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<MetricData>builder(storage, SignalTypes.metrics)
+            .setExportFunction(exporter::export, SignalDeserializer.ofMetrics())
             .build();
     return new MetricFromDiskExporter(delegate);
   }
 
-  private MetricFromDiskExporter(FromDiskExporterImpl<MetricData> delegate) {
+  public static MetricFromDiskExporter create(HttpExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<MetricData>builder(storage, SignalTypes.metrics)
+            .setExporter(exporter)
+            .build();
+    return new MetricFromDiskExporter(delegate);
+  }
+
+  // Private because untested.
+  @SuppressWarnings("unused")
+  private static MetricFromDiskExporter create(GrpcExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<MetricData>builder(storage, SignalTypes.metrics)
+            .setExporter(exporter)
+            .build();
+    return new MetricFromDiskExporter(delegate);
+  }
+
+  private MetricFromDiskExporter(FromDiskExporterImpl delegate) {
     this.delegate = delegate;
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/SpanFromDiskExporter.java

@@ -9,26 +9,49 @@
 import io.opentelemetry.contrib.disk.buffering.internal.exporter.FromDiskExporterImpl;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
+import io.opentelemetry.exporter.internal.grpc.GrpcExporter;
+import io.opentelemetry.exporter.internal.http.HttpExporter;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
 public class SpanFromDiskExporter implements FromDiskExporter {
 
-  private final FromDiskExporterImpl<SpanData> delegate;
+  private final FromDiskExporterImpl delegate;
 
   public static SpanFromDiskExporter create(SpanExporter exporter, Storage storage)
       throws IOException {
-    FromDiskExporterImpl<SpanData> delegate =
-        FromDiskExporterImpl.<SpanData>builder(storage)
-            .setDeserializer(SignalDeserializer.ofSpans())
-            .setExportFunction(exporter::export)
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<SpanData>builder(storage, SignalTypes.spans)
+            .setExportFunction(exporter::export, SignalDeserializer.ofSpans())
             .build();
     return new SpanFromDiskExporter(delegate);
   }
 
-  private SpanFromDiskExporter(FromDiskExporterImpl<SpanData> delegate) {
+  public static SpanFromDiskExporter create(HttpExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<SpanData>builder(storage, SignalTypes.spans)
+            .setExporter(exporter)
+            .build();
+    return new SpanFromDiskExporter(delegate);
+  }
+
+  // Private because untested.
+  @SuppressWarnings("unused")
+  private static SpanFromDiskExporter create(GrpcExporter<Marshaler> exporter, Storage storage)
+      throws IOException {
+    FromDiskExporterImpl delegate =
+        FromDiskExporterImpl.<SpanData>builder(storage, SignalTypes.spans)
+            .setExporter(exporter)
+            .build();
+    return new SpanFromDiskExporter(delegate);
+  }
+
+  private SpanFromDiskExporter(FromDiskExporterImpl delegate) {
     this.delegate = delegate;
   }
 

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

@@ -5,51 +5,80 @@
 
 package io.opentelemetry.contrib.disk.buffering.internal.exporter;
 
-import static java.util.Collections.emptyList;
-
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.DebugLogger;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
+import io.opentelemetry.exporter.internal.grpc.GrpcExporter;
+import io.opentelemetry.exporter.internal.http.HttpExporter;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import java.io.IOException;
 import java.util.Collection;
+import java.util.List;
+import java.util.function.BiFunction;
 import java.util.function.Function;
-import org.jetbrains.annotations.NotNull;
+import java.util.logging.Logger;
 
-public class FromDiskExporterBuilder<T> {
+public class FromDiskExporterBuilder<EXPORT_DATA> {
 
-  private SignalDeserializer<T> serializer = noopDeserializer();
+  private final DebugLogger logger;
   private final Storage storage;
+  private final SignalTypes signalType;
 
-  private Function<Collection<T>, CompletableResultCode> exportFunction =
-      x -> CompletableResultCode.ofFailure();
+  private BiFunction<ProtoByteArrayMarshaler, Integer, CompletableResultCode> exportFunction =
+      (x, i) -> CompletableResultCode.ofFailure();
 
-  public FromDiskExporterBuilder(Storage storage) {
+  public FromDiskExporterBuilder(Storage storage, SignalTypes signalType) {
     if (storage == null) {
       throw new NullPointerException("Storage cannot be null");
     }
     this.storage = storage;
+    this.signalType = signalType;
+    this.logger =
+        DebugLogger.wrap(
+            Logger.getLogger(FromDiskExporterImpl.class.getName()), storage.isDebugEnabled());
   }
 
-  @NotNull
-  private static <T> SignalDeserializer<T> noopDeserializer() {
-    return x -> emptyList();
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<EXPORT_DATA> setExportFunction(
+      Function<Collection<EXPORT_DATA>, CompletableResultCode> exportFunction,
+      SignalDeserializer<EXPORT_DATA> deserializer) {
+    if (!deserializer.signalType().equals(signalType.name())) {
+      throw new IllegalArgumentException(
+          deserializer.signalType() + " does not match " + signalType);
+    }
+    this.exportFunction =
+        (exportRequest, itemCount) -> {
+          try {
+            List<EXPORT_DATA> telemetry = deserializer.deserialize(exportRequest.getBytes());
+            return exportFunction.apply(telemetry);
+          } catch (IOException e) {
+            return CompletableResultCode.ofExceptionalFailure(e);
+          }
+        };
+    return this;
   }
 
+  /**
+   * The provided HttpExporter should _NOT_ be configured to send JSON. The data is serialized to
+   * disk in protobuf format and sent directly to the provided exporter as a serialized payload.
+   */
   @CanIgnoreReturnValue
-  public FromDiskExporterBuilder<T> setDeserializer(SignalDeserializer<T> serializer) {
-    this.serializer = serializer;
+  public FromDiskExporterBuilder<EXPORT_DATA> setExporter(HttpExporter<Marshaler> exporter) {
+    // Any way we can assert the exporter is not configured for JSON?
+    this.exportFunction = exporter::export;
     return this;
   }
 
   @CanIgnoreReturnValue
-  public FromDiskExporterBuilder<T> setExportFunction(
-      Function<Collection<T>, CompletableResultCode> exportFunction) {
-    this.exportFunction = exportFunction;
+  public FromDiskExporterBuilder<EXPORT_DATA> setExporter(GrpcExporter<Marshaler> exporter) {
+    this.exportFunction = exporter::export;
     return this;
   }
 
-  public FromDiskExporterImpl<T> build() throws IOException {
-    return new FromDiskExporterImpl<>(serializer, exportFunction, storage);
+  public FromDiskExporterImpl build() throws IOException {
+    return new FromDiskExporterImpl(exportFunction, storage, signalType);
   }
 }

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

@@ -6,43 +6,46 @@
 package io.opentelemetry.contrib.disk.buffering.internal.exporter;
 
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.DeserializationException;
-import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.files.reader.ProcessResult;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.responses.ReadableResult;
 import io.opentelemetry.contrib.disk.buffering.internal.utils.DebugLogger;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.ProtobufTools;
+import io.opentelemetry.contrib.disk.buffering.internal.utils.SignalTypes;
+import io.opentelemetry.exporter.internal.marshal.ProtoFieldInfo;
+import io.opentelemetry.proto.collector.logs.v1.internal.ExportLogsServiceRequest;
+import io.opentelemetry.proto.collector.metrics.v1.internal.ExportMetricsServiceRequest;
+import io.opentelemetry.proto.collector.trace.v1.internal.ExportTraceServiceRequest;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import java.io.IOException;
-import java.util.Collection;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 import java.util.logging.Logger;
 
 /**
  * Signal-type generic class that can read telemetry previously buffered on disk and send it to
  * another delegated exporter.
  */
-public final class FromDiskExporterImpl<EXPORT_DATA> implements FromDiskExporter {
+public final class FromDiskExporterImpl implements FromDiskExporter {
   private final DebugLogger logger;
   private final Storage storage;
-  private final SignalDeserializer<EXPORT_DATA> deserializer;
-  private final Function<Collection<EXPORT_DATA>, CompletableResultCode> exportFunction;
+  private final SignalTypes signalType;
+  private final BiFunction<ProtoByteArrayMarshaler, Integer, CompletableResultCode> exportFunction;
 
   FromDiskExporterImpl(
-      SignalDeserializer<EXPORT_DATA> deserializer,
-      Function<Collection<EXPORT_DATA>, CompletableResultCode> exportFunction,
-      Storage storage) {
-    this.deserializer = deserializer;
+      BiFunction<ProtoByteArrayMarshaler, Integer, CompletableResultCode> exportFunction,
+      Storage storage,
+      SignalTypes signalType) {
     this.exportFunction = exportFunction;
     this.storage = storage;
+    this.signalType = signalType;
     this.logger =
         DebugLogger.wrap(
             Logger.getLogger(FromDiskExporterImpl.class.getName()), storage.isDebugEnabled());
   }
 
-  public static <T> FromDiskExporterBuilder<T> builder(Storage storage) {
-    return new FromDiskExporterBuilder<>(storage);
+  public static <T> FromDiskExporterBuilder<T> builder(Storage storage, SignalTypes signalType) {
+    return new FromDiskExporterBuilder<>(storage, signalType);
   }
 
   /**
@@ -56,25 +59,37 @@ public static <T> FromDiskExporterBuilder<T> builder(Storage storage) {
    */
   @Override
   public boolean exportStoredBatch(long timeout, TimeUnit unit) throws IOException {
-    logger.log("Attempting to export " + deserializer.signalType() + " batch from disk.");
+    logger.log("Attempting to export " + signalType.name() + " batch from disk.");
     ReadableResult result =
         storage.readAndProcess(
             bytes -> {
-              logger.log(
-                  "Read "
-                      + bytes.length
-                      + " "
-                      + deserializer.signalType()
-                      + " bytes from storage.");
+              logger.log("Read " + bytes.length + " " + signalType.name() + " bytes from storage.");
+              ProtoFieldInfo field;
+              switch (signalType) {
+                case metrics:
+                  field = ExportMetricsServiceRequest.RESOURCE_METRICS;
+                  break;
+                case logs:
+                  field = ExportLogsServiceRequest.RESOURCE_LOGS;
+                  break;
+                case spans:
+                  field = ExportTraceServiceRequest.RESOURCE_SPANS;
+                  break;
+                default:
+                  throw new IllegalStateException("Unsupported signal type: " + signalType);
+              }
+              int itemCount = 0;
               try {
-                List<EXPORT_DATA> telemetry = deserializer.deserialize(bytes);
-                logger.log(
-                    "Now exporting batch of " + telemetry.size() + " " + deserializer.signalType());
-                CompletableResultCode join = exportFunction.apply(telemetry).join(timeout, unit);
-                return join.isSuccess() ? ProcessResult.SUCCEEDED : ProcessResult.TRY_LATER;
+                itemCount = ProtobufTools.countRepeatedField(bytes, field.getFieldNumber());
               } catch (DeserializationException e) {
                 return ProcessResult.CONTENT_INVALID;
               }
+              logger.log("Now exporting batch of " + itemCount + " " + signalType.name());
+              CompletableResultCode join =
+                  exportFunction
+                      .apply(new ProtoByteArrayMarshaler(bytes), itemCount)
+                      .join(timeout, unit);
+              return join.isSuccess() ? ProcessResult.SUCCEEDED : ProcessResult.TRY_LATER;
             });
     return result == ReadableResult.SUCCEEDED;
   }