diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index daf35c2c351..f8a281966af 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java @@ -28,7 +28,13 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni } public @Nullable BackendApi createBackendApi(Intake intake, boolean responseCompression) { - HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true); + return createBackendApi( + intake, responseCompression, new HttpRetryPolicy.Factory(5, 100, 2.0, true)); + } + + /** Creates a backend API with the retry policy required by the calling product. */ + public @Nullable BackendApi createBackendApi( + Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) { if (intake.isAgentlessEnabled(config)) { HttpUrl agentlessUrl = HttpUrl.get(intake.getAgentlessUrl(config)); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposurePayloads.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposurePayloads.java new file mode 100644 index 00000000000..e6fd843adac --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposurePayloads.java @@ -0,0 +1,169 @@ +package com.datadog.featureflag; + +import com.squareup.moshi.JsonAdapter; +import com.squareup.moshi.Moshi; +import com.squareup.moshi.Types; +import datadog.trace.api.featureflag.exposure.ExposureEvent; +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; + +final class ExposurePayloads { + + private static final byte[] PAYLOAD_SUFFIX = FeatureFlagEvpPublisher.utf8Bytes("]}"); + private static final byte[] JSON_COMMA = FeatureFlagEvpPublisher.utf8Bytes(","); + + private static final JsonAdapter EVENT_JSON_ADAPTER; + private static final JsonAdapter> CONTEXT_JSON_ADAPTER; + + static { + final Moshi moshi = new Moshi.Builder().build(); + EVENT_JSON_ADAPTER = moshi.adapter(ExposureEvent.class); + final Type contextType = Types.newParameterizedType(Map.class, String.class, String.class); + CONTEXT_JSON_ADAPTER = moshi.adapter(contextType); + } + + private ExposurePayloads() {} + + static EncodedPayloads buildPayloadsForTest( + final List events, + final Map context, + final int payloadSizeLimitBytes) { + final List payloads = new ArrayList<>(); + final EncodingResult result = + writePayloads(events, context, payloadSizeLimitBytes, payloads::add); + return new EncodedPayloads(payloads, result.droppedPayloadLimit, result.droppedSerialization); + } + + static EncodingResult writePayloads( + final List events, + final Map context, + final int payloadSizeLimitBytes, + final Consumer payloadConsumer) { + final byte[] prefix = payloadPrefix(context); + EncodedPayloadBuilder current = new EncodedPayloadBuilder(prefix); + long droppedPayloadLimit = 0; + long droppedSerialization = 0; + + for (final ExposureEvent event : events) { + final byte[] eventBytes; + try { + eventBytes = encodeEvent(event); + } catch (final RuntimeException ignored) { + droppedSerialization++; + continue; + } + if (!current.canAdd(eventBytes, payloadSizeLimitBytes) && !current.isEmpty()) { + payloadConsumer.accept(current.toPayload()); + current = new EncodedPayloadBuilder(prefix); + } + if (current.canAdd(eventBytes, payloadSizeLimitBytes)) { + current.add(eventBytes); + } else { + droppedPayloadLimit++; + } + } + + if (!current.isEmpty()) { + payloadConsumer.accept(current.toPayload()); + } + return new EncodingResult(droppedPayloadLimit, droppedSerialization); + } + + private static byte[] payloadPrefix(final Map context) { + return FeatureFlagEvpPublisher.utf8Bytes( + "{\"context\":" + CONTEXT_JSON_ADAPTER.toJson(context) + ",\"exposures\":["); + } + + private static byte[] encodeEvent(final ExposureEvent event) { + return FeatureFlagEvpPublisher.utf8Bytes(EVENT_JSON_ADAPTER.toJson(event)); + } + + static final class EncodedPayloads { + final List payloads; + final long droppedPayloadLimit; + final long droppedSerialization; + + private EncodedPayloads( + final List payloads, + final long droppedPayloadLimit, + final long droppedSerialization) { + this.payloads = payloads; + this.droppedPayloadLimit = droppedPayloadLimit; + this.droppedSerialization = droppedSerialization; + } + } + + static final class EncodingResult { + final long droppedPayloadLimit; + final long droppedSerialization; + + private EncodingResult(final long droppedPayloadLimit, final long droppedSerialization) { + this.droppedPayloadLimit = droppedPayloadLimit; + this.droppedSerialization = droppedSerialization; + } + } + + static final class EncodedPayload { + final byte[] body; + final int eventCount; + + private EncodedPayload(final byte[] body, final int eventCount) { + this.body = body; + this.eventCount = eventCount; + } + } + + private static final class EncodedPayloadBuilder { + private final byte[] prefix; + private final List events = new ArrayList<>(); + private int eventBytes; + + private EncodedPayloadBuilder(final byte[] prefix) { + this.prefix = prefix; + } + + private boolean isEmpty() { + return events.isEmpty(); + } + + private boolean canAdd(final byte[] event, final int payloadSizeLimitBytes) { + return sizeWith(event) <= payloadSizeLimitBytes; + } + + private long sizeWith(final byte[] event) { + return (long) prefix.length + + PAYLOAD_SUFFIX.length + + eventBytes + + event.length + + events.size(); + } + + private void add(final byte[] event) { + events.add(event); + eventBytes += event.length; + } + + private EncodedPayload toPayload() { + final int size = + prefix.length + PAYLOAD_SUFFIX.length + eventBytes + Math.max(0, events.size() - 1); + final byte[] body = new byte[size]; + int offset = 0; + System.arraycopy(prefix, 0, body, offset, prefix.length); + offset += prefix.length; + for (int index = 0; index < events.size(); index++) { + if (index > 0) { + System.arraycopy(JSON_COMMA, 0, body, offset, JSON_COMMA.length); + offset += JSON_COMMA.length; + } + final byte[] event = events.get(index); + System.arraycopy(event, 0, body, offset, event.length); + offset += event.length; + } + System.arraycopy(PAYLOAD_SUFFIX, 0, body, offset, PAYLOAD_SUFFIX.length); + return new EncodedPayload(body, events.size()); + } + } +} 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..e472cacc62a 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 @@ -8,16 +8,20 @@ import datadog.common.queue.MessagePassingBlockingQueue; import datadog.common.queue.Queues; import datadog.communication.BackendApiFactory; +import datadog.communication.EvpProxy; import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.http.HttpRetryPolicy; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.exposure.ExposureEvent; import datadog.trace.api.featureflag.exposure.ExposuresRequest; import datadog.trace.api.internal.VisibleForTesting; +import datadog.trace.api.telemetry.CoreMetricCollector; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,11 +31,25 @@ public class ExposureWriterImpl implements ExposureWriter { private static final int DEFAULT_CAPACITY = 1 << 16; // 65536 elements private static final int DEFAULT_FLUSH_INTERVAL_IN_SECONDS = 1; private static final int FLUSH_THRESHOLD = 100; + static final int MAX_BATCH_EVENTS = 1_000; + static final int EXPOSURE_PAYLOAD_SIZE_LIMIT_BYTES = EvpProxy.PAYLOAD_SIZE_LIMIT_BYTES; + static final String EXPOSURE_DROPPED_METRIC = "exposures.events.dropped"; + static final String DROP_REASON_QUEUE_OVERFLOW = "queue_overflow"; + static final String DROP_REASON_PAYLOAD_LIMIT = "payload_limit"; + static final String DROP_REASON_SERIALIZATION = "serialization"; + static final String DROP_REASON_DELIVERY_FAILURE = "delivery_failure"; private static final String EXPOSURES_ROUTE = "exposures"; + private static final CoreMetricCollector CORE_METRICS = CoreMetricCollector.getInstance(); private final MessagePassingBlockingQueue queue; + private final AtomicLong droppedQueueOverflow = new AtomicLong(); + private final ExposureSerializingHandler serializer; private final Thread serializerThread; + private static void countDropped(final long value, final String reason) { + CORE_METRICS.count(EXPOSURE_DROPPED_METRIC, value, "reason:" + reason); + } + public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config config) { this(DEFAULT_CAPACITY, DEFAULT_FLUSH_INTERVAL_IN_SECONDS, SECONDS, sco, config); } @@ -43,13 +61,14 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con final SharedCommunicationObjects sco, final Config config) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); - final ExposureSerializingHandler serializer = + this.serializer = new ExposureSerializingHandler( new BackendApiFactory(config, sco), queue, flushInterval, timeUnit, FeatureFlagEvpContext.from(config), + droppedQueueOverflow, this::close); this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer); } @@ -70,7 +89,9 @@ public void close() { @Override public void accept(final ExposureEvent event) { - queue.offer(event); + if (!queue.offer(event)) { + droppedQueueOverflow.incrementAndGet(); + } } @VisibleForTesting @@ -83,6 +104,16 @@ int queueSize() { return queue.size(); } + @VisibleForTesting + long droppedQueueOverflow() { + return droppedQueueOverflow.get(); + } + + @VisibleForTesting + void flushForTest() { + serializer.flushIfNecessary(); + } + private static class ExposureSerializingHandler implements Runnable { private final MessagePassingBlockingQueue queue; private final long ticksRequiredToFlush; @@ -92,7 +123,8 @@ private static class ExposureSerializingHandler implements Runnable { private final Map context; private final ExposureCache cache; - private final List buffer = new ArrayList<>(); + private final List buffer = new ArrayList<>(MAX_BATCH_EVENTS); + private final AtomicLong droppedQueueOverflow; private final Runnable errorCallback; public ExposureSerializingHandler( @@ -101,11 +133,15 @@ public ExposureSerializingHandler( final long flushInterval, final TimeUnit timeUnit, final Map context, + final AtomicLong droppedQueueOverflow, final Runnable errorCallback) { this.queue = queue; this.cache = new LRUExposureCache(queue.capacity()); - this.evpPublisher = new FeatureFlagEvpPublisher<>(backendApiFactory, ExposuresRequest.class); + this.evpPublisher = + new FeatureFlagEvpPublisher<>( + backendApiFactory, ExposuresRequest.class, true, HttpRetryPolicy.Factory.NEVER_RETRY); this.context = context; + this.droppedQueueOverflow = droppedQueueOverflow; this.lastTicks = System.nanoTime(); this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval); @@ -144,7 +180,8 @@ private void runDutyCycle() throws InterruptedException { } private void consumeBatch() { - queue.drain(this::addToBuffer, queue.size()); + final int remainingCapacity = MAX_BATCH_EVENTS - buffer.size(); + queue.drain(this::addToBuffer, Math.min(queue.size(), remainingCapacity)); } /** Adds an element to the buffer taking care of duplicated exposures thanks to the LRU cache */ @@ -157,32 +194,61 @@ private boolean addToBuffer(final ExposureEvent event) { } protected void flushIfNecessary() { + reportQueueDrops(); if (buffer.isEmpty()) { return; } if (shouldFlush()) { - final byte[] payload; + final ExposurePayloads.EncodingResult result; try { - final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer); - payload = evpPublisher.serialize(exposures); - } catch (RuntimeException e) { - LOGGER.error(EXCLUDE_TELEMETRY, "Could not serialize exposures; dropping batch", e); - this.buffer.clear(); - return; + result = + ExposurePayloads.writePayloads( + buffer, context, EXPOSURE_PAYLOAD_SIZE_LIMIT_BYTES, this::submitPayload); + } finally { + buffer.clear(); } - try { - evpPublisher.post(EXPOSURES_ROUTE, payload); - this.buffer.clear(); - } catch (Exception e) { - LOGGER.debug("Could not submit exposures", e); + if (result.droppedSerialization > 0) { + countDropped(result.droppedSerialization, DROP_REASON_SERIALIZATION); + LOGGER.error( + EXCLUDE_TELEMETRY, + "Could not serialize {} exposure event(s); dropping events", + result.droppedSerialization); + } + if (result.droppedPayloadLimit > 0) { + countDropped(result.droppedPayloadLimit, DROP_REASON_PAYLOAD_LIMIT); + LOGGER.warn( + "Exposure payload limit dropped {} event(s) (best-effort telemetry)", + result.droppedPayloadLimit); } } } + private void submitPayload(final ExposurePayloads.EncodedPayload payload) { + try { + evpPublisher.post(EXPOSURES_ROUTE, payload.body); + } catch (Exception e) { + countDropped(payload.eventCount, DROP_REASON_DELIVERY_FAILURE); + LOGGER.debug("Could not submit exposures; dropping attempted batch", e); + } + } + + private void reportQueueDrops() { + final long dropped = droppedQueueOverflow.getAndSet(0); + if (dropped > 0) { + countDropped(dropped, DROP_REASON_QUEUE_OVERFLOW); + LOGGER.warn( + "Exposure queue full - dropped {} event(s) under backpressure" + + " (best-effort telemetry)", + dropped); + } + } + private boolean shouldFlush() { long nanoTime = System.nanoTime(); long ticks = nanoTime - lastTicks; - if (ticks > ticksRequiredToFlush || queue.size() >= FLUSH_THRESHOLD) { + if (ticks > ticksRequiredToFlush + || buffer.size() >= MAX_BATCH_EVENTS + || queue.size() >= FLUSH_THRESHOLD) { lastTicks = nanoTime; return true; } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java index c871b18165e..b726c1046fd 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java @@ -4,9 +4,11 @@ import com.squareup.moshi.Moshi; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; +import datadog.communication.http.HttpRetryPolicy; import datadog.trace.api.intake.Intake; import java.io.IOException; import java.io.UnsupportedEncodingException; +import java.util.function.Supplier; import okhttp3.MediaType; import okhttp3.RequestBody; @@ -14,8 +16,7 @@ final class FeatureFlagEvpPublisher { private static final MediaType JSON = MediaType.parse("application/json"); - private final BackendApiFactory backendApiFactory; - private final boolean responseCompression; + private final Supplier backendApiSupplier; private final JsonAdapter jsonAdapter; private BackendApi evp; @@ -27,14 +28,32 @@ final class FeatureFlagEvpPublisher { final BackendApiFactory backendApiFactory, final Class requestType, final boolean responseCompression) { - this.backendApiFactory = backendApiFactory; - this.responseCompression = responseCompression; + this( + () -> backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM, responseCompression), + requestType); + } + + FeatureFlagEvpPublisher( + final BackendApiFactory backendApiFactory, + final Class requestType, + final boolean responseCompression, + final HttpRetryPolicy.Factory retryPolicyFactory) { + this( + () -> + backendApiFactory.createBackendApi( + Intake.EVENT_PLATFORM, responseCompression, retryPolicyFactory), + requestType); + } + + FeatureFlagEvpPublisher( + final Supplier backendApiSupplier, final Class requestType) { + this.backendApiSupplier = backendApiSupplier; this.jsonAdapter = new Moshi.Builder().build().adapter(requestType); } boolean start() { if (evp == null) { - evp = backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM, responseCompression); + evp = backendApiSupplier.get(); } return evp != null; } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposurePayloadsTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposurePayloadsTest.java new file mode 100644 index 00000000000..588867f3872 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposurePayloadsTest.java @@ -0,0 +1,101 @@ +package com.datadog.featureflag; + +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.squareup.moshi.Moshi; +import datadog.trace.api.featureflag.exposure.Allocation; +import datadog.trace.api.featureflag.exposure.ExposureEvent; +import datadog.trace.api.featureflag.exposure.ExposuresRequest; +import datadog.trace.api.featureflag.exposure.Flag; +import datadog.trace.api.featureflag.exposure.Subject; +import datadog.trace.api.featureflag.exposure.Variant; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class ExposurePayloadsTest { + + private static final Map CONTEXT = singletonMap("service", "test-service"); + + @Test + void splitsPayloadsAtByteLimit() throws Exception { + final ExposureEvent first = exposure("first", repeat('a', 128)); + final ExposureEvent second = exposure("second", repeat('b', 128)); + final ExposureEvent third = exposure("third", repeat('c', 128)); + final int oneEventBytes = + ExposurePayloads.buildPayloadsForTest(singletonList(first), CONTEXT, Integer.MAX_VALUE) + .payloads + .get(0) + .body + .length; + final int payloadLimit = oneEventBytes * 2; + + final ExposurePayloads.EncodedPayloads encoded = + ExposurePayloads.buildPayloadsForTest( + Arrays.asList(first, second, third), CONTEXT, payloadLimit); + + assertEquals(2, encoded.payloads.size()); + assertEquals(0, encoded.droppedPayloadLimit); + assertEquals(3, encoded.payloads.stream().mapToInt(payload -> payload.eventCount).sum()); + for (ExposurePayloads.EncodedPayload payload : encoded.payloads) { + assertTrue(payload.body.length <= payloadLimit); + assertEquals(payload.eventCount, decode(payload.body).exposures.size()); + } + } + + @Test + void dropsAnEventThatCannotFitInOnePayload() { + final ExposurePayloads.EncodedPayloads encoded = + ExposurePayloads.buildPayloadsForTest( + singletonList(exposure("large", repeat('x', 1_024))), CONTEXT, 256); + + assertTrue(encoded.payloads.isEmpty()); + assertEquals(1, encoded.droppedPayloadLimit); + } + + @Test + void dropsOnlyTheEventThatCannotBeSerialized() { + final ExposureEvent invalid = + new ExposureEvent( + 1, + new Allocation("allocation-invalid"), + new Flag("flag-invalid"), + new Variant("variant-invalid"), + new Subject("subject-invalid", singletonMap("not-a-number", (Object) Double.NaN))); + final ExposureEvent valid = exposure("valid", "value"); + + final ExposurePayloads.EncodedPayloads encoded = + ExposurePayloads.buildPayloadsForTest( + Arrays.asList(invalid, valid), CONTEXT, Integer.MAX_VALUE); + + assertEquals(1, encoded.droppedSerialization); + assertEquals(1, encoded.payloads.size()); + assertEquals(1, encoded.payloads.get(0).eventCount); + } + + private static ExposuresRequest decode(final byte[] body) throws Exception { + return new Moshi.Builder() + .build() + .adapter(ExposuresRequest.class) + .fromJson(new String(body, StandardCharsets.UTF_8)); + } + + private static ExposureEvent exposure(final String id, final String value) { + return new ExposureEvent( + 1, + new Allocation("allocation-" + id), + new Flag("flag-" + id), + new Variant("variant-" + id), + new Subject("subject-" + id, singletonMap("attribute", (Object) value))); + } + + private static String repeat(final char value, final int count) { + final char[] chars = new char[count]; + Arrays.fill(chars, value); + return new String(chars); + } +} 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..644b7f71a7b 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 @@ -1,11 +1,12 @@ package com.datadog.featureflag; +import static com.datadog.featureflag.FlagEvaluationTestSupport.clearCoreMetrics; +import static com.datadog.featureflag.FlagEvaluationTestSupport.metricSum; import static java.util.Collections.singletonList; import static java.util.Collections.singletonMap; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; @@ -26,9 +27,13 @@ import datadog.trace.api.featureflag.exposure.Flag; import datadog.trace.api.featureflag.exposure.Subject; import datadog.trace.api.featureflag.exposure.Variant; +import datadog.trace.api.telemetry.CoreMetricCollector; +import datadog.trace.api.telemetry.MetricCollector; import datadog.trace.test.util.PollingConditions; import java.io.ByteArrayInputStream; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -44,14 +49,13 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; import okio.Okio; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; import org.tabletest.junit.TableTest; class ExposureWriterTests { @@ -62,13 +66,22 @@ class ExposureWriterTests { private final PollingConditions poll = new PollingConditions(TIMEOUT_SECONDS); private Queue requests; private Set failed; + private AtomicInteger exposureAttempts; + private AtomicInteger attemptedExposureEvents; + private AtomicInteger largestAttemptBytes; + private AtomicInteger largestAttemptEvents; private JavaTestHttpServer server; private SharedCommunicationObjects sharedCommunicationObjects; @BeforeEach void setUp() { + clearCoreMetrics(); requests = new ConcurrentLinkedQueue<>(); failed = Collections.newSetFromMap(new ConcurrentHashMap()); + exposureAttempts = new AtomicInteger(); + attemptedExposureEvents = new AtomicInteger(); + largestAttemptBytes = new AtomicInteger(); + largestAttemptEvents = new AtomicInteger(); JsonAdapter adapter = new Moshi.Builder().build().adapter(ExposuresRequest.class); server = @@ -84,6 +97,7 @@ void cleanup() { if (server != null) { server.close(); } + clearCoreMetrics(); } private void handleExposureRequest(HandlerApi api, JsonAdapter adapter) @@ -92,6 +106,14 @@ private void handleExposureRequest(HandlerApi api, JsonAdapter adapter.fromJson( Okio.buffer(Okio.source(new ByteArrayInputStream(api.getRequest().getBody())))); String serviceName = exposuresRequest.context.get("service"); + exposureAttempts.incrementAndGet(); + attemptedExposureEvents.addAndGet(exposuresRequest.exposures.size()); + largestAttemptBytes.accumulateAndGet(api.getRequest().getBody().length, Math::max); + largestAttemptEvents.accumulateAndGet(exposuresRequest.exposures.size(), Math::max); + if ("reject-forever".equals(serviceName)) { + api.getResponse().status(400).send("Rejected"); + return; + } boolean failForever = "fail-forever".equals(serviceName); boolean fail = serviceName.startsWith("fail") && (failed.add(serviceName) || failForever); if (fail) { @@ -202,11 +224,75 @@ void testHighLoadScenario() throws Exception { } } - @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testFailuresAreRetried(boolean finallyFail) throws Exception { - String serviceName = finallyFail ? "fail-forever" : "fail-once"; - Config config = mockConfig(serviceName); + @Test + void batchesDoNotExceedEventLimit() throws Exception { + Config config = mockConfig("test-service"); + List exposures = buildExposures(ExposureWriterImpl.MAX_BATCH_EVENTS + 1); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(1 << 11, 100, MILLISECONDS, sharedCommunicationObjects, config)) { + writer.init(); + for (ExposureEvent exposure : exposures) { + writer.accept(exposure); + } + + poll.eventually(() -> assertEquals(exposures.size(), allExposures().size())); + assertTrue(largestAttemptEvents.get() <= ExposureWriterImpl.MAX_BATCH_EVENTS); + } + } + + @Test + void queueOverflowIsCountedAndReported() { + Config config = mockConfig("test-service"); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(2, 1, MILLISECONDS, sharedCommunicationObjects, config)) { + writer.accept(buildExposure()); + writer.accept(buildExposure()); + writer.accept(buildExposure()); + + assertEquals(1, writer.droppedQueueOverflow()); + final long queueDrops = writer.droppedQueueOverflow(); + writer.flushForTest(); + assertEquals(0, writer.droppedQueueOverflow()); + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + queueDrops, + metricSum( + metrics, + ExposureWriterImpl.EXPOSURE_DROPPED_METRIC, + "reason:" + ExposureWriterImpl.DROP_REASON_QUEUE_OVERFLOW)); + } + } + + @Test + void eventLargerThanPayloadLimitIsNotSent() throws Exception { + Config config = mockConfig("test-service"); + String largeAttribute = + repeat('x', ExposureWriterImpl.EXPOSURE_PAYLOAD_SIZE_LIMIT_BYTES + 1_024); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(2, 25, MILLISECONDS, sharedCommunicationObjects, config)) { + writer.init(); + writer.accept(buildExposure(singletonMap("large", (Object) largeAttribute))); + + MILLISECONDS.sleep(200); // wait for the oversized event to be processed + assertEquals(0, exposureAttempts.get()); + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + 1, + metricSum( + metrics, + ExposureWriterImpl.EXPOSURE_DROPPED_METRIC, + "reason:" + ExposureWriterImpl.DROP_REASON_PAYLOAD_LIMIT)); + } + } + + @Test + void testFailuresAreNotRetried() throws Exception { + Config config = mockConfig("fail-once"); try (ExposureWriterImpl writer = new ExposureWriterImpl(1 << 4, 100, MILLISECONDS, sharedCommunicationObjects, config)) { @@ -214,12 +300,41 @@ void testFailuresAreRetried(boolean finallyFail) throws Exception { writer.accept(buildExposure()); MILLISECONDS.sleep(500); // wait for a flush to happen - ExposuresRequest found = findRequest(serviceName); - if (finallyFail) { - assertNull(found, requests.toString()); - } else { - poll.eventually(() -> assertNotNull(findRequest(serviceName), requests.toString())); + assertNull(findRequest("fail-once"), requests.toString()); + assertEquals(1, exposureAttempts.get()); + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + 1, + metricSum( + metrics, + ExposureWriterImpl.EXPOSURE_DROPPED_METRIC, + "reason:" + ExposureWriterImpl.DROP_REASON_DELIVERY_FAILURE)); + } + } + + @Test + void persistentFailureDoesNotRetainOrRetryBatch() throws Exception { + final int rounds = 40; + final int eventsPerRound = 25; + Config config = mockConfig("reject-forever"); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(1 << 10, 25, MILLISECONDS, sharedCommunicationObjects, config)) { + writer.init(); + for (int round = 1; round <= rounds; round++) { + for (ExposureEvent exposure : buildExposures(eventsPerRound)) { + writer.accept(exposure); + } + final int expectedEvents = round * eventsPerRound; + poll.eventually(() -> assertEquals(expectedEvents, attemptedExposureEvents.get())); } + + final int attemptsAfterLastBatch = exposureAttempts.get(); + MILLISECONDS.sleep(100); + assertEquals(attemptsAfterLastBatch, exposureAttempts.get()); + assertEquals(rounds * eventsPerRound, attemptedExposureEvents.get()); + assertTrue(largestAttemptEvents.get() <= eventsPerRound); } } @@ -237,6 +352,14 @@ void testSerializationFailureDoesNotPoisonFollowingExposures() throws Exception writer.accept(validExposure); poll.eventually(() -> assertExposures(allExposures(), singletonList(validExposure))); + final Collection metrics = + CoreMetricCollector.getInstance().drain(); + assertEquals( + 1, + metricSum( + metrics, + ExposureWriterImpl.EXPOSURE_DROPPED_METRIC, + "reason:" + ExposureWriterImpl.DROP_REASON_SERIALIZATION)); } } @@ -431,4 +554,10 @@ private static ExposureEvent buildExposure(String id, Map attrib new Variant("Variant_" + id), new Subject("Subject_" + id, attributes)); } + + private static String repeat(final char value, final int count) { + final char[] chars = new char[count]; + Arrays.fill(chars, value); + return new String(chars); + } }