OTLP Stdout exporter

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

@@ -1,2 +1,12 @@
 Comparing source compatibility of opentelemetry-exporter-logging-otlp-1.42.0-SNAPSHOT.jar against opentelemetry-exporter-logging-otlp-1.41.0.jar
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingLogRecordExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) java.lang.String toString()
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingMetricExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.metrics.Aggregation getDefaultAggregation(io.opentelemetry.sdk.metrics.InstrumentType)
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.common.export.MemoryMode getMemoryMode()
+	+++  NEW METHOD: PUBLIC(+) java.lang.String toString()
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingSpanExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) java.lang.String toString()

exporters/common/src/main/java/io/opentelemetry/exporter/internal/ExporterBuilderUtil.java

@@ -13,6 +13,8 @@
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.metrics.Aggregation;
 import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
 import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.metrics.internal.aggregator.AggregationUtil;
 import java.net.URI;
@@ -97,4 +99,94 @@
   }
 
   private ExporterBuilderUtil() {}
+
+  /**
+   * Invoke the {@code aggregationTemporalitySelectorConsumer} with the configured {@link
+   * AggregationTemporality}.
+   */
+  public static void configureOtlpAggregationTemporality(
+      ConfigProperties config,
+      Consumer<AggregationTemporalitySelector> aggregationTemporalitySelectorConsumer) {
+    String temporalityStr = config.getString("otel.exporter.otlp.metrics.temporality.preference");
+    if (temporalityStr == null) {
+      return;
+    }
+    AggregationTemporalitySelector temporalitySelector;
+    switch (temporalityStr.toLowerCase(Locale.ROOT)) {
+      case "cumulative":
+        temporalitySelector = AggregationTemporalitySelector.alwaysCumulative();
+        break;
+      case "delta":
+        temporalitySelector = AggregationTemporalitySelector.deltaPreferred();
+        break;
+      case "lowmemory":
+        temporalitySelector = AggregationTemporalitySelector.lowMemory();
+        break;
+      default:
+        throw new ConfigurationException("Unrecognized aggregation temporality: " + temporalityStr);
+    }
+    aggregationTemporalitySelectorConsumer.accept(temporalitySelector);
+  }
+
+  public static void configureOtlpAggregationTemporality(
+      StructuredConfigProperties config,
+      Consumer<AggregationTemporalitySelector> aggregationTemporalitySelectorConsumer) {
+    String temporalityStr = config.getString("temporality_preference");
+    if (temporalityStr == null) {
+      return;
+    }
+    AggregationTemporalitySelector temporalitySelector;
+    switch (temporalityStr.toLowerCase(Locale.ROOT)) {
+      case "cumulative":
+        temporalitySelector = AggregationTemporalitySelector.alwaysCumulative();
+        break;
+      case "delta":
+        temporalitySelector = AggregationTemporalitySelector.deltaPreferred();
+        break;
+      case "lowmemory":
+        temporalitySelector = AggregationTemporalitySelector.lowMemory();
+        break;
+      default:
+        throw new ConfigurationException("Unrecognized temporality_preference: " + temporalityStr);
+    }
+    aggregationTemporalitySelectorConsumer.accept(temporalitySelector);
+  }
+
+  /**
+   * Invoke the {@code defaultAggregationSelectorConsumer} with the configured {@link
+   * DefaultAggregationSelector}.
+   */
+  public static void configureOtlpHistogramDefaultAggregation(
+      ConfigProperties config,
+      Consumer<DefaultAggregationSelector> defaultAggregationSelectorConsumer) {
+    String defaultHistogramAggregation =
+        config.getString("otel.exporter.otlp.metrics.default.histogram.aggregation");
+    if (defaultHistogramAggregation != null) {
+      configureHistogramDefaultAggregation(
+          defaultHistogramAggregation, defaultAggregationSelectorConsumer);
+    }
+  }
+
+  /**
+   * Invoke the {@code defaultAggregationSelectorConsumer} with the configured {@link
+   * DefaultAggregationSelector}.
+   */
+  public static void configureOtlpHistogramDefaultAggregation(
+      StructuredConfigProperties config,
+      Consumer<DefaultAggregationSelector> defaultAggregationSelectorConsumer) {
+    String defaultHistogramAggregation = config.getString("default_histogram_aggregation");
+    if (defaultHistogramAggregation == null) {
+      return;
+    }
+    if (AggregationUtil.aggregationName(Aggregation.base2ExponentialBucketHistogram())
+        .equalsIgnoreCase(defaultHistogramAggregation)) {
+      defaultAggregationSelectorConsumer.accept(
+          DefaultAggregationSelector.getDefault()
+              .with(InstrumentType.HISTOGRAM, Aggregation.base2ExponentialBucketHistogram()));
+    } else if (!AggregationUtil.aggregationName(explicitBucketHistogram())
+        .equalsIgnoreCase(defaultHistogramAggregation)) {
+      throw new ConfigurationException(
+          "Unrecognized default_histogram_aggregation: " + defaultHistogramAggregation);
+    }
+  }
 }

exporters/logging-otlp/build.gradle.kts

@@ -20,5 +20,6 @@ dependencies {
 
   testImplementation(project(":sdk:testing"))
 
+  testImplementation("com.google.guava:guava")
   testImplementation("org.skyscreamer:jsonassert")
 }

exporters/logging-otlp/src/main/java/io/opentelemetry/exporter/logging/otlp/OtlpJsonLoggingLogRecordExporter.java

@@ -5,17 +5,21 @@
 
 package io.opentelemetry.exporter.logging.otlp;
 
-import static io.opentelemetry.exporter.logging.otlp.JsonUtil.JSON_FACTORY;
-
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.core.io.SegmentedStringWriter;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
+import io.opentelemetry.exporter.internal.otlp.logs.LogReusableDataMarshaler;
 import io.opentelemetry.exporter.internal.otlp.logs.ResourceLogsMarshaler;
+import io.opentelemetry.exporter.logging.otlp.internal.InternalBuilder;
+import io.opentelemetry.exporter.logging.otlp.internal.logs.LogRecordBuilderAccessUtil;
+import io.opentelemetry.exporter.logging.otlp.internal.logs.OtlpJsonLoggingLogRecordExporterBuilder;
+import io.opentelemetry.exporter.logging.otlp.internal.writer.JsonWriter;
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.logs.export.LogRecordExporter;
-import java.io.IOException;
 import java.util.Collection;
+import java.util.StringJoiner;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -32,40 +36,80 @@
 
   private final AtomicBoolean isShutdown = new AtomicBoolean();
 
+  private final JsonWriter jsonWriter;
+
+  private final Function<Collection<LogRecordData>, CompletableResultCode> marshaler;
+  private final MemoryMode memoryMode;
+  private final boolean wrapperJsonObject;
+
+  static {
+    LogRecordBuilderAccessUtil.setToExporter(
+        builder ->
+            new OtlpJsonLoggingLogRecordExporter(
+                builder.getJsonWriter(), builder.getMemoryMode(), builder.isWrapperJsonObject()));
+    LogRecordBuilderAccessUtil.setToBuilder(
+        exporter ->
+            InternalBuilder.forLogs()
+                .setJsonWriter(exporter.jsonWriter)
+                .setWrapperJsonObject(exporter.wrapperJsonObject)
+                .setMemoryMode(exporter.memoryMode));
+  }
+
   /** Returns a new {@link OtlpJsonLoggingLogRecordExporter}. */
   public static LogRecordExporter create() {
-    return new OtlpJsonLoggingLogRecordExporter();
+    return OtlpJsonLoggingLogRecordExporterBuilder.create().build();
   }
 
-  private OtlpJsonLoggingLogRecordExporter() {}
+  OtlpJsonLoggingLogRecordExporter(
+      JsonWriter jsonWriter, MemoryMode memoryMode, boolean wrapperJsonObject) {
+    this.memoryMode = memoryMode;
+    this.wrapperJsonObject = wrapperJsonObject;
+    this.jsonWriter = jsonWriter;
+
+    marshaler = createMarshaler(jsonWriter, memoryMode, wrapperJsonObject);
+  }
+
+  private static Function<Collection<LogRecordData>, CompletableResultCode> createMarshaler(
+      JsonWriter jsonWriter, MemoryMode memoryMode, boolean wrapperJsonObject) {
+
+    if (wrapperJsonObject) {
+      LogReusableDataMarshaler reusableDataMarshaler =
+          new LogReusableDataMarshaler(memoryMode) {
+            @Override
+            public CompletableResultCode doExport(Marshaler exportRequest, int numItems) {
+              return jsonWriter.write(exportRequest);
+            }
+          };
+
+      return reusableDataMarshaler::export;
+    } else {
+      return logs -> {
+        // not support for low allocation marshaler
+
+        for (ResourceLogsMarshaler resourceLogs : ResourceLogsMarshaler.create(logs)) {
+          CompletableResultCode resultCode = jsonWriter.write(resourceLogs);
+          if (!resultCode.isSuccess()) {
+            // already logged
+            return resultCode;
+          }
+        }
+        return CompletableResultCode.ofSuccess();
+      };
+    }
+  }
 
   @Override
   public CompletableResultCode export(Collection<LogRecordData> logs) {
     if (isShutdown.get()) {
       return CompletableResultCode.ofFailure();
     }
 
-    ResourceLogsMarshaler[] allResourceLogs = ResourceLogsMarshaler.create(logs);
-    for (ResourceLogsMarshaler resourceLogs : allResourceLogs) {
-      SegmentedStringWriter sw = new SegmentedStringWriter(JSON_FACTORY._getBufferRecycler());
-      try (JsonGenerator gen = JsonUtil.create(sw)) {
-        resourceLogs.writeJsonTo(gen);
-      } catch (IOException e) {
-        // Shouldn't happen in practice, just skip it.
-        continue;
-      }
-      try {
-        logger.log(Level.INFO, sw.getAndClear());
-      } catch (IOException e) {
-        logger.log(Level.WARNING, "Unable to read OTLP JSON log records", e);
-      }
-    }
-    return CompletableResultCode.ofSuccess();
+    return marshaler.apply(logs);
   }
 
   @Override
   public CompletableResultCode flush() {
-    return CompletableResultCode.ofSuccess();
+    return jsonWriter.flush();
   }
 
   @Override
@@ -75,4 +119,13 @@
     }
     return CompletableResultCode.ofSuccess();
   }
+
+  @Override
+  public String toString() {
+    StringJoiner joiner = new StringJoiner(", ", "OtlpJsonLoggingLogRecordExporter{", "}");
+    joiner.add("memoryMode=" + memoryMode);
+    joiner.add("wrapperJsonObject=" + wrapperJsonObject);
+    joiner.add("jsonWriter=" + jsonWriter);
+    return joiner.toString();
+  }
 }