recreation of existing pr

open-telemetry · Dec 19, 2023 · 7c7ef56 · 7c7ef56
1 parent 0953181
commit 7c7ef56
Show file tree

Hide file tree

Showing 4 changed files with 58 additions and 12 deletions.
diff --git a/Sources/Exporters/OpenTelemetryProtocolHttp/logs/OtlpHttpLogExporter.swift b/Sources/Exporters/OpenTelemetryProtocolHttp/logs/OtlpHttpLogExporter.swift
@@ -6,6 +6,7 @@
 import Foundation
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
+import NIO
 
 public func defaultOltpHttpLoggingEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/logs")!
@@ -14,7 +15,7 @@ public func defaultOltpHttpLoggingEndpoint() -> URL {
 public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
 
   var pendingLogRecords: [ReadableLogRecord] = []
-
+  private let exporterLock = Lock()
   override public init(endpoint: URL = defaultOltpHttpLoggingEndpoint(),
                        config: OtlpConfiguration = OtlpConfiguration(),
                        useSession: URLSession? = nil,
@@ -24,8 +25,13 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
 
   public func export(logRecords: [OpenTelemetrySdk.ReadableLogRecord], explicitTimeout: TimeInterval? = nil) -> OpenTelemetrySdk.ExportResult {
     pendingLogRecords.append(contentsOf: logRecords)
-    let sendingLogRecords = pendingLogRecords
-    pendingLogRecords = []
+    let sendingLogRecords: [ReadableLogRecord] = []
+
+    exporterLock.withLockVoid {
+      pendingLogRecords.append(contentsOf: logRecords)
+      sendingLogRecords = pendingLogRecords
+      pendingLogRecords = []
+    }
 
     let body = Opentelemetry_Proto_Collector_Logs_V1_ExportLogsServiceRequest.with { request in
       request.resourceLogs = LogRecordAdapter.toProtoResourceRecordLog(logRecordList: sendingLogRecords)
@@ -38,7 +44,9 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
       case .success(_):
         break
       case .failure(let error):
-        self?.pendingLogRecords.append(contentsOf: sendingLogRecords)
+        self?.exporterLock.withLockVoid {
+          self?.pendingLogRecords.append(contentsOf: sendingLogRecords)
+        }
         print(error)
       }
     }
@@ -52,6 +60,10 @@ public class OtlpHttpLogExporter : OtlpHttpExporterBase, LogRecordExporter {
 
   public func flush(explicitTimeout: TimeInterval? = nil) -> ExportResult {
     var exporterResult: ExportResult = .success
+    var pendingLogRecords: [ReadableLogRecord] = []
+    exporterLock.withLockVoid {
+      pendingLogRecords = self.pendingLogRecords
+    }
 
     if !pendingLogRecords.isEmpty {
       let body = Opentelemetry_Proto_Collector_Logs_V1_ExportLogsServiceRequest.with { request in

diff --git a/Sources/Exporters/OpenTelemetryProtocolHttp/metric/OltpHTTPMetricExporter.swift b/Sources/Exporters/OpenTelemetryProtocolHttp/metric/OltpHTTPMetricExporter.swift
@@ -6,21 +6,28 @@
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
 import Foundation
+import NIO
 
 public func defaultOltpHTTPMetricsEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/metrics")!
 }
 
 public class OtlpHttpMetricExporter: OtlpHttpExporterBase, MetricExporter {
   var pendingMetrics: [Metric] = []
+  private let exporterLock = Lock()
 
   override
   public init(endpoint: URL = defaultOltpHTTPMetricsEndpoint(), config : OtlpConfiguration = OtlpConfiguration(), useSession: URLSession? = nil, envVarHeaders: [(String,String)]? = EnvVarHeaders.attributes) {
     super.init(endpoint: endpoint, config: config, useSession: useSession, envVarHeaders: envVarHeaders)
   }
 
   public func export(metrics: [Metric], shouldCancel: (() -> Bool)?) -> MetricExporterResultCode {
-    pendingMetrics.append(contentsOf: metrics)
+    var sendingMetrics: [Metric] = []
+    exporterLock.withLockVoid {
+      pendingMetrics.append(contentsOf: metrics)
+      sendingMetrics = pendingMetrics
+      pendingMetrics = []
+    }
     let sendingMetrics = pendingMetrics
     pendingMetrics = []
     let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with {
@@ -33,7 +40,9 @@ public class OtlpHttpMetricExporter: OtlpHttpExporterBase, MetricExporter {
       case .success(_):
         break
       case .failure(let error):
-        self?.pendingMetrics.append(contentsOf: sendingMetrics)
+        self?.exporterLock.withLockVoid {
+          self?.pendingMetrics.append(contentsOf: sendingMetrics)
+        }
         print(error)
       }
     }

diff --git a/Sources/Exporters/OpenTelemetryProtocolHttp/metric/StableOtlpHTTPMetricExporter.swift b/Sources/Exporters/OpenTelemetryProtocolHttp/metric/StableOtlpHTTPMetricExporter.swift
@@ -6,7 +6,7 @@
 import Foundation
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
-
+import NIO
 public func defaultStableOtlpHTTPMetricsEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/metrics")!
 }
@@ -16,6 +16,7 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
   var defaultAggregationSelector: DefaultAggregationSelector
 
   var pendingMetrics: [StableMetricData] = []
+  private let exporterLock = Lock()
 
   // MARK: - Init
 
@@ -31,7 +32,12 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
   // MARK: - StableMetricsExporter
 
   public func export(metrics : [StableMetricData]) -> ExportResult {
-    pendingMetrics.append(contentsOf: metrics)
+    var sendingMetrics: [StableMetricData] = []
+    exporterLock.withLockVoid {
+      pendingMetrics.append(contentsOf: metrics)
+      sendingMetrics = pendingMetrics
+      pendingMetrics = []
+    }
     let sendingMetrics = pendingMetrics
     pendingMetrics = []
     let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with {
@@ -44,7 +50,9 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
       case .success(_):
         break
       case .failure(let error):
-        self?.pendingMetrics.append(contentsOf: sendingMetrics)
+        self?.exporterLock.withLockVoid {
+          self?.pendingMetrics.append(contentsOf: sendingMetrics)
+        }
         print(error)
       }
     }
@@ -54,7 +62,10 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
 
   public func flush() -> ExportResult {
     var exporterResult: ExportResult = .success
-
+    var pendingMetrics: [Metric] = []
+    exporterLock.withLockVoid {
+      pendingMetrics = self.pendingMetrics
+    }
     if !pendingMetrics.isEmpty {
       let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with {
         $0.resourceMetrics = MetricsAdapter.toProtoResourceMetrics(stableMetricData: pendingMetrics)

diff --git a/Sources/Exporters/OpenTelemetryProtocolHttp/trace/OtlpHttpTraceExporter.swift b/Sources/Exporters/OpenTelemetryProtocolHttp/trace/OtlpHttpTraceExporter.swift
@@ -6,6 +6,7 @@
 import Foundation
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
+import NIO
 
 public func defaultOltpHttpTracesEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/traces")!
@@ -15,19 +16,26 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
 
 
   var pendingSpans: [SpanData] = []
+  private let exporterLock = Lock()
   override
   public init(endpoint: URL = defaultOltpHttpTracesEndpoint(), config: OtlpConfiguration = OtlpConfiguration(),
               useSession: URLSession? = nil,  envVarHeaders: [(String,String)]? = EnvVarHeaders.attributes) {
     super.init(endpoint: endpoint, config: config, useSession: useSession)
   }
 
   public func export(spans: [SpanData], explicitTimeout: TimeInterval? = nil) -> SpanExporterResultCode {
+    var sendingSpans: [SpandData] = []
+    exporterLock.withLockVoid {
+      pendingSpans.append(contentsOf: spans)
+      sendingSpans = pendingSpans
+      pendingSpans = []
+    }
     pendingSpans.append(contentsOf: spans)
     let sendingSpans = pendingSpans
     pendingSpans = []
 
     let body = Opentelemetry_Proto_Collector_Trace_V1_ExportTraceServiceRequest.with {
-      $0.resourceSpans = SpanAdapter.toProtoResourceSpans(spanDataList: spans)
+      $0.resourceSpans = SpanAdapter.toProtoResourceSpans(spanDataList: sendingSpans)
     }
     var request = createRequest(body: body, endpoint: endpoint)
     if let headers = envVarHeaders {
@@ -45,7 +53,9 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
       case .success:
         break
       case .failure(let error):
-        self?.pendingSpans.append(contentsOf: sendingSpans)
+        self?.exporterLock.withLockVoid {
+          self?.pendingSpans.append(contentsOf: sendingSpans)
+        }
         print(error)
       }
     }
@@ -54,6 +64,10 @@ public class OtlpHttpTraceExporter: OtlpHttpExporterBase, SpanExporter {
 
   public func flush(explicitTimeout: TimeInterval? = nil) -> SpanExporterResultCode {
     var resultValue: SpanExporterResultCode = .success
+    var pendingSpans: [SpanData] = []
+    exporterLock.withLockVoid {
+      pendingSpans = self.pendingSpans
+    }
     if !pendingSpans.isEmpty {
       let body = Opentelemetry_Proto_Collector_Trace_V1_ExportTraceServiceRequest.with {
         $0.resourceSpans = SpanAdapter.toProtoResourceSpans(spanDataList: pendingSpans)