diff --git a/products/feature-flagging/feature-flagging-api/src/jmh/java/datadog/trace/api/openfeature/ExposureDispatchHotPathBenchmark.java b/products/feature-flagging/feature-flagging-api/src/jmh/java/datadog/trace/api/openfeature/ExposureDispatchHotPathBenchmark.java new file mode 100644 index 00000000000..be7a6dc13f2 --- /dev/null +++ b/products/feature-flagging/feature-flagging-api/src/jmh/java/datadog/trace/api/openfeature/ExposureDispatchHotPathBenchmark.java @@ -0,0 +1,188 @@ +package datadog.trace.api.openfeature; + +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; + +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import datadog.trace.api.featureflag.exposure.ExposureEvent; +import dev.openfeature.sdk.ImmutableMetadata; +import dev.openfeature.sdk.ImmutableStructure; +import dev.openfeature.sdk.MutableContext; +import dev.openfeature.sdk.ProviderEvaluation; +import dev.openfeature.sdk.Reason; +import dev.openfeature.sdk.Value; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +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.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Measures the evaluation-thread cost of exposure admission and context capture. + * + *

The first path always captures an event. The duplicate path uses the same scalar identity for + * every invocation. This separates the required first-event copy from avoidable duplicate copies. + * + *

Run: {@code ./gradlew :products:feature-flagging:feature-flagging-api:jmh + * -PjmhIncludes=ExposureDispatchHotPathBenchmark -PjmhProf=gc}. + */ +@State(Scope.Benchmark) +@Warmup(iterations = 3, time = 2, timeUnit = SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = SECONDS) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(NANOSECONDS) +@Fork(1) +public class ExposureDispatchHotPathBenchmark { + + @Param({"empty", "flat/100attrs", "nested/10structs_10fields", "wide/1000attrs"}) + public String shape; + + @Param({"first", "duplicate"}) + public String path; + + private MutableContext context; + private ProviderEvaluation evaluation; + private BenchmarkExposureListener listener; + + @Setup(Level.Trial) + public void setUp() { + context = buildContext(shape); + evaluation = + ProviderEvaluation.builder() + .value("on-value") + .variant("on") + .reason(Reason.TARGETING_MATCH.name()) + .flagMetadata(ImmutableMetadata.builder().addString("allocationKey", "alloc-1").build()) + .build(); + listener = new BenchmarkExposureListener("first".equals(path)); + if ("duplicate".equals(path)) { + listener.record("bench-flag", "bench-user", "on", "alloc-1"); + } + FeatureFlaggingGateway.addExposureListener(listener); + } + + @TearDown(Level.Trial) + public void tearDown() { + FeatureFlaggingGateway.removeExposureListener(listener); + } + + @Benchmark + public void dispatchExposure() { + DDEvaluator.dispatchExposure("bench-flag", evaluation, context); + } + + private static MutableContext buildContext(final String shape) { + final MutableContext ctx = new MutableContext("bench-user"); + if ("empty".equals(shape)) { + return ctx; + } + if ("flat/100attrs".equals(shape)) { + return addFlat(ctx, 100); + } + if ("nested/10structs_10fields".equals(shape)) { + for (int i = 0; i < 10; i++) { + final Map inner = new HashMap<>(); + for (int j = 0; j < 10; j++) { + inner.put("field" + j, new Value("value" + j)); + } + ctx.add("struct" + i, new ImmutableStructure(inner)); + } + return ctx; + } + if ("wide/1000attrs".equals(shape)) { + return addFlat(ctx, 1_000); + } + throw new IllegalArgumentException("unknown benchmark shape: " + shape); + } + + private static MutableContext addFlat(final MutableContext ctx, final int count) { + for (int i = 0; i < count; i++) { + ctx.add("field" + i, "value" + i); + } + return ctx; + } + + private static final class BenchmarkExposureListener + implements FeatureFlaggingGateway.ExposureListener { + private final boolean alwaysCapture; + private final ConcurrentMap identities = new ConcurrentHashMap<>(); + + private BenchmarkExposureListener(final boolean alwaysCapture) { + this.alwaysCapture = alwaysCapture; + } + + @Override + public boolean shouldCapture( + final String flag, final String subject, final String variant, final String allocation) { + final IdentityValue current = identities.get(new Identity(flag, subject)); + if (alwaysCapture) { + return true; + } + return current == null || !current.matches(variant, allocation); + } + + @Override + public void accept(final ExposureEvent event) { + record(event.flag.key, event.subject.id, event.variant.key, event.allocation.key); + } + + private void record( + final String flag, final String subject, final String variant, final String allocation) { + identities.put(new Identity(flag, subject), new IdentityValue(variant, allocation)); + } + } + + private static final class Identity { + private final String flag; + private final String subject; + + private Identity(final String flag, final String subject) { + this.flag = flag; + this.subject = subject; + } + + @Override + public boolean equals(final Object other) { + if (this == other) { + return true; + } + if (!(other instanceof Identity)) { + return false; + } + final Identity identity = (Identity) other; + return Objects.equals(flag, identity.flag) && Objects.equals(subject, identity.subject); + } + + @Override + public int hashCode() { + return Objects.hash(flag, subject); + } + } + + private static final class IdentityValue { + private final String variant; + private final String allocation; + + private IdentityValue(final String variant, final String allocation) { + this.variant = variant; + this.allocation = allocation; + } + + private boolean matches(final String otherVariant, final String otherAllocation) { + return Objects.equals(variant, otherVariant) && Objects.equals(allocation, otherAllocation); + } + } +} diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java index e74074a640a..ef276d5ca7f 100644 --- a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java +++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java @@ -20,7 +20,6 @@ import dev.openfeature.sdk.ErrorCode; import dev.openfeature.sdk.EvaluationContext; import dev.openfeature.sdk.ImmutableMetadata; -import dev.openfeature.sdk.ImmutableStructure; import dev.openfeature.sdk.ProviderEvaluation; import dev.openfeature.sdk.Reason; import dev.openfeature.sdk.Structure; @@ -29,14 +28,10 @@ import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.Instant; -import java.util.AbstractMap; -import java.util.ArrayList; import java.util.Collections; -import java.util.Deque; import java.util.HashMap; import java.util.HashSet; import java.util.IdentityHashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; @@ -57,8 +52,7 @@ class DDEvaluator implements Evaluator, FeatureFlaggingGateway.ConfigListener { * caller's evaluation thread over a caller-owned Value tree, so an arbitrarily deep * list/structure would overflow that thread's stack - and a StackOverflowError is not caught by * the LinkageError | Exception guards that keep telemetry from breaking an evaluation. Values - * below the limit are truncated to null, the same way the cycle guard truncates. Kept aligned - * with the cross-SDK RFC target (4). + * below the limit are omitted. Kept aligned with the cross-SDK RFC target (4). */ static final int MAX_SNAPSHOT_DEPTH = 4; @@ -649,20 +643,26 @@ private static Double parseDouble(final Object value) { return Double.parseDouble(String.valueOf(value)); } - private static void dispatchExposure( + static void dispatchExposure( final String flag, final ProviderEvaluation evaluation, final EvaluationContext context) { final String allocationKey = allocationKey(evaluation); final String variantKey = evaluation.getVariant(); if (allocationKey == null || variantKey == null) { return; } + final String subjectKey = context.getTargetingKey(); + if (!FeatureFlaggingGateway.shouldCaptureExposure( + flag, subjectKey, variantKey, allocationKey)) { + return; + } + final Map attributes = copyExposureContext(context).attrs; final ExposureEvent event = new ExposureEvent( System.currentTimeMillis(), new datadog.trace.api.featureflag.exposure.Allocation(allocationKey), new datadog.trace.api.featureflag.exposure.Flag(flag), new datadog.trace.api.featureflag.exposure.Variant(variantKey), - new Subject(context.getTargetingKey(), flattenContext(context))); + new Subject(subjectKey, attributes)); FeatureFlaggingGateway.dispatch(event); } @@ -672,96 +672,6 @@ private static String allocationKey(final ProviderEvaluation resolution) return meta == null ? null : meta.getString("allocationKey"); } - static AbstractMap flattenContext(final EvaluationContext context) { - return flattenValues(snapshotValues(context)); - } - - static Map snapshotValues(final EvaluationContext context) { - final HashMap values = new HashMap<>(); - final Set seenContainers = Collections.newSetFromMap(new IdentityHashMap<>()); - for (final String key : context.keySet()) { - values.put(key, snapshotValue(context.getValue(key), seenContainers, 0)); - } - return values; - } - - private static Value snapshotValue( - final Value value, final Set seenContainers, final int depth) { - if (value == null) { - return null; - } else if (value.isNull()) { - return new Value(); - } else if (value.isBoolean()) { - return new Value(value.asBoolean()); - } else if (value.isNumber()) { - final Object number = value.asObject(); - return number instanceof Integer - ? new Value((Integer) number) - : new Value(((Number) number).doubleValue()); - } else if (value.isString()) { - return new Value(value.asString()); - } else if (value.isInstant()) { - return new Value(value.asInstant()); - } else if (value.isList()) { - final List list = value.asList(); - if (depth >= MAX_SNAPSHOT_DEPTH || !seenContainers.add(list)) { - return new Value(); - } - final List snapshot = new ArrayList<>(list.size()); - for (final Value item : list) { - snapshot.add(snapshotValue(item, seenContainers, depth + 1)); - } - seenContainers.remove(list); - return new Value(Collections.unmodifiableList(snapshot)); - } else if (value.isStructure()) { - final Structure structure = value.asStructure(); - if (depth >= MAX_SNAPSHOT_DEPTH || !seenContainers.add(structure)) { - return new Value(); - } - final Map snapshot = new HashMap<>(); - for (final String key : structure.keySet()) { - snapshot.put(key, snapshotValue(structure.getValue(key), seenContainers, depth + 1)); - } - seenContainers.remove(structure); - return new Value(new ImmutableStructure(snapshot)); - } - throw new IllegalArgumentException("Unsupported OpenFeature value type: " + value); - } - - static AbstractMap flattenValues(final Map values) { - final HashMap result = new HashMap<>(); - final Set seenContainers = Collections.newSetFromMap(new IdentityHashMap<>()); - for (final Map.Entry root : values.entrySet()) { - final Deque deque = new LinkedList<>(); - deque.push(new FlattenEntry(root.getKey(), root.getValue())); - while (!deque.isEmpty()) { - final FlattenEntry entry = deque.pop(); - final Value value = entry.value; - if (value == null) { - result.put(entry.key, null); - } else if (value.isList()) { - final List list = value.asList(); - if (seenContainers.add(list)) { - for (int i = 0; i < list.size(); i++) { - deque.push(new FlattenEntry(entry.key + "[" + i + "]", list.get(i))); - } - } - } else if (value.isStructure()) { - final Structure structure = value.asStructure(); - if (seenContainers.add(structure)) { - for (final String property : structure.keySet()) { - deque.push( - new FlattenEntry(entry.key + "." + property, structure.getValue(property))); - } - } - } else { - result.put(entry.key, convertValue(value)); - } - } - } - return result; - } - private static Object convertValue(final Value value) { if (value == null || value.isNull()) { return null; @@ -856,6 +766,19 @@ static final class CopyResult { * canonical-key sorting happens once in the aggregator, off the hot path. */ static CopyResult copyPrunedContext(final EvaluationContext context) { + return copyPrunedContext(context, false); + } + + /** + * Builds the bounded exposure attributes. The targeting key stays in the attributes to preserve + * the existing exposure payload. The subject ID also carries the targeting key. + */ + static CopyResult copyExposureContext(final EvaluationContext context) { + return copyPrunedContext(context, true); + } + + private static CopyResult copyPrunedContext( + final EvaluationContext context, final boolean includeTargetingKey) { if (context == null) { return new CopyResult(Collections.emptyMap(), null); } @@ -871,7 +794,7 @@ static CopyResult copyPrunedContext(final EvaluationContext context) { reasonMask[0] |= REASON_MAX_CONTEXT_FIELDS; break; } - if (EvaluationContext.TARGETING_KEY.equals(key)) { + if (!includeTargetingKey && EvaluationContext.TARGETING_KEY.equals(key)) { continue; } copyPrunedValue(out, key, context.getValue(key), seen, 0, reasonMask); @@ -973,14 +896,4 @@ private interface NumberComparator { private interface SemverComparator { boolean compare(int ordering); } - - private static class FlattenEntry { - private final String key; - private final Value value; - - private FlattenEntry(final String key, final Value value) { - this.key = key; - this.value = value; - } - } } diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java index aeddda0bfd2..a83bac37259 100644 --- a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java +++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java @@ -26,6 +26,7 @@ import com.squareup.moshi.Moshi; import com.squareup.moshi.Types; import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import datadog.trace.api.featureflag.exposure.ExposureEvent; import datadog.trace.api.featureflag.ufc.v1.Allocation; import datadog.trace.api.featureflag.ufc.v1.ConditionConfiguration; import datadog.trace.api.featureflag.ufc.v1.ConditionOperator; @@ -38,6 +39,7 @@ import datadog.trace.api.featureflag.ufc.v1.Variant; import dev.openfeature.sdk.ErrorCode; import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.ImmutableMetadata; import dev.openfeature.sdk.MutableContext; import dev.openfeature.sdk.ProviderEvaluation; import dev.openfeature.sdk.Value; @@ -641,11 +643,11 @@ private static Arguments[] flatteningTestCases() { @MethodSource("flatteningTestCases") @ParameterizedTest - public void testFlattening( + public void testExposureContextFlattening( final Map attributes, final Map expected) { final EvaluationContext context = new MutableContext(Value.objectToValue(attributes).asStructure().asMap()); - final Map result = DDEvaluator.flattenContext(context); + final Map result = DDEvaluator.copyExposureContext(context).attrs; assertThat(result.size(), equalTo(expected.size())); for (final Map.Entry entry : expected.entrySet()) { @@ -653,24 +655,6 @@ public void testFlattening( } } - @Test - public void testDeeplyNestedContextIsTruncatedRatherThanOverflowingTheStack() { - Value nested = new Value("leaf"); - for (int i = 0; i < 10_000; i++) { - nested = new Value(singletonList(nested)); - } - final EvaluationContext context = new MutableContext().add("deep", singletonList(nested)); - - final Map result = DDEvaluator.flattenContext(context); - - final StringBuilder truncatedKey = new StringBuilder("deep"); - for (int i = 0; i < DDEvaluator.MAX_SNAPSHOT_DEPTH; i++) { - truncatedKey.append("[0]"); - } - assertThat(result.size(), equalTo(1)); - assertThat(result, hasEntry(truncatedKey.toString(), null)); - } - @Test public void testCopyPrunedContextCapsTopLevelFieldCount() { final MutableContext context = new MutableContext(); @@ -780,6 +764,69 @@ public void testCopyPrunedContextExcludesTargetingKey() { assertThat(result.truncatedReason, equalTo(null)); } + @Test + public void testCopyExposureContextPreservesTargetingKey() { + final MutableContext context = new MutableContext("user-42").add("region", "us-east-1"); + + final DDEvaluator.CopyResult result = DDEvaluator.copyExposureContext(context); + + assertThat(result.attrs, hasEntry("targetingKey", "user-42")); + assertThat(result.attrs, hasEntry("region", "us-east-1")); + assertThat(result.truncatedReason, equalTo(null)); + } + + @Test + public void testCopyExposureContextCapsFieldCount() { + final MutableContext context = new MutableContext("user-42"); + for (int i = 0; i < DDEvaluator.MAX_CONTEXT_FIELDS + 100; i++) { + context.add(String.format("k%04d", i), "v"); + } + + final DDEvaluator.CopyResult result = DDEvaluator.copyExposureContext(context); + + assertThat(result.attrs.size(), equalTo(DDEvaluator.MAX_CONTEXT_FIELDS)); + assertThat(result.truncatedReason, equalTo("max_context_fields")); + } + + @Test + public void testDispatchExposureChecksAdmissionBeforeContextCapture() { + final List captured = new ArrayList<>(); + final FeatureFlaggingGateway.ExposureListener listener = + new FeatureFlaggingGateway.ExposureListener() { + @Override + public boolean shouldCapture( + final String flag, + final String subject, + final String variant, + final String allocation) { + return captured.isEmpty(); + } + + @Override + public void accept(final ExposureEvent event) { + captured.add(event); + } + }; + final MutableContext context = new MutableContext("user-42").add("region", "first"); + final ProviderEvaluation evaluation = + ProviderEvaluation.builder() + .value("on-value") + .variant("on") + .flagMetadata(ImmutableMetadata.builder().addString("allocationKey", "alloc-1").build()) + .build(); + FeatureFlaggingGateway.addExposureListener(listener); + try { + DDEvaluator.dispatchExposure("flag", evaluation, context); + context.add("region", "changed"); + DDEvaluator.dispatchExposure("flag", evaluation, context); + } finally { + FeatureFlaggingGateway.removeExposureListener(listener); + } + + assertThat(captured.size(), equalTo(1)); + assertThat(captured.get(0).subject.attributes, hasEntry("region", "first")); + } + @Test public void testCopyPrunedContextNoTruncationReturnsNullReason() { final MutableContext context = new MutableContext("user-1"); diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java index 2a823bd32ef..68c3cb67e3b 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java @@ -16,7 +16,16 @@ public interface ActivationListener { void activate(); } - public interface ExposureListener extends Consumer {} + public interface ExposureListener extends Consumer { + /** + * Returns whether this listener needs a complete event for the supplied exposure identity. The + * default preserves the behavior of listeners that do not implement early admission. + */ + default boolean shouldCapture( + final String flag, final String subject, final String variant, final String allocation) { + return true; + } + } public interface SpanEnrichmentListener extends Consumer {} @@ -80,6 +89,17 @@ public static void removeExposureListener(final ExposureListener listener) { EXPOSURE_LISTENERS.remove(listener); } + /** Returns whether at least one listener needs a complete event for this exposure identity. */ + public static boolean shouldCaptureExposure( + final String flag, final String subject, final String variant, final String allocation) { + for (final ExposureListener listener : EXPOSURE_LISTENERS) { + if (listener.shouldCapture(flag, subject, variant, allocation)) { + return true; + } + } + return false; + } + public static void dispatch(final ExposureEvent event) { EXPOSURE_LISTENERS.forEach(listener -> listener.accept(event)); } diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java index 887a153f0a1..ff79df0aa4a 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java @@ -1,5 +1,7 @@ package datadog.trace.api.featureflag; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -92,6 +94,43 @@ void testAttachingAnExposureListener() { verifyNoMoreInteractions(exposureListener); } + @Test + void testExposureAdmissionUsesRegisteredListeners() { + assertFalse( + FeatureFlaggingGateway.shouldCaptureExposure("flag", "subject", "variant", "allocation")); + + final FeatureFlaggingGateway.ExposureListener rejectingListener = + new FeatureFlaggingGateway.ExposureListener() { + @Override + public boolean shouldCapture( + final String flag, + final String subject, + final String variant, + final String allocation) { + return false; + } + + @Override + public void accept(final ExposureEvent event) {} + }; + FeatureFlaggingGateway.addExposureListener(rejectingListener); + try { + assertFalse( + FeatureFlaggingGateway.shouldCaptureExposure("flag", "subject", "variant", "allocation")); + } finally { + FeatureFlaggingGateway.removeExposureListener(rejectingListener); + } + + final FeatureFlaggingGateway.ExposureListener listener = event -> {}; + FeatureFlaggingGateway.addExposureListener(listener); + try { + assertTrue( + FeatureFlaggingGateway.shouldCaptureExposure("flag", "subject", "variant", "allocation")); + } finally { + FeatureFlaggingGateway.removeExposureListener(listener); + } + } + @Test void testAttachingASpanEnrichmentListener() { final SpanEnrichmentEvent firstEvent = SpanEnrichmentEvent.serialId(42, true, "user-1"); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureAdmissionCache.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureAdmissionCache.java new file mode 100644 index 00000000000..3fd60fc2ebd --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureAdmissionCache.java @@ -0,0 +1,149 @@ +package com.datadog.featureflag; + +import datadog.trace.api.featureflag.exposure.ExposureEvent; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ConcurrentMap; + +/** + * A bounded, thread-safe cache for exposure admission on application evaluation threads. + * + *

The serializer keeps the authoritative LRU cache. This cache can only avoid work for an exact + * recent match. An eviction or a concurrent miss creates an extra event, which the serializer + * removes. It does not remove a changed exposure. + */ +final class ExposureAdmissionCache { + + private static final int LOCK_COUNT = 64; + + private final int capacity; + private final ConcurrentMap identities = new ConcurrentHashMap<>(); + private final ConcurrentLinkedQueue insertionOrder = new ConcurrentLinkedQueue<>(); + private final Object[] locks = new Object[LOCK_COUNT]; + private volatile boolean closed; + + ExposureAdmissionCache(final int capacity) { + if (capacity <= 0) { + throw new IllegalArgumentException("capacity must be positive"); + } + this.capacity = capacity; + for (int i = 0; i < locks.length; i++) { + locks[i] = new Object(); + } + } + + boolean contains( + final String flag, final String subject, final String variant, final String allocation) { + final Value current = identities.get(new Key(flag, subject)); + return current != null && current.matches(variant, allocation); + } + + void add(final ExposureEvent event) { + if (closed) { + return; + } + final Key key = + new Key( + event.flag == null ? null : event.flag.key, + event.subject == null ? null : event.subject.id); + final Value value = + new Value( + event.variant == null ? null : event.variant.key, + event.allocation == null ? null : event.allocation.key); + final Value previous = identities.put(key, value); + if (previous == null) { + insertionOrder.offer(key); + evictExcess(); + } + } + + Object lockFor(final ExposureEvent event) { + final String flag = event.flag == null ? null : event.flag.key; + final String subject = event.subject == null ? null : event.subject.id; + int hash = flag == null ? 0 : flag.hashCode(); + hash = 31 * hash + (subject == null ? 0 : subject.hashCode()); + return locks[(hash ^ (hash >>> 16)) & (locks.length - 1)]; + } + + void clear() { + identities.clear(); + insertionOrder.clear(); + } + + void close() { + closeWithLocksHeld(0); + } + + boolean isClosed() { + return closed; + } + + int size() { + return identities.size(); + } + + private void evictExcess() { + while (identities.size() > capacity) { + final Key oldest = insertionOrder.poll(); + if (oldest == null) { + return; + } + identities.remove(oldest); + } + } + + private void closeWithLocksHeld(final int index) { + if (index == locks.length) { + closed = true; + clear(); + return; + } + synchronized (locks[index]) { + closeWithLocksHeld(index + 1); + } + } + + static final class Key { + private final String flag; + private final String subject; + + Key(final String flag, final String subject) { + this.flag = flag; + this.subject = subject; + } + + @Override + public boolean equals(final Object other) { + if (this == other) { + return true; + } + if (!(other instanceof Key)) { + return false; + } + final Key key = (Key) other; + return Objects.equals(flag, key.flag) && Objects.equals(subject, key.subject); + } + + @Override + public int hashCode() { + int result = flag == null ? 0 : flag.hashCode(); + result = 31 * result + (subject == null ? 0 : subject.hashCode()); + return result; + } + } + + private static final class Value { + private final String variant; + private final String allocation; + + private Value(final String variant, final String allocation) { + this.variant = variant; + this.allocation = allocation; + } + + private boolean matches(final String otherVariant, final String otherAllocation) { + return Objects.equals(variant, otherVariant) && Objects.equals(allocation, otherAllocation); + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java index 46b3fe5a486..ed216e0dee8 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java @@ -30,6 +30,7 @@ public class ExposureWriterImpl implements ExposureWriter { private static final String EXPOSURES_ROUTE = "exposures"; private final MessagePassingBlockingQueue queue; + private final ExposureAdmissionCache admissionCache; private final Thread serializerThread; public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config config) { @@ -43,6 +44,7 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con final SharedCommunicationObjects sco, final Config config) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); + this.admissionCache = new ExposureAdmissionCache(capacity); final ExposureSerializingHandler serializer = new ExposureSerializingHandler( new BackendApiFactory(config, sco), @@ -63,6 +65,7 @@ public void init() { @Override public void close() { FeatureFlaggingGateway.removeExposureListener(this); + admissionCache.close(); if (this.serializerThread.isAlive()) { this.serializerThread.interrupt(); } @@ -70,7 +73,21 @@ public void close() { @Override public void accept(final ExposureEvent event) { - queue.offer(event); + synchronized (admissionCache.lockFor(event)) { + if (admissionCache.isClosed()) { + return; + } + if (queue.offer(event)) { + admissionCache.add(event); + } + } + } + + @Override + public boolean shouldCapture( + final String flag, final String subject, final String variant, final String allocation) { + return !admissionCache.isClosed() + && !admissionCache.contains(flag, subject, variant, allocation); } @VisibleForTesting diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureAdmissionCacheTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureAdmissionCacheTest.java new file mode 100644 index 00000000000..372e65f8338 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureAdmissionCacheTest.java @@ -0,0 +1,117 @@ +package com.datadog.featureflag; + +import static java.util.Collections.emptyMap; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.api.featureflag.exposure.Allocation; +import datadog.trace.api.featureflag.exposure.ExposureEvent; +import datadog.trace.api.featureflag.exposure.Flag; +import datadog.trace.api.featureflag.exposure.Subject; +import datadog.trace.api.featureflag.exposure.Variant; +import org.junit.jupiter.api.Test; + +class ExposureAdmissionCacheTest { + + @Test + void rejectsNonPositiveCapacity() { + assertThrows(IllegalArgumentException.class, () -> new ExposureAdmissionCache(0)); + } + + @Test + void admitsOnlyExactRecentIdentity() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(4); + cache.add(event("flag", "subject", "variant", "allocation")); + + assertTrue(cache.contains("flag", "subject", "variant", "allocation")); + assertFalse(cache.contains("other", "subject", "variant", "allocation")); + assertFalse(cache.contains("flag", "other", "variant", "allocation")); + assertFalse(cache.contains("flag", "subject", "other", "allocation")); + assertFalse(cache.contains("flag", "subject", "variant", "other")); + } + + @Test + void changedValueReplacesPreviousValue() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(4); + cache.add(event("flag", "subject", "first", "allocation")); + cache.add(event("flag", "subject", "second", "allocation")); + + assertFalse(cache.contains("flag", "subject", "first", "allocation")); + assertTrue(cache.contains("flag", "subject", "second", "allocation")); + } + + @Test + void evictsOldestIdentityAtCapacity() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(2); + cache.add(event("first", "subject", "variant", "allocation")); + cache.add(event("second", "subject", "variant", "allocation")); + cache.add(event("third", "subject", "variant", "allocation")); + + assertEquals(2, cache.size()); + assertFalse(cache.contains("first", "subject", "variant", "allocation")); + assertTrue(cache.contains("second", "subject", "variant", "allocation")); + assertTrue(cache.contains("third", "subject", "variant", "allocation")); + } + + @Test + void clearRemovesRetainedCustomerValues() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(2); + cache.add(event("flag", "subject", "variant", "allocation")); + + cache.clear(); + + assertEquals(0, cache.size()); + assertFalse(cache.contains("flag", "subject", "variant", "allocation")); + } + + @Test + void closeRejectsFutureValues() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(2); + final ExposureEvent event = event("flag", "subject", "variant", "allocation"); + cache.add(event); + + cache.close(); + cache.add(event); + + assertTrue(cache.isClosed()); + assertEquals(0, cache.size()); + } + + @Test + void supportsNullIdentityFields() { + final ExposureAdmissionCache cache = new ExposureAdmissionCache(2); + final ExposureEvent event = new ExposureEvent(1, null, null, null, null); + + cache.add(event); + + assertTrue(cache.contains(null, null, null, null)); + assertEquals(cache.lockFor(event), cache.lockFor(event)); + } + + @Test + void identityKeyUsesFlagAndSubject() { + final ExposureAdmissionCache.Key key = new ExposureAdmissionCache.Key("flag", "subject"); + final ExposureAdmissionCache.Key equal = new ExposureAdmissionCache.Key("flag", "subject"); + + assertEquals(key, key); + assertEquals(key, equal); + assertEquals(key.hashCode(), equal.hashCode()); + assertNotEquals(key, null); + assertNotEquals(key, "not-a-key"); + assertNotEquals(key, new ExposureAdmissionCache.Key("other", "subject")); + assertNotEquals(key, new ExposureAdmissionCache.Key("flag", "other")); + } + + private static ExposureEvent event( + final String flag, final String subject, final String variant, final String allocation) { + return new ExposureEvent( + 1, + new Allocation(allocation), + new Flag(flag), + new Variant(variant), + new Subject(subject, emptyMap())); + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java index 76b9e2602d8..84575352a54 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java @@ -164,6 +164,38 @@ void testLruCache() throws Exception { } } + @Test + void testAdmissionCacheRecordsOnlyQueuedExposures() { + final Config config = mockConfig("test-service"); + final ExposureEvent accepted = buildExposure(); + final ExposureEvent rejected = buildExposure(); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(1, 100, MILLISECONDS, sharedCommunicationObjects, config)) { + assertTrue(shouldCapture(writer, accepted)); + writer.accept(accepted); + assertFalse(shouldCapture(writer, accepted)); + + writer.accept(rejected); + assertTrue(shouldCapture(writer, rejected)); + } + } + + @Test + void testCloseClearsAdmissionCache() { + final Config config = mockConfig("test-service"); + final ExposureEvent exposure = buildExposure(); + final ExposureWriterImpl writer = + new ExposureWriterImpl(1, 100, MILLISECONDS, sharedCommunicationObjects, config); + writer.accept(exposure); + + writer.close(); + writer.accept(buildExposure()); + + assertFalse(shouldCapture(writer, exposure)); + assertEquals(1, writer.queueSize()); + } + @Test void testHighLoadScenario() throws Exception { Config config = mockConfig("test-service"); @@ -431,4 +463,9 @@ private static ExposureEvent buildExposure(String id, Map attrib new Variant("Variant_" + id), new Subject("Subject_" + id, attributes)); } + + private static boolean shouldCapture(final ExposureWriterImpl writer, final ExposureEvent event) { + return writer.shouldCapture( + event.flag.key, event.subject.id, event.variant.key, event.allocation.key); + } }