refactor: O11Y-630 - Sampling and conditional logic for LOGS moved from exporters to processors (#288)

## Summary
To avoid buffering logs that should be discarded due to the sampling
config, the stop sampling logs in the exporter to do it in the
processors.
"Conditional logic" refers to the logic that was inside the
ConditionLogRecordExporter (Used to allow different filtering rules for
crashes vs normal logs)

## How did you test this change?
Unit tests

## Are there any deployment considerations?
No

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Replaces exporter-based log sampling and crash/normal filtering with
processor-based equivalents and updates the log pipeline and tests
accordingly.
> 
> - **Logs pipeline**:
> - Introduces `ConditionalLogRecordProcessor` to filter crash
(`io.opentelemetry.crash`) vs normal logs.
> - Introduces `SamplingLogProcessor` to apply sampling before
downstream processing.
> - Removes exporter wrappers `ConditionalLogRecordExporter` and
`SamplingLogExporter`.
> - Updates `InstrumentationManager.createLoggerProcessor` to chain
`SamplingLogProcessor` → `ConditionalLogRecordProcessor` →
`RoutingLogRecordProcessor`, and simplifies `createLogExporter` to only
build a composite in debug mode.
> - **Tests**:
> - Adds unit tests for `ConditionalLogRecordProcessor` and
`SamplingLogProcessor`.
>   - Removes tests for deleted exporter classes.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
2067fb3. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: agrognetti

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

@@ -0,0 +1,50 @@
+package com.launchdarkly.observability.client
+
+import io.opentelemetry.context.Context
+import io.opentelemetry.sdk.common.CompletableResultCode
+import io.opentelemetry.sdk.logs.LogRecordProcessor
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord
+
+/**
+ * Filters log records before they reach downstream processors based on their instrumentation scope.
+ *
+ * Crash logs emitted by OpenTelemetry's crash instrumentation use the "io.opentelemetry.crash" scope.
+ * This processor can drop those logs independently from normal application logs.
+ */
+class ConditionalLogRecordProcessor(
+    private val delegate: LogRecordProcessor,
+    private val allowNormalLogs: Boolean,
+    private val allowCrashes: Boolean
+) : LogRecordProcessor {
+
+    override fun onEmit(context: Context, logRecord: ReadWriteLogRecord) {
+        if (!shouldEmitLog(logRecord)) {
+            return
+        }
+
+        delegate.onEmit(context, logRecord)
+    }
+
+    override fun forceFlush(): CompletableResultCode = delegate.forceFlush()
+
+    override fun shutdown(): CompletableResultCode = delegate.shutdown()
+
+    override fun close() {
+        delegate.close()
+    }
+
+    private fun shouldEmitLog(logRecord: ReadWriteLogRecord): Boolean {
+        // Check if this is a crash log from OpenTelemetry's CrashReporterInstrumentation
+        val instrumentationScopeName = logRecord.instrumentationScopeInfo.name
+        val isCrashLog = instrumentationScopeName == CRASH_INSTRUMENTATION_SCOPE
+
+        return when {
+            isCrashLog -> allowCrashes
+            else -> allowNormalLogs
+        }
+    }
+
+    private companion object {
+        const val CRASH_INSTRUMENTATION_SCOPE = "io.opentelemetry.crash"
+    }
+}

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

@@ -9,7 +9,7 @@ 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.SamplingLogProcessor
 import com.launchdarkly.observability.sampling.SamplingTraceExporter
 import com.launchdarkly.observability.coroutines.DispatcherProviderHolder
 import io.opentelemetry.android.OpenTelemetryRum
@@ -415,20 +415,30 @@ class InstrumentationManager(
 
             val finalExporter = createLogExporter(
                 primaryLogExporter,
-                exportSampler,
                 logger,
                 telemetryInspector,
                 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)
+
+            val samplingProcessor = SamplingLogProcessor(
+                createBatchLogRecordProcessor(finalExporter),
+                exportSampler
+            )
+
+            val baseProcessor = ConditionalLogRecordProcessor(
+                delegate = samplingProcessor,
+                allowNormalLogs = !options.disableLogs,
+                allowCrashes = !options.disableErrorTracking
+            )
+
+            /*
+                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)
             options.instrumentations.forEach { instrumentation ->
                 instrumentation.getLogRecordProcessor(credential = sdkKey)?.let { processor ->
                     instrumentation.getLoggerScopeName().let { scopeName ->
@@ -449,12 +459,11 @@ class InstrumentationManager(
 
         private fun createLogExporter(
             primaryExporter: LogRecordExporter,
-            exportSampler: ExportSampler,
             logger: LDLogger,
             telemetryInspector: TelemetryInspector?,
             options: Options
         ): LogRecordExporter {
-            val baseExporter = if (options.debug) {
+            return if (options.debug) {
                 LogRecordExporter.composite(
                     buildList {
                         add(primaryExporter)
@@ -465,14 +474,6 @@ class InstrumentationManager(
             } else {
                 primaryExporter
             }
-
-            val conditionalExporter = ConditionalLogRecordExporter(
-                delegate = baseExporter,
-                allowNormalLogs = !options.disableLogs,
-                allowCrashes = !options.disableErrorTracking
-            )
-
-            return SamplingLogExporter(conditionalExporter, exportSampler)
         }
 
         fun createBatchLogRecordProcessor(logRecordExporter: LogRecordExporter): BatchLogRecordProcessor {

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

@@ -0,0 +1,60 @@
+package com.launchdarkly.observability.sampling
+
+import io.opentelemetry.api.common.Attributes
+import io.opentelemetry.context.Context
+import io.opentelemetry.sdk.common.CompletableResultCode
+import io.opentelemetry.sdk.logs.LogRecordProcessor
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord
+
+/**
+ * A [LogRecordProcessor] that applies sampling logic before delegating to another [LogRecordProcessor].
+ *
+ * This processor drops unsampled log records so they are never enqueued in downstream processors,
+ * reducing buffering and export overhead for logs that would have been filtered out later.
+ *
+ * @param delegate The underlying [LogRecordProcessor] that should receive sampled log records.
+ * @param sampler The sampler that decides whether a log record should be processed.
+ */
+class SamplingLogProcessor(
+    private val delegate: LogRecordProcessor,
+    private val sampler: ExportSampler
+) : LogRecordProcessor {
+
+    override fun onEmit(context: Context, logRecord: ReadWriteLogRecord) {
+        if (!sampler.isSamplingEnabled()) {
+            delegate.onEmit(context, logRecord)
+            return
+        }
+
+        val samplingResult = sampler.sampleLog(logRecord.toLogRecordData())
+        if (!samplingResult.sample) {
+            return
+        }
+
+        samplingResult.attributes?.let { addSamplingAttributes(logRecord, it) }
+        delegate.onEmit(context, logRecord)
+    }
+
+    override fun shutdown(): CompletableResultCode {
+        return delegate.shutdown()
+    }
+
+    override fun forceFlush(): CompletableResultCode {
+        return delegate.forceFlush()
+    }
+
+    override fun close() {
+        delegate.close()
+    }
+
+    private fun addSamplingAttributes(
+        logRecord: ReadWriteLogRecord,
+        samplingAttributes: Attributes
+    ) {
+        val mergedAttributes = Attributes.builder()
+            .putAll(logRecord.attributes)
+            .putAll(samplingAttributes)
+            .build()
+        logRecord.setAllAttributes(mergedAttributes)
+    }
+}