diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationAggregator.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationAggregator.java index 5d0eeaa0e82..94ce5655a11 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationAggregator.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationAggregator.java @@ -4,6 +4,7 @@ import static datadog.trace.util.HashingUtils.hash; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -11,6 +12,9 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +@SuppressFBWarnings( + value = {"AT_NONATOMIC_64BIT_PRIMITIVE", "AT_NONATOMIC_OPERATIONS_ON_SHARED_VARIABLE"}, + justification = "The aggregator is confined to the single flag-evaluation serializer thread") final class FlagEvaluationAggregator { // Design assumptions — document the scale we sized for @@ -32,6 +36,7 @@ final class FlagEvaluationAggregator { static final int GLOBAL_CAP = 131_072; // nearest power of two above FULL_BUCKET_SIZING_BASIS static final int PER_FLAG_CAP = PER_FLAG_BUCKET_SIZING_BASIS; static final int DEGRADED_CAP = 32_768; // nearest power of two above DEGRADED_BUCKET_SIZING_BASIS + static final long RETAINED_BYTE_BUDGET = 64L << 20; private static final byte CTX_TAG_STRING = 's'; private static final byte CTX_TAG_BOOL = 'b'; @@ -45,7 +50,20 @@ final class FlagEvaluationAggregator { final Map degradedTier = new HashMap<>(); final Map perFlagCount = new HashMap<>(); final AtomicLong droppedDegradedOverflow = new AtomicLong(0); + final AtomicLong droppedByteBudget = new AtomicLong(0); + final AtomicLong degradedCardinalityCap = new AtomicLong(0); + final AtomicLong degradedByteBudget = new AtomicLong(0); final AtomicInteger globalFullCount = new AtomicInteger(0); + private final long retainedByteBudget; + private long retainedBytes; + + FlagEvaluationAggregator() { + this(RETAINED_BYTE_BUDGET); + } + + FlagEvaluationAggregator(final long retainedByteBudget) { + this.retainedByteBudget = Math.max(0, retainedByteBudget); + } void aggregate(final FlagEvalEvent event) { final boolean isDefault = event.variant == null; @@ -66,50 +84,78 @@ void aggregate(final FlagEvalEvent event) { final int flagCount = perFlagCount.getOrDefault(event.flagKey, 0); if (globalFullCount.get() < GLOBAL_CAP && flagCount < PER_FLAG_CAP) { - fullTier.put( - fullKey, - new EvalBucket( - event.flagKey, - event.variant, - event.allocationKey, - event.targetingKey, - event.errorMessage, - event.evalTimeMs, - isDefault, - prunedAttrs, - observeFullEvaluationData)); - globalFullCount.incrementAndGet(); - perFlagCount.put(event.flagKey, flagCount + 1); - return; + final long bucketBytes = + FlagEvaluationMemoryEstimator.fullBucketBytes(event, prunedAttrs, ctxKey); + if (reserve(bucketBytes)) { + fullTier.put( + fullKey, + new EvalBucket( + event.flagKey, + event.variant, + event.allocationKey, + event.targetingKey, + event.errorMessage, + event.evalTimeMs, + isDefault, + prunedAttrs, + observeFullEvaluationData)); + globalFullCount.incrementAndGet(); + perFlagCount.put(event.flagKey, flagCount + 1); + return; + } } + final boolean degradedByByteBudget = + globalFullCount.get() < GLOBAL_CAP && flagCount < PER_FLAG_CAP; final DegradedKey degradedKey = buildDegradedKey(event); bucket = degradedTier.get(degradedKey); if (bucket != null) { bucket.merge(event.evalTimeMs, isDefault); bucket.observeFullEvaluationData &= observeFullEvaluationData; + countDegraded(degradedByByteBudget); return; } if (degradedTier.size() < DEGRADED_CAP) { - degradedTier.put( - degradedKey, - new EvalBucket( - event.flagKey, - event.variant, - event.allocationKey, - null, - event.errorMessage, - event.evalTimeMs, - isDefault, - null, - observeFullEvaluationData)); + if (reserve(FlagEvaluationMemoryEstimator.degradedBucketBytes(event))) { + degradedTier.put( + degradedKey, + new EvalBucket( + event.flagKey, + event.variant, + event.allocationKey, + null, + event.errorMessage, + event.evalTimeMs, + isDefault, + null, + observeFullEvaluationData)); + countDegraded(degradedByByteBudget); + return; + } + droppedByteBudget.incrementAndGet(); return; } droppedDegradedOverflow.incrementAndGet(); } + private void countDegraded(final boolean degradedByByteBudget) { + if (degradedByByteBudget) { + degradedByteBudget.incrementAndGet(); + } else { + degradedCardinalityCap.incrementAndGet(); + } + } + + private boolean reserve(final long bytes) { + if (bytes > retainedByteBudget - retainedBytes) { + return false; + } + retainedBytes += bytes; + return true; + } + boolean isEmpty() { return fullTier.isEmpty() && degradedTier.isEmpty(); } @@ -118,14 +164,6 @@ int fullTierSize() { return fullTier.size(); } - long degradedEvaluationCount() { - long count = 0; - for (final EvalBucket bucket : degradedTier.values()) { - count += bucket.count; - } - return count; - } - int bucketCount() { return fullTier.size() + degradedTier.size(); } @@ -143,11 +181,22 @@ void clear() { degradedTier.clear(); perFlagCount.clear(); globalFullCount.set(0); + retainedBytes = 0; + } + + long retainedBytes() { + return retainedBytes; } AggregatedState snapshot() { return new AggregatedState( - new HashMap<>(fullTier), new HashMap<>(degradedTier), droppedDegradedOverflow.get()); + new HashMap<>(fullTier), + new HashMap<>(degradedTier), + droppedDegradedOverflow.get(), + droppedByteBudget.get(), + degradedCardinalityCap.get(), + degradedByteBudget.get(), + retainedBytes); } void simulateFullTierAtCap() { @@ -441,14 +490,26 @@ static class AggregatedState { final Map fullTier; final Map degradedTier; final long droppedDegradedOverflow; + final long droppedByteBudget; + final long degradedCardinalityCap; + final long degradedByteBudget; + final long retainedBytes; AggregatedState( final Map fullTier, final Map degradedTier, - final long droppedDegradedOverflow) { + final long droppedDegradedOverflow, + final long droppedByteBudget, + final long degradedCardinalityCap, + final long degradedByteBudget, + final long retainedBytes) { this.fullTier = fullTier; this.degradedTier = degradedTier; this.droppedDegradedOverflow = droppedDegradedOverflow; + this.droppedByteBudget = droppedByteBudget; + this.degradedCardinalityCap = degradedCardinalityCap; + this.degradedByteBudget = degradedByteBudget; + this.retainedBytes = retainedBytes; } } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationMemoryEstimator.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationMemoryEstimator.java new file mode 100644 index 00000000000..7ff9968a3b1 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationMemoryEstimator.java @@ -0,0 +1,74 @@ +package com.datadog.featureflag; + +import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import java.util.Map; + +/** Conservatively estimates memory retained by one aggregation bucket. */ +final class FlagEvaluationMemoryEstimator { + + // These constants include aligned object headers, references, and amortized HashMap storage. + // They intentionally overestimate JOL measurements made with 100-bucket representative graphs. + private static final long BUCKET_AND_INDEX_BYTES = 256; + 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 = 64; + private static final long MAX_BYTES_PER_CHARACTER = 2; + + private FlagEvaluationMemoryEstimator() {} + + static long fullBucketBytes( + final FlagEvalEvent event, + final Map prunedAttrs, + final String canonicalContextKey) { + long bytes = BUCKET_AND_INDEX_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)); + bytes = add(bytes, stringBytes(canonicalContextKey)); + if (prunedAttrs == null || prunedAttrs.isEmpty()) { + return bytes; + } + + bytes = add(bytes, CONTEXT_MAP_BYTES); + for (final Map.Entry entry : prunedAttrs.entrySet()) { + bytes = add(bytes, CONTEXT_ENTRY_BYTES); + bytes = add(bytes, stringBytes(entry.getKey())); + bytes = add(bytes, contextValueBytes(entry.getValue())); + } + return bytes; + } + + static long degradedBucketBytes(final FlagEvalEvent event) { + long bytes = BUCKET_AND_INDEX_BYTES; + bytes = add(bytes, stringBytes(event.flagKey)); + bytes = add(bytes, stringBytes(event.variant)); + bytes = add(bytes, stringBytes(event.allocationKey)); + return add(bytes, stringBytes(event.errorMessage)); + } + + private static long contextValueBytes(final Object value) { + if (value instanceof String) { + return stringBytes((String) value); + } + if (value == null) { + return 0; + } + return add(OTHER_VALUE_BYTES, characterBytes(value.toString().length())); + } + + 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-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..83c4290861b 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 @@ -40,15 +40,15 @@ * string identity). Context pruning: deterministic (sort before cut), <=256 fields, string values * <=256 chars; the pruned attributes are what gets aggregated and serialized. Caps: * 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. + * firstEvalMs/lastEvalMs across events in the same bucket. Retained aggregation memory is limited + * to 64 MiB. Runtime default: absent variant means runtimeDefaultUsed=true. Flush interval: 10 + * seconds. Queue: bounded MessagePassingBlockingQueue (capacity 2^12), 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. */ public class FlagEvaluationWriterImpl implements FlagEvaluationWriter { @@ -65,8 +65,10 @@ public class FlagEvaluationWriterImpl implements FlagEvaluationWriter { static final String DROP_REASON_QUEUE_OVERFLOW = "queue_overflow"; static final String DROP_REASON_CLOSED = "closed"; static final String DROP_REASON_DEGRADED_CAP = "degraded_cap"; + static final String DROP_REASON_BYTE_BUDGET = "byte_budget"; static final String DROP_REASON_PAYLOAD_LIMIT = "payload_limit"; static final String DEGRADED_REASON_CARDINALITY_CAP = "cardinality_cap"; + static final String DEGRADED_REASON_BYTE_BUDGET = "byte_budget"; static final String DEGRADED_REASON_PAYLOAD_LIMIT = "payload_limit"; private static final String FLAG_EVALUATION_ROUTE = "flagevaluation"; private static final CoreMetricCollector CORE_METRICS = CoreMetricCollector.getInstance(); @@ -449,6 +451,22 @@ void flush() { + " (best-effort telemetry)", dgDrops); } + final long byteBudgetDrops = aggregator.droppedByteBudget.getAndSet(0); + countMetric(FLAG_EVALUATION_DROPPED_METRIC, byteBudgetDrops, DROP_REASON_BYTE_BUDGET); + if (byteBudgetDrops > 0) { + LOGGER.warn( + "flag evaluation aggregation byte budget full - dropped {} evaluation(s)" + + " (best-effort telemetry)", + byteBudgetDrops); + } + countMetric( + FLAG_EVALUATION_DEGRADED_METRIC, + aggregator.degradedCardinalityCap.getAndSet(0), + DEGRADED_REASON_CARDINALITY_CAP); + countMetric( + FLAG_EVALUATION_DEGRADED_METRIC, + aggregator.degradedByteBudget.getAndSet(0), + DEGRADED_REASON_BYTE_BUDGET); // Drain per-reason context-truncation counters and emit one metric per unique reason tag. for (final Map.Entry entry : contextTruncatedCounts.entrySet()) { @@ -462,10 +480,6 @@ void flush() { return; } try { - countMetric( - FLAG_EVALUATION_DEGRADED_METRIC, - aggregator.degradedEvaluationCount(), - DEGRADED_REASON_CARDINALITY_CAP); final List events = buildEventList(); if (events.isEmpty()) { return; diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationAggregatorTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationAggregatorTest.java index 06faeadefea..a1b9ac16c80 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationAggregatorTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationAggregatorTest.java @@ -80,6 +80,110 @@ void degradedCapOverflowIncrementsDroppedCounter() { assertTrue(state.droppedDegradedOverflow > 0); } + @Test + void byteBudgetOverflowRoutesToDegradedTierAndReportsReason() { + final Map attrs = context(10, 16, 32); + final FlagEvalEvent event = event("byte-flag", "on", "alloc1", "user-1", 1000L, true, attrs); + final long fullBucketBytes = + FlagEvaluationMemoryEstimator.fullBucketBytes( + event, attrs, FlagEvaluationAggregator.canonicalContextKey(attrs)); + final long degradedBucketBytes = FlagEvaluationMemoryEstimator.degradedBucketBytes(event); + final FlagEvaluationAggregator aggregator = new FlagEvaluationAggregator(fullBucketBytes - 1); + + aggregator.aggregate(event); + + final FlagEvaluationAggregator.AggregatedState state = aggregator.snapshot(); + assertEquals(0, state.fullTier.size()); + assertEquals(1, state.degradedTier.size()); + assertEquals(1, state.degradedByteBudget); + assertEquals(0, state.degradedCardinalityCap); + assertEquals(0, state.droppedByteBudget); + assertEquals(degradedBucketBytes, state.retainedBytes); + } + + @Test + void byteBudgetDropsNewDegradedBucketWhenNoSpaceRemains() { + final Map attrs = context(10, 16, 32); + final FlagEvalEvent event = + event("drop-byte-flag", "on", "alloc1", "user-1", 1000L, true, attrs); + final long degradedBucketBytes = FlagEvaluationMemoryEstimator.degradedBucketBytes(event); + final FlagEvaluationAggregator aggregator = + new FlagEvaluationAggregator(degradedBucketBytes - 1); + + aggregator.aggregate(event); + + final FlagEvaluationAggregator.AggregatedState state = aggregator.snapshot(); + assertTrue(state.fullTier.isEmpty()); + assertTrue(state.degradedTier.isEmpty()); + assertEquals(1, state.droppedByteBudget); + assertEquals(0, state.degradedByteBudget); + assertEquals(0, state.retainedBytes); + } + + @Test + void existingBucketMergesWithoutReservingMoreBytes() { + final Map attrs = context(10, 16, 32); + final FlagEvalEvent event = + event("merge-byte-flag", "on", "alloc1", "user-1", 1000L, true, attrs); + final long fullBucketBytes = + FlagEvaluationMemoryEstimator.fullBucketBytes( + event, attrs, FlagEvaluationAggregator.canonicalContextKey(attrs)); + final FlagEvaluationAggregator aggregator = new FlagEvaluationAggregator(fullBucketBytes); + + aggregator.aggregate(event); + aggregator.aggregate(event); + + final FlagEvaluationAggregator.AggregatedState state = aggregator.snapshot(); + assertEquals(1, state.fullTier.size()); + assertEquals(2, state.fullTier.values().iterator().next().count); + assertEquals(fullBucketBytes, state.retainedBytes); + } + + @Test + void clearReleasesRetainedByteBudget() { + final FlagEvalEvent event = simpleEvent("clear-byte-flag", "on"); + final long fullBucketBytes = FlagEvaluationMemoryEstimator.fullBucketBytes(event, null, ""); + final FlagEvaluationAggregator aggregator = new FlagEvaluationAggregator(fullBucketBytes); + + aggregator.aggregate(event); + assertEquals(fullBucketBytes, aggregator.retainedBytes()); + + aggregator.clear(); + assertEquals(0, aggregator.retainedBytes()); + aggregator.aggregate(event); + assertEquals(1, aggregator.fullTierSize()); + } + + @Test + void estimatorCoversMeasuredJolProfiles() { + final Map typical = context(256, 16, 32); + final FlagEvalEvent typicalEvent = + event("flag", "on", "allocation", fixedLength("subject", 64), 1L, true, typical); + final Map maximum = context(256, 256, 256); + final FlagEvalEvent maximumEvent = + event("flag", "on", "allocation", fixedLength("subject", 64), 1L, true, maximum); + final FlagEvalEvent protectedEvent = + event("flag", "on", "allocation", fixedLength("subject", 64), 1L, false, maximum); + final Map nullValue = new HashMap<>(); + nullValue.put("nullable", null); + final FlagEvalEvent nullValueEvent = + event("flag", "on", "allocation", "subject", 1L, true, nullValue); + + assertTrue( + FlagEvaluationMemoryEstimator.fullBucketBytes( + typicalEvent, typical, FlagEvaluationAggregator.canonicalContextKey(typical)) + >= 60_021); + assertTrue( + FlagEvaluationMemoryEstimator.fullBucketBytes( + maximumEvent, maximum, FlagEvaluationAggregator.canonicalContextKey(maximum)) + >= 297_589); + assertTrue(FlagEvaluationMemoryEstimator.fullBucketBytes(protectedEvent, null, "") >= 253); + assertTrue( + FlagEvaluationMemoryEstimator.fullBucketBytes( + nullValueEvent, nullValue, FlagEvaluationAggregator.canonicalContextKey(nullValue)) + > 0); + } + @Test void perFlagCapOverflowRoutesToDegradedTierAndMergesSameDegradedKey() { final FlagEvaluationAggregator aggregator = new FlagEvaluationAggregator(); @@ -201,6 +305,7 @@ void capSizingUsesNamedScaleConstants() { assertEquals(131_072, FlagEvaluationAggregator.GLOBAL_CAP); assertEquals(10_000, FlagEvaluationAggregator.PER_FLAG_CAP); assertEquals(32_768, FlagEvaluationAggregator.DEGRADED_CAP); + assertEquals(64L << 20, FlagEvaluationAggregator.RETAINED_BYTE_BUDGET); } @Test @@ -394,6 +499,23 @@ private static FlagEvalEvent simpleEvent(final String flagKey, final String vari return event(flagKey, variant, "alloc1", "user-1", 1000L, emptyMap()); } + private static Map context( + final int fieldCount, final int keyLength, final int valueLength) { + final Map attrs = new HashMap<>(); + for (int field = 0; field < fieldCount; field++) { + attrs.put(fixedLength("key-" + field, keyLength), fixedLength("value-" + field, valueLength)); + } + return attrs; + } + + private static String fixedLength(final String prefix, final int length) { + final char[] value = new char[length]; + final int prefixLength = Math.min(prefix.length(), length); + prefix.getChars(0, prefixLength, value, 0); + Arrays.fill(value, prefixLength, length, 'x'); + return new String(value); + } + private static FlagEvaluationAggregator.FullKey fullKey( final String flagKey, final String variant, 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..f7efcadcd9f 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 @@ -87,6 +87,38 @@ void degradedCapOverflowTelemetryIsEmittedOnFlush() { "reason:" + FlagEvaluationWriterImpl.DROP_REASON_DEGRADED_CAP)); } + @Test + void byteBudgetTelemetrySeparatesDegradedAndDroppedEvaluations() { + final BackendApi mockEvp = mock(BackendApi.class); + final FlagEvaluationTestSupport.TestWriterSetup setup = buildTestWriter(mockEvp); + setup.handler.aggregator.degradedCardinalityCap.addAndGet(2); + setup.handler.aggregator.degradedByteBudget.addAndGet(3); + setup.handler.aggregator.droppedByteBudget.addAndGet(4); + + setup.handler.flush(); + + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + 2, + metricSum( + metrics, + FlagEvaluationWriterImpl.FLAG_EVALUATION_DEGRADED_METRIC, + "reason:" + FlagEvaluationWriterImpl.DEGRADED_REASON_CARDINALITY_CAP)); + assertEquals( + 3, + metricSum( + metrics, + FlagEvaluationWriterImpl.FLAG_EVALUATION_DEGRADED_METRIC, + "reason:" + FlagEvaluationWriterImpl.DEGRADED_REASON_BYTE_BUDGET)); + assertEquals( + 4, + metricSum( + metrics, + FlagEvaluationWriterImpl.FLAG_EVALUATION_DROPPED_METRIC, + "reason:" + FlagEvaluationWriterImpl.DROP_REASON_BYTE_BUDGET)); + } + @Test void startRegistersWriterAndCloseDeregistersIt() { final BackendApi mockEvp = mock(BackendApi.class);