Telemetry Metrics

.gitignore

@@ -14,3 +14,4 @@
 .cxx
 local.properties
 .idea
+/core/bin

core/src/main/java/com/segment/analytics/kotlin/core/Analytics.kt

@@ -81,9 +81,6 @@ open class Analytics protected constructor(
 
     init {
         require(configuration.isValid()) { "invalid configuration" }
-        Telemetry.increment(Telemetry.INVOKE,
-            mapOf("message" to "configured", "apihost" to configuration.apiHost, "cdnhost" to configuration.cdnHost,
-                "flush" to "at:${configuration.flushAt} int:${configuration.flushInterval} pol:${configuration.flushPolicies.count()}"))
         build()
     }
 
@@ -96,10 +93,8 @@ open class Analytics protected constructor(
         object : CoroutineConfiguration {
             override val store = Store()
             val exceptionHandler = CoroutineExceptionHandler { _, t ->
-                Analytics.segmentLog(
-                    "Caught Exception in Analytics Scope: ${t}"
-                )
-                Telemetry.error(Telemetry.INVOKE_ERROR,
+                reportErrorWithMetrics(null, t,
+                    "Caught Exception in Analytics Scope", Telemetry.INVOKE_ERROR_METRIC,
                     mapOf("error" to t.toString(), "message" to "Exception in Analytics Scope"),
                     t.stackTraceToString()
                 )
@@ -121,6 +116,10 @@ open class Analytics protected constructor(
         add(ContextPlugin())
         add(UserInfoPlugin())
 
+        Telemetry.increment(Telemetry.INVOKE_METRIC,
+            mapOf("message" to "configured", "apihost" to configuration.apiHost, "cdnhost" to configuration.cdnHost,
+                "flush" to "at:${configuration.flushAt} int:${configuration.flushInterval} pol:${configuration.flushPolicies.count()}"))
+
         // Setup store
         analyticsScope.launch(analyticsDispatcher) {
             store.also {
@@ -136,6 +135,8 @@ open class Analytics protected constructor(
                 add(SegmentDestination())
             }
 
+            Telemetry.subscribe(store)
+
             checkSettings()
         }
     }

core/src/main/java/com/segment/analytics/kotlin/core/Errors.kt

@@ -10,6 +10,14 @@ fun Analytics.reportInternalError(error: Throwable) {
     Analytics.reportInternalError(error)
 }
 
+fun reportErrorWithMetrics(analytics: Analytics?, error: Throwable, message: String, metric:String, tags: Map<String, String>, log: String) {
+    analytics?.configuration?.errorHandler?.invoke(error)
+    var fullMessage = message
+    error.message?.let { fullMessage += ": $it"}
+    Analytics.segmentLog(fullMessage)
+    Telemetry.error(metric, tags, log)
+}
+
 fun Analytics.Companion.reportInternalError(error: Throwable) {
     error.message?.let {
         Analytics.segmentLog(it)

core/src/main/java/com/segment/analytics/kotlin/core/HTTPClient.kt

@@ -34,8 +34,8 @@ class HTTPClient(
             URL(url)
         } catch (e: MalformedURLException) {
             val error = IOException("Attempted to use malformed url: $url", e)
-            Analytics.reportInternalError(error)
-            Telemetry.error(Telemetry.INVOKE_ERROR,
+            reportErrorWithMetrics(null, e,
+                "Attempted to use malformed url: $url", Telemetry.INVOKE_ERROR_METRIC,
                 mapOf("error" to e.toString(), "writekey" to writeKey, "message" to "Malformed url"),
                 e.stackTraceToString()
             )

core/src/main/java/com/segment/analytics/kotlin/core/Settings.kt

@@ -13,8 +13,6 @@ import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
-import kotlinx.serialization.json.double
-import kotlinx.serialization.json.jsonPrimitive
 import kotlinx.serialization.serializer
 import java.io.BufferedReader
 
@@ -100,13 +98,6 @@ suspend fun Analytics.checkSettings() {
                 update(settingsObj)
             }
 
-            settingsObj?.metrics?.get("sampleRate")?.jsonPrimitive?.double.also {
-                if (it != null) {
-                    Telemetry.sampleRate = it
-                    Telemetry.start()
-                }
-            }
-
             // we're good to go back to a running state.
             store.dispatch(System.ToggleRunningAction(running = true), System::class)
         }
@@ -123,12 +114,8 @@ internal fun Analytics.fetchSettings(
     log("Fetched Settings: $settingsString")
     LenientJson.decodeFromString(settingsString)
 } catch (ex: Exception) {
-    reportInternalError(ex)
-    Analytics.segmentLog(
-        "${ex.message}: failed to fetch settings",
-        kind = LogKind.ERROR
-    )
-    Telemetry.error(Telemetry.INVOKE_ERROR,
+    reportErrorWithMetrics(this, ex, "Failed to fetch settings",
+        Telemetry.INVOKE_ERROR_METRIC,
         mapOf("error" to ex.toString(), "writekey" to writeKey, "message" to "Error retrieving settings"),
         ex.stackTraceToString()
     )

core/src/main/java/com/segment/analytics/kotlin/core/Telemetry.kt

@@ -4,9 +4,11 @@ import com.segment.analytics.kotlin.core.utilities.SegmentInstant
 import kotlinx.serialization.*
 import kotlinx.serialization.json.Json
 import kotlinx.coroutines.*
+import sovran.kotlin.Store
+import sovran.kotlin.Subscriber
 import java.net.HttpURLConnection
 import java.lang.System
-import java.time.LocalDateTime
+import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.Executors
 import kotlin.math.min
 import kotlin.math.roundToInt
@@ -36,15 +38,16 @@ fun logError(err: Throwable) {
     Analytics.reportInternalError(err)
 }
 
-object Telemetry {
+object Telemetry: Subscriber {
+    private const val METRICS_BASE_TAG = "analytics_mobile"
     // Metric class for Analytics SDK
-    const val INVOKE = "analytics_mobile.invoke"
+    const val INVOKE_METRIC = "$METRICS_BASE_TAG.invoke"
     // Metric class for Analytics SDK errors
-    const val INVOKE_ERROR = "analytics_mobile.invoke.error"
+    const val INVOKE_ERROR_METRIC = "$METRICS_BASE_TAG.invoke.error"
     // Metric class for Analytics SDK plugins
-    const val INTEGRATION = "analytics_mobile.integration.invoke"
+    const val INTEGRATION_METRIC = "$METRICS_BASE_TAG.integration.invoke"
     // Metric class for Analytics SDK plugin errors
-    const val INTEGRATION_ERROR = "analytics_mobile.integration.invoke.error"
+    const val INTEGRATION_ERROR_METRIC = "$METRICS_BASE_TAG.integration.invoke.error"
 
     var enable: Boolean = true
     var host: String = Constants.DEFAULT_API_HOST
@@ -64,39 +67,33 @@ object Telemetry {
             field = min(value, MAX_QUEUE_BYTES)
         }
 
-    private val queue = mutableListOf<RemoteMetric>()
+    private val queue = ConcurrentLinkedQueue<RemoteMetric>()
     private var queueBytes = 0
     private var queueSizeExceeded = false
     private val seenErrors = mutableMapOf<String, Int>()
     private var started = false
     private var rateLimitEndTime: Long = 0
-    private var telemetryScope: CoroutineScope? = null
-    private var telemetryDispatcher: ExecutorCoroutineDispatcher? = null
+    private val exceptionHandler = CoroutineExceptionHandler { _, t ->
+        errorHandler?.let {
+            it( Exception(
+                "Caught Exception in Telemetry Scope: ${t.message}",
+                t
+            ))
+        }
+    }
+    private var telemetryScope: CoroutineScope = CoroutineScope(SupervisorJob() + exceptionHandler)
+    private var telemetryDispatcher: ExecutorCoroutineDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
+    private var telemetryJob: Job? = null
     fun start() {
         if (started || sampleRate == 0.0) return
         started = true
 
         // Assume sampleRate is now set and everything in the queue hasn't had it applied
         if (Math.random() > sampleRate) {
             resetQueue()
-        } else {
-            queue.forEach {
-                it.value = (it.value / sampleRate).roundToInt()
-            }
         }
 
-        val exceptionHandler = CoroutineExceptionHandler { _, t ->
-            errorHandler?.let {
-                it( Exception(
-                    "Caught Exception in Telemetry Scope: ${t.message}",
-                    t
-                ))
-            }
-        }
-        telemetryScope = CoroutineScope(SupervisorJob() + exceptionHandler)
-        telemetryDispatcher = Executors.newCachedThreadPool().asCoroutineDispatcher()
-
-        telemetryScope?.launch(telemetryDispatcher!!) {
+        telemetryJob = telemetryScope.launch(telemetryDispatcher) {
             while (isActive) {
                 if (!enable) return@launch
                 try {
@@ -116,28 +113,27 @@ object Telemetry {
 
     fun reset()
     {
-        telemetryDispatcher?.close()
-        telemetryScope = null
-        telemetryDispatcher = null
+        telemetryJob?.cancel()
         resetQueue()
         seenErrors.clear()
         started = false
         rateLimitEndTime = 0
     }
 
     fun increment(metric: String, tags: Map<String, String>) {
+        start()
         if (!enable || sampleRate == 0.0) return
-        if (!metric.startsWith("analytics_mobile.")) return
+        if (!metric.startsWith(METRICS_BASE_TAG)) return
         if (tags.isEmpty()) return
         if (Math.random() > sampleRate) return
         if (queue.size >= maxQueueSize) return
 
-        addRemoteMetric(metric, tags, value = (1.0 / sampleRate).roundToInt())
+        addRemoteMetric(metric, tags)
     }
 
     fun error(metric:String, tags: Map<String, String>, log: String) {
         if (!enable || sampleRate == 0.0) return
-        if (!metric.startsWith("analytics_mobile.")) return
+        if (!metric.startsWith(METRICS_BASE_TAG)) return
         if (tags.isEmpty()) return
         if (queue.size >= maxQueueSize) return
 
@@ -157,8 +153,9 @@ object Telemetry {
             if (seenErrors.containsKey(errorKey)) {
                 seenErrors[errorKey] = seenErrors[errorKey]!! + 1
                 if (Math.random() > sampleRate) return
-                // Send how many we've seen after the first since we know for sure.
-                addRemoteMetric(metric, filteredTags, log=logData, value = seenErrors[errorKey]!!)
+                // Adjust how many we've seen after the first since we know for sure.
+                addRemoteMetric(metric, filteredTags, log=logData,
+                    value = (seenErrors[errorKey]!! * sampleRate).toInt())
                 seenErrors[errorKey] = 0
             } else {
                 addRemoteMetric(metric, filteredTags, log=logData)
@@ -189,18 +186,32 @@ object Telemetry {
     }
 
     private fun send() {
+        if (sampleRate == 0.0) return
+        var queueCount = queue.size
+        // Reset queue data size counter since all current queue items will be removed
+        queueBytes = 0
+        queueSizeExceeded = false
+        val sendQueue = mutableListOf<RemoteMetric>()
+        while (queueCount-- > 0 && !queue.isEmpty()) {
+            val m = queue.poll()
+            if(m != null) {
+                m.value = (m.value / sampleRate).roundToInt()
+                sendQueue.add(m)
+            }
+        }
         try {
             // Json.encodeToString by default does not include default values
             //  We're using this to leave off the 'log' parameter if unset.
-            val payload = Json.encodeToString(mapOf("series" to queue))
-            resetQueue()
+            val payload = Json.encodeToString(mapOf("series" to sendQueue))
 
             val connection = httpClient.upload(host)
             connection.outputStream?.use { outputStream ->
                 // Write the JSON string to the outputStream.
                 outputStream.write(payload.toByteArray(Charsets.UTF_8))
                 outputStream.flush() // Ensure all data is written
             }
+            connection.inputStream?.close()
+            connection.outputStream?.close()
             connection.close()
         } catch (e: HTTPException) {
             errorHandler?.invoke(e)
@@ -263,6 +274,22 @@ object Telemetry {
         }
     }
 
+    internal suspend fun subscribe(store: Store) {
+        store.subscribe(
+            this,
+            com.segment.analytics.kotlin.core.System::class,
+            initialState = true,
+            handler = ::systemUpdate,
+            queue = telemetryDispatcher
+        )
+    }
+    private suspend fun systemUpdate(system: com.segment.analytics.kotlin.core.System) {
+        system.settings?: let {
+            Telemetry.sampleRate = it.sampleRate
+            Telemetry.start()
+        }
+    }
+
     private fun resetQueue() {
         queue.clear()
         queueBytes = 0

core/src/main/java/com/segment/analytics/kotlin/core/platform/Mediator.kt

@@ -5,6 +5,7 @@ import com.segment.analytics.kotlin.core.BaseEvent
 import com.segment.analytics.kotlin.core.Telemetry
 import com.segment.analytics.kotlin.core.platform.plugins.logger.LogKind
 import com.segment.analytics.kotlin.core.platform.plugins.logger.segmentLog
+import com.segment.analytics.kotlin.core.reportErrorWithMetrics
 import com.segment.analytics.kotlin.core.reportInternalError
 import java.util.concurrent.CopyOnWriteArrayList
 import kotlin.reflect.KClass
@@ -34,7 +35,7 @@ internal class Mediator(internal var plugins: CopyOnWriteArrayList<Plugin> = Cop
             result?.let {
                 val copy = it.copy<BaseEvent>()
                 try {
-                    Telemetry.increment(Telemetry.INTEGRATION,
+                    Telemetry.increment(Telemetry.INTEGRATION_METRIC,
                         mapOf("message" to "event-${it.type}", "plugin" to "${plugin.type}-${plugin.javaClass}"))
                     when (plugin) {
                         is DestinationPlugin -> {
@@ -45,9 +46,9 @@ internal class Mediator(internal var plugins: CopyOnWriteArrayList<Plugin> = Cop
                         }
                     }
                 } catch (t: Throwable) {
-                    Analytics.reportInternalError("Caught Exception in plugin: $t")
                     Analytics.segmentLog("Skipping plugin due to Exception: $plugin", kind = LogKind.WARNING)
-                    Telemetry.error(Telemetry.INTEGRATION_ERROR,
+                    reportErrorWithMetrics(null, t,"Caught Exception in plugin",
+                        Telemetry.INTEGRATION_ERROR_METRIC,
                         mapOf("error" to t.toString(), "plugin" to "${plugin.type}-${plugin.javaClass}",
                         "writekey" to plugin.analytics.configuration.writeKey, "message" to "Exception executing plugin"),
                         t.stackTraceToString()
@@ -64,9 +65,8 @@ internal class Mediator(internal var plugins: CopyOnWriteArrayList<Plugin> = Cop
             try {
                 closure(it)
             } catch (t: Throwable) {
-                Analytics.reportInternalError(t)
-                Analytics.segmentLog("Caught Exception applying closure to plugin: $it: $t")
-                Telemetry.error(Telemetry.INTEGRATION_ERROR,
+                reportErrorWithMetrics(null, t,
+                    "Caught Exception applying closure to plugin: $it", Telemetry.INTEGRATION_ERROR_METRIC,
                     mapOf("error" to t.toString(), "plugin" to "${it.type}-${it.javaClass}",
                         "writekey" to it.analytics.configuration.writeKey, "message" to "Exception executing plugin"),
                     t.stackTraceToString()

core/src/main/java/com/segment/analytics/kotlin/core/platform/Timeline.kt

@@ -5,6 +5,7 @@ import com.segment.analytics.kotlin.core.BaseEvent
 import com.segment.analytics.kotlin.core.System
 import com.segment.analytics.kotlin.core.Telemetry
 import com.segment.analytics.kotlin.core.platform.plugins.logger.segmentLog
+import com.segment.analytics.kotlin.core.reportErrorWithMetrics
 import com.segment.analytics.kotlin.core.reportInternalError
 import kotlinx.coroutines.launch
 import kotlin.reflect.KClass
@@ -65,16 +66,15 @@ internal class Timeline {
         try {
             plugin.setup(analytics)
         } catch (t: Throwable) {
-            analytics.reportInternalError(t)
-            Analytics.segmentLog("Caught Exception while setting up plugin $plugin: $t")
-            Telemetry.error(Telemetry.INTEGRATION_ERROR,
+            reportErrorWithMetrics(analytics, t,
+                "Caught Exception while setting up plugin $plugin", Telemetry.INTEGRATION_ERROR_METRIC,
                 mapOf("plugin" to "${plugin.type.toString()}-${plugin.javaClass.toString()}",
                     "error" to t.toString(), "writekey" to analytics.configuration.writeKey,
                     "message" to "Exception setting up plugin"),
                 t.stackTraceToString()
             )
         }
-        Telemetry.increment(Telemetry.INTEGRATION,
+        Telemetry.increment(Telemetry.INTEGRATION_METRIC,
             mapOf("message" to "added",
                 "plugin" to "${plugin.type.toString()}-${plugin.javaClass.toString()}"))