sampler and span processor prototype implementations for consistent sampling

consistent-sampling/build.gradle.kts

@@ -0,0 +1,13 @@
+plugins {
+  id("otel.java-conventions")
+  id("otel.publish-conventions")
+}
+
+description = "Sampler and exporter implementations for consistent sampling"
+
+dependencies {
+  api("io.opentelemetry:opentelemetry-sdk")
+  testImplementation("com.google.guava:guava:31.0.1-jre")
+  testImplementation("org.hipparchus:hipparchus-core:2.0")
+  testImplementation("org.hipparchus:hipparchus-stat:2.0")
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/export/ConsistentReservoirSamplingBatchSpanProcessorBuilder.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.export;
+
+import static io.opentelemetry.api.internal.Utils.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.contrib.util.RandomGenerator;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+/** Builder class for {@link ConsistentReservoirSamplingBatchSpanProcessorBuilder}. */
+public final class ConsistentReservoirSamplingBatchSpanProcessorBuilder {
+
+  // Visible for testing
+  static final long DEFAULT_SCHEDULE_DELAY_MILLIS = 5000;
+  // Visible for testing
+  static final int DEFAULT_RESERVOIR_SIZE = 2048;
+  // Visible for testing
+  static final int DEFAULT_EXPORT_TIMEOUT_MILLIS = 30_000;
+
+  private final SpanExporter spanExporter;
+  private long scheduleDelayNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_SCHEDULE_DELAY_MILLIS);
+  private int reservoirSize = DEFAULT_RESERVOIR_SIZE;
+  private long exporterTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_EXPORT_TIMEOUT_MILLIS);
+  private MeterProvider meterProvider = MeterProvider.noop();
+  private RandomGenerator threadSafeRandomGenerator = () -> ThreadLocalRandom.current().nextLong();
+  private boolean useAlternativeReservoirImplementation = false;
+
+  ConsistentReservoirSamplingBatchSpanProcessorBuilder(SpanExporter spanExporter) {
+    this.spanExporter = requireNonNull(spanExporter, "spanExporter");
+  }
+
+  // TODO: Consider to add support for constant Attributes and/or Resource.
+
+  /**
+   * Sets the delay interval between two consecutive exports. If unset, defaults to {@value
+   * DEFAULT_SCHEDULE_DELAY_MILLIS}ms.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setScheduleDelay(
+      long delay, TimeUnit unit) {
+    requireNonNull(unit, "unit");
+    checkArgument(delay >= 0, "delay must be non-negative");
+    scheduleDelayNanos = unit.toNanos(delay);
+    return this;
+  }
+
+  /**
+   * Sets the delay interval between two consecutive exports. If unset, defaults to {@value
+   * DEFAULT_SCHEDULE_DELAY_MILLIS}ms.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setScheduleDelay(Duration delay) {
+    requireNonNull(delay, "delay");
+    return setScheduleDelay(delay.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
+  // Visible for testing
+  long getScheduleDelayNanos() {
+    return scheduleDelayNanos;
+  }
+
+  /**
+   * Sets the maximum time an export will be allowed to run before being cancelled. If unset,
+   * defaults to {@value DEFAULT_EXPORT_TIMEOUT_MILLIS}ms.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setExporterTimeout(
+      long timeout, TimeUnit unit) {
+    requireNonNull(unit, "unit");
+    checkArgument(timeout >= 0, "timeout must be non-negative");
+    exporterTimeoutNanos = unit.toNanos(timeout);
+    return this;
+  }
+
+  /**
+   * Sets the maximum time an export will be allowed to run before being cancelled. If unset,
+   * defaults to {@value DEFAULT_EXPORT_TIMEOUT_MILLIS}ms.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setExporterTimeout(Duration timeout) {
+    requireNonNull(timeout, "timeout");
+    return setExporterTimeout(timeout.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
+  // Visible for testing
+  long getExporterTimeoutNanos() {
+    return exporterTimeoutNanos;
+  }
+
+  /**
+   * Sets the reservoir size, themaximum number of Spans that can be collected.
+   *
+   * <p>See the ConsistentReservoirSamplingBatchSpanProcessor class description for a high-level
+   * design description of this class.
+   *
+   * <p>Default value is {@code 2048}.
+   *
+   * @param reservoirSize the reservoir size, the maximum number of Spans that are kept
+   * @return this.
+   * @see ConsistentReservoirSamplingBatchSpanProcessorBuilder#DEFAULT_RESERVOIR_SIZE
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setReservoirSize(int reservoirSize) {
+    this.reservoirSize = reservoirSize;
+    return this;
+  }
+
+  // Visible for testing
+  int getReservoirSize() {
+    return reservoirSize;
+  }
+
+  /**
+   * Sets the {@link MeterProvider} to use to collect metrics related to batch export. If not set,
+   * metrics will not be collected.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessorBuilder setMeterProvider(
+      MeterProvider meterProvider) {
+    requireNonNull(meterProvider, "meterProvider");
+    this.meterProvider = meterProvider;
+    return this;
+  }
+
+  // Visible for testing
+  ConsistentReservoirSamplingBatchSpanProcessorBuilder setThreadSafeRandomGenerator(
+      RandomGenerator threadSafeRandomGenerator) {
+    this.threadSafeRandomGenerator = threadSafeRandomGenerator;
+    return this;
+  }
+
+  // Visible for testing
+  ConsistentReservoirSamplingBatchSpanProcessorBuilder useAlternativeReservoirImplementation(
+      boolean useAlternativeReservoirImplementation) {
+    this.useAlternativeReservoirImplementation = useAlternativeReservoirImplementation;
+    return this;
+  }
+
+  /**
+   * Returns a new {@link ConsistentReservoirSamplingBatchSpanProcessorBuilder} that batches, then
+   * converts spans to proto and forwards them to the given {@code spanExporter}.
+   *
+   * @return a new {@link ConsistentReservoirSamplingBatchSpanProcessorBuilder}.
+   * @throws NullPointerException if the {@code spanExporter} is {@code null}.
+   */
+  public ConsistentReservoirSamplingBatchSpanProcessor build() {
+    return new ConsistentReservoirSamplingBatchSpanProcessor(
+        spanExporter,
+        meterProvider,
+        scheduleDelayNanos,
+        reservoirSize,
+        exporterTimeoutNanos,
+        threadSafeRandomGenerator,
+        useAlternativeReservoirImplementation);
+  }
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/samplers/ConsistentAlwaysOffSampler.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.samplers;
+
+import io.opentelemetry.contrib.state.OtelTraceState;
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+public final class ConsistentAlwaysOffSampler extends ConsistentSampler {
+
+  @Override
+  protected int getP(int parentP, boolean isRoot) {
+    return OtelTraceState.getMaxP();
+  }
+
+  @Override
+  public String getDescription() {
+    return "ConsistentAlwaysOffSampler";
+  }
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/samplers/ConsistentAlwaysOnSampler.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.samplers;
+
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+public class ConsistentAlwaysOnSampler extends ConsistentSampler {
+
+  @Override
+  protected int getP(int parentP, boolean isRoot) {
+    return 0;
+  }
+
+  @Override
+  public String getDescription() {
+    return "ConsistentAlwaysOnSampler";
+  }
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/samplers/ConsistentComposedAndSampler.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.samplers;
+
+import static java.util.Objects.requireNonNull;
+
+import io.opentelemetry.contrib.state.OtelTraceState;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * A consistent sampler composes two consistent samplers.
+ *
+ * <p>This sampler samples if both samplers would sample.
+ */
+@Immutable
+public final class ConsistentComposedAndSampler extends ConsistentSampler {
+
+  private final ConsistentSampler sampler1;
+  private final ConsistentSampler sampler2;
+  private final String description;
+
+  public static ConsistentComposedAndSampler create(
+      ConsistentSampler sampler1, ConsistentSampler sampler2) {
+    return new ConsistentComposedAndSampler(sampler1, sampler2);
+  }
+
+  private ConsistentComposedAndSampler(ConsistentSampler sampler1, ConsistentSampler sampler2) {
+    this.sampler1 = requireNonNull(sampler1);
+    this.sampler2 = requireNonNull(sampler2);
+    this.description =
+        "ConsistentComposedAndSampler{"
+            + "sampler1="
+            + sampler1.getDescription()
+            + ",sampler2="
+            + sampler2.getDescription()
+            + '}';
+  }
+
+  @Override
+  protected int getP(int parentP, boolean isRoot) {
+    int p1 = sampler1.getP(parentP, isRoot);
+    int p2 = sampler2.getP(parentP, isRoot);
+    if (OtelTraceState.isValidP(p1) && OtelTraceState.isValidP(p2)) {
+      return Math.max(p1, p2);
+    } else {
+      return OtelTraceState.getInvalidP();
+    }
+  }
+
+  @Override
+  public String getDescription() {
+    return description;
+  }
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/samplers/ConsistentComposedOrSampler.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.samplers;
+
+import static java.util.Objects.requireNonNull;
+
+import io.opentelemetry.contrib.state.OtelTraceState;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * A consistent sampler composes two consistent samplers.
+ *
+ * <p>This sampler samples if any of the two samplers would sample.
+ */
+@Immutable
+public final class ConsistentComposedOrSampler extends ConsistentSampler {
+
+  private final ConsistentSampler sampler1;
+  private final ConsistentSampler sampler2;
+  private final String description;
+
+  public static ConsistentComposedOrSampler create(
+      ConsistentSampler sampler1, ConsistentSampler sampler2) {
+    return new ConsistentComposedOrSampler(sampler1, sampler2);
+  }
+
+  private ConsistentComposedOrSampler(ConsistentSampler sampler1, ConsistentSampler sampler2) {
+    this.sampler1 = requireNonNull(sampler1);
+    this.sampler2 = requireNonNull(sampler2);
+    this.description =
+        "ConsistentComposedOrSampler{"
+            + "sampler1="
+            + sampler1.getDescription()
+            + ",sampler2="
+            + sampler2.getDescription()
+            + '}';
+  }
+
+  @Override
+  protected int getP(int parentP, boolean isRoot) {
+    int p1 = sampler1.getP(parentP, isRoot);
+    int p2 = sampler2.getP(parentP, isRoot);
+    if (OtelTraceState.isValidP(p1)) {
+      if (OtelTraceState.isValidP(p2)) {
+        return Math.min(p1, p2);
+      } else {
+        return p1;
+      }
+    } else {
+      if (OtelTraceState.isValidP(p2)) {
+        return p2;
+      } else {
+        return OtelTraceState.getInvalidP();
+      }
+    }
+  }
+
+  @Override
+  public String getDescription() {
+    return description;
+  }
+}