Add logs sampler

Author: oliver-zhang

sdk-extensions/autoconfigure-spi/src/main/java/io/opentelemetry/sdk/autoconfigure/spi/logs/ConfigurableLogSamplerProvider.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.autoconfigure.spi.logs;
+
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.logs.samplers.LogSampler;
+
+public interface ConfigurableLogSamplerProvider {
+
+  /**
+   * Returns a {@link LogSampler} that can be registered to OpenTelemetry by providing the property
+   * value specified by {@link #getName()}.
+   */
+  LogSampler createSampler(ConfigProperties config);
+
+  /**
+   * Returns the name of this sampler, which can be specified with the {@code otel.logs.sampler}
+   * property to enable it. The name returned should NOT be the same as any other exporter name. If
+   * the name does conflict with another exporter name, the resulting behavior is undefined and it
+   * is explicitly unspecified which exporter will actually be used.
+   */
+  String getName();
+}

sdk-extensions/autoconfigure/src/main/java/io/opentelemetry/sdk/autoconfigure/LoggerProviderConfiguration.java

@@ -8,8 +8,11 @@
 import static io.opentelemetry.sdk.autoconfigure.LogRecordExporterConfiguration.configureLogRecordExporters;
 
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.sdk.autoconfigure.internal.NamedSpiManager;
 import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException;
+import io.opentelemetry.sdk.autoconfigure.spi.logs.ConfigurableLogSamplerProvider;
 import io.opentelemetry.sdk.logs.LogLimits;
 import io.opentelemetry.sdk.logs.LogLimitsBuilder;
 import io.opentelemetry.sdk.logs.LogRecordProcessor;
@@ -18,6 +21,7 @@
 import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessorBuilder;
 import io.opentelemetry.sdk.logs.export.LogRecordExporter;
 import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor;
+import io.opentelemetry.sdk.logs.samplers.LogSampler;
 import java.io.Closeable;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -31,6 +35,7 @@ final class LoggerProviderConfiguration {
 
   private static final List<String> simpleProcessorExporterNames =
       Arrays.asList("console", "logging");
+  public static final String ALWAYS_ON = "always_on";
 
   static void configureLoggerProvider(
       SdkLoggerProviderBuilder loggerProviderBuilder,
@@ -48,6 +53,9 @@ static void configureLoggerProvider(
     Map<String, LogRecordExporter> exportersByName =
         configureLogRecordExporters(config, spiHelper, logRecordExporterCustomizer, closeables);
 
+    String sampler = config.getString("otel.logs.sampler", ALWAYS_ON);
+    loggerProviderBuilder.setLogSampler(configSampler(sampler, config, spiHelper));
+
     List<LogRecordProcessor> processors =
         configureLogRecordProcessors(config, exportersByName, meterProvider, closeables);
     for (LogRecordProcessor processor : processors) {
@@ -59,6 +67,27 @@ static void configureLoggerProvider(
     }
   }
 
+  static LogSampler configSampler(String sampler, ConfigProperties config, SpiHelper spiHelper) {
+    NamedSpiManager<LogSampler> spiSamplersManager =
+        spiHelper.loadConfigurable(
+            ConfigurableLogSamplerProvider.class,
+            ConfigurableLogSamplerProvider::getName,
+            ConfigurableLogSamplerProvider::createSampler,
+            config);
+    switch (sampler) {
+      case "always_on":
+        return LogSampler.alwaysOnSampler();
+      case "parentbased":
+        return LogSampler.parentBasedSampler();
+      default:
+        LogSampler spiSampler = spiSamplersManager.getByName(sampler);
+        if (spiSampler == null) {
+          throw new ConfigurationException("Unrecognized value for otel.logs.sampler: " + sampler);
+        }
+        return spiSampler;
+    }
+  }
+
   // Visible for testing
   static List<LogRecordProcessor> configureLogRecordProcessors(
       ConfigProperties config,

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

@@ -7,6 +7,7 @@
 
 import io.opentelemetry.sdk.common.Clock;
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.samplers.LogSampler;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -21,16 +22,19 @@ final class LoggerSharedState {
   private final Supplier<LogLimits> logLimitsSupplier;
   private final LogRecordProcessor logRecordProcessor;
   private final Clock clock;
+  private final LogSampler sampler;
   @Nullable private volatile CompletableResultCode shutdownResult = null;
 
   LoggerSharedState(
       Resource resource,
       Supplier<LogLimits> logLimitsSupplier,
       LogRecordProcessor logRecordProcessor,
+      LogSampler sampler,
       Clock clock) {
     this.resource = resource;
     this.logLimitsSupplier = logLimitsSupplier;
     this.logRecordProcessor = logRecordProcessor;
+    this.sampler = sampler;
     this.clock = clock;
   }
 
@@ -46,6 +50,10 @@ LogRecordProcessor getLogRecordProcessor() {
     return logRecordProcessor;
   }
 
+  LogSampler getSampler() {
+    return this.sampler;
+  }
+
   Clock getClock() {
     return clock;
   }

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

@@ -125,6 +125,11 @@ public void emit() {
         this.observedTimestampEpochNanos == 0
             ? this.loggerSharedState.getClock().now()
             : this.observedTimestampEpochNanos;
+    boolean isShouldSample =
+        loggerSharedState.getSampler().shouldSample(context, this.severity, body, attributes);
+    if (!isShouldSample) {
+      return;
+    }
     loggerSharedState
         .getLogRecordProcessor()
         .onEmit(

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

@@ -15,6 +15,7 @@
 import io.opentelemetry.sdk.internal.ComponentRegistry;
 import io.opentelemetry.sdk.internal.ScopeConfigurator;
 import io.opentelemetry.sdk.logs.internal.LoggerConfig;
+import io.opentelemetry.sdk.logs.samplers.LogSampler;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.Closeable;
 import java.util.List;
@@ -52,11 +53,12 @@ public static SdkLoggerProviderBuilder builder() {
       Resource resource,
       Supplier<LogLimits> logLimitsSupplier,
       List<LogRecordProcessor> processors,
+      LogSampler sampler,
       Clock clock,
       ScopeConfigurator<LoggerConfig> loggerConfigurator) {
     LogRecordProcessor logRecordProcessor = LogRecordProcessor.composite(processors);
     this.sharedState =
-        new LoggerSharedState(resource, logLimitsSupplier, logRecordProcessor, clock);
+        new LoggerSharedState(resource, logLimitsSupplier, logRecordProcessor, sampler, clock);
     this.loggerComponentRegistry =
         new ComponentRegistry<>(
             instrumentationScopeInfo ->

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

@@ -17,6 +17,7 @@
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.logs.internal.LoggerConfig;
 import io.opentelemetry.sdk.logs.internal.SdkLoggerProviderUtil;
+import io.opentelemetry.sdk.logs.samplers.LogSampler;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.ArrayList;
 import java.util.List;
@@ -38,6 +39,8 @@ public final class SdkLoggerProviderBuilder {
   private ScopeConfiguratorBuilder<LoggerConfig> loggerConfiguratorBuilder =
       LoggerConfig.configuratorBuilder();
 
+  private LogSampler sampler = LogSampler.alwaysOnSampler();
+
   SdkLoggerProviderBuilder() {}
 
   /**
@@ -96,6 +99,18 @@ public SdkLoggerProviderBuilder addLogRecordProcessor(LogRecordProcessor process
     return this;
   }
 
+  /**
+   * Set log sampler
+   *
+   * @param sampler the log sampler
+   * @return this
+   */
+  public SdkLoggerProviderBuilder setLogSampler(LogSampler sampler) {
+    requireNonNull(sampler, "sampler");
+    this.sampler = sampler;
+    return this;
+  }
+
   /**
    * Assign a {@link Clock}.
    *
@@ -156,6 +171,11 @@ SdkLoggerProviderBuilder addLoggerConfiguratorCondition(
    */
   public SdkLoggerProvider build() {
     return new SdkLoggerProvider(
-        resource, logLimitsSupplier, logRecordProcessors, clock, loggerConfiguratorBuilder.build());
+        resource,
+        logLimitsSupplier,
+        logRecordProcessors,
+        sampler,
+        clock,
+        loggerConfiguratorBuilder.build());
   }
 }

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/samplers/LogSampler.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs.samplers;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.Value;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.context.Context;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+@ThreadSafe
+public interface LogSampler {
+
+  static LogSampler alwaysOnSampler() {
+    return (parentContext, severity, value, attributes) -> true;
+  }
+
+  static LogSampler parentBasedSampler() {
+    return (parentContext, severity, value, attributes) -> {
+      SpanContext spanContext = Span.fromContext(parentContext).getSpanContext();
+      return spanContext.isSampled();
+    };
+  }
+
+  boolean shouldSample(
+      Context parentContext,
+      Severity severity,
+      @Nullable Value<?> value,
+      @Nullable Attributes attributes);
+}