chore(android): implement span processor and storage

closes #1407

android/measure/api/measure.api

@@ -80,6 +80,7 @@ public abstract interface class sh/measure/android/tracing/Span {
 	public abstract fun getDuration ()J
 	public abstract fun getName ()Ljava/lang/String;
 	public abstract fun getParentId ()Ljava/lang/String;
+	public abstract fun getSessionId ()Ljava/lang/String;
 	public abstract fun getSpanId ()Ljava/lang/String;
 	public abstract fun getStartTime ()J
 	public abstract fun getStatus ()Lsh/measure/android/tracing/SpanStatus;

android/measure/src/main/java/sh/measure/android/MeasureInitializer.kt

@@ -68,7 +68,9 @@ import sh.measure.android.storage.FileStorage
 import sh.measure.android.storage.FileStorageImpl
 import sh.measure.android.storage.PrefsStorage
 import sh.measure.android.storage.PrefsStorageImpl
+import sh.measure.android.tracing.MsrSpanProcessor
 import sh.measure.android.tracing.MsrTracer
+import sh.measure.android.tracing.SpanProcessor
 import sh.measure.android.tracing.Tracer
 import sh.measure.android.utils.AndroidSystemClock
 import sh.measure.android.utils.AndroidTimeProvider
@@ -357,10 +359,13 @@ internal class MeasureInitializerImpl(
         sessionManager = sessionManager,
         configProvider = configProvider,
     ),
+    private val spanProcessor: SpanProcessor = MsrSpanProcessor(eventProcessor),
     override val tracer: Tracer = MsrTracer(
         logger = logger,
         idProvider = idProvider,
         timeProvider = timeProvider,
+        spanProcessor = spanProcessor,
+        sessionManager = sessionManager,
     ),
 ) : MeasureInitializer
 

android/measure/src/main/java/sh/measure/android/events/EventProcessor.kt

@@ -14,6 +14,7 @@ import sh.measure.android.logger.Logger
 import sh.measure.android.screenshot.ScreenshotCollector
 import sh.measure.android.storage.EventStore
 import sh.measure.android.tracing.InternalTrace
+import sh.measure.android.tracing.SpanData
 import sh.measure.android.utils.IdProvider
 import sh.measure.android.utils.iso8601Timestamp
 import java.util.concurrent.RejectedExecutionException
@@ -87,6 +88,8 @@ internal interface EventProcessor {
         attributes: MutableMap<String, Any?> = mutableMapOf(),
         attachments: MutableList<Attachment> = mutableListOf(),
     )
+
+    fun trackSpan(spanData: SpanData)
 }
 
 internal class EventProcessorImpl(
@@ -166,6 +169,13 @@ internal class EventProcessorImpl(
         } ?: logger.log(LogLevel.Debug, "Event dropped: $type")
     }
 
+    override fun trackSpan(spanData: SpanData) {
+        ioExecutor.submit {
+            eventStore.store(spanData, sessionManager.getSessionId())
+            logger.log(LogLevel.Info, "Span processed: ${spanData.name}")
+        }
+    }
+
     private fun <T> track(
         data: T,
         timestamp: Long,

android/measure/src/main/java/sh/measure/android/storage/Database.kt

@@ -11,6 +11,7 @@ import sh.measure.android.exporter.AttachmentPacket
 import sh.measure.android.exporter.EventPacket
 import sh.measure.android.logger.LogLevel
 import sh.measure.android.logger.Logger
+import sh.measure.android.tracing.SpanData
 import java.io.Closeable
 
 internal interface Database : Closeable {
@@ -196,6 +197,11 @@ internal interface Database : Closeable {
      * Clears app exit for sessions which happened before the given [timestamp].
      */
     fun clearAppExitSessionsBefore(timestamp: Long)
+
+    /**
+     * Inserts a span into span table.
+     */
+    fun insertSpan(sessionId: String, spanData: SpanData): Boolean
 }
 
 /**
@@ -215,6 +221,7 @@ internal class DatabaseImpl(
             db.execSQL(Sql.CREATE_EVENTS_BATCH_TABLE)
             db.execSQL(Sql.CREATE_USER_DEFINED_ATTRIBUTES_TABLE)
             db.execSQL(Sql.CREATE_APP_EXIT_TABLE)
+            db.execSQL(Sql.CREATE_SPANS_TABLE)
             db.execSQL(Sql.CREATE_EVENTS_TIMESTAMP_INDEX)
             db.execSQL(Sql.CREATE_EVENTS_SESSION_ID_INDEX)
             db.execSQL(Sql.CREATE_EVENTS_BATCH_EVENT_ID_INDEX)
@@ -831,6 +838,26 @@ internal class DatabaseImpl(
         logger.log(LogLevel.Debug, "Cleared $result app_exit rows")
     }
 
+    override fun insertSpan(sessionId: String, spanData: SpanData): Boolean {
+        val values = ContentValues().apply {
+            put(SpansTable.COL_NAME, spanData.name)
+            put(SpansTable.COL_SESSION_ID, sessionId)
+            put(SpansTable.COL_SPAN_ID, spanData.spanId)
+            put(SpansTable.COL_TRACE_ID, spanData.traceId)
+            put(SpansTable.COL_PARENT_ID, spanData.parentId)
+            put(SpansTable.COL_START_TIME, spanData.startTime)
+            put(SpansTable.COL_END_TIME, spanData.endTime)
+            put(SpansTable.COL_DURATION, spanData.duration)
+            put(SpansTable.COL_STATUS, spanData.status.name)
+        }
+        val result = writableDatabase.insert(
+            SpansTable.TABLE_NAME,
+            null,
+            values,
+        )
+        return result != -1L
+    }
+
     override fun getAttachmentsForEvents(events: List<String>): List<String> {
         val attachmentIds = mutableListOf<String>()
         readableDatabase.rawQuery(Sql.getAttachmentsForEvents(events), null).use {

android/measure/src/main/java/sh/measure/android/storage/DbConstants.kt

@@ -8,6 +8,7 @@ internal object DbConstants {
 internal object DbVersion {
     const val V1 = 1
     const val V2 = 2
+    const val V3 = 3
 }
 
 internal object EventTable {
@@ -64,6 +65,19 @@ internal object AppExitTable {
     const val COL_CREATED_AT = "created_at"
 }
 
+internal object SpansTable {
+    const val TABLE_NAME = "spans"
+    const val COL_NAME = "name"
+    const val COL_SESSION_ID = "session_id"
+    const val COL_SPAN_ID = "span_id"
+    const val COL_TRACE_ID = "trace_id"
+    const val COL_PARENT_ID = "parent_id"
+    const val COL_START_TIME = "start_time"
+    const val COL_END_TIME = "end_time"
+    const val COL_DURATION = "duration"
+    const val COL_STATUS = "status"
+}
+
 internal object UserDefinedAttributesTable {
     const val TABLE_NAME = "user_defined_attributes"
     const val COL_KEY = "key"
@@ -144,6 +158,20 @@ internal object Sql {
         )
     """
 
+    const val CREATE_SPANS_TABLE = """
+        CREATE TABLE IF NOT EXISTS ${SpansTable.TABLE_NAME} (
+            ${SpansTable.COL_SPAN_ID} TEXT NOT NULL PRIMARY KEY,
+            ${SpansTable.COL_NAME} TEXT NOT NULL,
+            ${SpansTable.COL_SESSION_ID} TEXT NOT NULL,
+            ${SpansTable.COL_TRACE_ID} TEXT NOT NULL,
+            ${SpansTable.COL_PARENT_ID} TEXT,
+            ${SpansTable.COL_START_TIME} INTEGER NOT NULL,
+            ${SpansTable.COL_END_TIME} INTEGER NOT NULL,
+            ${SpansTable.COL_DURATION} INTEGER NOT NULL,
+            ${SpansTable.COL_STATUS} TEXT NOT NULL
+        )
+    """
+
     const val CREATE_SESSIONS_CREATED_AT_INDEX = """
         CREATE INDEX IF NOT EXISTS sessions_created_at_index ON ${SessionsTable.TABLE_NAME} (${SessionsTable.COL_CREATED_AT})
     """

android/measure/src/main/java/sh/measure/android/storage/DbMigrations.kt

@@ -13,6 +13,7 @@ internal object DbMigrations {
                 for (version in oldVersion + 1..newVersion) {
                     when (version) {
                         DbVersion.V2 -> migrateToV2(db)
+                        DbVersion.V3 -> migrateToV3(db)
                         else -> logger.log(
                             LogLevel.Warning,
                             "No migration found for version $version",
@@ -40,4 +41,8 @@ internal object DbMigrations {
             """.trimIndent(),
         )
     }
+
+    private fun migrateToV3(db: SQLiteDatabase) {
+        db.execSQL(Sql.CREATE_SPANS_TABLE)
+    }
 }

android/measure/src/main/java/sh/measure/android/storage/EventStore.kt

@@ -9,11 +9,13 @@ import sh.measure.android.events.EventType
 import sh.measure.android.logger.LogLevel
 import sh.measure.android.logger.Logger
 import sh.measure.android.okhttp.HttpData
+import sh.measure.android.tracing.SpanData
 import sh.measure.android.utils.IdProvider
 import java.io.File
 
 internal interface EventStore {
     fun <T> store(event: Event<T>)
+    fun store(spanData: SpanData, sessionId: String)
 }
 
 /**
@@ -31,6 +33,13 @@ internal class EventStoreImpl(
     private val idProvider: IdProvider,
 ) : EventStore {
 
+    override fun store(spanData: SpanData, sessionId: String) {
+        val result = database.insertSpan(sessionId, spanData)
+        if (!result) {
+            logger.log(LogLevel.Error, "Unable to store span(${spanData.name}) to database")
+        }
+    }
+
     override fun <T> store(event: Event<T>) {
         val serializedAttributes = event.serializeAttributes()
         val serializedUserDefAttributes = event.serializeUserDefinedAttributes()

android/measure/src/main/java/sh/measure/android/tracing/InvalidSpan.kt

@@ -6,6 +6,9 @@ internal class InvalidSpan : Span {
     override val name: String = "invalid"
 
     override val parentId: String? = null
+
+    override val sessionId: String = ""
+
     override val startTime: Long = 0
 
     override fun getStatus(): SpanStatus {

android/measure/src/main/java/sh/measure/android/tracing/MsrSpan.kt

@@ -1,6 +1,6 @@
 package sh.measure.android.tracing
 
-import android.util.Log
+import sh.measure.android.SessionManager
 import sh.measure.android.logger.LogLevel
 import sh.measure.android.logger.Logger
 import sh.measure.android.utils.IdProvider
@@ -12,39 +12,47 @@ import sh.measure.android.utils.TimeProvider
 internal class MsrSpan(
     private val logger: Logger,
     private val timeProvider: TimeProvider,
+    private val spanProcessor: SpanProcessor,
     override val name: String,
     override val spanId: String,
     override val traceId: String,
     override val parentId: String?,
+    override val sessionId: String,
     override val startTime: Long,
-) : Span {
+) : Span, ReadableSpan {
     private val lock = Any()
     private var status = SpanStatus.Unset
     private var endTime = 0L
     private var hasEnded: EndState = EndState.NotEnded
-    private var duration: Long = 0
 
     companion object {
         fun startSpan(
             name: String,
             logger: Logger,
             timeProvider: TimeProvider,
+            spanProcessor: SpanProcessor,
+            sessionManager: SessionManager,
             idProvider: IdProvider,
             parentSpan: Span?,
             timestamp: Long? = null,
         ): Span {
             val startTime = timestamp ?: timeProvider.now()
             val spanId: String = idProvider.spanId()
             val traceId = parentSpan?.traceId ?: idProvider.traceId()
-            return MsrSpan(
+            val sessionId = sessionManager.getSessionId()
+            val span = MsrSpan(
                 logger = logger,
                 timeProvider = timeProvider,
+                spanProcessor = spanProcessor,
                 name = name,
                 spanId = spanId,
                 traceId = traceId,
                 parentId = parentSpan?.spanId,
+                sessionId = sessionId,
                 startTime = startTime,
             )
+            spanProcessor.onStart(span)
+            return span
         }
     }
 
@@ -84,14 +92,11 @@ internal class MsrSpan(
             endTime = timestamp
             hasEnded = EndState.Ending
         }
-
-        // trigger onEnding
+        spanProcessor.onEnding(this)
         synchronized(lock) {
             hasEnded = EndState.Ended
         }
-
-        // trigger onEnded
-        Log.i("MsrSpan", "${this.toSpanData()}")
+        spanProcessor.onEnded(this)
     }
 
     override fun makeCurrent(): Scope {
@@ -103,28 +108,43 @@ internal class MsrSpan(
     }
 
     override fun getDuration(): Long {
-        return duration
-    }
-
-    private enum class EndState {
-        NotEnded,
-        Ending,
-        Ended,
+        synchronized(lock) {
+            if (hasEnded != EndState.Ended) {
+                logger.log(
+                    LogLevel.Warning,
+                    "Attempt to duration of a span($name) that has not ended",
+                )
+                return 0
+            } else {
+                return calculateDuration()
+            }
+        }
     }
 
-    private fun toSpanData(): SpanData {
+    override fun toSpanData(): SpanData {
         synchronized(lock) {
-            this.duration = (endTime - startTime).coerceAtLeast(0)
             return SpanData(
                 spanId = spanId,
+                traceId = traceId,
                 name = name,
                 startTime = startTime,
                 endTime = endTime,
                 status = status,
                 hasEnded = hasEnded == EndState.Ended,
                 parentId = parentId,
-                duration = duration,
+                sessionId = sessionId,
+                duration = calculateDuration(),
             )
         }
     }
+
+    private fun calculateDuration(): Long {
+        return (endTime - startTime).coerceAtLeast(0)
+    }
+
+    private enum class EndState {
+        NotEnded,
+        Ending,
+        Ended,
+    }
 }

android/measure/src/main/java/sh/measure/android/tracing/MsrSpanBuilder.kt

@@ -1,5 +1,6 @@
 package sh.measure.android.tracing
 
+import sh.measure.android.SessionManager
 import sh.measure.android.logger.Logger
 import sh.measure.android.utils.IdProvider
 import sh.measure.android.utils.TimeProvider
@@ -8,6 +9,8 @@ internal class MsrSpanBuilder(
     val name: String,
     private val idProvider: IdProvider,
     private val timeProvider: TimeProvider,
+    private val spanProcessor: SpanProcessor,
+    private val sessionManager: SessionManager,
     private val logger: Logger,
 ) : SpanBuilder {
     private var parentSpan: Span? = null
@@ -26,10 +29,12 @@ internal class MsrSpanBuilder(
     override fun startSpan(): Span {
         val parent = findSpanParent()
         return MsrSpan.startSpan(
-            name,
-            logger,
-            timeProvider,
-            idProvider,
+            name = name,
+            logger = logger,
+            timeProvider = timeProvider,
+            spanProcessor = spanProcessor,
+            sessionManager = sessionManager,
+            idProvider = idProvider,
             parentSpan = parent,
         )
     }
@@ -41,6 +46,8 @@ internal class MsrSpanBuilder(
             logger = logger,
             timeProvider = timeProvider,
             idProvider = idProvider,
+            spanProcessor = spanProcessor,
+            sessionManager = sessionManager,
             parentSpan = parent,
             timestamp = timestamp,
         )