recreation of existing pr

open-telemetry · Dec 19, 2023 · 79cc70e · 79cc70e
1 parent 0953181
commit 79cc70e
Show file tree

Hide file tree

Showing 5 changed files with 63 additions and 20 deletions.
diff --git a/[email protected] b/[email protected]
@@ -99,12 +99,14 @@ let package = Package(
                 path: "Sources/Exporters/OpenTelemetryProtocolCommon"),
         .target(name: "OpenTelemetryProtocolExporterHttp",
                 dependencies: ["OpenTelemetrySdk",
-                               "OpenTelemetryProtocolExporterCommon"],
+                               "OpenTelemetryProtocolExporterCommon",
+                               .product(name: "NIO", package: "swift-nio")],
                 path: "Sources/Exporters/OpenTelemetryProtocolHttp"),
         .target(name: "OpenTelemetryProtocolExporterGrpc",
                 dependencies: ["OpenTelemetrySdk",
                                "OpenTelemetryProtocolExporterCommon",
-                               .product(name: "GRPC", package: "grpc-swift")],
+                               .product(name: "GRPC", package: "grpc-swift"),
+                               .product(name: "NIO", package: "swift-nio")],
                 path: "Sources/Exporters/OpenTelemetryProtocolGrpc"),
         .target(name: "StdoutExporter",
                 dependencies: ["OpenTelemetrySdk"],

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 NIOConcurrencyHelpers
 
 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 = []
+    var 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,23 +6,29 @@
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
 import Foundation
+import NIOConcurrencyHelpers
+
 
 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)
-    let sendingMetrics = pendingMetrics
-    pendingMetrics = []
+    var sendingMetrics: [Metric] = []
+    exporterLock.withLockVoid {
+      pendingMetrics.append(contentsOf: metrics)
+      sendingMetrics = pendingMetrics
+      pendingMetrics = []
+    }
     let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with {
       $0.resourceMetrics = MetricsAdapter.toProtoResourceMetrics(metricDataList: sendingMetrics)
     }
@@ -33,7 +39,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,6 +6,7 @@
 import Foundation
 import OpenTelemetrySdk
 import OpenTelemetryProtocolExporterCommon
+import NIOConcurrencyHelpers
 
 public func defaultStableOtlpHTTPMetricsEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/metrics")!
@@ -16,6 +17,7 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
   var defaultAggregationSelector: DefaultAggregationSelector
 
   var pendingMetrics: [StableMetricData] = []
+  private let exporterLock = Lock()
 
   // MARK: - Init
 
@@ -31,9 +33,12 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
   // MARK: - StableMetricsExporter
 
   public func export(metrics : [StableMetricData]) -> ExportResult {
-    pendingMetrics.append(contentsOf: metrics)
-    let sendingMetrics = pendingMetrics
-    pendingMetrics = []
+    var sendingMetrics: [StableMetricData] = []
+    exporterLock.withLockVoid {
+      pendingMetrics.append(contentsOf: metrics)
+      sendingMetrics = pendingMetrics
+      pendingMetrics = []
+    }
     let body = Opentelemetry_Proto_Collector_Metrics_V1_ExportMetricsServiceRequest.with {
       $0.resourceMetrics = MetricsAdapter.toProtoResourceMetrics(stableMetricData: sendingMetrics)
     }
@@ -44,7 +49,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 +61,10 @@ public class StableOtlpHTTPMetricExporter: StableOtlpHTTPExporterBase, StableMet
 
   public func flush() -> ExportResult {
     var exporterResult: ExportResult = .success
-
+    var pendingMetrics: [StableMetricData] = []
+    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 NIOConcurrencyHelpers
 
 public func defaultOltpHttpTracesEndpoint() -> URL {
   URL(string: "http://localhost:4318/v1/traces")!
@@ -15,19 +16,23 @@ 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 {
-    pendingSpans.append(contentsOf: spans)
-    let sendingSpans = pendingSpans
-    pendingSpans = []
+    var sendingSpans: [SpanData] = []
+    exporterLock.withLockVoid {
+      pendingSpans.append(contentsOf: spans)
+      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 +50,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 +61,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)