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..08f573240d6 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 @@ -5,6 +5,7 @@ import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.exposure.ExposureEvent; import datadog.trace.api.featureflag.exposure.Subject; +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEventMemoryEstimator; import datadog.trace.api.featureflag.ufc.v1.Allocation; import datadog.trace.api.featureflag.ufc.v1.ConditionConfiguration; import datadog.trace.api.featureflag.ufc.v1.ConditionOperator; @@ -826,11 +827,18 @@ static String truncationReasonTag(final int reasonMask) { static final class CopyResult { final Map attrs; + /** Conservative retained-byte estimate for attrs, calculated during the bounded copy. */ + final long estimatedRetainedBytes; + /** Non-null when at least one cap fired; ready to use as the "reason:..." tag value. */ final String truncatedReason; - CopyResult(final Map attrs, final String truncatedReason) { + CopyResult( + final Map attrs, + final long estimatedRetainedBytes, + final String truncatedReason) { this.attrs = attrs; + this.estimatedRetainedBytes = estimatedRetainedBytes; this.truncatedReason = truncatedReason; } } @@ -857,27 +865,30 @@ static final class CopyResult { */ static CopyResult copyPrunedContext(final EvaluationContext context) { if (context == null) { - return new CopyResult(Collections.emptyMap(), null); + return new CopyResult(Collections.emptyMap(), 0, null); } final Set keys = context.keySet(); if (keys.isEmpty()) { - return new CopyResult(Collections.emptyMap(), null); + return new CopyResult(Collections.emptyMap(), 0, null); } final HashMap out = new HashMap<>(); final Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); - final int[] reasonMask = {0}; + final CopyState state = new CopyState(FlagEvalEventMemoryEstimator.contextMapRetainedBytes()); for (final String key : keys) { if (out.size() >= MAX_CONTEXT_FIELDS) { - reasonMask[0] |= REASON_MAX_CONTEXT_FIELDS; + state.reasonMask |= REASON_MAX_CONTEXT_FIELDS; break; } if (EvaluationContext.TARGETING_KEY.equals(key)) { continue; } - copyPrunedValue(out, key, context.getValue(key), seen, 0, reasonMask); + copyPrunedValue(out, key, context.getValue(key), seen, 0, state); } final Map attrs = out.isEmpty() ? Collections.emptyMap() : out; - return new CopyResult(attrs, truncationReasonTag(reasonMask[0])); + return new CopyResult( + attrs, + out.isEmpty() ? 0 : state.estimatedRetainedBytes, + truncationReasonTag(state.reasonMask)); } private static void copyPrunedValue( @@ -886,52 +897,52 @@ private static void copyPrunedValue( final Value value, final Set seen, final int depth, - final int[] reasonMask) { + final CopyState state) { if (out.size() >= MAX_CONTEXT_FIELDS) { - reasonMask[0] |= REASON_MAX_CONTEXT_FIELDS; + state.reasonMask |= REASON_MAX_CONTEXT_FIELDS; return; } if (key.length() > MAX_KEY_LENGTH) { - reasonMask[0] |= REASON_MAX_KEY_LENGTH; + state.reasonMask |= REASON_MAX_KEY_LENGTH; return; } if (value == null || value.isNull()) { - out.put(key, null); + putPrunedValue(out, key, null, state); return; } if (value.isString()) { final String s = value.asString(); if (s.length() > MAX_VALUE_LENGTH) { - reasonMask[0] |= REASON_MAX_VALUE_LENGTH; + state.reasonMask |= REASON_MAX_VALUE_LENGTH; return; } - out.put(key, s); + putPrunedValue(out, key, s, state); return; } if (value.isBoolean() || value.isNumber() || value.isInstant()) { - out.put(key, convertValue(value)); + putPrunedValue(out, key, convertValue(value), state); return; } if (value.isList()) { final List list = value.asList(); if (depth >= MAX_SNAPSHOT_DEPTH) { - reasonMask[0] |= REASON_MAX_SNAPSHOT_DEPTH; + state.reasonMask |= REASON_MAX_SNAPSHOT_DEPTH; return; } if (!seen.add(list)) { - reasonMask[0] |= REASON_CYCLE; + state.reasonMask |= REASON_CYCLE; return; } if (list.size() > MAX_LIST_ELEMENTS) { - reasonMask[0] |= REASON_MAX_LIST_ELEMENTS; + state.reasonMask |= REASON_MAX_LIST_ELEMENTS; } final int limit = Math.min(list.size(), MAX_LIST_ELEMENTS); for (int i = 0; i < limit; i++) { if (out.size() >= MAX_CONTEXT_FIELDS) { - reasonMask[0] |= REASON_MAX_CONTEXT_FIELDS; + state.reasonMask |= REASON_MAX_CONTEXT_FIELDS; break; } - copyPrunedValue(out, key + "[" + i + "]", list.get(i), seen, depth + 1, reasonMask); + copyPrunedValue(out, key + "[" + i + "]", list.get(i), seen, depth + 1, state); } seen.remove(list); return; @@ -939,31 +950,47 @@ private static void copyPrunedValue( if (value.isStructure()) { final Structure structure = value.asStructure(); if (depth >= MAX_SNAPSHOT_DEPTH) { - reasonMask[0] |= REASON_MAX_SNAPSHOT_DEPTH; + state.reasonMask |= REASON_MAX_SNAPSHOT_DEPTH; return; } if (!seen.add(structure)) { - reasonMask[0] |= REASON_CYCLE; + state.reasonMask |= REASON_CYCLE; return; } int walked = 0; for (final String property : structure.keySet()) { if (walked >= MAX_STRUCTURE_PROPERTIES) { - reasonMask[0] |= REASON_MAX_STRUCTURE_PROPERTIES; + state.reasonMask |= REASON_MAX_STRUCTURE_PROPERTIES; break; } if (out.size() >= MAX_CONTEXT_FIELDS) { - reasonMask[0] |= REASON_MAX_CONTEXT_FIELDS; + state.reasonMask |= REASON_MAX_CONTEXT_FIELDS; break; } walked++; copyPrunedValue( - out, key + "." + property, structure.getValue(property), seen, depth + 1, reasonMask); + out, key + "." + property, structure.getValue(property), seen, depth + 1, state); } seen.remove(structure); } } + private static void putPrunedValue( + final Map out, final String key, final Object value, final CopyState state) { + state.estimatedRetainedBytes += + FlagEvalEventMemoryEstimator.contextEntryRetainedBytes(key, value); + out.put(key, value); + } + + private static final class CopyState { + private int reasonMask; + private long estimatedRetainedBytes; + + private CopyState(final long estimatedRetainedBytes) { + this.estimatedRetainedBytes = estimatedRetainedBytes; + } + } + @FunctionalInterface private interface NumberComparator { boolean compare(double a, double b); diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/FlagEvalLoggingHook.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/FlagEvalLoggingHook.java index 322f11ac9e1..ac2c8ae08cd 100644 --- a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/FlagEvalLoggingHook.java +++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/FlagEvalLoggingHook.java @@ -138,6 +138,7 @@ public void finallyAfter( // consulted by the aggregator, so skip the bounded copy entirely — the copy cost only // applies under consent-on. final Map attrs; + final long estimatedContextRetainedBytes; if (observeFullEvaluationData && ctx != null && ctx.getCtx() != null) { // Bounded copy of the caller's mutable context (see DDEvaluator.copyPrunedContext for // every retained-size cap). Runs inline because the event is consumed asynchronously @@ -147,8 +148,10 @@ public void finallyAfter( w.countContextTruncated(copy.truncatedReason); } attrs = copy.attrs; + estimatedContextRetainedBytes = copy.estimatedRetainedBytes; } else { attrs = Collections.emptyMap(); + estimatedContextRetainedBytes = 0; } w.enqueue( @@ -160,7 +163,8 @@ public void finallyAfter( errorMessage, evalTimeMs, observeFullEvaluationData, - attrs)); + attrs, + estimatedContextRetainedBytes)); } catch (LinkageError e) { // Never let EVP recording break flag evaluation } 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..2604f24c46a 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.flagevaluation.FlagEvalEventMemoryEstimator; import datadog.trace.api.featureflag.ufc.v1.Allocation; import datadog.trace.api.featureflag.ufc.v1.ConditionConfiguration; import datadog.trace.api.featureflag.ufc.v1.ConditionOperator; @@ -790,6 +791,9 @@ public void testCopyPrunedContextNoTruncationReturnsNullReason() { assertThat(result.attrs, hasEntry("region", "us-east-1")); assertThat(result.truncatedReason, equalTo(null)); + assertThat( + result.estimatedRetainedBytes, + equalTo(FlagEvalEventMemoryEstimator.retainedContextBytes(result.attrs))); } @Test diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/FlagEvalLoggingHookTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/FlagEvalLoggingHookTest.java index 3b7910d4712..28bd9fb9c7e 100644 --- a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/FlagEvalLoggingHookTest.java +++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/FlagEvalLoggingHookTest.java @@ -568,6 +568,17 @@ void contextAttributesAreFlattenedAndConvertedInline() { assertEquals(42, attrs.get("score")); assertEquals("gold", attrs.get("profile.tier")); assertFalse(attrs.containsKey("targetingKey")); + final FlagEvalEvent fallbackEstimate = + new FlagEvalEvent( + captured.get().flagKey, + captured.get().variant, + captured.get().allocationKey, + captured.get().targetingKey, + captured.get().errorMessage, + captured.get().evalTimeMs, + captured.get().observeFullEvaluationData, + attrs); + assertEquals(fallbackEstimate.estimatedRetainedBytes, captured.get().estimatedRetainedBytes); assertTrue( attrs.values().stream().noneMatch(Value.class::isInstance), "context attrs must contain converted scalar values, not OpenFeature Value wrappers"); diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEvent.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEvent.java index f5f44b94d7b..e17953c341e 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEvent.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEvent.java @@ -54,6 +54,9 @@ public final class FlagEvalEvent { */ public final boolean observeFullEvaluationData; + /** Estimated retained bytes for this complete event while it waits in the queue. */ + public final long estimatedRetainedBytes; + /** Convenience constructor; consent defaults to the privacy-preserving false. */ public FlagEvalEvent( final String flagKey, @@ -62,7 +65,16 @@ public FlagEvalEvent( final String targetingKey, final long evalTimeMs, final Map attrs) { - this(flagKey, variant, allocationKey, targetingKey, null, evalTimeMs, false, attrs); + this( + flagKey, + variant, + allocationKey, + targetingKey, + null, + evalTimeMs, + false, + attrs, + FlagEvalEventMemoryEstimator.UNKNOWN_RETAINED_BYTES); } /** Convenience constructor; consent defaults to the privacy-preserving false. */ @@ -74,7 +86,16 @@ public FlagEvalEvent( final String errorMessage, final long evalTimeMs, final Map attrs) { - this(flagKey, variant, allocationKey, targetingKey, errorMessage, evalTimeMs, false, attrs); + this( + flagKey, + variant, + allocationKey, + targetingKey, + errorMessage, + evalTimeMs, + false, + attrs, + FlagEvalEventMemoryEstimator.UNKNOWN_RETAINED_BYTES); } public FlagEvalEvent( @@ -86,6 +107,28 @@ public FlagEvalEvent( final long evalTimeMs, final boolean observeFullEvaluationData, final Map attrs) { + this( + flagKey, + variant, + allocationKey, + targetingKey, + errorMessage, + evalTimeMs, + observeFullEvaluationData, + attrs, + FlagEvalEventMemoryEstimator.UNKNOWN_RETAINED_BYTES); + } + + public FlagEvalEvent( + final String flagKey, + final String variant, + final String allocationKey, + final String targetingKey, + final String errorMessage, + final long evalTimeMs, + final boolean observeFullEvaluationData, + final Map attrs, + final long estimatedContextRetainedBytes) { this.flagKey = flagKey; this.variant = variant; this.allocationKey = allocationKey; @@ -94,5 +137,7 @@ public FlagEvalEvent( this.evalTimeMs = evalTimeMs; this.observeFullEvaluationData = observeFullEvaluationData; this.attrs = attrs != null ? attrs : Collections.emptyMap(); + this.estimatedRetainedBytes = + FlagEvalEventMemoryEstimator.estimateRetainedBytes(this, estimatedContextRetainedBytes); } } diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimator.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimator.java new file mode 100644 index 00000000000..bbf2080eb9d --- /dev/null +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimator.java @@ -0,0 +1,78 @@ +package datadog.trace.api.featureflag.flagevaluation; + +import java.util.Map; + +/** Conservatively estimates memory retained while an event waits in the evaluation queue. */ +public final class FlagEvalEventMemoryEstimator { + + public static final long UNKNOWN_RETAINED_BYTES = -1; + + // These constants include aligned object headers, references, and amortized queue or map storage. + // They use the same string and context-entry model as the aggregation byte budget. + private static final long EVENT_AND_QUEUE_ENTRY_BYTES = 128; + private static final long CONTEXT_MAP_BYTES = 64; + private static final long CONTEXT_ENTRY_BYTES = 48; + private static final long STRING_BYTES = 40; + private static final long OTHER_VALUE_BYTES = 128; + private static final long MAX_BYTES_PER_CHARACTER = 2; + + private FlagEvalEventMemoryEstimator() {} + + public static long retainedBytes(final FlagEvalEvent event) { + return event.estimatedRetainedBytes; + } + + static long estimateRetainedBytes( + final FlagEvalEvent event, final long estimatedContextRetainedBytes) { + long bytes = EVENT_AND_QUEUE_ENTRY_BYTES; + bytes = add(bytes, stringBytes(event.flagKey)); + bytes = add(bytes, stringBytes(event.variant)); + bytes = add(bytes, stringBytes(event.allocationKey)); + bytes = add(bytes, stringBytes(event.targetingKey)); + bytes = add(bytes, stringBytes(event.errorMessage)); + final long contextBytes = + estimatedContextRetainedBytes < 0 + ? retainedContextBytes(event.attrs) + : estimatedContextRetainedBytes; + return add(bytes, contextBytes); + } + + public static long retainedContextBytes(final Map attrs) { + if (attrs.isEmpty()) { + return 0; + } + long bytes = contextMapRetainedBytes(); + for (final Map.Entry entry : attrs.entrySet()) { + bytes = add(bytes, contextEntryRetainedBytes(entry.getKey(), entry.getValue())); + } + return bytes; + } + + public static long contextMapRetainedBytes() { + return CONTEXT_MAP_BYTES; + } + + public static long contextEntryRetainedBytes(final String key, final Object value) { + return add(CONTEXT_ENTRY_BYTES, add(stringBytes(key), contextValueBytes(value))); + } + + private static long contextValueBytes(final Object value) { + if (value instanceof String) { + return stringBytes((String) value); + } + return value == null ? 0 : OTHER_VALUE_BYTES; + } + + private static long stringBytes(final String value) { + return value == null ? 0 : add(STRING_BYTES, characterBytes(value.length())); + } + + private static long characterBytes(final int length) { + return MAX_BYTES_PER_CHARACTER * length; + } + + private static long add(final long left, final long right) { + // Both inputs describe live Java objects. Their sum cannot approach the long range in one JVM. + return left + right; + } +} diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimatorTest.java b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimatorTest.java new file mode 100644 index 00000000000..362a164bb1c --- /dev/null +++ b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventMemoryEstimatorTest.java @@ -0,0 +1,63 @@ +package datadog.trace.api.featureflag.flagevaluation; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class FlagEvalEventMemoryEstimatorTest { + + @Test + void estimatesFallbackContextAndEverySupportedValueShape() { + final Map attrs = new LinkedHashMap<>(); + attrs.put("string", "value"); + attrs.put("number", 42L); + attrs.put("empty", null); + final FlagEvalEvent event = + new FlagEvalEvent("flag", "on", "allocation", "subject", "error", 1L, true, attrs); + + final long expectedContextBytes = + FlagEvalEventMemoryEstimator.contextMapRetainedBytes() + + FlagEvalEventMemoryEstimator.contextEntryRetainedBytes("string", "value") + + FlagEvalEventMemoryEstimator.contextEntryRetainedBytes("number", 42L) + + FlagEvalEventMemoryEstimator.contextEntryRetainedBytes("empty", null); + + assertEquals(expectedContextBytes, FlagEvalEventMemoryEstimator.retainedContextBytes(attrs)); + assertEquals( + 128 + + stringBytes("flag") + + stringBytes("on") + + stringBytes("allocation") + + stringBytes("subject") + + stringBytes("error") + + expectedContextBytes, + FlagEvalEventMemoryEstimator.retainedBytes(event)); + } + + @Test + void usesPrecomputedContextBytesWithoutReadingTheMap() { + final Map unreadableAttrs = + new java.util.AbstractMap() { + @Override + public java.util.Set> entrySet() { + throw new AssertionError("precomputed estimate must avoid a second context walk"); + } + }; + final FlagEvalEvent event = + new FlagEvalEvent("flag", null, null, null, null, 1L, true, unreadableAttrs, 512); + + assertEquals( + 128 + stringBytes("flag") + 512, FlagEvalEventMemoryEstimator.retainedBytes(event)); + } + + @Test + void emptyContextRetainsNoContextBytes() { + assertEquals( + 0, FlagEvalEventMemoryEstimator.retainedContextBytes(java.util.Collections.emptyMap())); + } + + private static long stringBytes(final String value) { + return 40L + 2L * value.length(); + } +} diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventTest.java b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventTest.java index cc614150178..c8b3a852a39 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventTest.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/flagevaluation/FlagEvalEventTest.java @@ -26,6 +26,7 @@ void storesFieldsWithContextAttributes() { assertNull(event.errorMessage); assertEquals(123L, event.evalTimeMs); assertSame(attrs, event.attrs); + assertTrue(event.estimatedRetainedBytes > 0); } @Test @@ -51,4 +52,16 @@ void storesExplicitObserveFullEvaluationData() { assertTrue( new FlagEvalEvent("f", "on", "a", "t", null, 1L, true, attrs).observeFullEvaluationData); } + + @Test + void precomputedContextBytesMatchFallbackEstimate() { + final Map attrs = Collections.singletonMap("tier", "gold"); + final long retainedBytes = FlagEvalEventMemoryEstimator.retainedContextBytes(attrs); + + final FlagEvalEvent event = + new FlagEvalEvent("f", "on", "a", "t", null, 1L, true, attrs, retainedBytes); + final FlagEvalEvent fallback = new FlagEvalEvent("f", "on", "a", "t", null, 1L, true, attrs); + + assertEquals(fallback.estimatedRetainedBytes, event.estimatedRetainedBytes); + } } diff --git a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java index 8da76d4a88e..5988c828b48 100644 --- a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java +++ b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java @@ -7,6 +7,7 @@ import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEventMemoryEstimator; import de.thetaphi.forbiddenapis.SuppressForbidden; import java.util.HashMap; import java.util.Map; @@ -64,6 +65,7 @@ public class FlagEvaluationEnqueueContentionBenchmark { private static final int NUM_FIELDS = 10; private Map attrs; + private long estimatedContextRetainedBytes; private String[] flagKeys; private String[] targetingKeys; private FlagEvaluationWriterImpl writer; @@ -84,6 +86,7 @@ public void setUp() { for (int i = 0; i < NUM_FIELDS; i++) { attrs.put("field" + i, "value"); } + estimatedContextRetainedBytes = FlagEvalEventMemoryEstimator.retainedContextBytes(attrs); flagKeys = keys("bench-flag-", NUM_FLAGS); targetingKeys = keys("bench-user-", NUM_USERS); @@ -177,7 +180,9 @@ private FlagEvalEvent nextEvent(final ProducerCursor c) { targetingKeys[Math.floorMod(i, targetingKeys.length)], null, 1_700_000_000_000L + i, - attrs); + true, + attrs, + estimatedContextRetainedBytes); } private static String[] keys(final String prefix, final int count) { diff --git a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java index e22d05bee07..ca7f8d66e9e 100644 --- a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java +++ b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java @@ -6,6 +6,7 @@ import datadog.communication.BackendApiFactory; import datadog.trace.api.Config; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEventMemoryEstimator; import java.util.HashMap; import java.util.Map; import org.openjdk.jmh.annotations.Benchmark; @@ -57,6 +58,7 @@ public class FlagEvaluationHotPathBenchmark { public String profile; private Map attrs; + private long estimatedContextRetainedBytes; private String[] flagKeys; private String[] targetingKeys; private int cursor; @@ -71,6 +73,7 @@ public void setUp() { for (int i = 0; i < p.numFields; i++) { attrs.put("field" + i, "value"); } + estimatedContextRetainedBytes = FlagEvalEventMemoryEstimator.retainedContextBytes(attrs); flagKeys = keys("bench-flag-", p.numFlags); targetingKeys = keys("bench-user-", p.numUsers); cursor = 0; @@ -121,7 +124,8 @@ private FlagEvalEvent nextEvent() { null, 1_700_000_000_000L + i, true, - attrs); + attrs, + estimatedContextRetainedBytes); } private static String[] keys(final String prefix, final int count) { diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java index e15666aa10a..4c8058419e3 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java @@ -42,19 +42,18 @@ * globalCap=131072, perFlagCap=10000, degradedCap=32768. Eval-time: min/max of * firstEvalMs/lastEvalMs across events in the same bucket. Runtime default: absent variant means * runtimeDefaultUsed=true. Flush interval: 10 seconds. Queue: bounded MessagePassingBlockingQueue - * (capacity 2^16), non-blocking offer; on overflow the event is dropped and the - * droppedQueueOverflow counter is incremented and surfaced on flush. Enqueue: lock-free. Producers - * contend only on the MPSC queue, never on a monitor, so evaluation threads do not serialize - * against each other. Shutdown: close() drains the queue and performs a final flush before the - * worker thread exits. Because enqueue is lock-free, a producer can still offer during shutdown; - * close() sweeps the queue once the worker has been joined, counting any remainder as a closed drop - * so shutdown loss is observable rather than silent. + * (capacity 4096) with a 16 MiB estimated retained-byte budget. Enqueue uses non-blocking count and + * byte admission. Shutdown drains the queue and performs a final flush before the worker exits. + * Because enqueue is lock-free, a producer can still offer during shutdown; close() sweeps the + * queue once the worker has been joined. The sweep releases byte reservations and reports closed + * drops. */ public class FlagEvaluationWriterImpl implements FlagEvaluationWriter { private static final Logger LOGGER = LoggerFactory.getLogger(FlagEvaluationWriterImpl.class); static final int DEFAULT_CAPACITY = 1 << 12; // 4096 elements, per cross-SDK RFC + static final long DEFAULT_QUEUE_RETAINED_BYTE_BUDGET = 16L << 20; static final int FLUSH_INTERVAL_SECONDS = 10; static final int FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES = EvpProxy.PAYLOAD_SIZE_LIMIT_BYTES; @@ -63,6 +62,7 @@ public class FlagEvaluationWriterImpl implements FlagEvaluationWriter { static final String FLAG_EVALUATION_SPLITS_METRIC = "flagevaluation.payload.splits"; static final String FLAG_EVALUATION_CONTEXT_TRUNCATED_METRIC = "flagevaluation.context.truncated"; static final String DROP_REASON_QUEUE_OVERFLOW = "queue_overflow"; + static final String DROP_REASON_QUEUE_BYTE_BUDGET = "queue_byte_budget"; static final String DROP_REASON_CLOSED = "closed"; static final String DROP_REASON_DEGRADED_CAP = "degraded_cap"; static final String DROP_REASON_PAYLOAD_LIMIT = "payload_limit"; @@ -72,6 +72,7 @@ public class FlagEvaluationWriterImpl implements FlagEvaluationWriter { private static final CoreMetricCollector CORE_METRICS = CoreMetricCollector.getInstance(); private final MessagePassingBlockingQueue queue; + private final QueueByteBudget queueByteBudget; private final FlagEvaluationSerializingHandler serializer; private final Thread serializerThread; private final Object lifecycleLock = new Object(); @@ -90,6 +91,8 @@ private static void countMetric(final String metricName, final long value, final */ private final AtomicLong droppedQueueOverflow = new AtomicLong(0); + private final AtomicLong droppedQueueByteBudget = new AtomicLong(0); + /** * Per-reason-tag counters for evaluations whose context was truncated by copyPrunedContext. Keyed * by the sorted comma-separated reason string (e.g. "max_key_length,max_value_length"). @@ -118,15 +121,34 @@ public FlagEvaluationWriterImpl(final SharedCommunicationObjects sco, final Conf final TimeUnit timeUnit, final BackendApiFactory backendApiFactory, final Config config) { + this( + capacity, + flushInterval, + timeUnit, + backendApiFactory, + config, + DEFAULT_QUEUE_RETAINED_BYTE_BUDGET); + } + + FlagEvaluationWriterImpl( + final int capacity, + final long flushInterval, + final TimeUnit timeUnit, + final BackendApiFactory backendApiFactory, + final Config config, + final long queueRetainedByteBudget) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); + this.queueByteBudget = new QueueByteBudget(queueRetainedByteBudget); this.serializer = new FlagEvaluationSerializingHandler( backendApiFactory, queue, + queueByteBudget, flushInterval, timeUnit, FeatureFlagEvpContext.from(config), droppedQueueOverflow, + droppedQueueByteBudget, contextTruncatedCounts, this::close); this.serializerThread = newAgentThread(FEATURE_FLAG_EVALUATION_PROCESSOR, serializer); @@ -170,13 +192,13 @@ public void close() { FeatureFlaggingGateway.setFlagEvaluationEnqueueEnabled(false); FeatureFlaggingGateway.setFlagEvalWriter(null); workerRunning = this.serializerThread.isAlive(); - if (workerRunning) { - // Ask the worker to drain the queue and final-flush, then interrupt to wake it from poll(). - serializer.requestShutdown(); - this.serializerThread.interrupt(); - } + } + if (workerRunning) { + serializer.requestShutdown(); + this.serializerThread.interrupt(); } if (Thread.currentThread() == this.serializerThread) { + sweepAndCountResidualEvents(); return; } if (workerRunning) { @@ -203,9 +225,21 @@ public void close() { */ private void sweepAndCountResidualEvents() { long residual = 0; - while (queue.poll() != null) { - residual++; + while (true) { + FlagEvalEvent queued; + while ((queued = queue.poll()) != null) { + queueByteBudget.release(queued.estimatedRetainedBytes); + residual++; + } + // A producer reserves bytes before its final closed check. A nonzero value with an empty + // queue therefore identifies a producer that can still cancel or complete an offer. A + // producer that reserves after this zero check observes closed and cannot offer. + if (queueByteBudget.retainedBytes() == 0) { + break; + } + Thread.yield(); } + queueByteBudget.flushReleases(); countMetric(FLAG_EVALUATION_DROPPED_METRIC, residual, DROP_REASON_CLOSED); } @@ -214,24 +248,27 @@ public void enqueue(final FlagEvalEvent event) { if (event == null) { return; } + // Reserve before the closed check. The reservation is also the shutdown admission token: a + // producer can offer only while its reservation remains visible to close(). + final long retainedBytes = event.estimatedRetainedBytes; + if (!queueByteBudget.reserve(retainedBytes)) { + if (isClosedOrEnqueueDisabled()) { + countClosedDrop(); + } else { + droppedQueueByteBudget.incrementAndGet(); + } + return; + } if (isClosedOrEnqueueDisabled()) { + queueByteBudget.cancel(retainedBytes); countClosedDrop(); return; } - // Deliberately lock-free: the hand-off queue is MPSC by design, so serializing producers on a - // monitor here would negate that and turn every evaluation in every application thread into - // contention on one lock. A producer that passed the check above can still offer after close() - // has started; that residue is accounted for by the worker's bounded post-drain passes and by - // close()'s post-join sweep, so shutdown loss stays observable. - // - // Safe publication of the event (including the context snapshot built by the hook) comes from - // the queue's own offer/poll ordering, not from any monitor held here. - // - // Non-blocking offer. Count overflow so loss is observable rather than silent; the count is - // surfaced on the next flush. The hook's pre-queue guard (see FlagEvalLoggingHook) samples the - // queue depth before doing any context-copy work, so a saturated queue costs an - // AtomicInteger.get(); the offer here still races with the worker and can legitimately fail. + // Deliberately lock-free: the hand-off queue is MPSC by design, so application threads do not + // serialize on a monitor. The queue's offer/poll ordering safely publishes the event and its + // context snapshot. The count capacity stays as a secondary, non-blocking bound. if (!queue.offer(event)) { + queueByteBudget.cancel(retainedBytes); droppedQueueOverflow.incrementAndGet(); } } @@ -274,9 +311,22 @@ long droppedQueueOverflow() { return droppedQueueOverflow.get(); } + long droppedQueueByteBudget() { + return droppedQueueByteBudget.get(); + } + + long queuedRetainedBytes() { + return queueByteBudget.retainedBytes(); + } + /** Test seam: returns one queued event without starting the worker. */ FlagEvalEvent pollQueuedEventForTest() { - return queue.poll(); + final FlagEvalEvent queued = queue.poll(); + if (queued == null) { + return null; + } + queueByteBudget.release(queued.estimatedRetainedBytes); + return queued; } /** Test seam: flushes serializer state without starting the worker. */ @@ -288,6 +338,7 @@ void flushForTest() { static class FlagEvaluationSerializingHandler implements Runnable { private final MessagePassingBlockingQueue queue; + private final QueueByteBudget queueByteBudget; private final long ticksRequiredToFlush; @SuppressFBWarnings( @@ -299,6 +350,7 @@ static class FlagEvaluationSerializingHandler implements Runnable { evpPublisher; final Map context; private final AtomicLong droppedQueueOverflow; + private final AtomicLong droppedQueueByteBudget; private final ConcurrentHashMap contextTruncatedCounts; private final Runnable errorCallback; private final int payloadSizeLimitBytes; @@ -311,19 +363,23 @@ static class FlagEvaluationSerializingHandler implements Runnable { FlagEvaluationSerializingHandler( final BackendApiFactory backendApiFactory, final MessagePassingBlockingQueue queue, + final QueueByteBudget queueByteBudget, final long flushInterval, final TimeUnit timeUnit, final Map context, final AtomicLong droppedQueueOverflow, + final AtomicLong droppedQueueByteBudget, final ConcurrentHashMap contextTruncatedCounts, final Runnable errorCallback) { this( backendApiFactory, queue, + queueByteBudget, flushInterval, timeUnit, context, droppedQueueOverflow, + droppedQueueByteBudget, contextTruncatedCounts, errorCallback, FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); @@ -332,19 +388,23 @@ static class FlagEvaluationSerializingHandler implements Runnable { FlagEvaluationSerializingHandler( final BackendApiFactory backendApiFactory, final MessagePassingBlockingQueue queue, + final QueueByteBudget queueByteBudget, final long flushInterval, final TimeUnit timeUnit, final Map context, final AtomicLong droppedQueueOverflow, + final AtomicLong droppedQueueByteBudget, final ConcurrentHashMap contextTruncatedCounts, final Runnable errorCallback, final int payloadSizeLimitBytes) { this.queue = queue; + this.queueByteBudget = queueByteBudget; this.evpPublisher = new FeatureFlagEvpPublisher<>( backendApiFactory, FlagEvaluationPayloads.FlagEvaluationsRequest.class, false); this.context = context; this.droppedQueueOverflow = droppedQueueOverflow; + this.droppedQueueByteBudget = droppedQueueByteBudget; this.contextTruncatedCounts = contextTruncatedCounts; this.payloadSizeLimitBytes = payloadSizeLimitBytes; this.lastTicks = System.nanoTime(); @@ -395,19 +455,24 @@ public void run() { private void runDutyCycle() throws InterruptedException { final Thread thread = Thread.currentThread(); while (!thread.isInterrupted() && !shutdownRequested.get()) { - final FlagEvalEvent event = queue.poll(100, TimeUnit.MILLISECONDS); - if (event != null) { - aggregateEvent(event); + final FlagEvalEvent queued = queue.poll(100, TimeUnit.MILLISECONDS); + if (queued != null) { + queueByteBudget.release(queued.estimatedRetainedBytes); + aggregateEvent(queued); + } else { + queueByteBudget.flushReleases(); } flushIfNecessary(); } } void drainAndFlush() { - FlagEvalEvent event; - while ((event = queue.poll()) != null) { - aggregateEvent(event); + FlagEvalEvent queued; + while ((queued = queue.poll()) != null) { + queueByteBudget.release(queued.estimatedRetainedBytes); + aggregateEvent(queued); } + queueByteBudget.flushReleases(); flush(); } @@ -441,6 +506,14 @@ void flush() { + " (best-effort telemetry)", qDrops); } + final long byteDrops = droppedQueueByteBudget.getAndSet(0); + countMetric(FLAG_EVALUATION_DROPPED_METRIC, byteDrops, DROP_REASON_QUEUE_BYTE_BUDGET); + if (byteDrops > 0) { + LOGGER.warn( + "flag evaluation queue byte budget full - dropped {} evaluation(s)" + + " (best-effort telemetry)", + byteDrops); + } final long dgDrops = aggregator.droppedDegradedOverflow.getAndSet(0); countMetric(FLAG_EVALUATION_DROPPED_METRIC, dgDrops, DROP_REASON_DEGRADED_CAP); if (dgDrops > 0) { @@ -524,7 +597,9 @@ private List buildEventList() { } private boolean shouldFlush() { - if (aggregator.isEmpty() && droppedQueueOverflow.get() == 0) { + if (aggregator.isEmpty() + && droppedQueueOverflow.get() == 0 + && droppedQueueByteBudget.get() == 0) { return false; } final long nanoTime = System.nanoTime(); @@ -556,10 +631,12 @@ static class SerializingHandlerForTest extends FlagEvaluationSerializingHandler super( factory, Queues.mpscBlockingConsumerArrayQueue(DEFAULT_CAPACITY), + new QueueByteBudget(DEFAULT_QUEUE_RETAINED_BYTE_BUDGET), Long.MAX_VALUE, // effectively never auto-flush TimeUnit.NANOSECONDS, context, new AtomicLong(0), + new AtomicLong(0), new ConcurrentHashMap<>(), () -> {}, payloadSizeLimitBytes); @@ -628,4 +705,61 @@ static SerializingHandlerForTest createHandlerForTest( final int payloadSizeLimitBytes) { return new SerializingHandlerForTest(factory, context, payloadSizeLimitBytes); } + + static final class QueueByteBudget { + private static final long RELEASE_BATCH_BYTES = 64L << 10; + + private final long limitBytes; + private final AtomicLong availableBytes; + + // Only the queue's single consumer updates this field. Batching prevents a cache-line handoff + // between the application producer and worker threads for every event. Admission remains + // conservative until a batch is published. + @SuppressFBWarnings( + value = {"AT_NONATOMIC_64BIT_PRIMITIVE", "AT_NONATOMIC_OPERATIONS_ON_SHARED_VARIABLE"}, + justification = "Only the queue's single consumer reads or writes this field") + private long pendingReleasedBytes; + + QueueByteBudget(final long limitBytes) { + this.limitBytes = Math.max(0, limitBytes); + this.availableBytes = new AtomicLong(this.limitBytes); + } + + boolean reserve(final long bytes) { + long available; + do { + available = availableBytes.get(); + if (bytes > available) { + return false; + } + } while (!availableBytes.compareAndSet(available, available - bytes)); + return true; + } + + void release(final long bytes) { + pendingReleasedBytes += bytes; + if (pendingReleasedBytes >= RELEASE_BATCH_BYTES) { + flushReleases(); + } + } + + void cancel(final long bytes) { + // A producer calls this method when the queue rejects an event after byte reservation. + // Publish the cancellation directly because pendingReleasedBytes is consumer-confined. + availableBytes.addAndGet(bytes); + } + + void flushReleases() { + final long pending = pendingReleasedBytes; + if (pending == 0) { + return; + } + pendingReleasedBytes = 0; + availableBytes.addAndGet(pending); + } + + long retainedBytes() { + return limitBytes - availableBytes.get() - pendingReleasedBytes; + } + } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationEventMemoryEstimatorTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationEventMemoryEstimatorTest.java new file mode 100644 index 00000000000..bb918cae157 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationEventMemoryEstimatorTest.java @@ -0,0 +1,99 @@ +package com.datadog.featureflag; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEventMemoryEstimator; +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class FlagEvaluationEventMemoryEstimatorTest { + + @Test + void estimatesScalarOnlyEvent() { + final FlagEvalEvent event = + new FlagEvalEvent("flag", "on", "allocation", "subject", "error", 1L, false, null); + + assertEquals( + 128 + + stringBytes("flag") + + stringBytes("on") + + stringBytes("allocation") + + stringBytes("subject") + + stringBytes("error"), + FlagEvalEventMemoryEstimator.retainedBytes(event)); + } + + @Test + void estimatesEverySupportedContextValueShape() { + final Map attrs = new LinkedHashMap<>(); + attrs.put("string", "value"); + attrs.put("number", 42L); + attrs.put("empty", null); + final FlagEvalEvent event = new FlagEvalEvent("flag", null, null, null, null, 1L, true, attrs); + + final long expectedContextBytes = + 64 + + 48 + + stringBytes("string") + + stringBytes("value") + + 48 + + stringBytes("number") + + 128 + + 48 + + stringBytes("empty"); + assertEquals( + 128 + stringBytes("flag") + expectedContextBytes, + FlagEvalEventMemoryEstimator.retainedBytes(event)); + } + + @Test + void selectedBudgetKeepsTypicalCapacityAndBoundsWideContexts() { + final long budget = FlagEvaluationWriterImpl.DEFAULT_QUEUE_RETAINED_BYTE_BUDGET; + final long typicalBytes = + FlagEvalEventMemoryEstimator.retainedBytes(eventWithContext(10, 16, 32)); + final long nestedBytes = + FlagEvalEventMemoryEstimator.retainedBytes(eventWithContext(100, 32, 32)); + final long maximumBytes = + FlagEvalEventMemoryEstimator.retainedBytes(eventWithContext(256, 256, 256)); + + assertEquals(2_880, typicalBytes); + assertEquals(26_240, nestedBytes); + assertEquals(295_552, maximumBytes); + assertEquals(4_096, Math.min(4_096, budget / typicalBytes)); + assertEquals(639, budget / nestedBytes); + assertEquals(56, budget / maximumBytes); + } + + private static FlagEvalEvent eventWithContext( + final int fields, final int keyLength, final int valueLength) { + final Map attrs = new LinkedHashMap<>(); + for (int field = 0; field < fields; field++) { + attrs.put( + fixed("key-" + field + "-", keyLength, 'k'), + fixed("value-" + field + "-", valueLength, 'v')); + } + return new FlagEvalEvent( + fixed("flag-", 32, 'f'), + fixed("variant-", 16, 'v'), + fixed("allocation-", 32, 'a'), + fixed("target-", 64, 't'), + null, + 1L, + true, + attrs); + } + + private static String fixed(final String prefix, final int length, final char fillCharacter) { + final StringBuilder value = new StringBuilder(length).append(prefix); + while (value.length() < length) { + value.append(fillCharacter); + } + return value.substring(0, length); + } + + private static long stringBytes(final String value) { + return 40L + 2L * value.length(); + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java index b354fdbb6c4..d7b15f42bb8 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java @@ -34,6 +34,7 @@ import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEventMemoryEstimator; import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; import datadog.trace.api.intake.Intake; import datadog.trace.api.telemetry.CoreMetricCollector; @@ -101,6 +102,7 @@ void startRegistersWriterAndCloseDeregistersIt() { writer.close(); writer.close(); writer.start(); + writer.startForTest(); assertNull(FeatureFlaggingGateway.getFlagEvalWriter()); } @@ -130,6 +132,164 @@ void queueOverflowIncrementsObservableDropCounter() { "reason:" + FlagEvaluationWriterImpl.DROP_REASON_QUEUE_OVERFLOW)); } + @Test + void queueByteBudgetDropsBeforeCountCapacityAndReportsSeparateReason() { + final BackendApi mockEvp = mock(BackendApi.class); + final BackendApiFactory factory = mock(BackendApiFactory.class); + when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); + final FlagEvalEvent event = simpleEvent("byte-budget-flag", "on"); + final long eventBytes = FlagEvalEventMemoryEstimator.retainedBytes(event); + final FlagEvaluationWriterImpl writer = + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg(), eventBytes); + + writer.enqueue(event); + writer.enqueue(event); + + assertEquals(eventBytes, writer.queuedRetainedBytes()); + assertEquals(1, writer.droppedQueueByteBudget()); + assertEquals(0, writer.droppedQueueOverflow()); + assertNotNull(writer.pollQueuedEventForTest()); + assertEquals(0, writer.queuedRetainedBytes()); + assertNull(writer.pollQueuedEventForTest()); + + writer.flushForTest(); + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + 1, + metricSum( + metrics, + FlagEvaluationWriterImpl.FLAG_EVALUATION_DROPPED_METRIC, + "reason:" + FlagEvaluationWriterImpl.DROP_REASON_QUEUE_BYTE_BUDGET)); + } + + @Test + void concurrentByteAdmissionNeverExceedsTheBudget() throws Exception { + final BackendApi mockEvp = mock(BackendApi.class); + final BackendApiFactory factory = mock(BackendApiFactory.class); + when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); + final FlagEvalEvent event = simpleEvent("concurrent-byte-budget-flag", "on"); + final long eventBytes = FlagEvalEventMemoryEstimator.retainedBytes(event); + final int admittedEvents = 64; + final long budget = eventBytes * admittedEvents; + final FlagEvaluationWriterImpl writer = + new FlagEvaluationWriterImpl( + 1 << 12, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg(), budget); + final int producers = 16; + final int offersPerProducer = 100; + final java.util.concurrent.ExecutorService executor = + java.util.concurrent.Executors.newFixedThreadPool(producers); + final java.util.concurrent.CountDownLatch start = new java.util.concurrent.CountDownLatch(1); + final AtomicLong admitted = new AtomicLong(); + + for (int producer = 0; producer < producers; producer++) { + executor.submit( + () -> { + start.await(); + for (int offer = 0; offer < offersPerProducer; offer++) { + writer.enqueue(event); + } + return null; + }); + } + start.countDown(); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + + assertEquals(budget, writer.queuedRetainedBytes()); + assertEquals(producers * offersPerProducer - admittedEvents, writer.droppedQueueByteBudget()); + int drained = 0; + while (writer.pollQueuedEventForTest() != null) { + drained++; + } + assertEquals(admittedEvents, drained); + assertEquals(0, writer.queuedRetainedBytes()); + } + + @Test + void concurrentByteReservationsKeepExactAccounting() throws Exception { + final int producers = 64; + final int reservationsPerProducer = 10_000; + final long totalReservations = (long) producers * reservationsPerProducer; + final FlagEvaluationWriterImpl.QueueByteBudget budget = + new FlagEvaluationWriterImpl.QueueByteBudget(totalReservations); + final java.util.concurrent.ExecutorService executor = + java.util.concurrent.Executors.newFixedThreadPool(producers); + final java.util.concurrent.CountDownLatch start = new java.util.concurrent.CountDownLatch(1); + final AtomicLong admitted = new AtomicLong(); + + for (int producer = 0; producer < producers; producer++) { + executor.submit( + () -> { + start.await(); + for (int reservation = 0; reservation < reservationsPerProducer; reservation++) { + if (budget.reserve(1)) { + admitted.incrementAndGet(); + } + } + return null; + }); + } + start.countDown(); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + + assertEquals(totalReservations, admitted.get()); + assertEquals(totalReservations, budget.retainedBytes()); + budget.release(totalReservations); + assertEquals(0, budget.retainedBytes()); + } + + @Test + void byteReleasePublishesFullBatchesAndFlushesPartialBatches() { + final FlagEvaluationWriterImpl.QueueByteBudget budget = + new FlagEvaluationWriterImpl.QueueByteBudget(100_000); + + assertTrue(budget.reserve(70_000)); + budget.release(70_000); + assertEquals(0, budget.retainedBytes()); + + assertTrue(budget.reserve(10)); + budget.release(10); + assertEquals(0, budget.retainedBytes()); + budget.flushReleases(); + assertEquals(0, budget.retainedBytes()); + } + + @Test + void concurrentProducerCancellationsKeepExactAccounting() throws Exception { + final int producers = 16; + final int reservationsPerProducer = 10_000; + final long totalReservations = (long) producers * reservationsPerProducer; + final FlagEvaluationWriterImpl.QueueByteBudget budget = + new FlagEvaluationWriterImpl.QueueByteBudget(totalReservations); + final java.util.concurrent.ExecutorService executor = + java.util.concurrent.Executors.newFixedThreadPool(producers); + final java.util.concurrent.CountDownLatch start = new java.util.concurrent.CountDownLatch(1); + final AtomicLong admitted = new AtomicLong(); + + for (int producer = 0; producer < producers; producer++) { + executor.submit( + () -> { + start.await(); + for (int reservation = 0; reservation < reservationsPerProducer; reservation++) { + if (budget.reserve(1)) { + admitted.incrementAndGet(); + budget.cancel(1); + } + } + return null; + }); + } + start.countDown(); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + + assertEquals(totalReservations, admitted.get()); + assertEquals(0, budget.retainedBytes()); + } + @Test void enqueueAfterCloseIsDroppedAndCounted() { final BackendApi mockEvp = mock(BackendApi.class); @@ -199,6 +359,7 @@ void closeSweepsAndCountsEventsLeftInTheQueue() { metrics, FlagEvaluationWriterImpl.FLAG_EVALUATION_DROPPED_METRIC, "reason:" + FlagEvaluationWriterImpl.DROP_REASON_CLOSED)); + assertEquals(0, writer.queuedRetainedBytes()); assertNull(writer.pollQueuedEventForTest()); } @@ -259,11 +420,14 @@ void flushIfNecessaryDoesNotReturnEarlyWhenOnlyQueueDropsArePending() { final FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler handler = new FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler( mock(BackendApiFactory.class), - Queues.mpscBlockingConsumerArrayQueue(16), + Queues.mpscBlockingConsumerArrayQueue(16), + new FlagEvaluationWriterImpl.QueueByteBudget( + FlagEvaluationWriterImpl.DEFAULT_QUEUE_RETAINED_BYTE_BUDGET), Long.MAX_VALUE, TimeUnit.NANOSECONDS, context(), queueDrops, + new AtomicLong(0), new java.util.concurrent.ConcurrentHashMap<>(), () -> {}); @@ -272,6 +436,28 @@ void flushIfNecessaryDoesNotReturnEarlyWhenOnlyQueueDropsArePending() { assertEquals(1, queueDrops.get()); } + @Test + void flushIfNecessaryDoesNotReturnEarlyWhenOnlyQueueByteDropsArePending() { + final AtomicLong queueByteDrops = new AtomicLong(1); + final FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler handler = + new FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler( + mock(BackendApiFactory.class), + Queues.mpscBlockingConsumerArrayQueue(16), + new FlagEvaluationWriterImpl.QueueByteBudget( + FlagEvaluationWriterImpl.DEFAULT_QUEUE_RETAINED_BYTE_BUDGET), + Long.MAX_VALUE, + TimeUnit.NANOSECONDS, + context(), + new AtomicLong(0), + queueByteDrops, + new java.util.concurrent.ConcurrentHashMap<>(), + () -> {}); + + handler.flushIfNecessary(); + + assertEquals(1, queueByteDrops.get()); + } + @Test @SuppressWarnings("unchecked") void workerHandlesEmptyPolls() throws Exception { @@ -290,10 +476,13 @@ void workerHandlesEmptyPolls() throws Exception { new FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler( factory, queue, + new FlagEvaluationWriterImpl.QueueByteBudget( + FlagEvaluationWriterImpl.DEFAULT_QUEUE_RETAINED_BYTE_BUDGET), Long.MAX_VALUE, TimeUnit.NANOSECONDS, context(), new AtomicLong(0), + new AtomicLong(0), new java.util.concurrent.ConcurrentHashMap<>(), () -> {});