Removing noisy metric tags and adding stress test for telemetry

segmentio · Nov 25, 2024 · b25c4aa · b25c4aa
1 parent 546808d
commit b25c4aa
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 7 deletions.
diff --git a/core/src/main/java/com/segment/analytics/kotlin/core/Analytics.kt b/core/src/main/java/com/segment/analytics/kotlin/core/Analytics.kt
@@ -98,7 +98,6 @@ open class Analytics protected constructor(
                     Telemetry.INVOKE_ERROR_METRIC, t.stackTraceToString()) {
                     it["error"] = t.toString()
                     it["message"] = "Exception in Analytics Scope"
-                    it["caller"] = t.stackTrace[0].toString()
                 }
             }
             override val analyticsScope = CoroutineScope(SupervisorJob() + exceptionHandler)

diff --git a/core/src/main/java/com/segment/analytics/kotlin/core/HTTPClient.kt b/core/src/main/java/com/segment/analytics/kotlin/core/HTTPClient.kt
@@ -39,7 +39,6 @@ class HTTPClient(
                 it["error"] = e.toString()
                 it["writekey"] = writeKey
                 it["message"] = "Malformed url"
-                it["caller"] = e.stackTrace[0].toString()
             }
             throw error
         }

diff --git a/core/src/main/java/com/segment/analytics/kotlin/core/Settings.kt b/core/src/main/java/com/segment/analytics/kotlin/core/Settings.kt
@@ -124,7 +124,6 @@ internal fun Analytics.fetchSettings(
             it["error"] = ex.toString()
             it["writekey"] = writeKey
             it["message"] = "Error retrieving settings"
-            it["caller"] = ex.stackTrace[0].toString()
         }
         configuration.defaultSettings
     }
diff --git a/core/src/main/java/com/segment/analytics/kotlin/core/platform/Mediator.kt b/core/src/main/java/com/segment/analytics/kotlin/core/platform/Mediator.kt
@@ -63,7 +63,6 @@ internal class Mediator(internal var plugins: CopyOnWriteArrayList<Plugin> = Cop
                         }
                         it["writekey"] = plugin.analytics.configuration.writeKey
                         it["message"] = "Exception executing plugin"
-                        it["caller"] = t.stackTrace[0].toString()
                     }
                 }
             }
@@ -88,7 +87,6 @@ internal class Mediator(internal var plugins: CopyOnWriteArrayList<Plugin> = Cop
                     }
                     it["writekey"] = plugin.analytics.configuration.writeKey
                     it["message"] = "Exception executing plugin"
-                    it["caller"] = t.stackTrace[0].toString()
                 }
             }
         }

diff --git a/core/src/test/kotlin/com/segment/analytics/kotlin/core/TelemetryTest.kt b/core/src/test/kotlin/com/segment/analytics/kotlin/core/TelemetryTest.kt
@@ -7,6 +7,10 @@ import org.junit.jupiter.api.Test
 import java.lang.reflect.Field
 import java.net.HttpURLConnection
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import kotlin.random.Random
 
 class TelemetryTest {
     fun TelemetryResetFlushFirstError() {
@@ -182,7 +186,7 @@ class TelemetryTest {
         Telemetry.start()
         for (i in 1..Telemetry.maxQueueSize + 1) {
             Telemetry.increment(Telemetry.INVOKE_METRIC) { it["test"] = "test" + i }
-            Telemetry.error(Telemetry.INVOKE_ERROR_METRIC, "error") { it["test"] = "test" + i }
+            Telemetry.error(Telemetry.INVOKE_ERROR_METRIC, "error") { it["error"] = "test" + i }
         }
         assertEquals(Telemetry.maxQueueSize, TelemetryQueueSize())
     }
@@ -195,6 +199,44 @@ class TelemetryTest {
         Telemetry.sendWriteKeyOnError = false
         Telemetry.sendErrorLogData = false
         Telemetry.error(Telemetry.INVOKE_ERROR_METRIC, longString) { it["writekey"] = longString }
-        assertTrue(TelemetryQueueSize() < 1000)
+        assertTrue(TelemetryQueueBytes() < 1000)
+    }
+
+    @Test
+    fun testConcurrentErrorReportingWithQueuePressure() {
+        val operationCount = 200
+        val latch = CountDownLatch(operationCount)
+        val executor = Executors.newFixedThreadPool(3)
+
+        try {
+            // Launch operations across multiple threads
+            repeat(operationCount) { i ->
+                executor.submit {
+                    try {
+                        Telemetry.error(
+                            metric = Telemetry.INVOKE_ERROR_METRIC,
+                            log = "High pressure test $i"
+                        ) {
+                            it["error"] = "pressure_test_key"
+                            it["iteration"] = "$i"
+                        }
+
+                        // Add random delays to increase race condition probability
+                        if (i % 5 == 0) {
+                            Thread.sleep(Random.nextLong(1, 3))
+                        }
+                    } finally {
+                        latch.countDown()
+                    }
+                }
+            }
+
+            // Wait for all operations to complete
+            latch.await(15, TimeUnit.SECONDS)
+
+        } finally {
+            executor.shutdown()
+        }
+        assertTrue(TelemetryQueueSize() == Telemetry.maxQueueSize)
     }
 }