[disk-buffering] - Single responsibility for disk exporters

disk-buffering/CONTRIBUTING.md → disk-buffering/DESIGN.md

@@ -1,11 +1,18 @@
-# Contributor Guide
-
-Each one of the three exporters provided by this
-tool ([LogRecordDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/LogRecordDiskExporter.java), [MetricDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/MetricDiskExporter.java)
-and [SpanDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/SpanDiskExporter.java))
-is responsible of performing 2 actions, `write` and `read/delegate`, the `write` one happens
-automatically as a set of signals are provided from the processor, while the `read/delegate` one has
-to be triggered manually by the consumer of this library as explained in the [README](README.md).
+# Design Overview
+
+There are three main disk-writing exporters provided by this module:
+* [LogRecordToDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/LogRecordToDiskExporter.java)
+* [MetricToDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/MetricToDiskExporter.java)
+* [SpanToDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/SpanToDiskExporter.java))
+
+Each is responsible for writing a specific type of telemetry to disk storage for later 
+harvest/ingest.
+
+For later reading, there is:
+* [FromDiskExporter](src/main/java/io/opentelemetry/contrib/disk/buffering/FromDiskExporter.java)
+which can be created with `FromDiskExporter.builder()....build()`. 
+As explained in the [README](README.md), this has to be triggered manually by the consumer of 
+* this library and does not happen automaticall
 
 ## Writing overview
 

disk-buffering/README.md

@@ -1,18 +1,18 @@
 # Disk buffering
 
-This module provides signal exporter wrappers that intercept and store signals in files which can be
+This module provides exporters that store telemetry data in files which can be
 sent later on demand. A high level description of how it works is that there are two separate
 processes in place, one for writing data in disk, and one for reading/exporting the previously
 stored data.
 
 * Each exporter stores the received data automatically in disk right after it's received from its
   processor.
 * The reading of the data back from disk and exporting process has to be done manually. At
-  the moment there's no automatic mechanism to do so. There's more information on it can be
+  the moment there's no automatic mechanism to do so. There's more information on how it can be
   achieved, under [Reading data](#reading-data).
 
 > For a more detailed information on how the whole process works, take a look at
-> the [CONTRIBUTING](CONTRIBUTING.md) file.
+> the [DESIGN.md](DESIGN.md) file.
 
 ## Configuration
 

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering;
+
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.storage.responses.ReadableResult;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.logging.Level;
+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 FromDiskExporter<EXPORT_DATA> {
+  private final Storage storage;
+  private final SignalSerializer<EXPORT_DATA> deserializer;
+  private final Function<Collection<EXPORT_DATA>, CompletableResultCode> exportFunction;
+  private static final Logger logger = Logger.getLogger(FromDiskExporter.class.getName());
+
+  FromDiskExporter(
+      SignalSerializer<EXPORT_DATA> deserializer,
+      Function<Collection<EXPORT_DATA>, CompletableResultCode> exportFunction,
+      Storage storage) {
+    this.deserializer = deserializer;
+    this.exportFunction = exportFunction;
+    this.storage = storage;
+  }
+
+  public static <T> FromDiskExporterBuilder<T> builder() {
+    return new FromDiskExporterBuilder<>();
+  }
+
+  /**
+   * Reads data from the disk and attempts to export it.
+   *
+   * @param timeout The amount of time to wait for the wrapped exporter to finish.
+   * @param unit The unit of the time provided.
+   * @return true if there was data available and it was successfully exported within the timeout
+   *     provided. false otherwise.
+   * @throws IOException If an unexpected error happens.
+   */
+  public boolean exportStoredBatch(long timeout, TimeUnit unit) throws IOException {
+    logger.log(Level.INFO, "Attempting to export batch from disk.");
+    ReadableResult result =
+        storage.readAndProcess(
+            bytes -> {
+              logger.log(Level.INFO, "About to export stored batch.");
+              CompletableResultCode join =
+                  exportFunction.apply(deserializer.deserialize(bytes)).join(timeout, unit);
+              return join.isSuccess();
+            });
+    return result == ReadableResult.SUCCEEDED;
+  }
+
+  public void onShutDown() throws IOException {
+    storage.close();
+  }
+}

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering;
+
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.opentelemetry.contrib.disk.buffering.internal.StorageConfiguration;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
+import io.opentelemetry.contrib.disk.buffering.internal.storage.StorageBuilder;
+import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+import org.jetbrains.annotations.NotNull;
+
+public class FromDiskExporterBuilder<T> {
+
+  private SignalSerializer<T> serializer = noopSerializer();
+  private Function<Collection<T>, CompletableResultCode> exportFunction =
+      x -> CompletableResultCode.ofFailure();
+
+  @NotNull
+  private static <T> SignalSerializer<T> noopSerializer() {
+    return new SignalSerializer<T>() {
+
+      @Override
+      public byte[] serialize(Collection<T> ts) {
+        return new byte[0];
+      }
+
+      @Override
+      public List<T> deserialize(byte[] source) {
+        return Collections.emptyList();
+      }
+    };
+  }
+
+  private final StorageBuilder storageBuilder = Storage.builder();
+
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<T> setFolderName(String folderName) {
+    storageBuilder.setFolderName(folderName);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<T> setStorageConfiguration(StorageConfiguration configuration) {
+    storageBuilder.setStorageConfiguration(configuration);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<T> setStorageClock(Clock clock) {
+    storageBuilder.setStorageClock(clock);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<T> setDeserializer(SignalSerializer<T> serializer) {
+    this.serializer = serializer;
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public FromDiskExporterBuilder<T> setExportFunction(
+      Function<Collection<T>, CompletableResultCode> exportFunction) {
+    this.exportFunction = exportFunction;
+    return this;
+  }
+
+  public FromDiskExporter<T> build() throws IOException {
+    Storage storage = storageBuilder.build();
+    return new FromDiskExporter<>(serializer, exportFunction, storage);
+  }
+}

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering;
+
+import io.opentelemetry.contrib.disk.buffering.internal.StorageConfiguration;
+import io.opentelemetry.contrib.disk.buffering.internal.exporter.ToDiskExporter;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.LogRecordExporter;
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * This class implements a {@link LogRecordExporter} that delegates to an instance of {@code
+ * ToDiskExporter<LogRecordData>}.
+ */
+public class LogRecordToDiskExporter implements LogRecordExporter {
+  private final ToDiskExporter<LogRecordData> delegate;
+
+  /**
+   * Creates a new LogRecordToDiskExporter that will buffer LogRecordData telemetry on
+   * disk storage.
+   * @param delegate - The LogRecordExporter to delegate to if disk writing fails.
+   * @param config - The StorageConfiguration that specifies how storage is managed.
+   * @return A new LogRecordToDiskExporter instance.
+   * @throws IOException if the delegate ToDiskExporter could not be created.
+   */
+  public static LogRecordToDiskExporter create(LogRecordExporter delegate, StorageConfiguration config) throws IOException {
+    ToDiskExporter<LogRecordData> toDisk = ToDiskExporter.<LogRecordData>builder()
+        .setFolderName("logs")
+        .setStorageConfiguration(config)
+        .setSerializer(SignalSerializer.ofLogs())
+        .setExportFunction(delegate::export)
+        .build();
+    return new LogRecordToDiskExporter(toDisk);
+  }
+
+  // Visible for testing
+  LogRecordToDiskExporter(ToDiskExporter<LogRecordData> delegate) {
+    this.delegate = delegate;
+  }
+
+  @Override
+  public CompletableResultCode export(Collection<LogRecordData> logs) {
+    return delegate.export(logs);
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    try {
+      delegate.shutdown();
+      return CompletableResultCode.ofSuccess();
+    } catch (IOException e) {
+      return CompletableResultCode.ofFailure();
+    }
+  }
+}