Composite Samplers prototype (#1443)

Co-authored-by: Otmar Ertl <otmar.ertl@gmail.com>

consistent-sampling/src/main/java/io/opentelemetry/contrib/sampler/consistent56/ComposableSampler.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.sampler.consistent56;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import java.util.List;
+
+/** An interface for components to be used by composite consistent probability samplers. */
+public interface ComposableSampler {
+
+  /**
+   * Returns the SamplingIntent that is used for the sampling decision. The SamplingIntent includes
+   * the threshold value which will be used for the sampling decision.
+   *
+   * <p>NOTE: Keep in mind, that in any case the returned threshold value must not depend directly
+   * or indirectly on the random value. In particular this means that the parent sampled flag must
+   * not be used for the calculation of the threshold as the sampled flag depends itself on the
+   * random value.
+   */
+  SamplingIntent getSamplingIntent(
+      Context parentContext,
+      String name,
+      SpanKind spanKind,
+      Attributes attributes,
+      List<LinkData> parentLinks);
+
+  /** Return the string providing a description of the implementation. */
+  String getDescription();
+}

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

@@ -5,6 +5,13 @@
 
 package io.opentelemetry.contrib.sampler.consistent56;
 
+import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.getInvalidThreshold;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import java.util.List;
 import javax.annotation.concurrent.Immutable;
 
 @Immutable
@@ -19,8 +26,14 @@ static ConsistentAlwaysOffSampler getInstance() {
   }
 
   @Override
-  protected long getThreshold(long parentThreshold, boolean isRoot) {
-    return ConsistentSamplingUtil.getMaxThreshold();
+  public SamplingIntent getSamplingIntent(
+      Context parentContext,
+      String name,
+      SpanKind spanKind,
+      Attributes attributes,
+      List<LinkData> parentLinks) {
+
+    return () -> getInvalidThreshold();
   }
 
   @Override

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

@@ -7,6 +7,11 @@
 
 import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.getMinThreshold;
 
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import java.util.List;
 import javax.annotation.concurrent.Immutable;
 
 @Immutable
@@ -21,8 +26,14 @@ static ConsistentAlwaysOnSampler getInstance() {
   }
 
   @Override
-  protected long getThreshold(long parentThreshold, boolean isRoot) {
-    return getMinThreshold();
+  public SamplingIntent getSamplingIntent(
+      Context parentContext,
+      String name,
+      SpanKind spanKind,
+      Attributes attributes,
+      List<LinkData> parentLinks) {
+
+    return () -> getMinThreshold();
   }
 
   @Override

consistent-sampling/src/main/java/io/opentelemetry/contrib/sampler/consistent56/ConsistentAnyOf.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.sampler.consistent56;
+
+import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.getInvalidThreshold;
+import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.isValidThreshold;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * A consistent sampler that queries all its delegate samplers for their sampling threshold, and
+ * uses the minimum threshold value received.
+ */
+@Immutable
+final class ConsistentAnyOf extends ConsistentSampler {
+
+  private final ComposableSampler[] delegates;
+
+  private final String description;
+
+  /**
+   * Constructs a new consistent AnyOf sampler using the provided delegate samplers.
+   *
+   * @param delegates the delegate samplers
+   */
+  ConsistentAnyOf(@Nullable ComposableSampler... delegates) {
+    if (delegates == null || delegates.length == 0) {
+      throw new IllegalArgumentException(
+          "At least one delegate must be specified for ConsistentAnyOf");
+    }
+
+    this.delegates = delegates;
+
+    this.description =
+        Stream.of(delegates)
+            .map(Object::toString)
+            .collect(Collectors.joining(",", "ConsistentAnyOf{", "}"));
+  }
+
+  @Override
+  public SamplingIntent getSamplingIntent(
+      Context parentContext,
+      String name,
+      SpanKind spanKind,
+      Attributes attributes,
+      List<LinkData> parentLinks) {
+
+    SamplingIntent[] intents = new SamplingIntent[delegates.length];
+    int k = 0;
+    long minimumThreshold = getInvalidThreshold();
+    for (ComposableSampler delegate : delegates) {
+      SamplingIntent delegateIntent =
+          delegate.getSamplingIntent(parentContext, name, spanKind, attributes, parentLinks);
+      long delegateThreshold = delegateIntent.getThreshold();
+      if (isValidThreshold(delegateThreshold)) {
+        if (isValidThreshold(minimumThreshold)) {
+          minimumThreshold = Math.min(delegateThreshold, minimumThreshold);
+        } else {
+          minimumThreshold = delegateThreshold;
+        }
+      }
+      intents[k++] = delegateIntent;
+    }
+
+    long resultingThreshold = minimumThreshold;
+
+    return new SamplingIntent() {
+      @Override
+      public long getThreshold() {
+        return resultingThreshold;
+      }
+
+      @Override
+      public Attributes getAttributes() {
+        AttributesBuilder builder = Attributes.builder();
+        for (SamplingIntent intent : intents) {
+          builder = builder.putAll(intent.getAttributes());
+        }
+        return builder.build();
+      }
+
+      @Override
+      public TraceState updateTraceState(TraceState previousState) {
+        for (SamplingIntent intent : intents) {
+          previousState = intent.updateTraceState(previousState);
+        }
+        return previousState;
+      }
+    };
+  }
+
+  @Override
+  public String getDescription() {
+    return description;
+  }
+}

consistent-sampling/src/main/java/io/opentelemetry/contrib/sampler/consistent56/ConsistentFixedThresholdSampler.java

@@ -7,6 +7,14 @@
 
 import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.calculateSamplingProbability;
 import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.checkThreshold;
+import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.getInvalidThreshold;
+import static io.opentelemetry.contrib.sampler.consistent56.ConsistentSamplingUtil.getMaxThreshold;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import java.util.List;
 
 public class ConsistentFixedThresholdSampler extends ConsistentSampler {
 
@@ -18,7 +26,7 @@ protected ConsistentFixedThresholdSampler(long threshold) {
     this.threshold = threshold;
 
     String thresholdString;
-    if (threshold == ConsistentSamplingUtil.getMaxThreshold()) {
+    if (threshold == getMaxThreshold()) {
       thresholdString = "max";
     } else {
       thresholdString =
@@ -41,7 +49,18 @@ public String getDescription() {
   }
 
   @Override
-  protected long getThreshold(long parentThreshold, boolean isRoot) {
-    return threshold;
+  public SamplingIntent getSamplingIntent(
+      Context parentContext,
+      String name,
+      SpanKind spanKind,
+      Attributes attributes,
+      List<LinkData> parentLinks) {
+
+    return () -> {
+      if (threshold == getMaxThreshold()) {
+        return getInvalidThreshold();
+      }
+      return threshold;
+    };
   }
 }