feat: Android observability plugin session replay support

e2e/android/app/src/main/java/com/example/androidobservability/BaseApplication.kt

@@ -9,6 +9,7 @@ import com.launchdarkly.sdk.android.Components
 import com.launchdarkly.sdk.android.LDClient
 import com.launchdarkly.sdk.android.LDConfig
 import com.launchdarkly.observability.plugin.Observability
+import com.launchdarkly.observability.replay.ReplayInstrumentation
 import com.launchdarkly.sdk.android.LDAndroidLogging
 import com.launchdarkly.sdk.android.integrations.Plugin
 import io.opentelemetry.api.common.AttributeKey
@@ -29,6 +30,10 @@ open class BaseApplication : Application() {
         ),
         debug = true,
         logAdapter = LDAndroidLogging.adapter(),
+        // TODO: consider these being factories so that the obs plugin can pass instantiation data, log adapter
+        instrumentations = listOf(
+            ReplayInstrumentation()
+        ),
     )
 
     var telemetryInspector: TelemetryInspector? = null

sdk/@launchdarkly/observability-android/lib/build.gradle.kts

@@ -52,6 +52,11 @@ dependencies {
     // Android crash instrumentation
     implementation("io.opentelemetry.android.instrumentation:crash:0.11.0-alpha")
 
+    // TODO: O11Y-626 - move replay instrumentation and associated compose dependencies into dedicated package
+    // Compose dependencies for capture functionality
+    implementation("androidx.compose.ui:ui:1.7.5")
+    implementation("androidx.compose.ui:ui-tooling:1.7.5")
+
     // Use JUnit Jupiter for testing.
     testImplementation(platform("org.junit:junit-bom:5.13.4"))
     testImplementation("org.junit.jupiter:junit-jupiter")

sdk/@launchdarkly/observability-android/lib/src/main/kotlin/com/launchdarkly/observability/api/Options.kt

@@ -2,13 +2,15 @@ package com.launchdarkly.observability.api
 
 import com.launchdarkly.logging.LDLogAdapter
 import com.launchdarkly.observability.BuildConfig
+import com.launchdarkly.observability.interfaces.LDExtendedInstrumentation
 import com.launchdarkly.sdk.android.LDTimberLogging
 import io.opentelemetry.api.common.Attributes
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.minutes
 
-private const val DEFAULT_OTLP_ENDPOINT = "https://otel.observability.app.launchdarkly.com:4318"
-private const val DEFAULT_BACKEND_URL = "https://pub.observability.app.launchdarkly.com"
+const val DEFAULT_SERVICE_NAME = "observability-android"
+const val DEFAULT_OTLP_ENDPOINT = "https://otel.observability.app.launchdarkly.com:4318"
+const val DEFAULT_BACKEND_URL = "https://pub.observability.app.launchdarkly.com"
 
 /**
  * Configuration options for the Observability plugin.
@@ -27,21 +29,23 @@ private const val DEFAULT_BACKEND_URL = "https://pub.observability.app.launchdar
  * @property disableMetrics Disables metrics if true. Defaults to false.
  * @property logAdapter The log adapter to use. Defaults to using the LaunchDarkly SDK's LDTimberLogging.adapter(). Use LDAndroidLogging.adapter() to use the Android logging adapter.
  * @property loggerName The name of the logger to use. Defaults to "LaunchDarklyObservabilityPlugin".
+ * @property instrumentations List of additional instrumentations to use
  */
 data class Options(
-    val serviceName: String = "observability-android",
+    val serviceName: String = DEFAULT_SERVICE_NAME,
     val serviceVersion: String = BuildConfig.OBSERVABILITY_SDK_VERSION,
     val otlpEndpoint: String = DEFAULT_OTLP_ENDPOINT,
     val backendUrl: String = DEFAULT_BACKEND_URL,
     val resourceAttributes: Attributes = Attributes.empty(),
     val customHeaders: Map<String, String> = emptyMap(),
     val sessionBackgroundTimeout: Duration = 15.minutes,
     val debug: Boolean = false,
-    // TODO O11Y-398: implement disable config options after all other instrumentations are implemented
     val disableErrorTracking: Boolean = false,
     val disableLogs: Boolean = false,
     val disableTraces: Boolean = false,
     val disableMetrics: Boolean = false,
     val logAdapter: LDLogAdapter = LDTimberLogging.adapter(), // this follows the LaunchDarkly SDK's default log adapter
-    val loggerName: String = "LaunchDarklyObservabilityPlugin"
+    val loggerName: String = "LaunchDarklyObservabilityPlugin",
+    // TODO: update this to provide a list of factories instead of instances
+    val instrumentations: List<LDExtendedInstrumentation> = emptyList()
 )

sdk/@launchdarkly/observability-android/lib/src/main/kotlin/com/launchdarkly/observability/client/InstrumentationManager.kt

@@ -7,6 +7,7 @@ import com.launchdarkly.observability.interfaces.Metric
 import com.launchdarkly.observability.network.GraphQLClient
 import com.launchdarkly.observability.network.SamplingApiService
 import com.launchdarkly.observability.sampling.CustomSampler
+import com.launchdarkly.observability.sampling.ExportSampler
 import com.launchdarkly.observability.sampling.SamplingConfig
 import com.launchdarkly.observability.sampling.SamplingLogExporter
 import com.launchdarkly.observability.sampling.SamplingTraceExporter
@@ -29,6 +30,7 @@ import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter
 import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter
 import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter
 import io.opentelemetry.sdk.common.CompletableResultCode
+import io.opentelemetry.sdk.logs.LogRecordProcessor
 import io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder
 import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor
 import io.opentelemetry.sdk.logs.export.LogRecordExporter
@@ -65,20 +67,6 @@ class InstrumentationManager(
     private val logger: LDLogger,
     private val options: Options,
 ) {
-    companion object {
-        private const val METRICS_PATH = "/v1/metrics"
-        private const val LOGS_PATH = "/v1/logs"
-        private const val TRACES_PATH = "/v1/traces"
-        private const val INSTRUMENTATION_SCOPE_NAME = "com.launchdarkly.observability"
-        const val ERROR_SPAN_NAME = "highlight.error"
-        private const val BATCH_MAX_QUEUE_SIZE = 100
-        private const val BATCH_SCHEDULE_DELAY_MS = 1000L
-        private const val BATCH_EXPORTER_TIMEOUT_MS = 5000L
-        private const val BATCH_MAX_EXPORT_SIZE = 10
-        private const val METRICS_EXPORT_INTERVAL_SECONDS = 10L
-        private const val FLUSH_TIMEOUT_SECONDS = 5L
-    }
-
     private val otelRUM: OpenTelemetryRum
     private var otelMeter: Meter
     private var otelLogger: Logger
@@ -91,7 +79,7 @@ class InstrumentationManager(
     private var inMemoryMetricExporter: InMemoryMetricExporter? = null
     private var telemetryInspector: TelemetryInspector? = null
     private var spanProcessor: BatchSpanProcessor? = null
-    private var logProcessor: BatchLogRecordProcessor? = null
+    private var logProcessor: LogRecordProcessor? = null
     private var metricsReader: PeriodicMetricReader? = null
     private val gaugeCache = ConcurrentHashMap<String, DoubleGauge>()
     private val counterCache = ConcurrentHashMap<String, LongCounter>()
@@ -104,12 +92,15 @@ class InstrumentationManager(
     init {
         val otelRumConfig = createOtelRumConfig()
 
-        otelRUM = OpenTelemetryRum.builder(application, otelRumConfig)
+        val builder = OpenTelemetryRum.builder(application, otelRumConfig)
             .addLoggerProviderCustomizer { sdkLoggerProviderBuilder, _ ->
+                // TODO: O11Y-627 - need to refactor this so that the disableLogs option is specific to core logging functionality. when logs are disabled, session replay logs should not be blocked
                 return@addLoggerProviderCustomizer if (options.disableLogs && options.disableErrorTracking) {
                     sdkLoggerProviderBuilder
                 } else {
-                    configureLoggerProvider(sdkLoggerProviderBuilder)
+                    val processor = createLoggerProcessor(sdkLoggerProviderBuilder, customSampler, sdkKey, resources, logger, options)
+                    logProcessor = processor
+                    sdkLoggerProviderBuilder.addLogRecordProcessor(processor)
                 }
             }
             .addTracerProviderCustomizer { sdkTracerProviderBuilder, _ ->
@@ -126,7 +117,12 @@ class InstrumentationManager(
                     configureMeterProvider(sdkMeterProviderBuilder)
                 }
             }
-            .build()
+
+        for (instrumentation in options.instrumentations) {
+            builder.addInstrumentation(instrumentation)
+        }
+
+        otelRUM = builder.build()
 
         initializeTelemetryInspector()
         loadSamplingConfigAsync()
@@ -159,17 +155,6 @@ class InstrumentationManager(
         return !options.disableLogs || !options.disableTraces || !options.disableMetrics || !options.disableErrorTracking
     }
 
-    private fun configureLoggerProvider(sdkLoggerProviderBuilder: SdkLoggerProviderBuilder): SdkLoggerProviderBuilder {
-        val primaryLogExporter = createOtlpLogExporter()
-        sdkLoggerProviderBuilder.setResource(resources)
-
-        val finalExporter = createLogExporter(primaryLogExporter)
-        val processor = createBatchLogRecordProcessor(finalExporter)
-
-        logProcessor = processor
-        return sdkLoggerProviderBuilder.addLogRecordProcessor(processor)
-    }
-
     private fun configureTracerProvider(sdkTracerProviderBuilder: SdkTracerProviderBuilder): SdkTracerProviderBuilder {
         val primarySpanExporter = createOtlpSpanExporter()
         sdkTracerProviderBuilder.setResource(resources)
@@ -193,13 +178,6 @@ class InstrumentationManager(
             .registerMetricReader(metricReader)
     }
 
-    private fun createOtlpLogExporter(): LogRecordExporter {
-        return OtlpHttpLogRecordExporter.builder()
-            .setEndpoint(options.otlpEndpoint + LOGS_PATH)
-            .setHeaders { options.customHeaders }
-            .build()
-    }
-
     private fun createOtlpSpanExporter(): SpanExporter {
         return OtlpHttpSpanExporter.builder()
             .setEndpoint(options.otlpEndpoint + TRACES_PATH)
@@ -214,28 +192,6 @@ class InstrumentationManager(
             .build()
     }
 
-    private fun createLogExporter(primaryExporter: LogRecordExporter): LogRecordExporter {
-        val baseExporter = if (options.debug) {
-            LogRecordExporter.composite(
-                buildList {
-                    add(primaryExporter)
-                    add(DebugLogExporter(logger))
-                    add(InMemoryLogRecordExporter.create().also { inMemoryLogExporter = it })
-                }
-            )
-        } else {
-            primaryExporter
-        }
-
-        val conditionalExporter = ConditionalLogRecordExporter(
-            delegate = baseExporter,
-            allowNormalLogs = !options.disableLogs,
-            allowCrashes = !options.disableErrorTracking
-        )
-
-        return SamplingLogExporter(conditionalExporter, customSampler)
-    }
-
     private fun createSpanExporter(primaryExporter: SpanExporter): SpanExporter {
         val baseExporter = if (options.debug) {
             SpanExporter.composite(
@@ -295,15 +251,6 @@ class InstrumentationManager(
         }
     }
 
-    private fun createBatchLogRecordProcessor(logRecordExporter: LogRecordExporter): BatchLogRecordProcessor {
-        return BatchLogRecordProcessor.builder(logRecordExporter)
-            .setMaxQueueSize(BATCH_MAX_QUEUE_SIZE)
-            .setScheduleDelay(BATCH_SCHEDULE_DELAY_MS, TimeUnit.MILLISECONDS)
-            .setExporterTimeout(BATCH_EXPORTER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
-            .setMaxExportBatchSize(BATCH_MAX_EXPORT_SIZE)
-            .build()
-    }
-
     private fun createBatchSpanProcessor(spanExporter: SpanExporter): BatchSpanProcessor {
         return BatchSpanProcessor.builder(spanExporter)
             .setMaxQueueSize(BATCH_MAX_QUEUE_SIZE)
@@ -423,4 +370,87 @@ class InstrumentationManager(
             null
         }
     }
+
+    companion object {
+        private const val METRICS_PATH = "/v1/metrics"
+        private const val LOGS_PATH = "/v1/logs"
+        private const val TRACES_PATH = "/v1/traces"
+        private const val INSTRUMENTATION_SCOPE_NAME = "com.launchdarkly.observability"
+        const val ERROR_SPAN_NAME = "highlight.error"
+        private const val BATCH_MAX_QUEUE_SIZE = 100
+        private const val BATCH_SCHEDULE_DELAY_MS = 1000L
+        private const val BATCH_EXPORTER_TIMEOUT_MS = 5000L
+        private const val BATCH_MAX_EXPORT_SIZE = 10
+        private const val METRICS_EXPORT_INTERVAL_SECONDS = 10L
+        private const val FLUSH_TIMEOUT_SECONDS = 5L
+
+        internal fun createLoggerProcessor(
+            sdkLoggerProviderBuilder: SdkLoggerProviderBuilder,
+            exportSampler: ExportSampler,
+            sdkKey: String,
+            resource: Resource,
+            logger: LDLogger,
+            options: Options,
+        ): LogRecordProcessor {
+            val primaryLogExporter = createOtlpLogExporter(options)
+            sdkLoggerProviderBuilder.setResource(resource)
+
+            val finalExporter = createLogExporter(primaryLogExporter, exportSampler, logger, options)
+            val baseProcessor = createBatchLogRecordProcessor(finalExporter)
+
+            // Here we set up a routing log processor that will route logs with a matching scope name to the
+            // respective instrumentation's log record processor.  If the log's scope name does not match
+            // an instrumentation's scope name, it will fall through to the base processor.  This was
+            // originally added to route replay instrumentation logs through a separate log processing
+            // pipeline to provide instrumentation specific caching and export.
+            val routingLogRecordProcessor = RoutingLogRecordProcessor(fallthroughProcessor = baseProcessor)
+            for (i in options.instrumentations) {
+                i.getLogRecordProcessor(credential = sdkKey)?.let {
+                    i.getLoggerScopeName().let { scopeName ->
+                        routingLogRecordProcessor.registerProcessor(scopeName, it)
+                    }
+                }
+            }
+
+            return routingLogRecordProcessor
+        }
+
+        private fun createOtlpLogExporter(options: Options): LogRecordExporter {
+            return OtlpHttpLogRecordExporter.builder()
+                .setEndpoint(options.otlpEndpoint + LOGS_PATH)
+                .setHeaders { options.customHeaders }
+                .build()
+        }
+
+        private fun createLogExporter(primaryExporter: LogRecordExporter, exportSampler: ExportSampler, logger: LDLogger, options: Options): LogRecordExporter {
+            val baseExporter = if (options.debug) {
+                LogRecordExporter.composite(
+                    buildList {
+                        add(primaryExporter)
+                        add(DebugLogExporter(logger))
+//                        add(InMemoryLogRecordExporter.create().also { inMemoryLogExporter = it }) // TODO: figure out how to factor this out so functions can be static
+                    }
+                )
+            } else {
+                primaryExporter
+            }
+
+            val conditionalExporter = ConditionalLogRecordExporter(
+                delegate = baseExporter,
+                allowNormalLogs = !options.disableLogs,
+                allowCrashes = !options.disableErrorTracking
+            )
+
+            return SamplingLogExporter(conditionalExporter, exportSampler)
+        }
+
+        fun createBatchLogRecordProcessor(logRecordExporter: LogRecordExporter): BatchLogRecordProcessor {
+            return BatchLogRecordProcessor.builder(logRecordExporter)
+                .setMaxQueueSize(BATCH_MAX_QUEUE_SIZE)
+                .setScheduleDelay(BATCH_SCHEDULE_DELAY_MS, TimeUnit.MILLISECONDS)
+                .setExporterTimeout(BATCH_EXPORTER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+                .setMaxExportBatchSize(BATCH_MAX_EXPORT_SIZE)
+                .build()
+        }
+    }
 }

sdk/@launchdarkly/observability-android/lib/src/main/kotlin/com/launchdarkly/observability/client/NoopLogRecordPrcoessor.kt

@@ -0,0 +1,19 @@
+package com.launchdarkly.observability.client
+
+import io.opentelemetry.context.Context
+import io.opentelemetry.sdk.logs.LogRecordProcessor
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord
+
+/**
+ * A [LogRecordProcessor] that, surprise, does nothing.
+ */
+internal class NoopLogRecordProcessor private constructor() : LogRecordProcessor {
+    override fun onEmit(context: Context, logRecord: ReadWriteLogRecord) {}
+
+    companion object {
+        private val INSTANCE = NoopLogRecordProcessor()
+
+        val instance: LogRecordProcessor
+            get() = INSTANCE
+    }
+}

sdk/@launchdarkly/observability-android/lib/src/main/kotlin/com/launchdarkly/observability/client/RoutingLogRecordProcessor.kt

@@ -0,0 +1,27 @@
+package com.launchdarkly.observability.client
+
+import io.opentelemetry.context.Context
+import io.opentelemetry.sdk.logs.LogRecordProcessor
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * A [LogRecordProcessor] that implements a routing pattern to other registered [LogRecordProcessor]s
+ * using scope name as routing criteria.  If no [LogRecordProcessor] is registered for the given
+ * scope name, the [fallthroughProcessor] is called to handle the log.
+ */
+class RoutingLogRecordProcessor(
+    private val fallthroughProcessor: LogRecordProcessor = NoopLogRecordProcessor.instance
+) : LogRecordProcessor {
+    private val processors = ConcurrentHashMap<String, LogRecordProcessor>()
+
+    fun registerProcessor(scopeName: String, processor: LogRecordProcessor) {
+        processors[scopeName] = processor
+    }
+
+    override fun onEmit(context: Context, logRecord: ReadWriteLogRecord) {
+        val scopeName = logRecord.instrumentationScopeInfo.name
+        val processor = processors[scopeName] ?: fallthroughProcessor
+        processor.onEmit(context, logRecord)
+    }
+}

sdk/@launchdarkly/observability-android/lib/src/main/kotlin/com/launchdarkly/observability/interfaces/LDExtendedInstrumentation.kt

@@ -0,0 +1,22 @@
+package com.launchdarkly.observability.interfaces
+
+import io.opentelemetry.android.instrumentation.AndroidInstrumentation
+import io.opentelemetry.sdk.logs.LogRecordProcessor
+
+// This interface is for internal LaunchDarkly use only.
+interface LDExtendedInstrumentation : AndroidInstrumentation {
+
+    /**
+     * @return the scope name that this instrumentation will use for its logs
+     */
+    fun getLoggerScopeName(): String
+
+    /**
+     * @param credential the credential that will be used by exporters for authenticating with
+     * services
+     *
+     * @return the instrumentation specific [LogRecordProcessor] for handling this instrumentations
+     * logs, or null if this instrumentation does not need to provide any specific handling.
+     */
+    fun getLogRecordProcessor(credential: String): LogRecordProcessor? = null
+}