fix: Otlp Trace Exporters to Event Exporter

Sources/Observability/Client/ObservabilityClientFactory.swift

@@ -74,14 +74,14 @@ public struct ObservabilityClientFactory {
         }
 
         if options.traces == .enabled {
-            let tracesExporter = SamplingTraceExporterDecorator(
-                exporter: OtlpHttpTraceExporter(
-                    endpoint: url,
-                    config: .init(headers: options.customHeaders.map({ ($0.key, $0.value) }))
-                ),
-                sampler: sampler
+            let traceEventExporter = OtlpTraceEventExporter(
+                endpoint: url,
+                config: .init(headers: options.customHeaders.map({ ($0.key, $0.value) }))
             )
-            let decorator = TracerDecorator(options: options, sessionManager: sessionManager, exporter: tracesExporter)
+            Task {
+                await batchWorker.addExporter(traceEventExporter)
+            }
+            let decorator = TracerDecorator(options: options, sessionManager: sessionManager, sampler: sampler, eventQueue: eventQueue)
             /// tracer is enabled
             if options.autoInstrumentation.contains(.urlSession) {
                 autoInstrumentation.append(NetworkInstrumentationManager(options: options, tracer: decorator, session: sessionManager))

Sources/Observability/Metrics/OtlpMetricEventExporter.swift

@@ -3,33 +3,33 @@ import Common
 import Foundation
 import OpenTelemetryProtocolExporterCommon
 
-public final class OtlpMetricEventExporter: EventExporting {
-    let otlpHttpClient: OtlpHttpClient
+final class OtlpMetricEventExporter: EventExporting {
+    private let otlpHttpClient: OtlpHttpClient
 
-    public init(endpoint: URL,
-                config: OtlpConfiguration = OtlpConfiguration(),
-                useSession: URLSession? = nil,
-                envVarHeaders: [(String, String)]? = EnvVarHeaders.attributes) {
+    init(endpoint: URL,
+         config: OtlpConfiguration = OtlpConfiguration(),
+         useSession: URLSession? = nil,
+         envVarHeaders: [(String, String)]? = EnvVarHeaders.attributes) {
         self.otlpHttpClient = OtlpHttpClient(endpoint: endpoint,
                                              config: config,
                                              useSession: useSession,
                                              envVarHeaders: envVarHeaders)
     }
-
-    public func export(items: [EventQueueItem]) async throws {
+    
+    func export(items: [EventQueueItem]) async throws {
         let metricDatas: [OpenTelemetrySdk.MetricData] = items.compactMap { item in
             (item.payload as? MetricItem)?.metricData
         }
         guard metricDatas.isNotEmpty else {
             return
         }
+
         try await export(metricDatas: metricDatas)
     }
 
     private func export(metricDatas: [MetricData],
                         explicitTimeout: TimeInterval? = nil) async throws {
-        let body =
-        Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with { request in
+        let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with { request in
             request.resourceMetrics = MetricsAdapter.toProtoResourceMetrics(
                 metricData: metricDatas)
         }

Sources/Observability/Metrics/OtlpMetricScheduleExporter.swift

@@ -3,9 +3,9 @@ import OpenTelemetryApi
 import OpenTelemetrySdk
 
 final class OtlpMetricScheduleExporter: MetricExporter {
-    let eventQueue: EventQueue
-    let aggregationTemporalitySelector: AggregationTemporalitySelector
-    let defaultAggregationSelector: DefaultAggregationSelector
+    private let eventQueue: EventQueue
+    private let aggregationTemporalitySelector: AggregationTemporalitySelector
+    private let defaultAggregationSelector: DefaultAggregationSelector
 
     init(eventQueue: EventQueue,
          aggregationTemporalitySelector: AggregationTemporalitySelector = AggregationTemporality.alwaysCumulative(),
@@ -33,17 +33,17 @@ final class OtlpMetricScheduleExporter: MetricExporter {
     }
 
     public func getAggregationTemporality(
-      for instrument: OpenTelemetrySdk.InstrumentType
+        for instrument: OpenTelemetrySdk.InstrumentType
     ) -> OpenTelemetrySdk.AggregationTemporality {
-      return aggregationTemporalitySelector.getAggregationTemporality(
-        for: instrument)
+        return aggregationTemporalitySelector.getAggregationTemporality(
+            for: instrument)
     }
-
+    
     // MARK: - DefaultAggregationSelector
-
+    
     public func getDefaultAggregation(
-      for instrument: OpenTelemetrySdk.InstrumentType
+        for instrument: OpenTelemetrySdk.InstrumentType
     ) -> OpenTelemetrySdk.Aggregation {
-      return defaultAggregationSelector.getDefaultAggregation(for: instrument)
+        return defaultAggregationSelector.getDefaultAggregation(for: instrument)
     }
 }

Sources/Observability/Traces/EventSpanProcessor.swift

@@ -0,0 +1,45 @@
+import Foundation
+import OpenTelemetryApi
+import OpenTelemetrySdk
+
+final class EventSpanProcessor: SpanProcessor {
+    private let eventQueue: EventQueue
+    private let sampler: ExportSampler
+    let isStartRequired = false
+    let isEndRequired = true
+
+    init(eventQueue: EventQueue, sampler: ExportSampler) {
+        self.eventQueue = eventQueue
+        self.sampler = sampler
+    }
+
+    func onStart(parentContext: OpenTelemetryApi.SpanContext?, span: any OpenTelemetrySdk.ReadableSpan) {
+        // No-op
+    }
+
+    func onEnd(span: any OpenTelemetrySdk.ReadableSpan) {
+        if !span.context.traceFlags.sampled {
+          return
+        }
+
+        let spanData = span.toSpanData()
+        let sampledItems = sampler.sampleSpans(items: [spanData])
+        guard !sampledItems.isEmpty else {
+            return
+        }
+
+        Task {
+            await eventQueue.send(SpanItem(spanData: spanData))
+        }
+    }
+
+    func shutdown(explicitTimeout: TimeInterval?) {
+        // No-op
+    }
+
+    func forceFlush(timeout: TimeInterval?) {
+        // No-op
+    }
+
+
+}

Sources/Observability/Traces/OtlpTraceEventExporter.swift

@@ -0,0 +1,39 @@
+import OpenTelemetrySdk
+import Common
+import Foundation
+import OpenTelemetryProtocolExporterCommon
+
+final class OtlpTraceEventExporter: EventExporting {
+    private let otlpHttpClient: OtlpHttpClient
+
+    init(endpoint: URL,
+                config: OtlpConfiguration = OtlpConfiguration(),
+                useSession: URLSession? = nil,
+                envVarHeaders: [(String, String)]? = EnvVarHeaders.attributes) {
+        self.otlpHttpClient = OtlpHttpClient(endpoint: endpoint,
+                                             config: config,
+                                             useSession: useSession,
+                                             envVarHeaders: envVarHeaders)
+    }
+
+    func export(items: [EventQueueItem]) async throws {
+        let spanDatas: [OpenTelemetrySdk.SpanData] = items.compactMap { item in
+            (item.payload as? SpanItem)?.spanData
+        }
+        guard spanDatas.isNotEmpty else {
+            return
+        }
+        try await export(spanDatas: spanDatas)
+    }
+
+    private func export(spanDatas: [OpenTelemetrySdk.SpanData],
+                        explicitTimeout: TimeInterval? = nil) async throws {
+        let body =
+          Opentelemetry_Proto_Collector_Trace_V1_ExportTraceServiceRequest.with {
+            $0.resourceSpans = SpanAdapter.toProtoResourceSpans(
+              spanDataList: spanDatas)
+          }
+
+        try await otlpHttpClient.send(body: body, explicitTimeout: explicitTimeout)
+    }
+}

Sources/Observability/Traces/SpanItem.swift

@@ -0,0 +1,20 @@
+import Foundation
+import OpenTelemetrySdk
+
+struct SpanItem: EventQueueItemPayload {
+    var exporterClass: AnyClass {
+        Observability.OtlpTraceEventExporter.self
+    }
+
+    let spanData: SpanData
+    var timestamp: TimeInterval
+
+    init(spanData: SpanData) {
+        self.spanData = spanData
+        self.timestamp = spanData.endTime.timeIntervalSince1970
+    }
+
+    func cost() -> Int {
+        300 + spanData.events.count * 100 + spanData.attributes.count * 100
+    }
+}

Sources/Observability/Traces/TracerDecorator.swift

@@ -5,30 +5,18 @@ final class TracerDecorator: Tracer {
     private let options: Options
     private let sessionManager: SessionManaging
     private let tracerProvider: any TracerProvider
-    private let spanProcessor: any SpanProcessor
     private let tracer: any Tracer
-
-    init(options: Options, sessionManager: SessionManaging, exporter: SpanExporter) {
+    private var activeSpan: Span?
+
+    init(options: Options, sessionManager: SessionManaging, sampler: ExportSampler, eventQueue: EventQueue) {
         self.options = options
         self.sessionManager = sessionManager
-        /// Using the default values from OpenTelemetry for Swift
-        /// For reference check:
-        ///https://github.com/open-telemetry/opentelemetry-swift/blob/main/Sources/OpenTelemetrySdk/Trace/SpanProcessors/BatchSpanProcessor.swift
-        let processor = BatchSpanProcessor(
-            spanExporter: exporter,
-            scheduleDelay: 5,
-            exportTimeout: 30,
-            maxQueueSize: 2048,
-            maxExportBatchSize: 512,
-        )
-        self.spanProcessor = processor
+        let processor = EventSpanProcessor(eventQueue: eventQueue, sampler: sampler)
         let provider = TracerProviderBuilder()
             .add(spanProcessor: processor)
             .with(resource: Resource(attributes: options.resourceAttributes))
             .build()
         self.tracerProvider = provider
-
-
         self.tracer = tracerProvider.get(
             instrumentationName: options.serviceName,
             instrumentationVersion: options.serviceVersion,
@@ -84,20 +72,14 @@ extension TracerDecorator: TracesApi {
     }
 
     func flush() -> Bool {
-        /// span processor flush method differs from metrics and logs, it doesn't return a Result type
-        /// Processes all span events that have not yet been processed.
-        /// This method is executed synchronously on the calling thread
-        /// - Parameter timeout: Maximum time the flush complete or abort. If nil, it will wait indefinitely
-        /// func forceFlush(timeout: TimeInterval?)
-        spanProcessor.forceFlush(timeout: 5.0)
         return true
     }
 }
 
 /// Internal method used to set span start date
-extension TracerDecorator {
-    func startSpan(name: String, attributes: [String : AttributeValue], startTime: Date = Date()) -> any Span {
-        let builder = tracer.spanBuilder(spanName: name)
+extension Tracer {
+    func startSpan(name: String, attributes: [String : AttributeValue], startTime: Date) -> any Span {
+        let builder = spanBuilder(spanName: name)
 
         if let parent = OpenTelemetry.instance.contextProvider.activeSpan {
             builder.setParent(parent)

TestApp/Sources/AppDelegate.swift

@@ -29,7 +29,7 @@ let config = { () -> LDConfig in
 //            serviceName: "i-os-sessions",
 
             sessionBackgroundTimeout: 3,
-                      autoInstrumentation: [.memory])),
+            autoInstrumentation: [.memory, .urlSession])),
         SessionReplay(options: .init(
             isEnabled: true,
             privacy: .init(