fix: metric submission failure should keep metrics for the next attempt (#130)

* fix: metric submission failure should keep metrics for the next attempt

* Try to make tests more resilient

Author: gastonfournier

unleashandroidsdk/src/main/java/io/getunleash/android/metrics/MetricsBucket.kt

@@ -18,8 +18,8 @@ data class Bucket(
 )
 
 interface UnleashMetricsBucket {
-    fun count(featureName: String, enabled: Boolean): Boolean
-    fun countVariant(featureName: String, variant: Variant): Variant
+    fun count(featureName: String, enabled: Boolean, increment: Int = 1): Boolean
+    fun countVariant(featureName: String, variant: Variant, increment: Int = 1): Variant
     fun isEmpty(): Boolean
 }
 
@@ -30,14 +30,14 @@ data class CountBucket(
     val variants: ConcurrentHashMap<Pair<String, String>, AtomicInteger> = ConcurrentHashMap()
 ): UnleashMetricsBucket {
 
-    override fun count(featureName: String, enabled: Boolean): Boolean {
+    override fun count(featureName: String, enabled: Boolean, increment: Int): Boolean {
         (if (enabled) yes else no)
-            .getOrPut(featureName) { AtomicInteger(0) }.incrementAndGet()
+            .getOrPut(featureName) { AtomicInteger(0) }.addAndGet(increment)
         return enabled
     }
 
-    override fun countVariant(featureName: String, variant: Variant): Variant {
-        variants.getOrPut(Pair(featureName, variant.name)) { AtomicInteger(0) }.incrementAndGet()
+    override fun countVariant(featureName: String, variant: Variant, increment: Int): Variant {
+        variants.getOrPut(Pair(featureName, variant.name)) { AtomicInteger(0) }.addAndGet(increment)
         return variant
     }
 

unleashandroidsdk/src/main/java/io/getunleash/android/metrics/MetricsReporter.kt

@@ -2,5 +2,5 @@ package io.getunleash.android.metrics
 
 interface MetricsReporter {
 
-    suspend fun sendMetrics()
+    suspend fun sendMetrics(onComplete: ((Result<Unit>) -> Unit)? = null)
 }

unleashandroidsdk/src/main/java/io/getunleash/android/metrics/MetricsSender.kt

@@ -16,6 +16,7 @@ import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
 import java.io.IOException
 import java.util.Date
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.TimeUnit
 
 class MetricsSender(
@@ -30,14 +31,15 @@ class MetricsSender(
     private val metricsUrl = config.proxyUrl?.toHttpUrl()?.newBuilder()?.addPathSegment("client")
         ?.addPathSegment("metrics")?.build()
     private var bucket: CountBucket = CountBucket(start = Date())
+    private val inFlight = AtomicBoolean(false)
     private val throttler =
         Throttler(
             TimeUnit.MILLISECONDS.toSeconds(config.metricsStrategy.interval),
             longestAcceptableIntervalSeconds = 300,
             metricsUrl.toString()
         )
 
-    override suspend fun sendMetrics() {
+    override suspend fun sendMetrics(onComplete: ((Result<Unit>) -> Unit)?) {
         if (metricsUrl == null) {
             Log.d(TAG, "No proxy URL configured, skipping metrics reporting")
             return
@@ -46,12 +48,16 @@ class MetricsSender(
             Log.d(TAG, "No metrics to report")
             return
         }
+        if (!inFlight.compareAndSet(false, true)) {
+            Log.d(TAG, "Metrics report already in-flight, skipping this send")
+            return
+        }
         if (throttler.performAction()) {
             val toReport = swapAndFreeze()
             val payload = MetricsPayload(
                 appName = config.appName,
                 instanceId = config.instanceId,
-                bucket = toReport
+                bucket = toReport.first
             )
             val request = Request.Builder()
                 .headers(applicationHeaders.toHeaders())
@@ -61,6 +67,13 @@ class MetricsSender(
                 ).build()
             httpClient.newCall(request).enqueue(object : Callback {
                 override fun onFailure(call: Call, e: IOException) {
+                    mergeBack(toReport.second)
+                    inFlight.set(false)
+                    try {
+                        onComplete?.invoke(Result.failure(e))
+                    } catch (t: Throwable) {
+                        Log.w(TAG, "onComplete callback threw", t)
+                    }
                     Log.i(TAG, "Failed to report metrics for interval", e)
                 }
 
@@ -72,17 +85,37 @@ class MetricsSender(
                     throttler.handle(response.code)
                     response.body.use { // Need to consume body to ensure we don't keep connection open
                     }
+                    inFlight.set(false)
+                    try {
+                        onComplete?.invoke(Result.success(Unit))
+                    } catch (t: Throwable) {
+                        Log.w(TAG, "onComplete callback threw", t)
+                    }
                 }
             })
         } else {
             throttler.skipped()
+            inFlight.set(false)
         }
     }
 
-    private fun swapAndFreeze(): Bucket {
-        val bucketRef = bucket
+    private fun swapAndFreeze(): Pair<Bucket, CountBucket> {
+        val snapshot = bucket.copy()
         bucket = CountBucket(start = Date())
-        return bucketRef.copy().toBucket(bucket.start)
+        return Pair(snapshot.toBucket(bucket.start), snapshot)
+    }
+
+    // Note: this does not maintain the initial start time of the snapshot
+    private fun mergeBack(snapshot: CountBucket) {
+        for ((feature, count) in snapshot.yes) {
+            bucket.count(feature, true, count.get())
+        }
+        for ((feature, count) in snapshot.no) {
+            bucket.count(feature, false, count.get())
+        }
+        for ((pair, count) in snapshot.variants) {
+            bucket.countVariant(pair.first, Variant(pair.second), count.get())
+        }
     }
 
     override fun count(featureName: String, enabled: Boolean): Boolean {

unleashandroidsdk/src/main/java/io/getunleash/android/metrics/NoOpMetrics.kt

@@ -11,6 +11,6 @@ class NoOpMetrics: MetricsHandler {
         return variant
     }
 
-    override suspend fun sendMetrics() {
+    override suspend fun sendMetrics(onComplete: ((Result<Unit>) -> Unit)?) {
     }
 }

unleashandroidsdk/src/test/java/io/getunleash/android/DefaultUnleashTest.kt

@@ -403,9 +403,13 @@ class DefaultUnleashTest : BaseTest() {
 
         // change context to force a refresh
         unleash.setContext(UnleashContext(userId = "2"))
-        assertThat(togglesChecked).isEqualTo(1)
+        await().atMost(1, TimeUnit.SECONDS).until {
+            togglesChecked == 1
+        }
         unleash.setContext(UnleashContext(userId = "3"))
-        assertThat(togglesChecked).isEqualTo(2)
+        await().atMost(1, TimeUnit.SECONDS).until {
+            togglesChecked == 2
+        }
     }
 
     @Test

unleashandroidsdk/src/test/java/io/getunleash/android/metrics/MetricsSenderTest.kt

@@ -14,7 +14,12 @@ import org.junit.Test
 import org.mockito.Mockito.mock
 import java.util.concurrent.TimeUnit
 import net.javacrumbs.jsonunit.assertj.assertThatJson
+import okhttp3.mockwebserver.MockResponse
+import okhttp3.mockwebserver.SocketPolicy
+import org.awaitility.Awaitility.await
+import java.math.BigDecimal
 import java.math.BigDecimal.valueOf
+import java.util.concurrent.CountDownLatch
 
 class MetricsSenderTest : BaseTest() {
     var server: MockWebServer  = MockWebServer()
@@ -91,4 +96,53 @@ class MetricsSenderTest : BaseTest() {
             }
         }
     }
+
+    @Test
+    fun `failed send merges metrics back and next successful send includes both old and new`() = runTest {
+        val config = configBuilder.build()
+        val httpClient = ClientBuilder(config, mock(Context::class.java)).build("test", config.metricsStrategy)
+        val metricsSender = MetricsSender(config, httpClient)
+
+        // Initial metrics that will be part of the failed request
+        metricsSender.count("featureA", true)
+        metricsSender.count("featureB", true)
+
+        val latch = CountDownLatch(1)
+        var calls = 0
+        // Make the server drop the connection to simulate a network failure (500, 429, etc. would throttle further requests)
+        server.enqueue(MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AT_START))
+        metricsSender.sendMetrics { calls++; latch.countDown() }
+        metricsSender.count("featureA", true) // in between the failed send and the retry, add another metric
+        metricsSender.sendMetrics { calls++ } // This should be skipped since a send is in-flight
+        metricsSender.sendMetrics { calls++ } // This should be skipped since a send is in-flight
+
+        // Ensure the server received the (failed) request (may be disconnected)
+        server.takeRequest(1, TimeUnit.SECONDS)
+        latch.await(1, TimeUnit.SECONDS)
+        assertThat(calls).isEqualTo(1) // Only the first call should have been executed
+
+        // Add another metric after the failed send - should be merged with the snapshot
+        metricsSender.count("featureB", true)
+
+        // Now enqueue a successful response for the retry/send that should contain both counts
+        server.enqueue(MockResponse().setResponseCode(200))
+        metricsSender.sendMetrics()
+
+        val request = server.takeRequest(1, TimeUnit.SECONDS)!!
+        assertThat(request.method).isEqualTo("POST")
+        assertThat(request.path).isEqualTo("/proxy/client/metrics")
+        // The combined count for featureA should be 2 (one from failed snapshot + one new) and for featureB should be 1 (from failed snapshot)
+        assertThatJson(request.body.readUtf8()) {
+            node("bucket").apply {
+                node("toggles").apply {
+                    node("featureA").apply {
+                        node("yes").isNumber().isEqualTo(BigDecimal(2))
+                    }
+                    node("featureB").apply {
+                        node("yes").isNumber().isEqualTo(BigDecimal(2))
+                    }
+                }
+            }
+        }
+    }
 }