POC for severity_based and trace_based log record processors

sdk-extensions/incubator/src/main/java/io/opentelemetry/sdk/extension/incubator/fileconfig/DeclarativeConfiguration.java

@@ -15,8 +15,10 @@
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper;
 import io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider;
+import io.opentelemetry.sdk.extension.incubator.fileconfig.internal.model.LogRecordProcessorModel;
 import io.opentelemetry.sdk.extension.incubator.fileconfig.internal.model.OpenTelemetryConfigurationModel;
 import io.opentelemetry.sdk.extension.incubator.fileconfig.internal.model.SamplerModel;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
 import io.opentelemetry.sdk.trace.samplers.Sampler;
 import java.io.Closeable;
 import java.io.IOException;
@@ -210,6 +212,29 @@ public static Sampler createSampler(DeclarativeConfigProperties genericSamplerMo
         samplerModel);
   }
 
+  /**
+   * Create a {@link LogRecordProcessor} from the {@code logRecordProcessorModel} representing the
+   * log record processor config.
+   *
+   * <p>This is used when log record processors are composed, with one processor accepting one or
+   * more additional processors as config properties. The {@link ComponentProvider} implementation
+   * can call this to configure a delegate {@link LogRecordProcessor} from the {@link
+   * DeclarativeConfigProperties} corresponding to a particular config property.
+   */
+  public static LogRecordProcessor createLogRecordProcessor(
+      DeclarativeConfigProperties genericLogRecordProcessorModel) {
+    YamlDeclarativeConfigProperties yamlDeclarativeConfigProperties =
+        requireYamlDeclarativeConfigProperties(genericLogRecordProcessorModel);
+    LogRecordProcessorModel logRecordProcessorModel =
+        MAPPER.convertValue(
+            DeclarativeConfigProperties.toMap(yamlDeclarativeConfigProperties),
+            LogRecordProcessorModel.class);
+    return createAndMaybeCleanup(
+        LogRecordProcessorFactory.getInstance(),
+        SpiHelper.create(yamlDeclarativeConfigProperties.getComponentLoader()),
+        logRecordProcessorModel);
+  }
+
   private static YamlDeclarativeConfigProperties requireYamlDeclarativeConfigProperties(
       DeclarativeConfigProperties declarativeConfigProperties) {
     if (!(declarativeConfigProperties instanceof YamlDeclarativeConfigProperties)) {

sdk-extensions/incubator/src/main/java/io/opentelemetry/sdk/extension/incubator/fileconfig/internal/SeverityBasedLogRecordProcessorComponentProvider.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.extension.incubator.fileconfig.internal;
+
+import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider;
+import io.opentelemetry.sdk.extension.incubator.fileconfig.DeclarativeConfiguration;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.SeverityBasedLogRecordProcessor;
+
+/**
+ * ComponentProvider for SeverityBasedLogRecordProcessor to support declarative configuration.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+public class SeverityBasedLogRecordProcessorComponentProvider
+    implements ComponentProvider<LogRecordProcessor> {
+
+  @Override
+  public Class<LogRecordProcessor> getType() {
+    return LogRecordProcessor.class;
+  }
+
+  @Override
+  public String getName() {
+    return "severity_based";
+  }
+
+  @Override
+  public LogRecordProcessor create(DeclarativeConfigProperties config) {
+    String minimumSeverityStr = config.getString("minimum_severity");
+    if (minimumSeverityStr == null) {
+      throw new IllegalArgumentException(
+          "minimum_severity is required for severity_based log processors");
+    }
+
+    Severity minimumSeverity;
+    try {
+      minimumSeverity = Severity.valueOf(minimumSeverityStr);
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException("Invalid severity value: " + minimumSeverityStr, e);
+    }
+
+    DeclarativeConfigProperties delegateConfig = config.getStructured("delegate");
+    if (delegateConfig == null) {
+      throw new IllegalArgumentException("delegate is required for severity_based log processors");
+    }
+
+    LogRecordProcessor delegate = DeclarativeConfiguration.createLogRecordProcessor(delegateConfig);
+
+    return SeverityBasedLogRecordProcessor.builder(minimumSeverity, delegate).build();
+  }
+}

sdk-extensions/incubator/src/main/java/io/opentelemetry/sdk/extension/incubator/fileconfig/internal/TraceBasedLogRecordProcessorComponentProvider.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.extension.incubator.fileconfig.internal;
+
+import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider;
+import io.opentelemetry.sdk.extension.incubator.fileconfig.DeclarativeConfiguration;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.TraceBasedLogRecordProcessor;
+
+/**
+ * ComponentProvider for TraceBasedLogRecordProcessor to support declarative configuration.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+public class TraceBasedLogRecordProcessorComponentProvider
+    implements ComponentProvider<LogRecordProcessor> {
+
+  @Override
+  public Class<LogRecordProcessor> getType() {
+    return LogRecordProcessor.class;
+  }
+
+  @Override
+  public String getName() {
+    return "trace_based";
+  }
+
+  @Override
+  public LogRecordProcessor create(DeclarativeConfigProperties config) {
+    DeclarativeConfigProperties delegateConfig = config.getStructured("delegate");
+    if (delegateConfig == null) {
+      throw new IllegalArgumentException("delegate is required for trace_based log processors");
+    }
+
+    LogRecordProcessor delegate = DeclarativeConfiguration.createLogRecordProcessor(delegateConfig);
+
+    return TraceBasedLogRecordProcessor.builder(delegate).build();
+  }
+}

sdk-extensions/incubator/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider

@@ -1 +1,3 @@
 io.opentelemetry.sdk.extension.incubator.fileconfig.ServiceResourceDetector
+io.opentelemetry.sdk.extension.incubator.fileconfig.internal.SeverityBasedLogRecordProcessorComponentProvider
+io.opentelemetry.sdk.extension.incubator.fileconfig.internal.TraceBasedLogRecordProcessorComponentProvider

sdk-extensions/incubator/src/test/java/io/opentelemetry/sdk/extension/incubator/fileconfig/SeverityBasedLogRecordProcessorComponentProviderTest.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.extension.incubator.fileconfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties;
+import io.opentelemetry.common.ComponentLoader;
+import io.opentelemetry.sdk.extension.incubator.fileconfig.internal.SeverityBasedLogRecordProcessorComponentProvider;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.SeverityBasedLogRecordProcessor;
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+
+class SeverityBasedLogRecordProcessorComponentProviderTest {
+
+  @Test
+  void createSeverityBasedProcessor_DirectComponentProvider() {
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    assertThat(provider.getType()).isEqualTo(LogRecordProcessor.class);
+    assertThat(provider.getName()).isEqualTo("severity_based");
+  }
+
+  @Test
+  void createSeverityBasedProcessor_ValidConfig() {
+    DeclarativeConfigProperties config =
+        getConfig(
+            "minimum_severity: \"WARN\"\n"
+                + "delegate:\n"
+                + "  simple:\n"
+                + "    exporter:\n"
+                + "      console: {}\n");
+
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    LogRecordProcessor processor = provider.create(config);
+
+    assertThat(processor).isInstanceOf(SeverityBasedLogRecordProcessor.class);
+
+    assertThat(processor.toString())
+        .contains("minimumSeverity=WARN")
+        .contains("delegate=SimpleLogRecordProcessor")
+        .contains("logRecordExporter=SystemOutLogRecordExporter");
+  }
+
+  @Test
+  void createSeverityBasedProcessor_MissingMinimumSeverity() {
+    DeclarativeConfigProperties config =
+        getConfig(
+            "delegate:\n" // this comment exists only to influence spotless formatting
+                + "  simple:\n"
+                + "    exporter:\n"
+                + "      console: {}\n");
+
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    assertThatThrownBy(() -> provider.create(config))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("minimum_severity is required for severity_based log processors");
+  }
+
+  @Test
+  void createSeverityBasedProcessor_InvalidSeverity() {
+
+    DeclarativeConfigProperties config =
+        getConfig(
+            "minimum_severity: \"INVALID\"\n"
+                + "delegate:\n"
+                + "  simple:\n"
+                + "    exporter:\n"
+                + "      console: {}\n");
+
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    assertThatThrownBy(() -> provider.create(config))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid severity value: INVALID");
+  }
+
+  @Test
+  void createSeverityBasedProcessor_MissingDelegate() {
+    DeclarativeConfigProperties config = getConfig("minimum_severity: \"WARN\"\n");
+
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    assertThatThrownBy(() -> provider.create(config))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("delegate is required for severity_based log processors");
+  }
+
+  @Test
+  void createSeverityBasedProcessor_SingleDelegate() {
+    DeclarativeConfigProperties config =
+        getConfig(
+            "minimum_severity: \"INFO\"\n"
+                + "delegate:\n"
+                + "  simple:\n"
+                + "    exporter:\n"
+                + "      console: {}\n");
+
+    SeverityBasedLogRecordProcessorComponentProvider provider =
+        new SeverityBasedLogRecordProcessorComponentProvider();
+
+    LogRecordProcessor processor = provider.create(config);
+
+    assertThat(processor).isInstanceOf(SeverityBasedLogRecordProcessor.class);
+    assertThat(processor.toString()).contains("SeverityBasedLogRecordProcessor");
+  }
+
+  private static DeclarativeConfigProperties getConfig(String yaml) {
+    Object yamlObj =
+        DeclarativeConfiguration.loadYaml(
+            new ByteArrayInputStream(yaml.getBytes(StandardCharsets.UTF_8)),
+            Collections.emptyMap());
+
+    return DeclarativeConfiguration.toConfigProperties(
+        yamlObj,
+        ComponentLoader.forClassLoader(
+            SeverityBasedLogRecordProcessorComponentProviderTest.class.getClassLoader()));
+  }
+}

sdk-extensions/incubator/src/test/java/io/opentelemetry/sdk/extension/incubator/fileconfig/TraceBasedLogRecordProcessorComponentProviderTest.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.extension.incubator.fileconfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.opentelemetry.api.incubator.config.DeclarativeConfigProperties;
+import io.opentelemetry.common.ComponentLoader;
+import io.opentelemetry.sdk.extension.incubator.fileconfig.internal.TraceBasedLogRecordProcessorComponentProvider;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.TraceBasedLogRecordProcessor;
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+
+class TraceBasedLogRecordProcessorComponentProviderTest {
+
+  @Test
+  void createTraceBasedProcessor_DirectComponentProvider() {
+    TraceBasedLogRecordProcessorComponentProvider provider =
+        new TraceBasedLogRecordProcessorComponentProvider();
+
+    assertThat(provider.getType()).isEqualTo(LogRecordProcessor.class);
+    assertThat(provider.getName()).isEqualTo("trace_based");
+  }
+
+  @Test
+  void createTraceBasedProcessor_ValidConfig() {
+    DeclarativeConfigProperties config =
+        getConfig(
+            "delegate:\n" // this comment exists only to influence spotless formatting
+                + "  simple:\n"
+                + "    exporter:\n"
+                + "      console: {}\n");
+
+    TraceBasedLogRecordProcessorComponentProvider provider =
+        new TraceBasedLogRecordProcessorComponentProvider();
+
+    LogRecordProcessor processor = provider.create(config);
+
+    assertThat(processor).isInstanceOf(TraceBasedLogRecordProcessor.class);
+
+    assertThat(processor.toString())
+        .contains("TraceBasedLogRecordProcessor")
+        .contains("delegate=SimpleLogRecordProcessor")
+        .contains("logRecordExporter=SystemOutLogRecordExporter");
+  }
+
+  @Test
+  void createTraceBasedProcessor_MissingDelegate() {
+    DeclarativeConfigProperties config = getConfig("");
+
+    TraceBasedLogRecordProcessorComponentProvider provider =
+        new TraceBasedLogRecordProcessorComponentProvider();
+
+    assertThatThrownBy(() -> provider.create(config))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("delegate is required for trace_based log processors");
+  }
+
+  @Test
+  void createTraceBasedProcessor_SingleDelegate() {
+    DeclarativeConfigProperties config =
+        getConfig("delegate:\n" + "  simple:\n" + "    exporter:\n" + "      console: {}\n");
+
+    TraceBasedLogRecordProcessorComponentProvider provider =
+        new TraceBasedLogRecordProcessorComponentProvider();
+
+    LogRecordProcessor processor = provider.create(config);
+
+    assertThat(processor).isInstanceOf(TraceBasedLogRecordProcessor.class);
+    assertThat(processor.toString()).contains("TraceBasedLogRecordProcessor");
+  }
+
+  private static DeclarativeConfigProperties getConfig(String yaml) {
+    Object yamlObj =
+        DeclarativeConfiguration.loadYaml(
+            new ByteArrayInputStream(yaml.getBytes(StandardCharsets.UTF_8)),
+            Collections.emptyMap());
+
+    return DeclarativeConfiguration.toConfigProperties(
+        yamlObj,
+        ComponentLoader.forClassLoader(
+            TraceBasedLogRecordProcessorComponentProviderTest.class.getClassLoader()));
+  }
+}