diff --git a/.gitignore b/.gitignore index 92e132338..61652c446 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,5 @@ .gradle -/local.properties +local.properties .DS_Store build /captures diff --git a/example/build.gradle b/example/build.gradle index f5286b24b..b29f6a623 100644 --- a/example/build.gradle +++ b/example/build.gradle @@ -2,8 +2,6 @@ import java.util.Properties plugins { id("com.android.application") - // make sure this line comes *after* you apply the Android plugin - id("com.getkeepsafe.dexcount") } // local.properties is not checked in, so it is where machine-specific settings such as your mobile diff --git a/example/src/main/java/com/launchdarkly/example/MainActivity.java b/example/src/main/java/com/launchdarkly/example/MainActivity.java index 9b6043240..512be492c 100644 --- a/example/src/main/java/com/launchdarkly/example/MainActivity.java +++ b/example/src/main/java/com/launchdarkly/example/MainActivity.java @@ -26,10 +26,6 @@ import java.util.Date; import java.util.Locale; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import timber.log.Timber; @@ -45,6 +41,9 @@ public class MainActivity extends AppCompatActivity { private static final String DEFAULT_USER_KEY = "user key"; + /** How long startup blocks waiting for the first flags to arrive. */ + private static final int INIT_WAIT_SECONDS = 10; + private LDClient ldClient; private LDStatusListener ldStatusListener; private LDAllFlagsListener allFlagsListener; @@ -142,15 +141,12 @@ public void onCreate(Bundle savedInstanceState) { .set("email", "fake@example.com") .build(); - Future initFuture = LDClient.init(this.getApplication(), ldConfig, context); - try { - ldClient = initFuture.get(10, TimeUnit.SECONDS); - updateStatusString(ldClient.getConnectionInformation()); - ldClient.registerStatusListener(ldStatusListener); - ldClient.registerAllFlagsListener(allFlagsListener); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - Timber.e(e, "Exception when awaiting LaunchDarkly Client initialization"); - } + // Returns the client either way: if the flags have not arrived within the wait, it is usable + // with whatever it has cached. + ldClient = LDClient.init(this.getApplication(), ldConfig, context, INIT_WAIT_SECONDS); + updateStatusString(ldClient.getConnectionInformation()); + ldClient.registerStatusListener(ldStatusListener); + ldClient.registerAllFlagsListener(allFlagsListener); } private void setupListeners() { diff --git a/launchdarkly-android-client-sdk/build.gradle b/launchdarkly-android-client-sdk/build.gradle index 3b7566cb1..daa235fb0 100644 --- a/launchdarkly-android-client-sdk/build.gradle +++ b/launchdarkly-android-client-sdk/build.gradle @@ -79,7 +79,7 @@ ext.versions = [ "jacksonDatabind": "2.10.5.1", "junit": "4.13", "launchdarklyJavaSdkCommon": "2.4.0", - "launchdarklyJavaSdkInternal": "1.9.0", + "launchdarklyJavaSdkInternal": "1.12.0", "launchdarklyLogging": "1.1.1", "okhttp": "4.12.0", "testHelpers": "2.1.0", diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java index 8c9cc0454..890e45dd7 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java @@ -20,12 +20,9 @@ import com.launchdarkly.sdk.android.subsystems.HookConfiguration; import com.launchdarkly.sdk.android.subsystems.HttpConfiguration; import com.launchdarkly.sdk.android.subsystems.PluginsConfiguration; -import com.launchdarkly.sdk.internal.events.DefaultEventProcessor; import com.launchdarkly.sdk.internal.events.DefaultEventSender; -import com.launchdarkly.sdk.internal.events.Event; -import com.launchdarkly.sdk.internal.events.EventsConfiguration; +import com.launchdarkly.sdk.internal.events.EventSender; -import java.io.IOException; import java.util.HashMap; import java.util.Map; @@ -101,33 +98,33 @@ static final class EventProcessorBuilderImpl extends EventProcessorBuilder @Override public EventProcessor build(ClientContext clientContext) { ClientContextImpl clientContextImpl = ClientContextImpl.get(clientContext); - EventsConfiguration eventsConfig = new EventsConfiguration( - allAttributesPrivate, - capacity, - null, // contextDeduplicator - not needed for client-side use - diagnosticRecordingIntervalMillis, - clientContextImpl.getDiagnosticStore(), - new DefaultEventSender( - LDUtil.makeHttpProperties(clientContext), - StandardEndpoints.ANALYTICS_EVENTS_REQUEST_PATH, - StandardEndpoints.DIAGNOSTIC_EVENTS_REQUEST_PATH, - 0L, // use default retry delay - false, // disable gzip compression for Android + EventSender eventSender = new DefaultEventSender( + LDUtil.makeHttpProperties(clientContext), + StandardEndpoints.ANALYTICS_EVENTS_REQUEST_PATH, + StandardEndpoints.DIAGNOSTIC_EVENTS_REQUEST_PATH, + 0L, // use default retry delay + false, // disable gzip compression for Android + clientContext.getBaseLogger()); + return new DirectEventProcessor( + new OutboundEventBuffer( + allAttributesPrivate, + privateAttributes, + true, // perContextSummarization - enable for client SDK + capacity, clientContext.getBaseLogger()), - 1, // eventSendingThreadPoolSize + eventSender, clientContext.getServiceEndpoints().getEventsBaseUri(), + clientContextImpl.getDiagnosticStore(), + capacity, flushIntervalMillis, + diagnosticRecordingIntervalMillis, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, clientContext.isInBackground(), true, // initiallyOffline - privateAttributes, - true // perContextSummarization - enable for client SDK - ); - return new DefaultEventProcessorWrapper(new DefaultEventProcessor( - eventsConfig, EventUtil.makeEventsTaskExecutor(), - Thread.NORM_PRIORITY, // note, we may want to make this configurable as it is in java-server-sdk + EventUtil.makeDiagnosticsTaskExecutor(), clientContext.getBaseLogger() - )); + ); } @Override @@ -140,72 +137,6 @@ public LDValue describeConfiguration(ClientContext clientContext) { .put("eventsFlushIntervalMillis", flushIntervalMillis) .build(); } - - /** - * Adapter from the public component interface of EventProcessor to the internal - * implementation class from java-sdk-internal. - */ - private final class DefaultEventProcessorWrapper implements EventProcessor { - private final DefaultEventProcessor eventProcessor; - - DefaultEventProcessorWrapper(DefaultEventProcessor eventProcessor) { - this.eventProcessor = eventProcessor; - } - - @Override - public void recordEvaluationEvent( - LDContext context, - String flagKey, - int flagVersion, - int variation, - LDValue value, - EvaluationReason reason, - LDValue defaultValue, - boolean requireFullEvent, - Long debugEventsUntilDate - ) { - eventProcessor.sendEvent(new Event.FeatureRequest( - System.currentTimeMillis(), flagKey, context, flagVersion, variation, - value, defaultValue, reason, null, requireFullEvent, - debugEventsUntilDate, false)); - } - - @Override - public void recordIdentifyEvent(LDContext context) { - eventProcessor.sendEvent(new Event.Identify(System.currentTimeMillis(), context)); - } - - @Override - public void recordCustomEvent(LDContext context, String eventKey, LDValue data, Double metricValue) { - eventProcessor.sendEvent(new Event.Custom(System.currentTimeMillis(), eventKey, - context, data, metricValue)); - } - - @Override - public void setInBackground(boolean inBackground) { - eventProcessor.setInBackground(inBackground); - } - - @Override - public void setOffline(boolean offline) { - eventProcessor.setOffline(offline); - } - - @Override - public void flush() { - eventProcessor.flushAsync(); - } - - @Override - public void blockingFlush() { - eventProcessor.flushBlocking(); - } - - @Override - public void close() throws IOException { - eventProcessor.close(); - } - } } static final class HttpConfigurationBuilderImpl extends HttpConfigurationBuilder diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/DirectEventProcessor.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/DirectEventProcessor.java new file mode 100644 index 000000000..147a9fb87 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/DirectEventProcessor.java @@ -0,0 +1,714 @@ +package com.launchdarkly.sdk.android; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; +import com.launchdarkly.sdk.EvaluationReason; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.subsystems.EventProcessor; +import com.launchdarkly.sdk.internal.events.DiagnosticEvent; +import com.launchdarkly.sdk.internal.events.DiagnosticStore; +import com.launchdarkly.sdk.internal.events.Event; +import com.launchdarkly.sdk.internal.events.EventSender; +import com.launchdarkly.sdk.internal.events.EventSummarizer; +import com.launchdarkly.sdk.internal.events.Sampler; + +import java.io.IOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** + * The Android SDK's analytics event processor. + *

+ * Recording an event summarizes it immediately and, only if it has to be delivered in full, + * buffers it. There is no queue between the calling thread and the summarizer, so a burst of flag + * evaluations with trackEvent = false cannot displace anything and costs a counter increment, + * The configured capacity limits only the events that have trackEvents = true. + */ +final class DirectEventProcessor implements EventProcessor { + /** + * How long {@link #close()} spends waiting for the final delivery before it gives up and returns. + *

+ * {@code close()} runs on the caller's thread, which for an application shutting down is usually the + * main one, so this has to stay well inside the five seconds Android allows before an unanswered + * input event becomes an ANR. Two seconds is where the return on waiting longer falls off: the + * events client keeps pooled connections for only five seconds against a thirty second flush + * interval, so this post nearly always pays a full DNS, TCP and TLS handshake -- about three round + * trips, which two seconds covers up to roughly a 600ms RTT. A network slower than that is one the + * post is likely to fail on anyway. + *

+ * Overshooting the budget is cheaper than it looks, because the delivery is not cancelled when the + * budget expires; see {@link #close()}. + */ + static final long DEFAULT_CLOSE_BUDGET_MILLIS = 2_000; + + private final OutboundEventBuffer buffer; + private final EventSender eventSender; + private final URI eventsUri; + private final DiagnosticStore diagnosticStore; + private final long flushIntervalMillis; + private final long diagnosticRecordingIntervalMillis; + private final long closeBudgetMillis; + private final ScheduledExecutorService scheduler; + private final ExecutorService diagnosticExecutor; + private final LDLogger logger; + + private final AtomicBoolean inBackground; + private final AtomicBoolean offline; + private final AtomicBoolean closed = new AtomicBoolean(false); + // Set when the service tells us to stop, e.g. because the mobile key is invalid. + private volatile boolean disabled = false; + private volatile boolean diagnosticInitSent = false; + private final AtomicLong lastKnownPastTime = new AtomicLong(0); + + private final Object stateLock = new Object(); + private ScheduledFuture flushTask; + private ScheduledFuture diagnosticTask; + + /** + * Full events recorded but not yet encoded, guarded by {@link #recordLock}. + */ + private final List pending = new ArrayList<>(); + + /** + * Guards everything one recording writes: {@link #pending} and the summary counters behind + * {@link #buffer}. + *

+ * One evaluation can produce a counter, a full event and a debug event, and the three have to + * land together. Taken separately, a flush landing between them splits one evaluation across two + * payloads, and a {@link #close()} landing between them delivers the counter and then refuses the + * full event -- leaving a summary that says an evaluation happened and no event to go with it. + *

+ * Held for the appends and the handover of the run, never across the encode. Recording runs on + * whichever thread evaluated a flag, which on Android is usually the main one. The worst it may + * wait for is another thread's memory operation; if the encoder ran under this lock, an + * evaluation would instead wait on the dominant cost of the whole path. + */ + private final Object recordLock = new Object(); + + /** + * How many events the SDK will hold between flushes. + */ + private final int capacity; + + private final AtomicBoolean capacityExceeded = new AtomicBoolean(false); + + /** Whether the summarizer being full has already been reported, so it is logged once per run. */ + private final AtomicBoolean summaryContextsExceeded = new AtomicBoolean(false); + + private final AtomicLong droppedEvents = new AtomicLong(0); + + /** + * True while a diagnostic event is on its way to the service, so that a later one is dropped + * rather than queued behind it. + */ + private final AtomicBoolean diagnosticPostInFlight = new AtomicBoolean(false); + + /** + * The two threads that post through {@link #eventSender}, counted down at shutdown so that + * whichever finishes last is the one that closes it. + */ + private final AtomicInteger sendersStillDraining = new AtomicInteger(2); + + /** + * Guards the handover from running to shut down: held for the length of a submit, and by + * {@link #close()} while it queues the release of the sender and stops the executors accepting + * work. Nothing blocking happens under it. + */ + private final Object submitLock = new Object(); + + /** Set under {@link #submitLock} once close() has queued the release of the sender. */ + private boolean shuttingDown = false; + + DirectEventProcessor( + OutboundEventBuffer buffer, + EventSender eventSender, + URI eventsUri, + DiagnosticStore diagnosticStore, + int capacity, + long flushIntervalMillis, + long diagnosticRecordingIntervalMillis, + long closeBudgetMillis, + boolean initiallyInBackground, + boolean initiallyOffline, + ScheduledExecutorService scheduler, + ExecutorService diagnosticExecutor, + LDLogger logger + ) { + this.buffer = buffer; + this.eventSender = eventSender; + this.eventsUri = eventsUri; + this.diagnosticStore = diagnosticStore; + this.capacity = capacity; + this.flushIntervalMillis = flushIntervalMillis; + this.diagnosticRecordingIntervalMillis = diagnosticRecordingIntervalMillis; + this.closeBudgetMillis = closeBudgetMillis; + this.scheduler = scheduler; + this.diagnosticExecutor = diagnosticExecutor; + this.logger = logger; + this.inBackground = new AtomicBoolean(initiallyInBackground); + this.offline = new AtomicBoolean(initiallyOffline); + + synchronized (stateLock) { + updateScheduledTasks(initiallyInBackground, initiallyOffline); + } + } + + @Override + public void recordEvaluationEvent( + LDContext context, + String flagKey, + int flagVersion, + int variation, + LDValue value, + EvaluationReason reason, + LDValue defaultValue, + boolean requireFullEvent, + Long debugEventsUntilDate + ) { + try { + if (isStopped() || context == null) { + return; + } + Event.FeatureRequest event = new Event.FeatureRequest(System.currentTimeMillis(), flagKey, + context, flagVersion, variation, value, defaultValue, reason, null, + requireFullEvent, debugEventsUntilDate, false); + // Built before the lock is taken, so that the critical section is only the writes. + Event debugEvent = shouldDebugEvent(debugEventsUntilDate) ? event.toDebugEvent() : null; + boolean contextsExceeded; + boolean warnContextsExceeded; + synchronized (recordLock) { + if (closed.get()) { + return; + } + contextsExceeded = !buffer.summarize(event); + // Claimed under the lock that the delivery resets it under, so the warning belongs to + // the run whose summarizer turned this evaluation away rather than to the next one. + warnContextsExceeded = contextsExceeded + && summaryContextsExceeded.compareAndSet(false, true); + if (requireFullEvent) { + addPending(event); + } + if (debugEvent != null) { + addPending(debugEvent); + } + } + if (contextsExceeded) { + reportContextsExceeded(warnContextsExceeded); + } + } catch (RuntimeException e) { + // This runs on the application's thread, usually inside a flag evaluation, and an + // analytics failure must not become the application's failure. Errors such as + // OutOfMemoryError are left to propagate, so the application's crash reporting sees them. + logUnexpectedError(e); + } + } + + @Override + public void recordIdentifyEvent(LDContext context) { + if (isStopped() || context == null) { + return; + } + record(new Event.Identify(System.currentTimeMillis(), context)); + } + + @Override + public void recordCustomEvent(LDContext context, String eventKey, LDValue data, Double metricValue) { + if (isStopped() || context == null) { + return; + } + record(new Event.Custom(System.currentTimeMillis(), eventKey, context, data, metricValue)); + } + + /** + * Holds an event for the next flush to encode, counting it as dropped if the SDK is already full. + *

+ * Capacity is consulted before anything is encoded, so an event that will not be kept is never + * encoded. + * That ordering is what bounds an application re-evaluating a tracked flag in a render loop: once + * the limit is reached the cost of an evaluation falls back to its summary counter, however fast + * the loop runs. + */ + void record(Event event) { + try { + synchronized (recordLock) { + // The close check that decides the outcome, as against the fast path the public record + // methods take before building the event. deliverPayload lifts the run out under this + // same lock, so testing the flag here orders a record against close()'s final delivery: + // either the event is in the list before that delivery takes it, or it is refused. + // Tested outside the lock the two interleave, and an event can be left in a list that + // nothing will drain again. + if (closed.get()) { + return; + } + addPending(event); + } + } catch (RuntimeException e) { + // As in recordEvaluationEvent: on the caller's thread, so a failure is logged, not thrown. + logUnexpectedError(e); + } + } + + /** + * Holds one event for the next flush, unless it was sampled out or there is no room. Requires + * {@link #recordLock}. + */ + private void addPending(Event event) { + // Ahead of the capacity check, because an event the SDK was never going to send is not a + // loss and must not be counted as one. Sampling and capacity are different reasons not to + // keep an event, and only the second is one the SDK owes anyone a count of. + if (!Sampler.shouldSample(event.getSamplingRatio())) { + return; + } + if (pending.size() >= capacity) { + if (capacityExceeded.compareAndSet(false, true)) { + logger.warn("Exceeded event queue capacity. Increase capacity to avoid dropping events."); + } + droppedEvents.incrementAndGet(); + return; + } + capacityExceeded.set(false); + pending.add(event); + } + + /** + * Counts an evaluation the summarizer turned away, which it does when counting it would have meant + * holding a context beyond the configured capacity. + *

+ * A refused evaluation is a loss in the same sense a refused event is -- nothing later reconstructs + * a counter -- so it is reported the same way, through the dropped count diagnostics carry. + */ + private void reportContextsExceeded(boolean warn) { + // Warned once per delivery run rather than once per process: the summarizer's contexts are + // cleared with each run, so a later overflow is a new one worth hearing about. + if (warn) { + logger.warn("Exceeded the number of contexts that can be summarized at once." + + " Increase capacity to avoid dropping evaluations."); + } + droppedEvents.incrementAndGet(); + } + + /** + * @return the number of events dropped for capacity since this was last called, counting + * evaluations the summarizer turned away + */ + long getAndClearDroppedCount() { + return droppedEvents.getAndSet(0); + } + + @Override + public void setInBackground(boolean inBackground) { + synchronized (stateLock) { + if (this.inBackground.getAndSet(inBackground) == inBackground) { + return; + } + updateScheduledTasks(inBackground, offline.get()); + } + } + + @Override + public void setOffline(boolean offline) { + synchronized (stateLock) { + if (this.offline.getAndSet(offline) == offline) { + return; + } + updateScheduledTasks(inBackground.get(), offline); + } + } + + @Override + public void flush() { + if (isStopped()) { + return; + } + submit(this::deliverPayload); + } + + @Override + public void blockingFlush() { + if (isStopped()) { + return; + } + Future delivery = submit(this::deliverPayload); + if (delivery == null) { + return; + } + try { + delivery.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException e) { + logUnexpectedError(e.getCause() == null ? e : e.getCause()); + } + } + + @Override + public void close() throws IOException { + if (!closed.compareAndSet(false, true)) { + return; + } + synchronized (stateLock) { + updateScheduledTasks(inBackground.get(), offline.get()); + } + // Deliver what is still buffered before we let go of the sender. This waits rather than + // firing and forgetting because it is the last chance these events get: nothing is kept + // once the processor is gone. While offline that chance is not taken, and whatever is held + // is discarded. Offline is the application telling the SDK to stay off the network, and + // shutting down does not revoke that. + Future delivery = submit(this::deliverPayload); + if (delivery != null) { + try { + delivery.get(closeBudgetMillis, TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + // Deliberately not cancelled. The run has already been drained into a payload, so + // interrupting now would make the loss certain, while leaving it to run costs + // nothing: the scheduler thread is a daemon, and returning from close() does not + // end an Android process. The budget bounds the caller, not the delivery. + logger.warn("Gave up waiting for the final event delivery after {}ms;" + + " it continues in the background", closeBudgetMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException e) { + logUnexpectedError(e.getCause() == null ? e : e.getCause()); + } + } + // Queued on both of the threads that post through the sender, so that it is released by + // whichever of them finishes last. Closing it here instead would pull the HTTP client out + // from under a delivery we just decided not to wait for, or out from under a diagnostic post + // still in flight. shutdown() then refuses new work while letting what is already queued + // finish -- unlike shutdownNow(), which would interrupt those posts and strand the futures of + // anything it discarded. + // + // Queueing and shutting down are one step under submitLock, so that a flush cannot land in + // between them and put a delivery behind the release, where it would find the sender closed. + synchronized (submitLock) { + shuttingDown = true; + queueSenderRelease(scheduler); + queueSenderRelease(diagnosticExecutor); + scheduler.shutdown(); + diagnosticExecutor.shutdown(); + } + } + + /** Queues the release on one of the posting threads, or runs it here if that thread has gone. */ + private void queueSenderRelease(ExecutorService executor) { + try { + executor.submit(guarded(this::releaseSenderWhenLast)); + } catch (RuntimeException e) { // the executor was shut down under us + releaseSenderWhenLast(); // so nothing can still be posting on it + } + } + + /** Closes the sender once both of the threads that post through it have got this far. */ + private void releaseSenderWhenLast() { + if (sendersStillDraining.decrementAndGet() > 0) { + return; + } + try { + eventSender.close(); + } catch (IOException e) { + logUnexpectedError(e); + } + } + + /** + * Serializes and sends everything buffered. Runs on the scheduler thread, which is + * single-threaded, so only one payload is ever in flight and the run is taken exactly once per + * delivery. + *

+ * The run and the counters are taken together under {@link #recordLock}, so an evaluation is + * never split across two payloads, and encoded outside it, so recording does not wait on the + * encoder. + */ + private void deliverPayload() { + if (disabled || offline.get()) { + return; + } + List run; + List summaries; + synchronized (recordLock) { + run = pending.isEmpty() ? Collections.emptyList() : new ArrayList<>(pending); + pending.clear(); + summaries = buffer.takeSummaries(); + summaryContextsExceeded.set(false); + } + OutboundEventBuffer.Payload payload; + try { + payload = buffer.encode(run, summaries); + } catch (IOException e) { + logUnexpectedError(e); + return; + } + if (payload == null) { + return; + } + if (diagnosticStore != null) { + diagnosticStore.recordEventsInBatch(payload.getEventCount()); + } + try { + handleResponse(eventSender.sendAnalyticsEvents(payload.getData(), + payload.getEventCount(), eventsUri)); + } catch (Exception e) { + logUnexpectedError(e); + } + } + + /** + * Posts an event that has already been built. Requires the caller to have checked + * {@link #diagnosticsSuspended()}, because for the periodic event building it is what consumes + * the period. + */ + private void sendDiagnosticEvent(DiagnosticEvent diagnosticEvent, boolean isInit) { + try { + byte[] data = diagnosticEvent.getJsonValue().toJsonString() + .getBytes(StandardCharsets.UTF_8); + handleResponse(eventSender.sendDiagnosticEvent(data, eventsUri)); + if (isInit) { + // Attempted, not delivered. A failed post gives back an unsuccessful Result rather + // than throwing, so this marks the init event done either way and the process never + // retries it. That matches DefaultEventProcessor, which is the behaviour to keep: + // diagnostics are best-effort telemetry about the SDK, and a retry that outlived + // its own init would describe a configuration the application has moved on from. + diagnosticInitSent = true; + } + } catch (Exception e) { + logUnexpectedError(e); + } + } + + private void sendDiagnosticStats() { + if (diagnosticsSuspended() || diagnosticStore == null || !claimDiagnosticPost()) { + return; + } + DiagnosticStore store = diagnosticStore; + postDiagnostic(() -> { + // The check that decides the outcome, and it has to come before the event is built rather + // than after. createEventAndReset hands back the period and clears it -- the stream inits, + // the events-in-batch count and the period start all move -- and getAndClearDroppedCount + // does the same for the dropped count. This runs on the diagnostics thread, which may have + // been busy with an earlier post for as long as that post took, so the state can easily + // have changed since the checks above. Bailing out after the event was built would discard + // a period outright and leave the next event describing a window that begins after the + // reset; bailing out here leaves everything where it is, for the next period to carry. + if (diagnosticsSuspended()) { + return; + } + sendDiagnosticEvent(store.createEventAndReset(getAndClearDroppedCount(), 0), false); + }); + } + + /** + * Takes the diagnostic posting thread, or reports that the last event is still on it. + *

+ * A new event is dropped rather than queued behind the old one. Diagnostics are best-effort + * telemetry, a post can take tens of seconds against a network that never answers, and queueing + * would let an outage accumulate events describing an SDK state the application has since moved + * past -- the same reasoning that makes a failed init event final rather than retried. + */ + private boolean claimDiagnosticPost() { + if (diagnosticPostInFlight.compareAndSet(false, true)) { + return true; + } + logger.debug("Skipped a diagnostic event because the previous one is still being posted"); + return false; + } + + /** + * Hands a claimed post to the diagnostics thread, releasing the claim once it ends. + * + * @param post what to run there, which must hold a claim from {@link #claimDiagnosticPost()} + */ + private void postDiagnostic(Runnable post) { + Runnable releasing = () -> { + try { + post.run(); + } finally { + diagnosticPostInFlight.set(false); + } + }; + try { + diagnosticExecutor.submit(guarded(releasing)); + } catch (RuntimeException e) { // the executor was shut down under us + diagnosticPostInFlight.set(false); + } + } + + /** + * Unlike analytics events, diagnostics are not sent while offline or in the background. + *

+ * {@link #updateScheduledTasks} cancels the periodic task when either becomes true, but that is + * not enough on its own. Cancelling does not stop a run already underway, and the init event is + * submitted before it reaches the executor. Either can arrive here after the state changed. + */ + private boolean diagnosticsSuspended() { + return isStopped() || offline.get() || inBackground.get(); + } + + private void handleResponse(EventSender.Result result) { + if (result == null) { + return; + } + if (result.getTimeFromServer() != null) { + recordPastTime(result.getTimeFromServer().getTime()); + } + if (result.isMustShutDown()) { + disabled = true; + } + } + + /** + * Moves the threshold forwards only. Analytics and diagnostic responses are handled on separate + * threads, so a plain assignment would let an older reading of the service clock overwrite a + * newer one and keep debug events alive past the date the service set. A loop rather than + * {@code accumulateAndGet}, which needs API 24. + */ + private void recordPastTime(long timeFromServer) { + long known; + do { + known = lastKnownPastTime.get(); + if (timeFromServer <= known) { + return; + } + } while (!lastKnownPastTime.compareAndSet(known, timeFromServer)); + } + + /** + * A debug event is emitted until the date the service gave us passes. We compare against the + * last date we know to be in the past according to the service as well as the device clock, so + * that a device whose clock is wrong errs on the side of stopping sooner. + */ + private boolean shouldDebugEvent(Long debugEventsUntilDate) { + if (debugEventsUntilDate == null || debugEventsUntilDate <= 0) { + return false; + } + return debugEventsUntilDate > lastKnownPastTime.get() + && debugEventsUntilDate > System.currentTimeMillis(); + } + + /** + * Must be called holding {@code stateLock}. Once closed, this only ever cancels: close() sets the + * flag and then calls this under the same lock, so a call that got here first has its tasks + * cancelled by close(), and any call after it finds the flag set. + */ + private void updateScheduledTasks(boolean inBackground, boolean offline) { + boolean stopped = closed.get(); + // Flushing stays scheduled whether or not we are offline or in the background; a run while + // offline returns without doing anything. Cancelling it for an outage would restart the + // interval on every reconnect, and a run of brief outages would then hold events back for + // far longer than one interval. Left running, what an outage buffered goes out at the first + // run after it ends. + flushTask = enableOrDisableTask(!stopped, flushTask, flushIntervalMillis, + this::deliverPayload); + boolean diagnosticsEnabled = !stopped && diagnosticStore != null && !offline && !inBackground; + diagnosticTask = enableOrDisableTask(diagnosticsEnabled, diagnosticTask, + diagnosticRecordingIntervalMillis, this::sendDiagnosticStats); + if (diagnosticsEnabled && !diagnosticInitSent && claimDiagnosticPost()) { + DiagnosticStore store = diagnosticStore; + postDiagnostic(() -> { + // Re-checked on the posting thread: going online and coming to the foreground are + // two separate calls, and both want to send the init event we never got to send. + // Suspension is re-checked for the same reason it is for the periodic event, and + // costs nothing here: getInitEvent consumes nothing, so a skipped init is simply + // built again the next time diagnostics are enabled. + if (!diagnosticInitSent && !diagnosticsSuspended()) { + sendDiagnosticEvent(store.getInitEvent(), true); + } + }); + } + } + + private ScheduledFuture enableOrDisableTask( + boolean shouldEnable, + ScheduledFuture currentTask, + long intervalMillis, + Runnable task + ) { + if (!shouldEnable) { + if (currentTask != null) { + currentTask.cancel(false); + } + return null; + } + if (currentTask != null && !currentTask.isDone()) { + return currentTask; + } + if (currentTask != null) { + // Backstop for a throwable that escaped guarded() anyway, such as one thrown while + // logging the first: the executor marks the repeating future done and never fires it + // again, and holding that future here would make every later enable a no-op. + currentTask.cancel(false); + } + try { + // Fixed delay rather than fixed rate: a cached process stops running its tasks without + // stopping the clock, so at a fixed rate it would come back owing every run it missed and + // fire them one after another. + return scheduler.scheduleWithFixedDelay(guarded(task), intervalMillis, intervalMillis, + TimeUnit.MILLISECONDS); + } catch (RuntimeException e) { // the executor was shut down under us + return null; + } + } + + /** + * Puts work on the delivery thread, and is the one place that decides whether there is still a + * thread willing to take it. The {@code isStopped} tests the callers make first are a fast path + * that saves the work of getting here, not a guarantee; this is what a delivery is actually + * ordered against close(). + * + * @return the submitted task, or null if the processor is shutting down or already has + */ + private Future submit(Runnable task) { + synchronized (submitLock) { + if (shuttingDown) { + // close() has already queued the release of the sender. Work accepted now would be + // behind it in the queue and would run against a sender that had been closed. + return null; + } + try { + return scheduler.submit(guarded(task)); + } catch (RuntimeException e) { // the executor was shut down under us + return null; + } + } + } + + /** + * Keeps an unexpected failure from killing a repeating task or bubbling out of the executor. + * Anything that escapes a run suppresses the rest of a {@code scheduleWithFixedDelay} series, + * so this catches {@code Throwable} and not just {@code Exception}: a + * {@code StackOverflowError} from nested {@code LDValue} data or an {@code OutOfMemoryError} + * growing the payload stream would otherwise stop flushing for the life of the process with + * nothing logged. + */ + private Runnable guarded(Runnable task) { + return () -> { + try { + task.run(); + } catch (Throwable t) { + logUnexpectedError(t); + } + }; + } + + private boolean isStopped() { + return closed.get() || disabled; + } + + private void logUnexpectedError(Throwable e) { + logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e)); + logger.debug("{}", LogValues.exceptionTrace(e)); + } +} diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventUtil.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventUtil.java index e605de085..a534a74e9 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventUtil.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventUtil.java @@ -15,6 +15,7 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; @@ -22,17 +23,37 @@ abstract class EventUtil { static ScheduledExecutorService makeEventsTaskExecutor() { - return Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { + return Executors.newSingleThreadScheduledExecutor( + makeThreadFactory("LaunchDarkly-DefaultEventProcessor-%d")); + } + + /** + * A thread of its own for posting diagnostic events, kept apart from the one that delivers + * analytics events. + *

+ * A diagnostic post is an HTTP request like any other and can hold its thread for tens of seconds + * against a network that accepts connections and never answers. Sharing a thread with analytics + * delivery would let that stall the flushes, and an event buffer that is not being drained fills + * up and starts dropping what the application asked to send. This mirrors the separation + * {@code DefaultEventProcessor} has, where analytics go out on dedicated delivery workers and + * diagnostics on the shared executor. + */ + static ExecutorService makeDiagnosticsTaskExecutor() { + return Executors.newSingleThreadExecutor( + makeThreadFactory("LaunchDarkly-DiagnosticEventPoster-%d")); + } + + private static ThreadFactory makeThreadFactory(String nameFormat) { + return new ThreadFactory() { final AtomicLong count = new AtomicLong(0); @Override public Thread newThread(@NonNull Runnable r) { Thread thread = Executors.defaultThreadFactory().newThread(r); - thread.setName(String.format(Locale.ROOT, "LaunchDarkly-DefaultEventProcessor-%d", - count.getAndIncrement())); + thread.setName(String.format(Locale.ROOT, nameFormat, count.getAndIncrement())); thread.setDaemon(true); return thread; } - }); + }; } static DiagnosticStore.SdkDiagnosticParams makeDiagnosticParams(ClientContext clientContext) { diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/OutboundEventBuffer.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/OutboundEventBuffer.java new file mode 100644 index 000000000..47085ab16 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/OutboundEventBuffer.java @@ -0,0 +1,341 @@ +package com.launchdarkly.sdk.android; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; +import com.launchdarkly.sdk.AttributeRef; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.internal.events.AggregatedEventSummarizer; +import com.launchdarkly.sdk.internal.events.Event; +import com.launchdarkly.sdk.internal.events.EventOutputFormatter; +import com.launchdarkly.sdk.internal.events.EventSummarizer; +import com.launchdarkly.sdk.internal.events.EventSummarizerInterface; +import com.launchdarkly.sdk.internal.events.EventsConfiguration; +import com.launchdarkly.sdk.internal.events.PerContextEventSummarizer; + +import java.io.BufferedWriter; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.io.Writer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * The counters and the encoder behind {@link DirectEventProcessor}: evaluations are folded into + * summary counters as they are recorded, and a flush turns a run of full-fidelity events, together + * with the counters accumulated beside them, into the payload they will be sent as. + *

+ * The full events are held by the processor rather than here. Capacity is counted once, against + * everything the SDK is holding, and the processor is the only place that can see all of it. + *

+ * The summarization and the wire format come from java-sdk-internal rather than being + * reimplemented here, so there is one definition of what an event looks like on the wire. + * {@code DefaultEventProcessor} is not reused along with them because it only accepts individual + * events and summarizes them itself, on the far side of the bounded queue this is meant to get in + * front of. + */ +final class OutboundEventBuffer { + private static final int INITIAL_OUTPUT_BUFFER_SIZE = 2000; + private static final Event[] NO_EVENTS = new Event[0]; + private static final List NO_SUMMARIES = Collections.emptyList(); + + private final EventOutputFormatter formatter; + private final EventSummarizerInterface summarizer; + private final LDLogger logger; + + /** + * How many distinct contexts may be counted between two drains, and which ones already are. + *

+ * The per-context summarizer keeps a counter set per context, keyed on the whole context with + * every attribute retained, and offers no way to ask how many it holds. Without this the only + * bound on it is how long the SDK goes without a drain -- and it is not drained at all while the + * client is offline, which is exactly when an application is free to go on evaluating. + */ + private final int maxContexts; + private final Set countedContexts; + + /** + * @param allAttributesPrivate true to redact every context attribute except the key + * @param privateAttributes the individual context attributes to redact + * @param perContextSummarization true to emit one summary per context rather than one overall + * @param maxContexts how many distinct contexts may be counted between two drains + * @param logger where to report an event that cannot be serialized + */ + OutboundEventBuffer( + boolean allAttributesPrivate, + Collection privateAttributes, + boolean perContextSummarization, + int maxContexts, + LDLogger logger + ) { + // Only the private-attribute settings affect the output; the rest of EventsConfiguration + // describes the delivery behavior that the processor now handles itself, capacity included. + EventsConfiguration outputConfig = new EventsConfiguration(allAttributesPrivate, 0, + null, 0, null, null, 1, null, 0, false, false, privateAttributes, + perContextSummarization); + this.formatter = new EventOutputFormatter(outputConfig); + this.summarizer = perContextSummarization + ? new PerContextEventSummarizer() + : new AggregatedEventSummarizer(); + // One bucket overall, so there is no cardinality to bound and nothing to track it with. + this.maxContexts = perContextSummarization ? maxContexts : Integer.MAX_VALUE; + this.countedContexts = perContextSummarization ? new HashSet<>() : null; + this.logger = logger; + } + + /** + * Folds an evaluation into the summary counters, unless the evaluation asked to be left out of + * them. + *

+ * A counter is an aggregate rather than a buffered event, so no number of evaluations of a context + * already being counted can make this drop anything. What capacity does bound is how many distinct + * contexts are counted at once, because each one costs a retained context and its own counters. + * + * @param event the evaluation + * @return false if this evaluation was not counted, because counting it would have meant holding + * a context beyond the configured capacity + */ + synchronized boolean summarize(Event.FeatureRequest event) { + // Checked here rather than in DirectEventProcessor, for the same reason the sampling ratio + // is: this is where an event arrives from outside. The processor builds its own through the + // constructor overload that leaves this false, so a guard there could never fire and would + // read as dead. java-sdk-internal's DefaultEventProcessor, which this path replaced, honored + // the flag, and a counter is the one thing no later stage can reconstruct. + if (event.isExcludeFromSummaries()) { + return true; + } + // Only a context that is not being counted yet can be turned away, so reaching the limit costs + // an application evaluating against one context nothing, however many evaluations it does. + if (countedContexts != null && !countedContexts.contains(event.getContext())) { + if (countedContexts.size() >= maxContexts) { + return false; + } + countedContexts.add(event.getContext()); + } + summarizer.summarizeEvent( + event.getCreationDate(), + event.getKey(), + event.getVersion(), + event.getVariation(), + event.getValue(), + event.getDefaultVal(), + event.getContext() + ); + return true; + } + + /** + * Takes the counters and forgets which contexts they were counted for. + *

+ * The two have to move together. Every path that resets the summarizer has to come through here, + * because one that reset the counters alone would spend the cardinality limit on contexts whose + * counters had already gone out, and the limit would never lift. + *

+ * Separate from {@link #encode} so that the caller can take the counters in the same critical + * section it lifts the full events in. An evaluation writes a counter and a full event, and a + * flush that took the two at different moments could split one evaluation across two payloads. + */ + synchronized List takeSummaries() { + List summaries = summarizer.getSummariesAndReset(); + if (countedContexts != null) { + countedContexts.clear(); + } + return summaries; + } + + /** + * Serializes a run of events, together with the evaluations counted beside it, into the payload + * they will be sent as. + *

+ * Deliberately takes no lock. The caller has already taken both the run and the counters, so + * there is nothing left here to protect, and the encode is the dominant cost on this path -- + * holding a lock across it would make a thread recording an event wait for it. Recording happens + * on whichever thread evaluated a flag, which on Android is usually the main one. + *

+ * A run that cannot be serialized as a whole is retried per event and per summary. Anything that + * still fails is dropped and logged; the rest is sent. Everything the encoder can fail on is a + * property of the data it was handed -- the output stream is a byte array and cannot fail + * transiently -- so putting a failed item back would only make every later flush fail too. + * + * @param run the full events to send, in the order they were recorded + * @param summaries the counters taken alongside that run + * @return the payload to send, or null if there was nothing to send + * @throws IOException if the events could not be serialized + */ + Payload encode(List run, List summaries) throws IOException { + if (run.isEmpty() && summaries.isEmpty()) { + return null; + } + try { + return encodeAll(run, summaries); + } catch (Exception e) { + logger.error("Dropping unserializable analytics event(s): {}", + LogValues.exceptionSummary(e)); + logger.debug("{}", LogValues.exceptionTrace(e)); + return encodeSkippingFailures(run, summaries); + } + } + + private Payload encodeAll(List run, List summaries) + throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE); + int outputEventCount = write(run.toArray(NO_EVENTS), summaries, buffer); + if (outputEventCount == 0) { + return null; + } + return new Payload(buffer.toByteArray(), outputEventCount); + } + + private Payload encodeSkippingFailures(List run, + List summaries) { + List objects = new ArrayList<>(); + int outputEventCount = 0; + for (Event event : run) { + EncodedPiece piece = tryEncode(new Event[] { event }, NO_SUMMARIES); + if (piece == null) { + logger.error("Dropping unserializable event of type {}", event.getClass().getSimpleName()); + continue; + } + objects.add(piece.jsonObject); + outputEventCount += piece.eventCount; + } + for (EventSummarizer.EventSummary summary : summaries) { + EncodedPiece piece = tryEncode(NO_EVENTS, Collections.singletonList(summary)); + if (piece == null) { + logger.error("Dropping unserializable summary event"); + continue; + } + objects.add(piece.jsonObject); + outputEventCount += piece.eventCount; + } + if (objects.isEmpty()) { + return null; + } + return new Payload(joinObjects(objects), outputEventCount); + } + + private EncodedPiece tryEncode(Event[] events, List summaries) { + try { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE); + int count = write(events, summaries, buffer); + if (count == 0) { + return null; + } + byte[] jsonObject = objectFromArray(buffer.toByteArray()); + if (jsonObject == null) { + return null; + } + return new EncodedPiece(jsonObject, count); + } catch (Exception e) { + logger.debug("{}", LogValues.exceptionTrace(e)); + return null; + } + } + + private int write(Event[] events, List summaries, + ByteArrayOutputStream buffer) throws IOException { + Writer writer = new BufferedWriter( + new OutputStreamWriter(buffer, StandardCharsets.UTF_8), INITIAL_OUTPUT_BUFFER_SIZE); + int outputEventCount = formatter.writeOutputEvents(events, summaries, writer); + writer.flush(); + return outputEventCount; + } + + /** + * {@code EventOutputFormatter} always writes a JSON array. For a single successful event that + * is {@code [{...}]}, and the payload we are assembling needs the object in the middle. + */ + static byte[] objectFromArray(byte[] arrayJson) { + int start = 0; + int end = arrayJson.length - 1; + while (start <= end && arrayJson[start] <= ' ') { + start++; + } + while (end >= start && arrayJson[end] <= ' ') { + end--; + } + if (start > end || arrayJson[start] != '[' || arrayJson[end] != ']') { + return null; + } + start++; + end--; + while (start <= end && arrayJson[start] <= ' ') { + start++; + } + while (end >= start && arrayJson[end] <= ' ') { + end--; + } + if (start > end || arrayJson[start] != '{') { + return null; + } + int length = end - start + 1; + byte[] object = new byte[length]; + System.arraycopy(arrayJson, start, object, 0, length); + return object; + } + + private static byte[] joinObjects(List objects) { + int size = 2; + for (int i = 0; i < objects.size(); i++) { + if (i > 0) { + size++; + } + size += objects.get(i).length; + } + byte[] out = new byte[size]; + int offset = 0; + out[offset++] = '['; + for (int i = 0; i < objects.size(); i++) { + if (i > 0) { + out[offset++] = ','; + } + byte[] object = objects.get(i); + System.arraycopy(object, 0, out, offset, object.length); + offset += object.length; + } + out[offset] = ']'; + return out; + } + + private static final class EncodedPiece { + final byte[] jsonObject; + final int eventCount; + + EncodedPiece(byte[] jsonObject, int eventCount) { + this.jsonObject = jsonObject; + this.eventCount = eventCount; + } + } + + /** + * A serialized batch of analytics events. + */ + static final class Payload { + private final byte[] data; + private final int eventCount; + + Payload(byte[] data, int eventCount) { + this.data = data; + this.eventCount = eventCount; + } + + /** + * @return the JSON request body + */ + byte[] getData() { + return data; + } + + /** + * @return how many events the body represents, including summaries + */ + int getEventCount() { + return eventCount; + } + } +} diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventProcessorBuilder.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventProcessorBuilder.java index 989222ae2..b0080b17d 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventProcessorBuilder.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventProcessorBuilder.java @@ -93,13 +93,14 @@ public EventProcessorBuilder allAttributesPrivate(boolean allAttributesPrivate) * the buffer is flushed (see {@link #flushIntervalMillis(int)}, events will be discarded. Increasing the * capacity means that events are less likely to be discarded, at the cost of consuming more memory. *

- * The default value is {@link #DEFAULT_CAPACITY}. + * The default value is {@link #DEFAULT_CAPACITY}. A capacity below one is treated as one; to + * stop sending events altogether, use {@link Components#noEvents()} instead. * * @param capacity the capacity of the event buffer * @return the builder */ public EventProcessorBuilder capacity(int capacity) { - this.capacity = capacity; + this.capacity = capacity < 1 ? 1 : capacity; return this; } diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/subsystems/EventProcessor.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/subsystems/EventProcessor.java index d1c98af02..e65a8e2b7 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/subsystems/EventProcessor.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/subsystems/EventProcessor.java @@ -90,7 +90,8 @@ void recordCustomEvent( * Specifies that any buffered events should be sent as soon as possible, rather than waiting * for the next flush interval. This method is asynchronous, so events still may not be sent * until a later time. However, calling {@link Closeable#close()} will synchronously deliver - * any events that were not yet delivered prior to shutting down. + * any events that were not yet delivered prior to shutting down, unless the SDK is offline, in + * which case those events are discarded. */ void flush(); diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DiagnosticConfigTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DiagnosticConfigTest.java index 8243052b0..b144ef070 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DiagnosticConfigTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DiagnosticConfigTest.java @@ -88,6 +88,21 @@ public void customDiagnosticConfigurationEvents() throws Exception { Assert.assertEquals(expected.build(), diagnosticJson); } + @Test + public void diagnosticConfigurationReportsTheCapacityTheSdkWillUse() throws Exception { + // A capacity of nought or less runs at one, so that is what the service should be told. + for (int capacity : new int[] { 0, -5 }) { + LDConfig ldConfig = new LDConfig.Builder(AutoEnvAttributes.Disabled) + .events(Components.sendEvents().capacity(capacity)) + .build(); + + LDValue diagnosticJson = makeDiagnosticJson(ldConfig); + + Assert.assertEquals("capacity " + capacity, + 1, diagnosticJson.get("eventsCapacity").intValue()); + } + } + @Test public void customDiagnosticConfigurationStreaming() throws Exception { LDConfig ldConfig = new LDConfig.Builder(AutoEnvAttributes.Disabled) diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java new file mode 100644 index 000000000..a4acfd39c --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java @@ -0,0 +1,1285 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import com.launchdarkly.sdk.EvaluationReason; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.subsystems.EventProcessor; +import com.launchdarkly.sdk.internal.events.DiagnosticStore; +import com.launchdarkly.sdk.internal.events.Event; +import com.launchdarkly.sdk.internal.events.EventSender; +import com.launchdarkly.testhelpers.httptest.Handlers; +import com.launchdarkly.testhelpers.httptest.HttpServer; +import com.launchdarkly.testhelpers.httptest.RequestInfo; + +import org.junit.After; +import org.junit.Test; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Behavior of the SDK's own event processor, covering the parts that are not about buffering under + * load (see {@link EventProcessorBufferingTest} for those). + */ +public class DirectEventProcessorTest extends EventProcessorTestBase { + private static final String FLAG_KEY = "flag-key"; + private static final int FLAG_VERSION = 10; + private static final int VARIATION = 1; + private static final LDValue FLAG_VALUE = LDValue.of(true); + private static final LDValue DEFAULT_VALUE = LDValue.of(false); + + private static final int DEFAULT_CAPACITY = 100; + + // Enough concurrent evaluations that close() reliably lands between the two writes one evaluation + // makes. Against the unfixed code this failed in the first trial of every run, by two to four + // events -- roughly the number of recorders that can sit in the gap at once. + private static final int RACE_TRIALS = 20; + private static final int RACE_RECORDERS = 4; + private static final int EVALUATIONS_PER_RACE_RECORDER = 2_000; + private static final int EVALUATIONS_BEFORE_CLOSE = 200; + private static final int RACE_CAPACITY = 30; + // Past anything the recorders can produce, so that a payload short of a feature event is short + // because the evaluation was split rather than because the buffer was full. + private static final int NO_DROP_CAPACITY = RACE_RECORDERS * EVALUATIONS_PER_RACE_RECORDER * 2; + private static final int SPLIT_TRIALS = 8; + + // Long enough that the only delivery in a test is the one it asks for. + private static final long NO_PERIODIC_FLUSH_MILLIS = 600_000; + // Short enough to keep the close tests quick; the production value is chosen for a real network. + private static final long CLOSE_BUDGET_MILLIS = 200; + + /** Created by makeEventProcessor, which the tests call instead of building a processor. */ + private final List diagnosticExecutors = new ArrayList<>(); + + @After + public void shutDownDiagnosticExecutors() { + for (ExecutorService executor : diagnosticExecutors) { + executor.shutdownNow(); + } + } + + @Test + public void untrackedEvaluationProducesOnlyASummary() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + recordEvaluation(eventProcessor, false, null); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, events.size()); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void trackedEvaluationProducesAFeatureEventAndASummary() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION, + FLAG_VALUE, EvaluationReason.off(), DEFAULT_VALUE, true, null); + + List events = flushAndCollect(eventProcessor, server); + + LDValue featureEvent = requireEventOfKind(events, "feature"); + assertEquals(LDValue.of(FLAG_KEY), featureEvent.get("key")); + assertEquals(LDValue.of(VARIATION), featureEvent.get("variation")); + assertEquals(FLAG_VALUE, featureEvent.get("value")); + assertEquals(LDValue.of(FLAG_VERSION), featureEvent.get("version")); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void identifyAndCustomEventsAreDelivered() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordIdentifyEvent(CONTEXT); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.of("data"), 2.5); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(2, events.size()); + requireEventOfKind(events, "identify"); + LDValue customEvent = requireEventOfKind(events, "custom"); + assertEquals(LDValue.of("an-event"), customEvent.get("key")); + assertEquals(LDValue.of("data"), customEvent.get("data")); + assertEquals(LDValue.of(2.5), customEvent.get("metricValue")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void unserializableMetricDoesNotPoisonLaterDeliveries() throws Exception { + try (HttpServer server = startEventsServer()) { + // One slot makes this cover both recovery and release of the poisoned event's capacity. + EventProcessor eventProcessor = makeEventProcessor(server, 1); + try { + // This is the event produced by LDClient.trackMetric(..., Double.NaN). Gson's + // strict writer rejects the non-finite metric. + eventProcessor.recordCustomEvent( + CONTEXT, "poison", LDValue.ofNull(), Double.NaN); + eventProcessor.blockingFlush(); + server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + + eventProcessor.recordCustomEvent( + CONTEXT, "after-poison", LDValue.ofNull(), 1.0); + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, events.size()); + assertEquals(LDValue.of("after-poison"), requireEventOfKind(events, "custom").get("key")); + logging.assertErrorLogged("Dropping unserializable"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void unserializableMetricDoesNotDropSiblingEvents() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent( + CONTEXT, "poison", LDValue.ofNull(), Double.NaN); + eventProcessor.recordCustomEvent( + CONTEXT, "kept", LDValue.ofNull(), 1.0); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, events.size()); + assertEquals(LDValue.of("kept"), requireEventOfKind(events, "custom").get("key")); + logging.assertErrorLogged("Dropping unserializable"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void unserializableSummaryDoesNotPoisonLaterDeliveries() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + // Java callers can violate boolVariation's @NonNull contract. A null flag key can + // enter the summarizer, but Gson cannot use it as a JSON object member name. + eventProcessor.recordEvaluationEvent(CONTEXT, null, FLAG_VERSION, VARIATION, + FLAG_VALUE, null, DEFAULT_VALUE, false, null); + eventProcessor.blockingFlush(); + server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + + recordEvaluation(eventProcessor, false, null); + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, events.size()); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + logging.assertErrorLogged("Dropping unserializable"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void unserializableSummaryDoesNotDropSiblingEvents() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordEvaluationEvent(CONTEXT, null, FLAG_VERSION, VARIATION, + FLAG_VALUE, null, DEFAULT_VALUE, false, null); + eventProcessor.recordIdentifyEvent(CONTEXT); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, countEventsOfKind(events, "identify")); + assertEquals(0, countEventsOfKind(events, "summary")); + logging.assertErrorLogged("Dropping unserializable"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void capacityLimitsFullEventsButNotSummaries() throws Exception { + try (HttpServer server = startEventsServer()) { + int capacity = 3; + EventProcessor eventProcessor = makeEventProcessor(server, capacity); + try { + for (int i = 0; i < capacity + 2; i++) { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + } + // Evaluations well past capacity still have to be counted in full, because a + // counter does not occupy a buffer slot. + int evaluations = capacity * 100; + for (int i = 0; i < evaluations; i++) { + recordEvaluation(eventProcessor, false, null); + } + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(capacity, countEventsOfKind(events, "custom")); + assertEquals(evaluations, summaryCountFor(events, FLAG_KEY)); + logging.assertWarnLogged("Exceeded event queue capacity"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void aNonPositiveCapacityStillReportsRatherThanGoingSilent() throws Exception { + // Nought and negatives both mean one, and the processor and the buffer have to agree on + // that: disagreeing leaves the SDK holding an event it will never summarize, or summarizing + // for a buffer that will never send. Disabling events entirely is what noEvents() is for. + for (int capacity : new int[] { 0, -5 }) { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, capacity); + try { + recordEvaluation(eventProcessor, false, null); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals("capacity " + capacity + " summarized nothing", + 1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + } + + @Test + public void closeDeliversBufferedEventsWithoutAnExplicitFlush() throws Exception { + // This is the case the SDK previously lost: a short session that records something and + // then shuts down before the periodic flush comes around. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + recordEvaluation(eventProcessor, false, null); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + + eventProcessor.close(); + + List events = collectDelivered(server); + assertEquals(1, countEventsOfKind(events, "custom")); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } + } + + @Test + public void closeNeverDeliversASummaryWithoutTheFeatureEventItCounted() throws Exception { + // An evaluation of a tracked flag writes a summary counter and a feature event. close() has + // to take both or neither. If it can take the counter and then refuse the event, it reports + // an evaluation that no event describes -- for experimentation traffic, a data point that + // disappears while the summary insists it happened. + // + // The window is the gap between those two writes, so this races them rather than asserting + // on a single ordering. Nothing here is timing-tolerant: the invariant holds for every + // interleaving, so any trial that breaks it is a real defect. + int featureEventsSeen = 0; + for (int trial = 0; trial < RACE_TRIALS; trial++) { + try (HttpServer server = startEventsServer()) { + // Deliberately small. The recorders outrun the buffer, so capacity is the only thing + // bounding the payload, and the test server records the body with a single unlooped + // read -- so anything past one socket read's worth comes back truncated, at a point + // that moves from run to run. A few kilobytes stays well clear of that. + DirectEventProcessor eventProcessor = + (DirectEventProcessor) makeEventProcessor(server, RACE_CAPACITY); + // Guarantees the final delivery has something in it, so collectDelivered always has + // a request to read even if every recorder loses the race. + eventProcessor.recordIdentifyEvent(CONTEXT); + + AtomicInteger recorded = new AtomicInteger(); + List recorders = new ArrayList<>(); + for (int i = 0; i < RACE_RECORDERS; i++) { + Thread recorder = new Thread(() -> { + for (int n = 0; n < EVALUATIONS_PER_RACE_RECORDER; n++) { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, true, null); + recorded.incrementAndGet(); + } + }); + recorders.add(recorder); + recorder.start(); + } + // Close in the middle of the burst rather than at the edge of it. Closing before the + // recorders are going would resolve the race trivially every time. + while (recorded.get() < EVALUATIONS_BEFORE_CLOSE) { + Thread.yield(); + } + eventProcessor.close(); + for (Thread recorder : recorders) { + recorder.join(); + } + + long dropped = eventProcessor.getAndClearDroppedCount(); + List events = collectDelivered(server); + int featureEvents = countEventsOfKind(events, "feature"); + // Every evaluation is accounted for one of three ways: delivered in full, dropped + // for capacity, or refused outright before anything was written. Only the first two + // may leave a counter behind, so a counter that matches neither is one that was + // taken from a half-recorded evaluation. + assertEquals("trial " + trial + ": a summary counted an evaluation whose feature" + + " event was neither delivered nor dropped", + featureEvents + dropped, summaryCounters(events, FLAG_KEY)); + featureEventsSeen += featureEvents; + } + } + // Guards against the whole thing passing because nothing ever got as far as being delivered. + assertTrue("no evaluation survived to be delivered, so nothing was actually compared", + featureEventsSeen > 0); + } + + /** + * Like {@code summaryCountFor}, but returns zero rather than failing when no summary was + * delivered. Here a trial in which close() beat every recorder is a legitimate outcome. + */ + private int summaryCounters(List events, String flagKey) { + int total = 0; + for (LDValue event : events) { + if (!"summary".equals(event.get("kind").stringValue())) { + continue; + } + for (LDValue counter : event.get("features").get(flagKey).get("counters").values()) { + total += counter.get("count").intValue(); + } + } + return total; + } + + @Test + public void eventsRecordedWhileOfflineAreRetainedAndSentOnceOnline() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.setOffline(true); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + recordEvaluation(eventProcessor, false, null); + + eventProcessor.blockingFlush(); + server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + + eventProcessor.setOffline(false); + List events = flushAndCollect(eventProcessor, server); + + assertEquals(1, countEventsOfKind(events, "custom")); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void closingWhileOfflineStaysOffTheNetwork() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + eventProcessor.setOffline(true); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + recordEvaluation(eventProcessor, false, null); + + eventProcessor.close(); + + server.getRecorder().requireNoRequests(500, TimeUnit.MILLISECONDS); + } + } + + @Test + public void goingOnlineForTheFirstTimeDoesNotDeliverOnItsOwn() throws Exception { + // What LDClient does at startup: the initial identify is recorded while the processor is + // still offline, and initialization then turns it on. The identify waits to be batched with + // what follows rather than going out alone. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = buildOfflineEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY), true); + try { + eventProcessor.recordIdentifyEvent(CONTEXT); + + eventProcessor.setOffline(false); + + server.getRecorder().requireNoRequests(500, TimeUnit.MILLISECONDS); + List events = flushAndCollect(eventProcessor, server); + assertEquals(1, countEventsOfKind(events, "identify")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void closeCancelsThePeriodicFlushAndNothingAfterItSchedulesAnother() throws Exception { + List> scheduled = Collections.synchronizedList(new ArrayList<>()); + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, + long delay, TimeUnit unit) { + ScheduledFuture future = super.scheduleWithFixedDelay(command, initialDelay, delay, unit); + scheduled.add(future); + return future; + } + }; + // Otherwise shutting the executor down would cancel the task for close(), and this could not + // tell whether close() did. + scheduler.setContinueExistingPeriodicTasksAfterShutdownPolicy(true); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.close(); + eventProcessor.setOffline(true); + eventProcessor.setOffline(false); + eventProcessor.setInBackground(true); + eventProcessor.setInBackground(false); + + assertEquals(1, scheduled.size()); + assertTrue("close() left the periodic flush running", scheduled.get(0).isCancelled()); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void anOutageDoesNotRestartTheFlushInterval() throws Exception { + // Restarting it on every reconnect would let a run of brief outages hold events back for + // far longer than one interval. + AtomicInteger scheduled = new AtomicInteger(); + ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, + long delay, TimeUnit unit) { + scheduled.incrementAndGet(); + return super.scheduleWithFixedDelay(command, initialDelay, delay, unit); + } + }; + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + for (int i = 0; i < 5; i++) { + eventProcessor.setOffline(true); + eventProcessor.setOffline(false); + } + + assertEquals(1, scheduled.get()); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void eventsHeldDuringAnOutageGoOutWithTheNextPeriodicFlush() throws Exception { + BlockingQueue delivered = new LinkedBlockingQueue<>(); + EventSender sender = new StubEventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + delivered.add(data); + return new Result(true, false, null); + } + }; + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, 50, scheduler); + try { + eventProcessor.setOffline(true); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + assertNull("a periodic flush sent events while offline", + delivered.poll(300, TimeUnit.MILLISECONDS)); + + eventProcessor.setOffline(false); + + assertNotNull("the periodic flush did not deliver after the outage", + delivered.poll(2, TimeUnit.SECONDS)); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void debugEventIsSentWhileDebuggingIsActive() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + long debugUntil = System.currentTimeMillis() + 3_600_000; + recordEvaluation(eventProcessor, false, debugUntil); + + List events = flushAndCollect(eventProcessor, server); + + LDValue debugEvent = requireEventOfKind(events, "debug"); + assertEquals(LDValue.of(FLAG_KEY), debugEvent.get("key")); + // A debug event carries the full context rather than just its keys. + assertEquals(LDValue.of(CONTEXT.getKey()), debugEvent.get("context").get("key")); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void debugEventIsNotSentOnceDebuggingHasExpired() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + long debugUntil = System.currentTimeMillis() - 3_600_000; + recordEvaluation(eventProcessor, false, debugUntil); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(0, countEventsOfKind(events, "debug")); + assertEquals(1, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void recordingAfterCloseIsIgnored() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + eventProcessor.recordCustomEvent(CONTEXT, "before-close", LDValue.ofNull(), null); + eventProcessor.close(); + collectDelivered(server); + + eventProcessor.recordCustomEvent(CONTEXT, "after-close", LDValue.ofNull(), null); + eventProcessor.flush(); + + server.getRecorder().requireNoRequests(100, TimeUnit.MILLISECONDS); + } + } + + @Test + public void diagnosticInitEventIsSentWhenComingOnline() throws Exception { + try (HttpServer server = startEventsServer()) { + // makeEventProcessor takes the processor online, which is what triggers the init event. + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY, false); + try { + RequestInfo request = server.getRecorder().requireRequest(10, TimeUnit.SECONDS); + + assertEquals("/mobile/events/diagnostic", request.getPath()); + LDValue body = LDValue.parse(request.getBody()); + assertEquals(LDValue.of("diagnostic-init"), body.get("kind")); + // Only one, even though going online and coming to the foreground both ask for it. + returnToTheForegroundWithTheDiagnosticsThreadFree(eventProcessor); + server.getRecorder().requireNoRequests(500, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void diagnosticInitEventIsNotSentWhileOffline() throws Exception { + // Built offline and left that way. Waiting for the init first, as this used to, means the + // only thing stopping a second one is that the first already went -- so the test passes just + // as happily against a processor that sends init events while offline. + BlockingQueue posted = new LinkedBlockingQueue<>(); + EventSender sender = new StubEventSender() { + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + posted.add(LDValue.parse(new String(data, StandardCharsets.UTF_8))); + return new Result(true, false, null); + } + }; + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, makeDiagnosticStore(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + assertNull("a diagnostic event was sent while offline", + posted.poll(500, TimeUnit.MILLISECONDS)); + + // Proof that the silence above was the offline state and not a fixture that could never + // have sent anything. + eventProcessor.setOffline(false); + assertEquals(LDValue.of("diagnostic-init"), requirePosted(posted).get("kind")); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void beingToldToShutDownStopsRecordingAndDelivery() throws Exception { + // A 401 means the mobile key will not start working again, so the SDK is supposed to stop + // for the life of the process rather than keep posting events nobody will accept. + try (HttpServer server = HttpServer.start(Handlers.status(401))) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent(CONTEXT, "before", LDValue.ofNull(), null); + eventProcessor.blockingFlush(); + server.getRecorder().requireRequest(10, TimeUnit.SECONDS); + + eventProcessor.recordCustomEvent(CONTEXT, "after", LDValue.ofNull(), null); + eventProcessor.blockingFlush(); + + server.getRecorder().requireNoRequests(500, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void aFlushNeverSplitsAnEvaluationAcrossTwoPayloads() throws Exception { + // The other half of the atomicity invariant. close() only ever delivers once, so it can show + // an evaluation being stranded but not one being split: a counter going out in payload N with + // its feature event following in N+1. That leaves the totals correct and each payload wrong, + // so this checks payloads one at a time, against a flush running while recording continues. + // + // Every evaluation is tracked and the capacity is far beyond what the run produces, so within + // a payload the counter for the flag and the number of feature events are the same number. + // + // Repeated because the window is narrow -- the two writes are adjacent, and a flush has to + // land between them. A single run catches a split lock about two times in three. + int payloadsWithCounters = 0; + for (int trial = 0; trial < SPLIT_TRIALS; trial++) { + Queue payloads = new ConcurrentLinkedQueue<>(); + EventSender sender = new StubEventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + payloads.add(LDValue.parse(new String(data, StandardCharsets.UTF_8))); + return new Result(true, false, null); + } + }; + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = + makeEventProcessorWithCapacity(sender, NO_DROP_CAPACITY, scheduler); + try { + eventProcessor.setOffline(false); + AtomicInteger recorded = new AtomicInteger(); + List recorders = new ArrayList<>(); + for (int i = 0; i < RACE_RECORDERS; i++) { + Thread recorder = new Thread(() -> { + for (int n = 0; n < EVALUATIONS_PER_RACE_RECORDER; n++) { + recordEvaluation(eventProcessor, true, null); + recorded.incrementAndGet(); + } + }); + recorders.add(recorder); + recorder.start(); + } + int target = RACE_RECORDERS * EVALUATIONS_PER_RACE_RECORDER; + while (recorded.get() < target) { + eventProcessor.blockingFlush(); + } + for (Thread recorder : recorders) { + recorder.join(); + } + eventProcessor.blockingFlush(); + + assertEquals("trial " + trial + ": events were dropped, so a payload may be short" + + " for that reason instead", + 0, eventProcessor.getAndClearDroppedCount()); + int counted = 0; + for (LDValue payload : payloads) { + List events = new ArrayList<>(); + for (LDValue event : payload.values()) { + events.add(event); + } + int counters = summaryCounters(events, FLAG_KEY); + assertEquals("trial " + trial + ": a payload counted evaluations whose feature" + + " events went out separately", + countEventsOfKind(events, "feature"), counters); + counted += counters; + if (counters > 0) { + payloadsWithCounters++; + } + } + assertEquals("trial " + trial + ": some evaluations never reached a payload", + target, counted); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + // Otherwise a single delivery per trial would satisfy everything above without a flush ever + // having overlapped a recording. + assertTrue("every evaluation went out in one payload, so nothing was interleaved", + payloadsWithCounters > SPLIT_TRIALS); + } + + @Test + public void periodicFlushSurvivesErrorFromSender() throws Exception { + // An Error (not Exception) from a scheduled run used to cancel the repeating future with + // nothing logged, after which enableOrDisableTask kept returning that dead future forever. + CountDownLatch firstAttempt = new CountDownLatch(1); + BlockingQueue delivered = new LinkedBlockingQueue<>(); + EventSender sender = new StubEventSender() { + private final AtomicInteger attempts = new AtomicInteger(); + + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + if (attempts.getAndIncrement() == 0) { + firstAttempt.countDown(); + throw new Error("periodic flush"); + } + delivered.add(data); + return new Result(true, false, null); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, 40, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.blockingFlush(); + + eventProcessor.recordCustomEvent(CONTEXT, "before-error", LDValue.ofNull(), null); + assertTrue("sender never saw the first periodic flush", + firstAttempt.await(2, TimeUnit.SECONDS)); + + // A background toggle stays online, so nothing here reschedules on the processor's + // behalf. The periodic series has to still be alive on its own. + eventProcessor.setInBackground(true); + eventProcessor.setInBackground(false); + + eventProcessor.recordCustomEvent(CONTEXT, "after-error", LDValue.ofNull(), null); + byte[] payload = delivered.poll(2, TimeUnit.SECONDS); + assertNotNull("periodic flush did not run again after Error", payload); + assertTrue(new String(payload, StandardCharsets.UTF_8).contains("after-error")); + logging.assertErrorLogged("Unexpected error in event processor"); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void unexpectedRecordingErrorDoesNotBubbleToCallerAndLogs() throws Exception { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + // Must not throw if record throws: + eventProcessor.record(new Event(System.currentTimeMillis(), CONTEXT) { + @Override + public long getSamplingRatio() { + throw new RuntimeException("simulated record crash"); + } + }); + logging.assertErrorLogged("Unexpected error in event processor: java.lang.RuntimeException: simulated record crash"); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void anErrorWhileRecordingStillReachesTheCaller() throws Exception { + // Only exceptions are the SDK's to absorb. An Error such as OutOfMemoryError belongs to the + // application's crash reporting, and swallowing it on the caller's thread would hide it. + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + eventProcessor.record(new Event(System.currentTimeMillis(), CONTEXT) { + @Override + public long getSamplingRatio() { + throw new StackOverflowError("simulated"); + } + }); + fail("the Error was swallowed"); + } catch (StackOverflowError expected) { + // what the application's handler would see + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void closeGivesUpWaitingOnAStalledDelivery() throws Exception { + // close() runs on the caller's thread, usually the main one, so a send that never comes back + // used to park the application there for as long as the HTTP timeouts allowed. + CountDownLatch sendStarted = new CountDownLatch(1); + CountDownLatch releaseSend = new CountDownLatch(1); + EventSender sender = new StubEventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + sendStarted.countDown(); + // Bounded so that a regression fails on the elapsed time rather than hanging until + // the suite's global timeout. + awaitQuietly(releaseSend, 5, TimeUnit.SECONDS); + return new Result(true, false, null); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, NO_PERIODIC_FLUSH_MILLIS, + CLOSE_BUDGET_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordCustomEvent(CONTEXT, "stalled", LDValue.ofNull(), null); + + long startedAtNanos = System.nanoTime(); + eventProcessor.close(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAtNanos); + + assertTrue("the sender was never asked to send anything", + sendStarted.await(2, TimeUnit.SECONDS)); + assertTrue("close() waited " + elapsedMillis + "ms on a budget of " + CLOSE_BUDGET_MILLIS + + "ms", elapsedMillis < CLOSE_BUDGET_MILLIS * 5); + logging.assertWarnLogged("Gave up waiting for the final event delivery"); + } finally { + releaseSend.countDown(); + scheduler.shutdownNow(); + } + } + + @Test + public void closeReleasesTheSenderOnlyAfterTheLastDeliveryFinishes() throws Exception { + // Giving up on the wait must not turn into pulling the HTTP client out from under the + // delivery we just decided not to wait for. + CountDownLatch releaseSend = new CountDownLatch(1); + CountDownLatch senderClosed = new CountDownLatch(1); + AtomicBoolean posting = new AtomicBoolean(false); + AtomicBoolean closedMidPost = new AtomicBoolean(false); + EventSender sender = new StubEventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + posting.set(true); + awaitQuietly(releaseSend, 5, TimeUnit.SECONDS); + posting.set(false); + return new Result(true, false, null); + } + + @Override + public void close() { + if (posting.get()) { + closedMidPost.set(true); + } + senderClosed.countDown(); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, NO_PERIODIC_FLUSH_MILLIS, + CLOSE_BUDGET_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordCustomEvent(CONTEXT, "stalled", LDValue.ofNull(), null); + + eventProcessor.close(); + assertEquals("the sender was closed while a delivery was still in flight", + 1, senderClosed.getCount()); + + releaseSend.countDown(); + assertTrue("the sender was never closed once the delivery finished", + senderClosed.await(2, TimeUnit.SECONDS)); + assertFalse("the sender was closed while a delivery was posting through it", + closedMidPost.get()); + } finally { + releaseSend.countDown(); + scheduler.shutdownNow(); + } + } + + @Test + public void flushAfterCloseDoesNotPostThroughTheReleasedSender() throws Exception { + AtomicBoolean postedAfterRelease = new AtomicBoolean(false); + EventSender sender = releaseTrackingSender(postedAfterRelease); + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, NO_PERIODIC_FLUSH_MILLIS, + CLOSE_BUDGET_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.close(); + + eventProcessor.recordCustomEvent(CONTEXT, "after-close", LDValue.ofNull(), null); + eventProcessor.flush(); + eventProcessor.blockingFlush(); + + assertFalse("a flush after close posted through a sender that had been released", + postedAfterRelease.get()); + } finally { + scheduler.shutdownNow(); + } + } + + /** Reports through {@code postedAfterRelease} if it is asked to send once it has been closed. */ + private static EventSender releaseTrackingSender(AtomicBoolean postedAfterRelease) { + AtomicBoolean released = new AtomicBoolean(false); + return new StubEventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + if (released.get()) { + postedAfterRelease.set(true); + } + return new Result(true, false, null); + } + + @Override + public void close() { + released.set(true); + } + }; + } + + @Test + public void stalledDiagnosticPostDoesNotHoldUpAnalyticsDelivery() throws Exception { + // Diagnostics used to share the one thread that delivers analytics, so a post against a + // network that accepts connections and never answers stalled every flush behind it, and a + // buffer that is not being drained fills up and drops what the application asked to send. + CountDownLatch diagnosticStarted = new CountDownLatch(1); + CountDownLatch releaseDiagnostic = new CountDownLatch(1); + BlockingQueue analyticsDelivered = new LinkedBlockingQueue<>(); + EventSender sender = new StubEventSender() { + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + diagnosticStarted.countDown(); + awaitQuietly(releaseDiagnostic, 5, TimeUnit.SECONDS); + return new Result(true, false, null); + } + + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + analyticsDelivered.add(data); + return new Result(true, false, null); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(sender, makeDiagnosticStore(), + NO_PERIODIC_FLUSH_MILLIS, scheduler); + try { + // Coming online posts the diagnostic init event, which then never comes back. + eventProcessor.setOffline(false); + assertTrue("the diagnostic event was never posted", + diagnosticStarted.await(2, TimeUnit.SECONDS)); + + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + eventProcessor.flush(); + + byte[] payload = analyticsDelivered.poll(2, TimeUnit.SECONDS); + assertNotNull("analytics delivery was stuck behind the diagnostic post", payload); + assertTrue(new String(payload, StandardCharsets.UTF_8).contains("an-event")); + } finally { + releaseDiagnostic.countDown(); + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void diagnosticEventIsDroppedRatherThanQueuedBehindOneStillPosting() throws Exception { + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + AtomicInteger posts = new AtomicInteger(); + EventSender sender = new StubEventSender() { + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + posts.incrementAndGet(); + firstStarted.countDown(); + awaitQuietly(releaseFirst, 5, TimeUnit.SECONDS); + return new Result(true, false, null); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + // A diagnostic interval short enough that the periodic task fires repeatedly while the init + // event is still stuck on the posting thread. + DirectEventProcessor eventProcessor = makeEventProcessor(sender, makeDiagnosticStore(), + NO_PERIODIC_FLUSH_MILLIS, 20, scheduler); + try { + eventProcessor.setOffline(false); + assertTrue("the diagnostic event was never posted", + firstStarted.await(2, TimeUnit.SECONDS)); + + // Long enough for many periodic runs, every one of which has to be turned away rather + // than left on the posting thread's queue. Counting posts is not enough on its own: a + // queued one would not have started yet either, so the skips are what distinguishes + // being dropped from merely waiting. + Thread.sleep(300); + assertTrue("no diagnostic event was turned away, so they were queueing up instead", + countLogged("Skipped a diagnostic event") > 0); + assertEquals("more than one diagnostic event reached the sender", 1, posts.get()); + } finally { + releaseFirst.countDown(); + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void goingToTheBackgroundDefersADiagnosticPeriodRatherThanDestroyingIt() throws Exception { + // createEventAndReset hands the period's statistics back and clears them in the same call, so + // whoever consumes it owns it. The post reaches the diagnostics thread through a queue it may + // have to wait in, and the SDK can go offline or to the background in the meantime. If that is + // noticed after the event was built, the period is gone: nothing retries it, and the next + // event reports a window starting after the reset, so the statistics are not merely late but + // absent. + BlockingQueue posted = new LinkedBlockingQueue<>(); + EventSender sender = new StubEventSender() { + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + posted.add(LDValue.parse(new String(data, StandardCharsets.UTF_8))); + return new Result(true, false, null); + } + }; + + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + ExecutorService diagnosticExecutor = EventUtil.makeDiagnosticsTaskExecutor(); + diagnosticExecutors.add(diagnosticExecutor); + // Short enough that the periodic task fires while the diagnostics thread is held. + DirectEventProcessor eventProcessor = makeEventProcessor(sender, makeDiagnosticStore(), + NO_PERIODIC_FLUSH_MILLIS, 20, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler, diagnosticExecutor); + try { + eventProcessor.setOffline(false); + assertEquals(LDValue.of("diagnostic-init"), + requirePosted(posted).get("kind")); + + // Something for the period to have in it. Capacity is what makes these countable, and the + // dropped count is carried by the periodic event and nothing else. Far past capacity so + // that drops happen whatever else is draining the buffer. + for (int i = 0; i < DEFAULT_CAPACITY * 10; i++) { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + } + logging.assertWarnLogged("Exceeded event queue capacity"); + + // Holds the diagnostics thread so the next periodic post has to queue behind it, which is + // the window the state can change in. + CountDownLatch releaseThread = new CountDownLatch(1); + diagnosticExecutor.submit(() -> awaitQuietly(releaseThread, 5, TimeUnit.SECONDS)); + // A later run being turned away is proof that an earlier one took the claim and is sitting + // on the queue, which is what we need before changing the state underneath it. + awaitLogged("Skipped a diagnostic event"); + + eventProcessor.setInBackground(true); + releaseThread.countDown(); + // The queued post now runs suspended. Nothing should reach the sender. + assertNull("a diagnostic event was posted from the background", + posted.poll(300, TimeUnit.MILLISECONDS)); + + eventProcessor.setInBackground(false); + LDValue statistics = requirePosted(posted); + + assertEquals(LDValue.of("diagnostic"), statistics.get("kind")); + // Nothing is recorded after the suspension, so every drop this could report happened + // before it. A destroyed period therefore reports exactly zero, which is what separates + // the two outcomes. The exact figure is not asserted because it is not the same on every + // tier: where events are staged to a store, reaching it frees capacity as we go. + assertTrue("the suspended period was destroyed rather than carried forward", + statistics.get("droppedEvents").longValue() > 0); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + + private LDValue requirePosted(BlockingQueue posted) throws InterruptedException { + LDValue event = posted.poll(5, TimeUnit.SECONDS); + assertNotNull("no diagnostic event was posted", event); + return event; + } + + /** + * Goes to the background and back, leaving the processor at the point where it decides whether + * to send a second init event. + *

+ * Two things have to be true for that decision to be reached, and only one of them is under the + * test's control. Going to the foreground has to be a real transition, because setInBackground + * returns immediately when the value is unchanged. And the diagnostics thread has to be free: + * the first init event holds a claim on it until its post returns, which is after the request + * reaches the server, so a transition right after the request arrives is turned away before it + * gets anywhere near the decision. Being turned away is logged, which is what this waits out. + */ + private void returnToTheForegroundWithTheDiagnosticsThreadFree(EventProcessor eventProcessor) + throws InterruptedException { + long deadline = System.currentTimeMillis() + 5000; + while (System.currentTimeMillis() < deadline) { + long skipsBefore = countLogged("Skipped a diagnostic event"); + eventProcessor.setInBackground(true); + eventProcessor.setInBackground(false); + // Logged by the claim itself, on this thread, so it is already there if it happened. + if (countLogged("Skipped a diagnostic event") == skipsBefore) { + return; + } + Thread.sleep(10); + } + throw new AssertionError("the diagnostics thread never came free"); + } + + private void awaitLogged(String messageSubstring) throws InterruptedException { + long deadline = System.currentTimeMillis() + 5000; + while (System.currentTimeMillis() < deadline) { + if (countLogged(messageSubstring) > 0) { + return; + } + Thread.sleep(10); + } + throw new AssertionError("never logged: " + messageSubstring); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, long flushIntervalMillis, + ScheduledExecutorService scheduler) { + return makeEventProcessor(sender, flushIntervalMillis, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, scheduler); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, long flushIntervalMillis, + long closeBudgetMillis, + ScheduledExecutorService scheduler) { + return makeEventProcessor(sender, null, flushIntervalMillis, 60_000, closeBudgetMillis, + scheduler); + } + + /** + * For the one test whose subject is what a payload contains rather than how much fits. Named + * rather than overloaded: an int alongside the flush-interval long would quietly take over the + * calls that pass an interval as a literal. + */ + private DirectEventProcessor makeEventProcessorWithCapacity(EventSender sender, int capacity, + ScheduledExecutorService scheduler) { + ExecutorService diagnosticExecutor = EventUtil.makeDiagnosticsTaskExecutor(); + diagnosticExecutors.add(diagnosticExecutor); + return makeEventProcessor(sender, null, NO_PERIODIC_FLUSH_MILLIS, 60_000, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, scheduler, diagnosticExecutor, + capacity); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, + DiagnosticStore diagnosticStore, + long flushIntervalMillis, + ScheduledExecutorService scheduler) { + return makeEventProcessor(sender, diagnosticStore, flushIntervalMillis, 60_000, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, scheduler); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, + DiagnosticStore diagnosticStore, + long flushIntervalMillis, + long diagnosticIntervalMillis, + ScheduledExecutorService scheduler) { + return makeEventProcessor(sender, diagnosticStore, flushIntervalMillis, + diagnosticIntervalMillis, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, + DiagnosticStore diagnosticStore, + long flushIntervalMillis, + long diagnosticIntervalMillis, + long closeBudgetMillis, + ScheduledExecutorService scheduler) { + ExecutorService diagnosticExecutor = EventUtil.makeDiagnosticsTaskExecutor(); + diagnosticExecutors.add(diagnosticExecutor); + return makeEventProcessor(sender, diagnosticStore, flushIntervalMillis, + diagnosticIntervalMillis, closeBudgetMillis, scheduler, diagnosticExecutor); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, + DiagnosticStore diagnosticStore, + long flushIntervalMillis, + long diagnosticIntervalMillis, + long closeBudgetMillis, + ScheduledExecutorService scheduler, + ExecutorService diagnosticExecutor) { + return makeEventProcessor(sender, diagnosticStore, flushIntervalMillis, + diagnosticIntervalMillis, closeBudgetMillis, scheduler, diagnosticExecutor, + DEFAULT_CAPACITY); + } + + private DirectEventProcessor makeEventProcessor(EventSender sender, + DiagnosticStore diagnosticStore, + long flushIntervalMillis, + long diagnosticIntervalMillis, + long closeBudgetMillis, + ScheduledExecutorService scheduler, + ExecutorService diagnosticExecutor, + int capacity) { + return new DirectEventProcessor( + new OutboundEventBuffer(false, Collections.emptyList(), true, capacity, + logging.logger), + sender, + URI.create("https://events.example"), + diagnosticStore, + capacity, + flushIntervalMillis, + diagnosticIntervalMillis, + closeBudgetMillis, + false, + true, // initiallyOffline, as the SDK builds it + scheduler, + diagnosticExecutor, + logging.logger); + } + + private long countLogged(String messageSubstring) { + long count = 0; + for (String message : logging.logCapture.getMessageStrings()) { + if (message.contains(messageSubstring)) { + count++; + } + } + return count; + } + + private DiagnosticStore makeDiagnosticStore() { + return new DiagnosticStore(new DiagnosticStore.SdkDiagnosticParams(MOBILE_KEY, + "android-client-sdk", "0.0.0", "Android", null, Collections.emptyMap(), null)); + } + + private static void awaitQuietly(CountDownLatch latch, long timeout, TimeUnit unit) { + try { + latch.await(timeout, unit); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** Accepts everything, so that a test overrides only the one method it is about. */ + private static class StubEventSender implements EventSender { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + return new Result(true, false, null); + } + + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + return new Result(true, false, null); + } + + @Override + public void close() {} + } + + private void recordEvaluation(EventProcessor eventProcessor, boolean requireFullEvent, + Long debugEventsUntilDate) { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION, FLAG_VALUE, + null, DEFAULT_VALUE, requireFullEvent, debugEventsUntilDate); + } +} diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorBufferingTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorBufferingTest.java new file mode 100644 index 000000000..fc2a14729 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorBufferingTest.java @@ -0,0 +1,299 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; + +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.subsystems.EventProcessor; +import com.launchdarkly.testhelpers.httptest.HttpServer; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +/** + * Tests for what the event processor is allowed to drop when flag evaluations arrive faster than + * it can consume them. + *

+ * The processor used to be a thin wrapper around java-sdk-internal's + * {@code DefaultEventProcessor}, which buffers in two stages: every recorded event is first + * offered to a bounded "inbox" queue, and only after a background dispatcher thread picks it up is + * it either summarized or placed in the outbox that a flush actually drains. The inbox is sized to + * the configured event capacity, and it carries the processor's own FLUSH and SHUTDOWN control + * messages as well as events. So a burst of evaluations could fill it and cause silent drops even + * though those evaluations were only ever going to contribute a counter to a summary event, and + * even though unrelated custom events were competing for the same slots. + *

+ * These tests pin the behavior we want instead: summarization happens when the evaluation is + * recorded, so an evaluation never occupies a queue slot, and capacity only ever limits + * full-fidelity events. With summarization ahead of buffering there is no queue in this path, so + * the counts asserted below are exact rather than approximate. + */ +public class EventProcessorBufferingTest extends EventProcessorTestBase { + private static final String FLAG_KEY = "burst-flag"; + private static final int FLAG_VERSION = 10; + private static final int VARIATION = 1; + private static final LDValue FLAG_VALUE = LDValue.of(true); + private static final LDValue DEFAULT_VALUE = LDValue.of(false); + + // Deliberately far smaller than the number of evaluations each test records, so that any + // capacity-limited queue in the evaluation path would be guaranteed to overflow. + private static final int CAPACITY = 100; + + // For the tests that fill the summarizer to its context limit. Each counted context becomes its + // own summary event, so the flush is roughly this many summaries; at CAPACITY that came to about + // 28 KB, past what the test server reliably records with its single unlooped read of the body. + private static final int CONTEXT_LIMIT = 10; + + private static final int BURST_THREADS = 4; + private static final int EVALUATIONS_PER_THREAD = 25_000; + private static final int TOTAL_EVALUATIONS = BURST_THREADS * EVALUATIONS_PER_THREAD; + + // How many full-fidelity events the interleaving tests space through the burst. Comfortably + // under CAPACITY, so the buffer is never legitimately entitled to drop any of them. + private static final int INTERLEAVED_EVENTS = 50; + + @Test + public void summaryCountsSurviveEvaluationBurst() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CAPACITY); + try { + new EvaluationBurst(eventProcessor).join(); + + List events = flushAndCollect(eventProcessor, server); + + // Every evaluation contributes to the summary regardless of capacity, because a + // summary counter is not a buffered event. + assertEquals(TOTAL_EVALUATIONS, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void customEventsSurviveEvaluationBurst() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CAPACITY); + try { + // Record custom events spaced evenly through the burst. These are full-fidelity + // events and there are far fewer of them than the configured capacity, so none of + // them may be lost no matter how many evaluations are happening alongside them. + EvaluationBurst burst = new EvaluationBurst(eventProcessor); + for (int i = 1; i <= INTERLEAVED_EVENTS; i++) { + burst.awaitFraction(i, INTERLEAVED_EVENTS + 1); + eventProcessor.recordCustomEvent(CONTEXT, "burst-custom", LDValue.ofNull(), null); + } + burst.join(); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(INTERLEAVED_EVENTS, countEventsOfKind(events, "custom")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void fullFidelityEvaluationEventsSurviveEvaluationBurst() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CAPACITY); + try { + // A flag with trackEvents on produces a full feature event in addition to its + // summary counter. Evaluations of untracked flags must not crowd these out. + EvaluationBurst burst = new EvaluationBurst(eventProcessor); + for (int i = 1; i <= INTERLEAVED_EVENTS; i++) { + burst.awaitFraction(i, INTERLEAVED_EVENTS + 1); + eventProcessor.recordEvaluationEvent(CONTEXT, "tracked-flag", FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, true, null); + } + burst.join(); + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(INTERLEAVED_EVENTS, countEventsOfKind(events, "feature")); + // The tracked evaluations are summarized too, alongside the burst. + assertEquals(INTERLEAVED_EVENTS, summaryCountFor(events, "tracked-flag")); + assertEquals(TOTAL_EVALUATIONS, summaryCountFor(events, FLAG_KEY)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void distinctSummarizedContextsAreBoundedByCapacity() throws Exception { + // A summary counter costs a retained context and a set of counters per distinct context, and + // nothing drains it while the client is offline. Capacity has to bound that, or an application + // that goes on identifying through an outage grows the summarizer for as long as it lasts. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CONTEXT_LIMIT); + try { + for (int i = 0; i < CONTEXT_LIMIT * 3; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + + List events = flushAndCollect(eventProcessor, server); + + // One summary event per context that was counted, so this is the cardinality the + // summarizer was holding. + assertEquals(CONTEXT_LIMIT, countEventsOfKind(events, "summary")); + logging.assertWarnLogged("Exceeded the number of contexts"); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void exceedingContextLimitIsWarnedOncePerFlushRun() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CONTEXT_LIMIT); + try { + for (int i = 0; i < CONTEXT_LIMIT * 3; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + flushAndCollect(eventProcessor, server); + + for (int i = 0; i < CONTEXT_LIMIT * 3; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(100 + i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + flushAndCollect(eventProcessor, server); + + int warnCount = 0; + for (String msg : logging.logCapture.getMessageStrings()) { + if (msg.contains("Exceeded the number of contexts")) { + warnCount++; + } + } + assertEquals(2, warnCount); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void aContextAlreadyBeingCountedKeepsCountingOnceTheLimitIsReached() throws Exception { + // The limit is on how many contexts are held, not on how many evaluations are counted. An + // application evaluating against one context must not start losing counts because some other + // part of it churned through contexts. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CONTEXT_LIMIT); + try { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION, + FLAG_VALUE, null, DEFAULT_VALUE, false, null); + for (int i = 0; i < CONTEXT_LIMIT * 3; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + int evaluationsAfterTheLimit = 500; + for (int i = 0; i < evaluationsAfterTheLimit; i++) { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION, + FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + + List events = flushAndCollect(eventProcessor, server); + + assertEquals(evaluationsAfterTheLimit + 1, summaryCountForContext(events, CONTEXT)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void theLimitLiftsOnceTheSummariesHaveBeenDelivered() throws Exception { + // The bound is on how many contexts are held at once, not on how many an application may ever + // use, so a delivery has to make room for the next set. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, CONTEXT_LIMIT); + try { + for (int i = 0; i < CONTEXT_LIMIT * 3; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + flushAndCollect(eventProcessor, server); + + for (int i = CONTEXT_LIMIT * 3; i < CONTEXT_LIMIT * 4; i++) { + eventProcessor.recordEvaluationEvent(contextNumber(i), FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + } + List events = flushAndCollect(eventProcessor, server); + + assertEquals(CONTEXT_LIMIT, countEventsOfKind(events, "summary")); + } finally { + eventProcessor.close(); + } + } + } + + private static LDContext contextNumber(int i) { + return LDContext.builder("burst-user-" + i).set("email", "user" + i + "@example.com").build(); + } + + /** @return how many evaluations the summary for this context counted, across every flag in it */ + private static int summaryCountForContext(List events, LDContext context) { + int total = 0; + for (LDValue event : events) { + if (!"summary".equals(event.get("kind").stringValue()) + || !context.getKey().equals(event.get("context").get("key").stringValue())) { + continue; + } + LDValue features = event.get("features"); + for (String flagKey : features.keys()) { + for (LDValue counter : features.get(flagKey).get("counters").values()) { + total += counter.get("count").intValue(); + } + } + } + return total; + } + + /** + * Records evaluations of an untracked flag from several threads at once, faster than any + * single consumer thread could drain them. Tests that need to interleave other events use + * {@link #awaitFraction} rather than sleeping, so the interleaved events land at the same + * points in the burst regardless of how fast the machine running the test is. + */ + private final class EvaluationBurst { + private final List workers = new ArrayList<>(); + private volatile int progress; + + EvaluationBurst(EventProcessor eventProcessor) { + for (int i = 0; i < BURST_THREADS; i++) { + boolean reportsProgress = i == 0; + Thread worker = new Thread(() -> { + for (int j = 0; j < EVALUATIONS_PER_THREAD; j++) { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, + VARIATION, FLAG_VALUE, null, DEFAULT_VALUE, false, null); + if (reportsProgress) { + progress = j + 1; + } + } + }); + worker.start(); + workers.add(worker); + } + } + + /** Blocks until the burst is {@code numerator/denominator} of the way through. */ + void awaitFraction(int numerator, int denominator) { + int target = (int) ((long) EVALUATIONS_PER_THREAD * numerator / denominator); + while (progress < target && workers.get(0).isAlive()) { + Thread.yield(); + } + } + + void join() throws InterruptedException { + for (Thread worker : workers) { + worker.join(); + } + } + } +} diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorFlagsTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorFlagsTest.java new file mode 100644 index 000000000..5528e1ed5 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorFlagsTest.java @@ -0,0 +1,173 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.internal.events.Event; +import com.launchdarkly.testhelpers.httptest.HttpServer; + +import org.junit.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Tests for the two per-event flags the SDK is expected to honour: {@code excludeFromSummaries}, + * which keeps an evaluation out of the counters, and {@code samplingRatio}, which decides whether a + * full event is kept at all. + *

+ * The two are enforced at different stages, because that is where each one's subject matter lives. + * The counters are in {@link OutboundEventBuffer}, so the exclusion is checked there. The run of + * full events and the capacity counted against it are in {@link DirectEventProcessor}, so sampling + * is checked there, ahead of the capacity check that would otherwise record a sampled-out event as + * a drop. + *

+ * Neither flag can be reached through the SDK's own recording path. {@link DirectEventProcessor} + * builds its feature events with the constructor overload that fixes them at {@code false} and + * {@code 1}, so every evaluation an application performs leaves both at their defaults. That is why + * these tests hand an event straight to the stage that enforces the flag rather than going through + * {@code recordEvaluationEvent}: those are the only seams at which the flags can vary. + *

+ * They are worth pinning even so. java-sdk-internal's {@code DefaultEventProcessor}, which this + * path replaced, honoured both, and anything that later constructs feature events through the + * fourteen-argument constructor — adopting sampling is the obvious candidate — makes both live at + * once. The failure would be silent and would show up as wrong analytics rather than as an error, + * since an event carries its sampling ratio on the wire so the receiver can scale by it. + *

+ * Both ratios used here are deterministic rather than probabilistic: {@code Sampler.shouldSample} + * short-circuits to {@code true} at 1 and to {@code false} at 0, so nothing below is flaky. + */ +public class EventProcessorFlagsTest extends EventProcessorTestBase { + private static final String FLAG_KEY = "flags-test-flag"; + private static final int FLAG_VERSION = 3; + private static final int VARIATION = 1; + private static final LDValue FLAG_VALUE = LDValue.of(true); + private static final LDValue DEFAULT_VALUE = LDValue.of(false); + private static final long CREATION_DATE = 1_700_000_000_000L; + + // Far more than any test here records, so that capacity can never be what discarded an event. + private static final int CAPACITY = 100; + + private static final List NO_EVENTS = Collections.emptyList(); + + @Test + public void anEvaluationExcludedFromSummariesIsNotCounted() throws IOException { + OutboundEventBuffer buffer = makeBuffer(); + + buffer.summarize(evaluation(true)); + + assertNull("an excluded evaluation should leave nothing to send", encodeAll(buffer)); + } + + @Test + public void anEvaluationNotExcludedFromSummariesIsCounted() throws IOException { + OutboundEventBuffer buffer = makeBuffer(); + + buffer.summarize(evaluation(false)); + + assertEquals(1, summaryCountFor(drainToEvents(buffer), FLAG_KEY)); + } + + @Test + public void excludingOneEvaluationLeavesTheOthersCounted() throws IOException { + OutboundEventBuffer buffer = makeBuffer(); + + buffer.summarize(evaluation(false)); + buffer.summarize(evaluation(true)); + buffer.summarize(evaluation(false)); + + // The exclusion applies to the event that carried it and to nothing else, which is what + // distinguishes honouring the flag from dropping the whole summary. + assertEquals(2, summaryCountFor(drainToEvents(buffer), FLAG_KEY)); + } + + @Test + public void aFullEventSampledOutIsNotKept() throws Exception { + try (HttpServer server = startEventsServer()) { + DirectEventProcessor processor = processor(server); + try { + processor.record(evaluation(false, 0)); + // Something the SDK will certainly send, so the flush has a payload to look at. + // Without it there would be no request at all, and nothing to distinguish an event + // that was dropped from one that simply has not been sent yet. + processor.recordIdentifyEvent(CONTEXT); + + List events = flushAndCollect(processor, server); + + assertEquals(0, countEventsOfKind(events, "feature")); + assertEquals(1, countEventsOfKind(events, "identify")); + } finally { + processor.close(); + } + } + } + + @Test + public void aFullEventSampledInIsKept() throws Exception { + try (HttpServer server = startEventsServer()) { + DirectEventProcessor processor = processor(server); + try { + processor.record(evaluation(false, 1)); + + assertEquals(1, countEventsOfKind(flushAndCollect(processor, server), "feature")); + } finally { + processor.close(); + } + } + } + + @Test + public void aSampledOutEventIsNotCountedAsADrop() throws Exception { + try (HttpServer server = startEventsServer()) { + DirectEventProcessor processor = processor(server); + try { + processor.record(evaluation(false, 0)); + + // Sampling and capacity are different reasons not to keep an event, and only the + // second is a loss the SDK owes anyone a count of. + assertEquals(0, processor.getAndClearDroppedCount()); + } finally { + processor.close(); + } + } + } + + private DirectEventProcessor processor(HttpServer server) { + return (DirectEventProcessor) makeEventProcessor(server, CAPACITY); + } + + private OutboundEventBuffer makeBuffer() { + return new OutboundEventBuffer(false, Collections.emptyList(), false, 100, logging.logger); + } + + private Event.FeatureRequest evaluation(boolean excludeFromSummaries) { + return evaluation(excludeFromSummaries, 1); + } + + private Event.FeatureRequest evaluation(boolean excludeFromSummaries, long samplingRatio) { + return new Event.FeatureRequest(CREATION_DATE, FLAG_KEY, CONTEXT, FLAG_VERSION, VARIATION, + FLAG_VALUE, DEFAULT_VALUE, null, null, true, null, false, samplingRatio, + excludeFromSummaries); + } + + private OutboundEventBuffer.Payload encodeAll(OutboundEventBuffer buffer) throws IOException { + return buffer.encode(NO_EVENTS, buffer.takeSummaries()); + } + + private List drainToEvents(OutboundEventBuffer buffer) throws IOException { + OutboundEventBuffer.Payload payload = encodeAll(buffer); + if (payload == null) { + return Collections.emptyList(); + } + List events = new ArrayList<>(); + for (LDValue event : LDValue.parse(new String(payload.getData(), StandardCharsets.UTF_8)) + .values()) { + events.add(event); + } + return events; + } +} diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorPrivacyTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorPrivacyTest.java new file mode 100644 index 000000000..5ff0bcba3 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorPrivacyTest.java @@ -0,0 +1,269 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.launchdarkly.sdk.ContextKind; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.subsystems.EventProcessor; +import com.launchdarkly.testhelpers.httptest.HttpServer; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +/** + * Context attribute redaction, as observed in the payload the processor actually posts. + *

+ * Redaction is not reimplemented by the SDK's event processor - it reuses java-sdk-internal's + * context formatter through {@link OutboundEventBuffer}. These + * tests exist because that reuse is the whole argument for the current design, and nothing else in + * this repository asserts that an attribute an application marked private stays out of the wire + * format. Beyond the redaction itself they pin the two rules that are easy to get wrong when + * events are assembled somewhere new: which events inline a context at all, and which of them + * additionally redact the attributes of an anonymous context. + */ +public class EventProcessorPrivacyTest extends EventProcessorTestBase { + private static final String FLAG_KEY = "flag-key"; + private static final int DEFAULT_CAPACITY = 100; + + private static final LDContext PERSON = LDContext.builder("user-key") + .name("Sandy") + .set("email", "sandy@example.com") + .set("address", LDValue.buildObject() + .put("city", "Oakland") + .put("street", "123 Main St") + .build()) + .build(); + + @Test + public void namedPrivateAttributesAreRedactedAndReported() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).privateAttributes("email"), true); + try { + eventProcessor.recordIdentifyEvent(PERSON); + + LDValue context = requireEventOfKind(flushAndCollect(eventProcessor, server), + "identify").get("context"); + + assertEquals(LDValue.of("user-key"), context.get("key")); + assertEquals("email must not appear in the payload", + LDValue.ofNull(), context.get("email")); + // Attributes that were not marked private are still sent. + assertEquals(LDValue.of("Sandy"), context.get("name")); + assertEquals(redacted("email"), redactedAttributes(context)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void allAttributesPrivateLeavesOnlyTheKey() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).allAttributesPrivate(true), true); + try { + eventProcessor.recordIdentifyEvent(PERSON); + + LDValue context = requireEventOfKind(flushAndCollect(eventProcessor, server), + "identify").get("context"); + + // The key and kind are identifiers rather than attributes, so they survive. + assertEquals(LDValue.of("user-key"), context.get("key")); + assertEquals(LDValue.of("user"), context.get("kind")); + assertEquals(LDValue.ofNull(), context.get("name")); + assertEquals(LDValue.ofNull(), context.get("email")); + assertEquals(LDValue.ofNull(), context.get("address")); + assertEquals(redacted("address", "email", "name"), redactedAttributes(context)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void aPrivateSubAttributeRedactsOnlyThatProperty() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).privateAttributes("/address/street"), true); + try { + eventProcessor.recordIdentifyEvent(PERSON); + + LDValue context = requireEventOfKind(flushAndCollect(eventProcessor, server), + "identify").get("context"); + + assertEquals(LDValue.of("Oakland"), context.get("address").get("city")); + assertEquals(LDValue.ofNull(), context.get("address").get("street")); + assertEquals(redacted("/address/street"), redactedAttributes(context)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void attributesMarkedPrivateOnTheContextItselfAreRedacted() throws Exception { + // An application can mark an attribute private per-context instead of globally; the + // processor is not configured with anything in this case. + LDContext context = LDContext.builder("user-key") + .name("Sandy") + .set("email", "sandy@example.com") + .privateAttributes("email") + .build(); + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordIdentifyEvent(context); + + LDValue delivered = requireEventOfKind(flushAndCollect(eventProcessor, server), + "identify").get("context"); + + assertEquals(LDValue.ofNull(), delivered.get("email")); + assertEquals(LDValue.of("Sandy"), delivered.get("name")); + assertEquals(redacted("email"), redactedAttributes(delivered)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void anonymousContextAttributesAreRedactedFromFeatureEventsButNotIdentifyEvents() throws Exception { + LDContext anonymous = LDContext.builder(ContextKind.DEFAULT, "anon-key") + .anonymous(true) + .set("email", "sandy@example.com") + .build(); + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordIdentifyEvent(anonymous); + eventProcessor.recordEvaluationEvent(anonymous, FLAG_KEY, 10, 1, LDValue.of(true), + null, LDValue.of(false), true, null); + + List events = flushAndCollect(eventProcessor, server); + + // An identify event is the application deliberately registering this context. + LDValue identifyContext = requireEventOfKind(events, "identify").get("context"); + assertEquals(LDValue.of("sandy@example.com"), identifyContext.get("email")); + + // A feature event redacts every attribute of an anonymous context, without the + // application having to mark anything private. + LDValue featureContext = requireEventOfKind(events, "feature").get("context"); + assertEquals(LDValue.of("anon-key"), featureContext.get("key")); + assertEquals(LDValue.ofNull(), featureContext.get("email")); + assertEquals(redacted("email"), redactedAttributes(featureContext)); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void summaryEventsRedactPrivateAttributesOfTheContextTheyInline() throws Exception { + // Because this SDK summarizes per context, a summary event inlines the context it counted + // rather than just its key, and so it is subject to redaction like any other event. This + // is the case most easily missed, since summaries are produced by the buffer rather than + // by an explicit record call. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).privateAttributes("email"), true); + try { + eventProcessor.recordEvaluationEvent(PERSON, FLAG_KEY, 10, 1, LDValue.of(true), + null, LDValue.of(false), false, null); + + LDValue summary = requireEventOfKind(flushAndCollect(eventProcessor, server), + "summary"); + + LDValue context = summary.get("context"); + assertEquals(LDValue.of("user-key"), context.get("key")); + assertEquals(LDValue.ofNull(), context.get("email")); + assertEquals(redacted("email"), redactedAttributes(context)); + // Attributes that were not marked private are inlined here, same as anywhere else. + assertEquals(LDValue.of("Oakland"), context.get("address").get("city")); + assertFalse("summary should not contain the email value anywhere", + summary.toJsonString().contains("sandy@example.com")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void redactionAppliesToEveryKindOfAMultiContext() throws Exception { + LDContext multi = LDContext.createMulti( + LDContext.builder(ContextKind.of("user"), "user-key") + .set("email", "sandy@example.com") + .build(), + LDContext.builder(ContextKind.of("org"), "org-key") + .set("email", "billing@example.com") + .build()); + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).privateAttributes("email"), true); + try { + eventProcessor.recordIdentifyEvent(multi); + + LDValue delivered = requireEventOfKind(flushAndCollect(eventProcessor, server), + "identify").get("context"); + + assertEquals(LDValue.of("multi"), delivered.get("kind")); + assertEquals(LDValue.ofNull(), delivered.get("user").get("email")); + assertEquals(LDValue.ofNull(), delivered.get("org").get("email")); + assertEquals(redacted("email"), redactedAttributes(delivered.get("user"))); + assertEquals(redacted("email"), redactedAttributes(delivered.get("org"))); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void noPayloadContainsAPrivateValueAnywhere() throws Exception { + // A belt-and-braces check over the whole request body rather than one field, so that a + // private value cannot slip through in some event kind these tests do not name. + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, + eventsBuilder(DEFAULT_CAPACITY).privateAttributes("email"), true); + try { + eventProcessor.recordIdentifyEvent(PERSON); + eventProcessor.recordCustomEvent(PERSON, "an-event", LDValue.ofNull(), null); + eventProcessor.recordEvaluationEvent(PERSON, FLAG_KEY, 10, 1, LDValue.of(true), + null, LDValue.of(false), true, null); + + List events = flushAndCollect(eventProcessor, server); + + assertTrue("expected identify, custom, feature and summary events", + events.size() >= 4); + for (LDValue event : events) { + assertFalse("private value leaked into " + event.get("kind") + ": " + event, + event.toJsonString().contains("sandy@example.com")); + } + } finally { + eventProcessor.close(); + } + } + } + + private static List redacted(String... names) { + List result = new ArrayList<>(); + for (String name : names) { + result.add(name); + } + return result; + } + + /** The names the payload itself reports as redacted, sorted so the comparison is stable. */ + private static List redactedAttributes(LDValue context) { + List names = new ArrayList<>(); + for (LDValue name : context.get("_meta").get("redactedAttributes").values()) { + names.add(name.stringValue()); + } + java.util.Collections.sort(names); + return names; + } +} diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java new file mode 100644 index 000000000..95efa1f06 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java @@ -0,0 +1,150 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.LDConfig.Builder.AutoEnvAttributes; +import com.launchdarkly.sdk.android.env.EnvironmentReporterBuilder; +import com.launchdarkly.sdk.android.env.IEnvironmentReporter; +import com.launchdarkly.sdk.android.integrations.EventProcessorBuilder; +import com.launchdarkly.sdk.android.subsystems.ClientContext; +import com.launchdarkly.sdk.android.subsystems.EventProcessor; +import com.launchdarkly.testhelpers.httptest.Handlers; +import com.launchdarkly.testhelpers.httptest.HttpServer; +import com.launchdarkly.testhelpers.httptest.RequestInfo; + +import org.junit.Rule; +import org.junit.rules.Timeout; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * Fixture for tests that drive the SDK's event processor directly and read back the analytics + * payload it posts. + */ +public abstract class EventProcessorTestBase { + protected static final String MOBILE_KEY = "test-mobile-key"; + protected static final LDContext CONTEXT = LDContext.create("user-key"); + + // Long enough that the only payload in a test is the one it asks for explicitly. + private static final int NO_PERIODIC_FLUSH_MILLIS = 600_000; + + @Rule + public Timeout globalTimeout = Timeout.seconds(60); + @Rule + public LogCaptureRule logging = new LogCaptureRule(); + + private final IEnvironmentReporter environmentReporter = new EnvironmentReporterBuilder().build(); + + protected HttpServer startEventsServer() { + return HttpServer.start(Handlers.status(202)); + } + + protected EventProcessor makeEventProcessor(HttpServer server, int capacity) { + return makeEventProcessor(server, capacity, true); + } + + protected EventProcessor makeEventProcessor(HttpServer server, int capacity, + boolean diagnosticOptOut) { + return makeEventProcessor(server, eventsBuilder(capacity), diagnosticOptOut); + } + + /** + * @return an events builder with the capacity set and periodic flushing effectively off, for + * tests that need to configure something else on top such as private attributes + */ + protected EventProcessorBuilder eventsBuilder(int capacity) { + return Components.sendEvents() + .capacity(capacity) + .flushIntervalMillis(NO_PERIODIC_FLUSH_MILLIS); + } + + protected EventProcessor makeEventProcessor(HttpServer server, EventProcessorBuilder events, + boolean diagnosticOptOut) { + EventProcessor eventProcessor = buildOfflineEventProcessor(server, events, diagnosticOptOut); + // LDClient turns the processor on once initialization decides the SDK is not in offline mode. + eventProcessor.setOffline(false); + return eventProcessor; + } + + /** @return the processor as the SDK builds it, before initialization has turned it on */ + protected EventProcessor buildOfflineEventProcessor(HttpServer server, EventProcessorBuilder events, + boolean diagnosticOptOut) { + LDConfig config = new LDConfig.Builder(AutoEnvAttributes.Disabled) + .mobileKey(MOBILE_KEY) + .diagnosticOptOut(diagnosticOptOut) + .events(events) + .serviceEndpoints(Components.serviceEndpoints().events(server.getUri())) + .build(); + ClientContext clientContext = ClientContextImpl.fromConfig(config, MOBILE_KEY, "", + null, null, CONTEXT, logging.logger, null, environmentReporter, null); + return config.events.build(clientContext); + } + + /** + * Flushes and returns every event the processor posted. {@code blockingFlush} does not return + * until delivery completes, so there is no settling delay here. + */ + protected List flushAndCollect(EventProcessor eventProcessor, HttpServer server) { + eventProcessor.blockingFlush(); + return collectDelivered(server); + } + + /** Returns every event posted so far, without asking for a flush. */ + protected List collectDelivered(HttpServer server) { + List events = new ArrayList<>(); + collectRequest(server.getRecorder().requireRequest(10, TimeUnit.SECONDS), events); + while (server.getRecorder().count() > 0) { + collectRequest(server.getRecorder().requireRequest(10, TimeUnit.SECONDS), events); + } + return events; + } + + private void collectRequest(RequestInfo request, List events) { + assertEquals("POST", request.getMethod()); + assertEquals("/mobile/events/bulk", request.getPath()); + for (LDValue event : LDValue.parse(request.getBody()).values()) { + events.add(event); + } + } + + protected int countEventsOfKind(List events, String kind) { + int count = 0; + for (LDValue event : events) { + if (kind.equals(event.get("kind").stringValue())) { + count++; + } + } + return count; + } + + protected LDValue requireEventOfKind(List events, String kind) { + for (LDValue event : events) { + if (kind.equals(event.get("kind").stringValue())) { + return event; + } + } + throw new AssertionError("no event of kind " + kind + " in " + events); + } + + protected int summaryCountFor(List events, String flagKey) { + int total = 0; + boolean sawSummary = false; + for (LDValue event : events) { + if (!"summary".equals(event.get("kind").stringValue())) { + continue; + } + sawSummary = true; + // Per-context summarization means a flag can have more than one counter entry. + for (LDValue counter : event.get("features").get(flagKey).get("counters").values()) { + total += counter.get("count").intValue(); + } + } + assertTrue("no summary event was delivered", sawSummary); + return total; + } +} diff --git a/settings.gradle b/settings.gradle index 36c077f5b..fa5361a28 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,4 +1,5 @@ include(":launchdarkly-android-client-sdk") include(":shared-test-code") include(":example") +include(":test-app") include(":contract-tests") diff --git a/shared-test-code/build.gradle b/shared-test-code/build.gradle index c3fb0e054..50f66687c 100644 --- a/shared-test-code/build.gradle +++ b/shared-test-code/build.gradle @@ -12,7 +12,7 @@ ext.versions = [ "androidAnnotation": "1.2.0", "gson": "2.13.2", "junit": "4.13", - "launchdarklyJavaSdkInternal": "1.9.0", + "launchdarklyJavaSdkInternal": "1.12.0", "launchdarklyLogging": "1.1.1", ] diff --git a/test-app/README.md b/test-app/README.md new file mode 100644 index 000000000..e7d624ae3 --- /dev/null +++ b/test-app/README.md @@ -0,0 +1,19 @@ +# LaunchDarkly Android SDK test app + +This app is an internal quality harness. For customer integration examples, use `example`. + +Add the following to the repository's root `local.properties`: + +```properties +launchdarkly.mobileKey=mob-... +launchdarkly.environment=production +``` + +Set `launchdarkly.environment=staging` to use LaunchDarkly's staging endpoints. + +## Tier 1 event-loss scenario + +Create a boolean flag named `kill-flag`, or enter another flag key in the app. Tap +**Eval+track+kill** to evaluate the flag, track a stand-in error event, request a flush, and +terminate the process five seconds later. This exercises the interval between recording and +delivery without Android lifecycle callbacks masking the result. diff --git a/test-app/build.gradle b/test-app/build.gradle new file mode 100644 index 000000000..d5b3d28a9 --- /dev/null +++ b/test-app/build.gradle @@ -0,0 +1,50 @@ +import java.util.Properties + +plugins { + id("com.android.application") + // Method counts are an internal quality signal rather than customer setup. + id("com.getkeepsafe.dexcount") +} + +// local.properties is not checked in, so it is where machine-specific settings such as the mobile +// key belong. See test-app/README.md for the keys this app reads. +def localProperties = new Properties() +def localPropertiesFile = rootProject.file("local.properties") +if (localPropertiesFile.exists()) { + localPropertiesFile.withInputStream { localProperties.load(it) } +} + +android { + namespace "com.launchdarkly.sdk.testapp" + compileSdk = 34 + + defaultConfig { + applicationId = "com.launchdarkly.sdk.testapp" + minSdk = 21 + targetSdk = 34 + versionCode = 1 + versionName = "1.0" + + buildConfigField("String", "MOBILE_KEY", + "\"${localProperties.getProperty('launchdarkly.mobileKey', '')}\"") + buildConfigField("String", "LD_ENVIRONMENT", + "\"${localProperties.getProperty('launchdarkly.environment', 'production')}\"") + } + + buildFeatures { + buildConfig = true + } + + buildTypes { + release { + minifyEnabled = true + proguardFiles(getDefaultProguardFile("proguard-android.txt"), "proguard-rules.pro") + } + } +} + +dependencies { + implementation("androidx.appcompat:appcompat:1.2.0") + implementation("com.jakewharton.timber:timber:5.0.1") + implementation(project(":launchdarkly-android-client-sdk")) +} diff --git a/test-app/proguard-rules.pro b/test-app/proguard-rules.pro new file mode 100644 index 000000000..8788708d0 --- /dev/null +++ b/test-app/proguard-rules.pro @@ -0,0 +1,2 @@ +# Allow compile-time-only error-prone annotations to be stripped without breaking shrink. +-dontwarn com.google.errorprone.annotations.** diff --git a/test-app/src/main/AndroidManifest.xml b/test-app/src/main/AndroidManifest.xml new file mode 100644 index 000000000..b9729d0ff --- /dev/null +++ b/test-app/src/main/AndroidManifest.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + + + + + + diff --git a/test-app/src/main/java/com/launchdarkly/sdk/testapp/ExposureCountingHook.java b/test-app/src/main/java/com/launchdarkly/sdk/testapp/ExposureCountingHook.java new file mode 100644 index 000000000..4ace6bce6 --- /dev/null +++ b/test-app/src/main/java/com/launchdarkly/sdk/testapp/ExposureCountingHook.java @@ -0,0 +1,64 @@ +package com.launchdarkly.sdk.testapp; + +import com.launchdarkly.sdk.EvaluationDetail; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.android.integrations.EvaluationSeriesContext; +import com.launchdarkly.sdk.android.integrations.Hook; + +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Counts the evaluation series stages it observes, so the example can show what exposure + * deduplication does. Deduplication is configured at registration by wrapping this hook in a + * {@link com.launchdarkly.sdk.android.integrations.DedupingHook}, the same way a customer would + * wrap any other hook. + *

+ * Deduplication skips the whole series, so both counts stay equal and both stop climbing while + * repeated evaluations resolve to the same result. + */ +class ExposureCountingHook extends Hook { + private final String label; + private final Runnable onStage; + private final AtomicInteger befores = new AtomicInteger(); + private final AtomicInteger afters = new AtomicInteger(); + + /** + * @param label a name for this hook, shown in the app's dedupe status + * @param onStage run after each stage so the app can refresh its display + */ + ExposureCountingHook(String label, Runnable onStage) { + super(label); + this.label = label; + this.onStage = onStage; + } + + @Override + public Map beforeEvaluation(EvaluationSeriesContext seriesContext, Map seriesData) { + befores.incrementAndGet(); + onStage.run(); + return seriesData; + } + + @Override + public Map afterEvaluation(EvaluationSeriesContext seriesContext, Map seriesData, + EvaluationDetail evaluationDetail) { + afters.incrementAndGet(); + onStage.run(); + return seriesData; + } + + /** + * @param windowMillis the dedupe window this hook was registered with, for the status line + * @return a line describing this hook's window and how many evaluations have reached it + */ + String status(int windowMillis) { + return String.format(Locale.US, "%s (%d ms): %d (before %d / after %d)", + label, + windowMillis, + afters.get(), + befores.get(), + afters.get()); + } +} diff --git a/test-app/src/main/java/com/launchdarkly/sdk/testapp/MainActivity.java b/test-app/src/main/java/com/launchdarkly/sdk/testapp/MainActivity.java new file mode 100644 index 000000000..bbdd1e7a1 --- /dev/null +++ b/test-app/src/main/java/com/launchdarkly/sdk/testapp/MainActivity.java @@ -0,0 +1,319 @@ +package com.launchdarkly.sdk.testapp; + +import android.os.Bundle; +import android.os.Handler; +import android.os.Looper; +import android.widget.ArrayAdapter; +import android.widget.Button; +import android.widget.EditText; +import android.widget.Spinner; +import android.widget.Switch; +import android.widget.TextView; +import android.widget.Toast; + +import androidx.appcompat.app.AppCompatActivity; + +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.android.Components; +import com.launchdarkly.sdk.android.ConnectionInformation; +import com.launchdarkly.sdk.android.LDAllFlagsListener; +import com.launchdarkly.sdk.android.LDClient; +import com.launchdarkly.sdk.android.LDConfig; +import com.launchdarkly.sdk.android.LDConfig.Builder.AutoEnvAttributes; +import com.launchdarkly.sdk.android.LDFailure; +import com.launchdarkly.sdk.android.LDStatusListener; +import com.launchdarkly.sdk.android.integrations.DedupingHook; + +import java.util.Date; +import java.util.Locale; +import java.util.concurrent.atomic.AtomicInteger; + +import timber.log.Timber; + +public class MainActivity extends AppCompatActivity { + + // Two hooks with different windows, to show that each one is deduplicated on its own. + private static final int FAST_DEDUPE_WINDOW_MILLIS = 5_000; + private static final int SLOW_DEDUPE_WINDOW_MILLIS = 10_000; + + // The staging hosts mirror the production ones in StandardEndpoints under the ld-stg domain. + private static final String STAGING_DOMAIN = "ld-stg.launchdarkly.com"; + + private static final String DEFAULT_USER_KEY = "user key"; + + /** How long startup blocks waiting for the first flags to arrive. */ + private static final int INIT_WAIT_SECONDS = 10; + + private LDClient ldClient; + private LDStatusListener ldStatusListener; + private LDAllFlagsListener allFlagsListener; + + private final ExposureCountingHook fastHook = + new ExposureCountingHook("fast", this::updateDedupeStatus); + private final ExposureCountingHook slowHook = + new ExposureCountingHook("slow", this::updateDedupeStatus); + private final AtomicInteger evaluationsRequested = new AtomicInteger(); + + private static boolean isStaging() { + return "staging".equalsIgnoreCase(BuildConfig.LD_ENVIRONMENT); + } + + private void updateDedupeStatus() { + if (Looper.myLooper() != MainActivity.this.getMainLooper()) { + new Handler(MainActivity.this.getMainLooper()).post(this::updateDedupeStatus); + return; + } + + String result = String.format(Locale.US, + "Environment: %s\nEvaluations requested: %d\n%s\n%s", + isStaging() ? "staging" : "production", + evaluationsRequested.get(), + fastHook.status(FAST_DEDUPE_WINDOW_MILLIS), + slowHook.status(SLOW_DEDUPE_WINDOW_MILLIS)); + ((TextView) MainActivity.this.findViewById(R.id.dedupe_status)).setText(result); + } + + private void updateStatusString(final ConnectionInformation connectionInformation) { + if (Looper.myLooper() != MainActivity.this.getMainLooper()) { + new Handler(MainActivity.this.getMainLooper()).post(() -> updateStatusString(connectionInformation)); + } else { + TextView connection = MainActivity.this.findViewById(R.id.connection_status); + Long lastSuccess = connectionInformation.getLastSuccessfulConnection(); + Long lastFailure = connectionInformation.getLastFailedConnection(); + + String result = String.format(Locale.US, "Mode: %s\nSuccess at: %s\nFailure at: %s\nFailure type: %s", + connectionInformation.getConnectionMode().toString(), + lastSuccess == null ? "Never" : new Date(lastSuccess).toString(), + lastFailure == null ? "Never" : new Date(lastFailure).toString(), + connectionInformation.getLastFailure() != null ? + connectionInformation.getLastFailure().getFailureType() + : ""); + connection.setText(result); + } + } + + @Override + public void onCreate(Bundle savedInstanceState) { + super.onCreate(savedInstanceState); + setContentView(R.layout.activity_main); + + setupEval(); + setupFlushButton(); + setupTrackButton(); + setupIdentifyButton(); + setupKillUnsentButton(); + setupOfflineSwitch(); + setupListeners(); + updateDedupeStatus(); + + if (BuildConfig.MOBILE_KEY.isEmpty()) { + String message = "Set launchdarkly.mobileKey in local.properties and rebuild."; + Timber.e(message); + Toast.makeText(this, message, Toast.LENGTH_LONG).show(); + return; + } + + LDConfig.Builder configBuilder = new LDConfig.Builder(AutoEnvAttributes.Enabled) + .mobileKey(BuildConfig.MOBILE_KEY) + .http( + Components.httpConfiguration().useReport(false) + // change useReport to `true` if the request is to be REPORT'ed instead of GET'ed + ) + .hooks( + // Same shape a customer uses for any hook: wrap it at registration. Each + // wrapper has its own window, so neither suppresses the other. + Components.hooks() + .addHook(new DedupingHook(fastHook, FAST_DEDUPE_WINDOW_MILLIS)) + .addHook(new DedupingHook(slowHook, SLOW_DEDUPE_WINDOW_MILLIS)) + ); + + if (isStaging()) { + configBuilder.serviceEndpoints( + Components.serviceEndpoints() + .streaming("https://clientstream." + STAGING_DOMAIN) + .polling("https://clientsdk." + STAGING_DOMAIN) + .events("https://mobile." + STAGING_DOMAIN) + ); + } + + LDConfig ldConfig = configBuilder.build(); + + LDContext context = LDContext.builder(DEFAULT_USER_KEY) + .set("email", "fake@example.com") + .build(); + + // Returns the client either way: if the flags have not arrived within the wait, it is usable + // with whatever it has cached. + ldClient = LDClient.init(this.getApplication(), ldConfig, context, INIT_WAIT_SECONDS); + updateStatusString(ldClient.getConnectionInformation()); + ldClient.registerStatusListener(ldStatusListener); + ldClient.registerAllFlagsListener(allFlagsListener); + } + + private void setupListeners() { + ldStatusListener = new LDStatusListener() { + @Override + public void onConnectionModeChanged(final ConnectionInformation connectionInformation) { + updateStatusString(connectionInformation); + } + + @Override + public void onInternalFailure(final LDFailure ldFailure) { + new Handler(MainActivity.this.getMainLooper()).post(() -> { + Toast.makeText(MainActivity.this, ldFailure.toString(), Toast.LENGTH_SHORT).show(); + }); + updateStatusString(ldClient.getConnectionInformation()); + } + }; + + allFlagsListener = flagKey -> { + new Handler(MainActivity.this.getMainLooper()).post(() -> { + StringBuilder flags = new StringBuilder("Updated flags: "); + for (String flag : flagKey) { + flags.append(flag).append(" "); + } + Toast.makeText(MainActivity.this, flags.toString(), Toast.LENGTH_SHORT).show(); + }); + updateStatusString(ldClient.getConnectionInformation()); + }; + } + + private void setupFlushButton() { + Button flushButton = findViewById(R.id.flush_button); + flushButton.setOnClickListener(v -> { + Timber.i("flush onClick"); + MainActivity.this.doSafeClientAction(() -> ldClient.flush()); + }); + } + + private interface LDClientAction { + void call(); + } + + private void doSafeClientAction(LDClientAction function) { + if (ldClient != null) { + function.call(); + } + } + + private interface LDClientGet { + V get(); + } + + private V doSafeClientGet(LDClientGet function) { + return ldClient != null ? function.get() : null; + } + + private void setupTrackButton() { + Button trackButton = findViewById(R.id.track_button); + trackButton.setOnClickListener(v -> { + Timber.i("track onClick"); + MainActivity.this.doSafeClientAction(() -> ldClient.track("Android event name")); + }); + } + + /** + * Reproduces in-memory event loss: evaluate (exposure) and track (stand-in for an error), + * wait 5s so both calls are queued, then kill the process before the 30s flush. + * {@code finish()} or backgrounding would run the SDK's background flush, so this uses + * {@link android.os.Process#killProcess}. + */ + private void setupKillUnsentButton() { + Button killUnsentButton = findViewById(R.id.kill_unsent_button); + killUnsentButton.setOnClickListener(v -> { + final String typedKey = ((EditText) findViewById(R.id.feature_flag_key)).getText().toString().trim(); + final String flagKey = typedKey.isEmpty() ? "kill-flag" : typedKey; + Timber.w("eval+track+kill flag=%s", flagKey); + doSafeClientAction(() -> { + ldClient.boolVariation(flagKey, false); + ldClient.track("$ld:telemetry:error"); + ldClient.flush(); + new Handler(Looper.getMainLooper()).postDelayed( + () -> android.os.Process.killProcess(android.os.Process.myPid()), + 5_000); + }); + }); + } + + private void setupIdentifyButton() { + Button identify = findViewById(R.id.identify_button); + identify.setOnClickListener(v -> { + Timber.i("identify onClick"); + String typedKey = ((EditText) MainActivity.this.findViewById(R.id.userKey_editText)) + .getText().toString().trim(); + // An empty key builds an invalid context, which identify rejects before it resets the + // hooks' dedupe caches. Fall back to the key the client started with, so identifying to + // an unchanged context still demonstrates that the reset happens either way. + final String userKey = typedKey.isEmpty() ? DEFAULT_USER_KEY : typedKey; + final LDContext updatedContext = LDContext.create(userKey); + MainActivity.this.doSafeClientAction(() -> { + ldClient.identify(updatedContext); + }); + MainActivity.this.updateDedupeStatus(); + }); + } + + private void setupOfflineSwitch() { + Switch offlineSwitch = findViewById(R.id.offlineSwitch); + offlineSwitch.setOnCheckedChangeListener((compoundButton, isChecked) -> + MainActivity.this.doSafeClientAction(isChecked ? () -> ldClient.setOffline() : () -> ldClient.setOnline()) + ); + } + + private void setupEval() { + final Spinner spinner = findViewById(R.id.type_spinner); + ArrayAdapter adapter = ArrayAdapter.createFromResource(this, + R.array.types_array, android.R.layout.simple_spinner_item); + adapter.setDropDownViewResource(android.R.layout.simple_spinner_dropdown_item); + spinner.setAdapter(adapter); + + Button evalButton = findViewById(R.id.eval_button); + evalButton.setOnClickListener(v -> { + Timber.i("eval onClick"); + final String flagKey = ((EditText) MainActivity.this.findViewById(R.id.feature_flag_key)).getText().toString(); + evaluationsRequested.incrementAndGet(); + + String type = spinner.getSelectedItem().toString(); + final String result; + String logResult; + switch (type) { + case "String": + result = MainActivity.this.doSafeClientGet(() -> ldClient.stringVariation(flagKey, "default")); + logResult = result == null ? "no result" : result; + Timber.i(logResult); + ((TextView) MainActivity.this.findViewById(R.id.result_textView)).setText(result); + MainActivity.this.doSafeClientAction(() -> { + ldClient.registerFeatureFlagListener(flagKey, flagKey1 -> { + evaluationsRequested.incrementAndGet(); + ((TextView) MainActivity.this.findViewById(R.id.result_textView)) + .setText(ldClient.stringVariation(flagKey1, "default")); + MainActivity.this.updateDedupeStatus(); + }); + }); + MainActivity.this.updateDedupeStatus(); + return; + case "Boolean": + result = MainActivity.this.doSafeClientGet(() -> String.valueOf(ldClient.boolVariation(flagKey, false))); + break; + case "Integer": + result = MainActivity.this.doSafeClientGet(() -> String.valueOf(ldClient.intVariation(flagKey, 0))); + break; + case "Float": + result = MainActivity.this.doSafeClientGet(() -> String.valueOf(ldClient.doubleVariation(flagKey, 0.0))); + break; + case "Value": + result = MainActivity.this.doSafeClientGet(() -> String.valueOf(ldClient.jsonValueVariation(flagKey, null))); + break; + default: + result = null; + break; + } + + logResult = result == null ? "no result" : result; + Timber.i(logResult); + ((TextView) MainActivity.this.findViewById(R.id.result_textView)).setText(result); + MainActivity.this.updateDedupeStatus(); + }); + } + +} diff --git a/test-app/src/main/res/layout/activity_main.xml b/test-app/src/main/res/layout/activity_main.xml new file mode 100644 index 000000000..dfa8fac02 --- /dev/null +++ b/test-app/src/main/res/layout/activity_main.xml @@ -0,0 +1,173 @@ + + + +