Add dd-openfeature SDK exposure and flag evaluation pipelines

dd-openfeature/build.gradle.kts

@@ -6,6 +6,7 @@ plugins {
   idea
   id("dd-trace-java.conventions.java")
   id("dd-trace-java.version-file")
+  id("dd-trace-java.jmh-conventions")
 }
 
 description = "Datadog OpenFeature provider SDK."
@@ -37,3 +38,7 @@ dependencies {
 tasks.withType<JavaCompile>().configureEach {
   configureCompiler(JavaVersion.VERSION_11)
 }
+
+jmh {
+  jmhVersion = libs.versions.jmh.get()
+}

dd-openfeature/src/jmh/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationEnqueueContentionBenchmark.java

@@ -0,0 +1,189 @@
+package com.datadog.openfeature.internal.flagevaluation;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import com.datadog.openfeature.internal.RuntimeServices;
+import com.datadog.openfeature.internal.connector.HealthMetrics;
+import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.util.HashMap;
+import java.util.Map;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.GroupThreads;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Producer-contention benchmark for {@link FlagEvaluationPipeline#enqueue}.
+ *
+ * <p>Every flag evaluation in every application thread calls {@code enqueue}, so producer-side
+ * scaling is the property that matters. {@link FlagEvaluationHotPathBenchmark#writerEnqueue}
+ * measures the single-threaded cost and cannot see contention; this benchmark measures how enqueue
+ * behaves as producer count grows.
+ *
+ * <p><strong>Why groups rather than {@code @Threads}:</strong> the hand-off queue ({@code
+ * EventQueue}) is multi-producer, <em>single</em>-consumer. Calling {@code poll()} from several
+ * threads breaks that contract, so producer threads cannot drain their own events. Each group
+ * therefore pairs N producers with exactly one consumer.
+ *
+ * <p><strong>Why the consumer drains in batches:</strong> with more producers than consumers the
+ * queue would otherwise saturate, after which {@code offer} fast-fails and the measurement turns
+ * into {@code AtomicLong} contention on the overflow counter instead of the enqueue path. The
+ * consumer polls up to {@link #DRAIN_BATCH} events per invocation to keep the queue off its
+ * capacity limit. Overflow is reported at the end of each iteration so a saturated - and therefore
+ * invalid - run is visible rather than silent.
+ *
+ * <p>Run: {@code ./gradlew :dd-openfeature:jmh
+ * -Pjmh.includes=FlagEvaluationEnqueueContentionBenchmark}.
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 3, time = 2, timeUnit = SECONDS)
+@Measurement(iterations = 5, time = 1, timeUnit = SECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(NANOSECONDS)
+@Fork(value = 1)
+public class FlagEvaluationEnqueueContentionBenchmark {
+
+  /** Events the single consumer drains per invocation, so it can keep up with N producers. */
+  static final int DRAIN_BATCH = 32;
+
+  private static final int NUM_FLAGS = 100;
+  private static final int NUM_USERS = 50;
+  private static final int NUM_FIELDS = 10;
+
+  private Map<String, Object> attrs;
+  private String[] flagKeys;
+  private String[] targetingKeys;
+  private FlagEvaluationPipeline writer;
+
+  /** Per-producer cursor: a shared counter would add its own cache-line contention. */
+  @State(Scope.Thread)
+  public static class ProducerCursor {
+    int cursor;
+  }
+
+  @Setup(Level.Iteration)
+  public void setUp() {
+    attrs = new HashMap<>();
+    for (int i = 0; i < NUM_FIELDS; i++) {
+      attrs.put("field" + i, "value");
+    }
+    flagKeys = keys("bench-flag-", NUM_FLAGS);
+    targetingKeys = keys("bench-user-", NUM_USERS);
+
+    // Capacity well above what the batch-draining consumer should ever let build up.
+    writer =
+        new FlagEvaluationPipeline(
+            1 << 20,
+            Long.MAX_VALUE,
+            NANOSECONDS,
+            (route, json) -> {},
+            new HashMap<>(),
+            new RuntimeServices(HealthMetrics.NOOP));
+  }
+
+  /**
+   * Reports queue overflow so a saturated run - where the consumer failed to keep up and the
+   * numbers no longer describe the enqueue path - is visible in the benchmark output.
+   */
+  @org.openjdk.jmh.annotations.TearDown(Level.Iteration)
+  @SuppressForbidden
+  public void reportOverflow() {
+    final long dropped = writer.droppedQueueOverflow();
+    if (dropped > 0) {
+      System.out.println(
+          "\nWARNING: queue overflowed "
+              + dropped
+              + " times - consumer could not keep up, enqueue timings for this iteration are"
+              + " measuring overflow accounting, not the enqueue path.");
+    }
+  }
+
+  // ---- 1 producer: uncontended baseline ----
+
+  @Benchmark
+  @Group("producers1")
+  @GroupThreads(1)
+  public void enqueue1(final ProducerCursor c) {
+    writer.enqueue(nextEvent(c));
+  }
+
+  @Benchmark
+  @Group("producers1")
+  @GroupThreads(1)
+  public void drain1(final Blackhole blackhole) {
+    drain(blackhole);
+  }
+
+  // ---- 4 producers ----
+
+  @Benchmark
+  @Group("producers4")
+  @GroupThreads(4)
+  public void enqueue4(final ProducerCursor c) {
+    writer.enqueue(nextEvent(c));
+  }
+
+  @Benchmark
+  @Group("producers4")
+  @GroupThreads(1)
+  public void drain4(final Blackhole blackhole) {
+    drain(blackhole);
+  }
+
+  // ---- 16 producers ----
+
+  @Benchmark
+  @Group("producers16")
+  @GroupThreads(16)
+  public void enqueue16(final ProducerCursor c) {
+    writer.enqueue(nextEvent(c));
+  }
+
+  @Benchmark
+  @Group("producers16")
+  @GroupThreads(1)
+  public void drain16(final Blackhole blackhole) {
+    drain(blackhole);
+  }
+
+  private void drain(final Blackhole blackhole) {
+    for (int i = 0; i < DRAIN_BATCH; i++) {
+      final FlagEvalEvent event = writer.pollQueuedEventForTest();
+      if (event == null) {
+        break;
+      }
+      blackhole.consume(event);
+    }
+  }
+
+  private FlagEvalEvent nextEvent(final ProducerCursor c) {
+    final int i = c.cursor++;
+    return new FlagEvalEvent(
+        flagKeys[Math.floorMod(i, flagKeys.length)],
+        "variant-" + Math.floorMod(i, 4),
+        "alloc-" + Math.floorMod(i, flagKeys.length),
+        targetingKeys[Math.floorMod(i, targetingKeys.length)],
+        null,
+        1_700_000_000_000L + i,
+        attrs);
+  }
+
+  private static String[] keys(final String prefix, final int count) {
+    final String[] out = new String[count];
+    for (int i = 0; i < count; i++) {
+      out[i] = prefix + i;
+    }
+    return out;
+  }
+}

dd-openfeature/src/jmh/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationHotPathBenchmark.java

@@ -0,0 +1,160 @@
+package com.datadog.openfeature.internal.flagevaluation;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import com.datadog.openfeature.internal.RuntimeServices;
+import com.datadog.openfeature.internal.connector.EventTransport;
+import com.datadog.openfeature.internal.connector.HealthMetrics;
+import java.util.HashMap;
+import java.util.Map;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Writer-side benchmark for EVP {@code flagevaluation} recording: the bounded-queue enqueue and the
+ * worker-thread aggregation of already-captured events.
+ *
+ * <p><strong>Scope limitation:</strong> this benchmark starts from a pre-built FlagEvalEvent
+ * holding a ready-made flat attribute map. It therefore does NOT cover the OpenFeature hook's own
+ * inline capture cost - in particular the bounded evaluation-context copy that
+ * FlagEvalLoggingHook.finallyAfter performs synchronously on the evaluation thread under
+ * consent-on. That cost is measured separately by FlagEvalHookHotPathBenchmark in the root package,
+ * where the hook and the OpenFeature context types live. Do not read writerEnqueue as the total
+ * per-evaluation cost paid by the caller.
+ *
+ * <p>Events are built with observeFullEvaluationData=true so the aggregator actually canonicalizes
+ * the context. Under the consent-off default it drops attrs and skips canonicalization, which makes
+ * every field-count profile measure the same scalar-only work.
+ *
+ * <p>Run: {@code ./gradlew :dd-openfeature:jmh -Pjmh.includes=FlagEvaluationHotPathBenchmark}.
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 3, time = 2, timeUnit = SECONDS)
+@Measurement(iterations = 5, time = 1, timeUnit = SECONDS)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(NANOSECONDS)
+@Fork(value = 1)
+public class FlagEvaluationHotPathBenchmark {
+
+  @Param({
+    "typical/100flags_50users_10fields",
+    "stress/10flags_1000users_250fields",
+    "scale/2500flags_500users_20fields"
+  })
+  public String profile;
+
+  private Map<String, Object> attrs;
+  private String[] flagKeys;
+  private String[] targetingKeys;
+  private int cursor;
+  private FlagEvaluationPipeline writer;
+  private FlagEvaluationPipeline.SerializingHandlerForTest handler;
+
+  @Setup(Level.Iteration)
+  public void setUp() {
+    final Profile p = Profile.fromName(profile);
+
+    attrs = new HashMap<>();
+    for (int i = 0; i < p.numFields; i++) {
+      attrs.put("field" + i, "value");
+    }
+    flagKeys = keys("bench-flag-", p.numFlags);
+    targetingKeys = keys("bench-user-", p.numUsers);
+    cursor = 0;
+
+    final EventTransport transport = (route, json) -> {};
+    final RuntimeServices services = new RuntimeServices(HealthMetrics.NOOP);
+    final Map<String, String> ddContext = new HashMap<>();
+    ddContext.put("service", "bench-service");
+    handler = new FlagEvaluationPipeline.SerializingHandlerForTest(transport, ddContext, services);
+
+    // Capacity large enough that the benchmark never overflows within a measurement window.
+    writer =
+        new FlagEvaluationPipeline(
+            1 << 20, Long.MAX_VALUE, NANOSECONDS, transport, ddContext, services);
+  }
+
+  /**
+   * Queue-mechanics cost only: event allocation plus the non-blocking bounded-queue offer. Excludes
+   * the hook's inline context snapshot - see the class javadoc.
+   */
+  @Benchmark
+  public void writerEnqueue(final Blackhole blackhole) {
+    final FlagEvalEvent event = nextEvent();
+    writer.enqueue(event);
+    blackhole.consume(writer.pollQueuedEventForTest());
+    blackhole.consume(event);
+  }
+
+  /** Worker-thread cost: materialize context, prune, canonicalize, and aggregate. */
+  @Benchmark
+  public void workerAggregate(final Blackhole blackhole) {
+    final FlagEvalEvent event = nextEvent();
+    handler.aggregateEvent(event);
+    if ((cursor % 10_000) == 0) {
+      handler.clearAggregationForTest();
+    }
+    blackhole.consume(handler.fullTierSizeForTest());
+  }
+
+  private FlagEvalEvent nextEvent() {
+    final int i = cursor++;
+    // observeFullEvaluationData=true is required for this benchmark to mean anything: under the
+    // consent-off default the aggregator drops attrs and skips canonicalContextKey entirely, so
+    // every field-count profile would collapse to the same scalar-only cost.
+    return new FlagEvalEvent(
+        flagKeys[Math.floorMod(i, flagKeys.length)],
+        "variant-" + Math.floorMod(i, 4),
+        "alloc-" + Math.floorMod(i, flagKeys.length),
+        targetingKeys[Math.floorMod(i, targetingKeys.length)],
+        null,
+        1_700_000_000_000L + i,
+        true,
+        attrs);
+  }
+
+  private static String[] keys(final String prefix, final int count) {
+    final String[] out = new String[count];
+    for (int i = 0; i < count; i++) {
+      out[i] = prefix + i;
+    }
+    return out;
+  }
+
+  private static final class Profile {
+    private final int numFlags;
+    private final int numUsers;
+    private final int numFields;
+
+    private Profile(final int numFlags, final int numUsers, final int numFields) {
+      this.numFlags = numFlags;
+      this.numUsers = numUsers;
+      this.numFields = numFields;
+    }
+
+    private static Profile fromName(final String name) {
+      if ("typical/100flags_50users_10fields".equals(name)) {
+        return new Profile(100, 50, 10);
+      }
+      if ("stress/10flags_1000users_250fields".equals(name)) {
+        return new Profile(10, 1_000, 250);
+      }
+      if ("scale/2500flags_500users_20fields".equals(name)) {
+        return new Profile(2_500, 500, 20);
+      }
+      throw new IllegalArgumentException("unknown benchmark profile: " + name);
+    }
+  }
+}

dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/Allocation.java

@@ -0,0 +1,9 @@
+package com.datadog.openfeature.internal.exposure;
+
+public class Allocation {
+  public final String key;
+
+  public Allocation(final String key) {
+    this.key = key;
+  }
+}