Refactor Log SDK to implement OTEP-0150

docs/apidiffs/current_vs_latest/opentelemetry-exporter-logging-otlp.txt

@@ -1,2 +1,7 @@
 Comparing source compatibility of  against 
-No changes.
++++  NEW CLASS: PUBLIC(+) FINAL(+) io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingLogExporter  (not serializable)
+	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
+	+++  NEW SUPERCLASS: java.lang.Object
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.logs.export.LogExporter create()
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.common.CompletableResultCode export(java.util.Collection)
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.common.CompletableResultCode shutdown()

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LogProcessor.java

@@ -6,24 +6,68 @@
 package io.opentelemetry.sdk.logs;
 
 import io.opentelemetry.sdk.common.CompletableResultCode;
-import io.opentelemetry.sdk.logs.data.LogRecord;
-import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.logs.data.LogData;
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
 
-public interface LogProcessor {
+public interface LogProcessor extends Closeable {
 
-  void addLogRecord(LogRecord record);
+  /**
+   * Process a log.
+   *
+   * @param logData the log
+   */
+  void process(LogData logData);
 
   /**
-   * Called when {@link SdkTracerProvider#shutdown()} is called.
+   * Shutdown the log processor.
    *
    * @return result
    */
-  CompletableResultCode shutdown();
+  default CompletableResultCode shutdown() {
+    return forceFlush();
+  }
 
   /**
-   * Processes all span events that have not yet been processed.
+   * Process all logs that have not yet been processed.
    *
    * @return result
    */
-  CompletableResultCode forceFlush();
+  default CompletableResultCode forceFlush() {
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  default void close() {
+    shutdown().join(10, TimeUnit.SECONDS);
+  }
+
+  /**
+   * Returns a {@link LogProcessor} which simply delegates to all processing to the {@code
+   * processors} in order.
+   */
+  static LogProcessor composite(LogProcessor... processors) {
+    return composite(Arrays.asList(processors));
+  }
+
+  /**
+   * Returns a {@link LogProcessor} which simply delegates to all processing to the {@code
+   * processors} in order.
+   */
+  static LogProcessor composite(Iterable<LogProcessor> processors) {
+    List<LogProcessor> processorList = new ArrayList<>();
+    for (LogProcessor processor : processors) {
+      processorList.add(processor);
+    }
+    if (processorList.isEmpty()) {
+      return NoopLogProcessor.getInstance();
+    }
+    if (processorList.size() == 1) {
+      return processorList.get(0);
+    }
+    return MultiLogProcessor.create(processorList);
+  }
 }

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LogSink.java

@@ -5,14 +5,23 @@
 
 package io.opentelemetry.sdk.logs;
 
-import io.opentelemetry.sdk.logs.data.LogRecord;
+import io.opentelemetry.sdk.logs.data.LogBuilder;
+import io.opentelemetry.sdk.logs.data.LogData;
 
 /** A LogSink accepts logging records for transmission to an aggregator or log processing system. */
 public interface LogSink {
+
+  /**
+   * Create a log builder. {@link LogBuilder#build()} can be passed to {@link #offer(LogData)}.
+   *
+   * @return the builder
+   */
+  LogBuilder builder();
+
   /**
-   * Pass a record to the SDK for transmission to a logging exporter.
+   * Pass the {@link LogData} to the sink.
    *
-   * @param record record to transmit
+   * @param logData the log
    */
-  void offer(LogRecord record);
+  void offer(LogData logData);
 }

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LogSinkSdkProviderBuilder.java

@@ -5,10 +5,49 @@
 
 package io.opentelemetry.sdk.logs;
 
+import static java.util.Objects.requireNonNull;
+
+import io.opentelemetry.sdk.resources.Resource;
+import java.util.ArrayList;
+import java.util.List;
+
 public final class LogSinkSdkProviderBuilder {
+
+  private final List<LogProcessor> logProcessors = new ArrayList<>();
+  private Resource resource = Resource.getDefault();
+
   LogSinkSdkProviderBuilder() {}
 
-  public LogSinkSdkProvider build() {
-    return new LogSinkSdkProvider();
+  /**
+   * Set the resource.
+   *
+   * @param resource the resource
+   * @return this
+   */
+  public LogSinkSdkProviderBuilder setResource(Resource resource) {
+    requireNonNull(resource, "resource");
+    this.resource = resource;
+    return this;
+  }
+
+  /**
+   * Add a log processor.
+   *
+   * @param processor the log processor
+   * @return this
+   */
+  public LogSinkSdkProviderBuilder addLogProcessor(LogProcessor processor) {
+    requireNonNull(processor, "processor");
+    logProcessors.add(processor);
+    return this;
+  }
+
+  /**
+   * Create a {@link SdkLogSinkProvider} instance.
+   *
+   * @return an instance configured with the provided options
+   */
+  public SdkLogSinkProvider build() {
+    return new SdkLogSinkProvider(resource, logProcessors);
   }
 }

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/LogSinkSharedState.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.resources.Resource;
+import java.util.List;
+import javax.annotation.Nullable;
+
+final class LogSinkSharedState {
+  private final Object lock = new Object();
+  private final Resource resource;
+  private final LogProcessor activeLogProcessor;
+  @Nullable private volatile CompletableResultCode shutdownResult = null;
+
+  LogSinkSharedState(Resource resource, List<LogProcessor> logProcessors) {
+    this.resource = resource;
+    this.activeLogProcessor = LogProcessor.composite(logProcessors);
+  }
+
+  Resource getResource() {
+    return resource;
+  }
+
+  LogProcessor getActiveLogProcessor() {
+    return activeLogProcessor;
+  }
+
+  boolean hasBeenShutdown() {
+    return shutdownResult != null;
+  }
+
+  CompletableResultCode shutdown() {
+    synchronized (lock) {
+      if (shutdownResult != null) {
+        return shutdownResult;
+      }
+      shutdownResult = activeLogProcessor.shutdown();
+      return shutdownResult;
+    }
+  }
+}

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/MultiLogProcessor.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.data.LogData;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+final class MultiLogProcessor implements LogProcessor {
+
+  private final List<LogProcessor> logProcessors;
+  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
+
+  static LogProcessor create(List<LogProcessor> logProcessorsList) {
+    return new MultiLogProcessor(
+        new ArrayList<>(Objects.requireNonNull(logProcessorsList, "logProcessorsList")));
+  }
+
+  @Override
+  public void process(LogData logData) {
+    for (LogProcessor logProcessor : logProcessors) {
+      logProcessor.process(logData);
+    }
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    if (isShutdown.getAndSet(true)) {
+      return CompletableResultCode.ofSuccess();
+    }
+    List<CompletableResultCode> results = new ArrayList<>(logProcessors.size());
+    for (LogProcessor logProcessor : logProcessors) {
+      results.add(logProcessor.shutdown());
+    }
+    return CompletableResultCode.ofAll(results);
+  }
+
+  @Override
+  public CompletableResultCode forceFlush() {
+    List<CompletableResultCode> results = new ArrayList<>(logProcessors.size());
+    for (LogProcessor logProcessor : logProcessors) {
+      results.add(logProcessor.forceFlush());
+    }
+    return CompletableResultCode.ofAll(results);
+  }
+
+  private MultiLogProcessor(List<LogProcessor> logProcessorsList) {
+    this.logProcessors = logProcessorsList;
+  }
+}

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/NoopLogProcessor.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs;
+
+import io.opentelemetry.sdk.logs.data.LogData;
+
+final class NoopLogProcessor implements LogProcessor {
+  private static final NoopLogProcessor INSTANCE = new NoopLogProcessor();
+
+  static LogProcessor getInstance() {
+    return INSTANCE;
+  }
+
+  private NoopLogProcessor() {}
+
+  @Override
+  public void process(LogData logData) {}
+}