Merge branch 'main' into network-instrumentation

Author: Ignacio Bonafonte

Sources/OpenTelemetrySdk/Metrics/Configuration/MeterSharedState.swift

@@ -19,12 +19,48 @@ public enum MetricExporterResultCode {
     case success
     case failureNotRetryable
     case failureRetryable
+
+    /// Merges the current result code with other result code
+    /// - Parameter newResultCode: the result code to merge with
+    mutating func mergeResultCode(newResultCode: MetricExporterResultCode) {
+        // If both results are success then return success.
+        if self == .success, newResultCode == .success {
+            self = .success
+            return
+        } else if self == .failureRetryable || self == .success,
+                  newResultCode == .failureRetryable || newResultCode == .success
+        {
+            self = .failureRetryable
+        }
+        self = .failureNotRetryable
+    }
 }
 
 public protocol MetricExporter {
     func export(metrics: [Metric], shouldCancel: (() -> Bool)?) -> MetricExporterResultCode
 }
 
+/// Implementation of the SpanExporter that simply forwards all received spans to a list of
+/// SpanExporter.
+/// Can be used to export to multiple backends using the same SpanProcessor} like a impleSampledSpansProcessor
+///  or a BatchSampledSpansProcessor.
+struct MultiMetricExporter: MetricExporter {
+
+    var metricExporters: [MetricExporter]
+
+    init(metricExporters: [MetricExporter]) {
+        self.metricExporters = metricExporters
+    }
+
+    func export(metrics: [Metric], shouldCancel: (() -> Bool)?) -> MetricExporterResultCode {
+        var currentResultCode = MetricExporterResultCode.success
+        metricExporters.forEach {
+            currentResultCode.mergeResultCode(newResultCode: $0.export(metrics: metrics, shouldCancel: shouldCancel))
+        }
+        return currentResultCode
+    }
+}
+
 struct NoopMetricExporter: MetricExporter {
     func export(metrics: [Metric], shouldCancel: (() -> Bool)?) -> MetricExporterResultCode {
         return .success

Sources/OpenTelemetrySdk/Metrics/Export/MetricExporter.swift

@@ -13,18 +13,17 @@
 // limitations under the License.
 //
 
-
 import Foundation
 
-public class MetricProcessorSdk : MetricProcessor {
-    private let lock : Lock
-    var metrics : [Metric]
-    
+public class MetricProcessorSdk: MetricProcessor {
+    private let lock: Lock
+    var metrics: [Metric]
+
     public init() {
         metrics = [Metric]()
         lock = Lock()
     }
-    
+
     /// Finish the current collection cycle and return the metrics it holds.
     /// This is called at the end of one collection cycle by the Controller.
     /// MetricProcessor can use this to clear its Metrics (in case of stateless).
@@ -36,7 +35,7 @@ public class MetricProcessorSdk : MetricProcessor {
         }
         return metrics
     }
-    
+
     /// Process the metric. This method is called once every collection interval.
     /// - Parameters:
     ///   - metric: the metric record.
@@ -47,8 +46,5 @@ public class MetricProcessorSdk : MetricProcessor {
         }
 
         metrics.append(metric)
-        
     }
-
-    
 }

Sources/OpenTelemetrySdk/Metrics/Export/MetricProcessorSdk.swift

@@ -15,35 +15,31 @@
 import Foundation
 
 class PushMetricController {
-    public private(set) var pushInterval: TimeInterval
-    var metricExporter: MetricExporter
-    var metricProcessor: MetricProcessor
+    var meterSharedState: MeterSharedState
     var meterProvider: MeterProviderSdk
 
     let pushMetricQueue = DispatchQueue(label: "org.opentelemetry.PushMetricController.pushMetricQueue")
 
-    init(meterProvider: MeterProviderSdk, metricProcessor: MetricProcessor, metricExporter: MetricExporter, pushInterval: TimeInterval, shouldCancel: (() -> Bool)? = nil) {
+    init(meterProvider: MeterProviderSdk, meterSharedState: MeterSharedState, shouldCancel: (() -> Bool)? = nil) {
         self.meterProvider = meterProvider
-        self.metricProcessor = metricProcessor
-        self.metricExporter = metricExporter
-        self.pushInterval = pushInterval
-        pushMetricQueue.asyncAfter(deadline: .now() + pushInterval) { [weak self] in
+        self.meterSharedState = meterSharedState
+        pushMetricQueue.asyncAfter(deadline: .now() + meterSharedState.metricPushInterval) { [weak self] in
             guard let self = self else {
                 return
             }
             while !(shouldCancel?() ?? false) {
                 autoreleasepool {
                     let start = Date()
                     let values = self.meterProvider.getMeters().values
-                    for index in values.indices {
-                        values[index].collect()
+                    values.forEach {
+                        $0.collect()
                     }
 
-                    let metricToExport = self.metricProcessor.finishCollectionCycle()
+                    let metricToExport = self.meterSharedState.metricProcessor.finishCollectionCycle()
 
-                    _ = metricExporter.export(metrics: metricToExport, shouldCancel: shouldCancel)
+                    _ = meterSharedState.metricExporter.export(metrics: metricToExport, shouldCancel: shouldCancel)
                     let timeInterval = Date().timeIntervalSince(start)
-                    let remainingWait = pushInterval - timeInterval
+                    let remainingWait = meterSharedState.metricPushInterval - timeInterval
                     if remainingWait > 0 {
                         usleep(UInt32(remainingWait * 1000000))
                     }

Sources/OpenTelemetrySdk/Metrics/Export/PushMetricController.swift

@@ -18,38 +18,35 @@ import OpenTelemetryApi
 
 public class MeterProviderSdk: MeterProvider {
     private let lock = Lock()
-    static public let defaultPushInterval: TimeInterval = 60
+    public static let defaultPushInterval: TimeInterval = 60
 
     var meterRegistry = [InstrumentationLibraryInfo: MeterSdk]()
 
-    var meterSharedState : MeterSharedState
+    var meterSharedState: MeterSharedState
     var pushMetricController: PushMetricController!
     var defaultMeter: MeterSdk
 
     public convenience init() {
         self.init(metricProcessor: NoopMetricProcessor(),
                   metricExporter: NoopMetricExporter())
     }
-    
+
     public init(metricProcessor: MetricProcessor,
                 metricExporter: MetricExporter,
                 metricPushInterval: TimeInterval = MeterProviderSdk.defaultPushInterval,
-                resource: Resource = EnvVarResource.resource) {
-        self.meterSharedState = MeterSharedState(metricProcessor:metricProcessor, metricPushInterval: metricPushInterval, resource: resource)
+                resource: Resource = EnvVarResource.resource)
+    {
+        self.meterSharedState = MeterSharedState(metricProcessor: metricProcessor, metricPushInterval: metricPushInterval, metricExporter: metricExporter, resource: resource)
 
         defaultMeter = MeterSdk(meterSharedState: self.meterSharedState, instrumentationLibraryInfo: InstrumentationLibraryInfo())
 
         pushMetricController = PushMetricController(
             meterProvider: self,
-            metricProcessor: metricProcessor,
-            metricExporter: metricExporter,
-            pushInterval: meterSharedState.metricPushInterval) {
-            false
+            meterSharedState: self.meterSharedState) {
+                false
         }
     }
 
-
-
     public func get(instrumentationName: String, instrumentationVersion: String? = nil) -> Meter {
         if instrumentationName.isEmpty {
             return defaultMeter
@@ -59,7 +56,7 @@ public class MeterProviderSdk: MeterProvider {
         defer {
             lock.unlock()
         }
-        let instrumentationLibraryInfo  = InstrumentationLibraryInfo(name: instrumentationName, version: instrumentationVersion)
+        let instrumentationLibraryInfo = InstrumentationLibraryInfo(name: instrumentationName, version: instrumentationVersion)
         var meter: MeterSdk! = meterRegistry[instrumentationLibraryInfo]
         if meter == nil {
             meter = MeterSdk(meterSharedState: self.meterSharedState, instrumentationLibraryInfo: instrumentationLibraryInfo)
@@ -76,6 +73,30 @@ public class MeterProviderSdk: MeterProvider {
         return meterRegistry
     }
 
+    public func setMetricProcessor(_ metricProcessor: MetricProcessor) {
+        pushMetricController.pushMetricQueue.sync {
+            meterSharedState.metricProcessor = metricProcessor
+        }
+    }
+
+    public func addMetricExporter(_ metricExporter: MetricExporter) {
+        pushMetricController.pushMetricQueue.sync {
+            meterSharedState.addMetricExporter(metricExporter: metricExporter)
+        }
+    }
+
+    public func setMetricPushInterval(_ interval: TimeInterval) {
+        pushMetricController.pushMetricQueue.sync {
+            meterSharedState.metricPushInterval = interval
+        }
+    }
+
+    public func setResource(_ resource: Resource) {
+        pushMetricController.pushMetricQueue.sync {
+            meterSharedState.resource = resource
+        }
+    }
+
     private static func createLibraryResourceLabels(name: String, version: String) -> [String: String] {
         var labels = ["name": name]
         if !version.isEmpty {

Sources/OpenTelemetrySdk/Metrics/Configuration/MeterProviderSdk.swift renamed to Sources/OpenTelemetrySdk/Metrics/MeterProviderSdk.swift

@@ -0,0 +1,44 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import Foundation
+
+class MeterSharedState {
+    /// Configures metric processor. (aka batcher).
+    var metricProcessor: MetricProcessor
+    /// Sets the push interval.
+    var metricPushInterval: TimeInterval
+    /// Sets the exporter
+    var metricExporter: MetricExporter
+
+    var resource: Resource
+
+    init(metricProcessor: MetricProcessor, metricPushInterval: TimeInterval, metricExporter: MetricExporter, resource: Resource) {
+        self.metricProcessor = metricProcessor
+        self.metricPushInterval = metricPushInterval
+        self.metricExporter = metricExporter
+        self.resource = resource
+    }
+
+    func addMetricExporter(metricExporter: MetricExporter) {
+        if metricExporter is NoopMetricExporter {
+            self.metricExporter = metricExporter
+        } else if var multiMetricExporter = metricExporter as? MultiMetricExporter {
+            multiMetricExporter.metricExporters.append(metricExporter)
+        } else {
+            self.metricExporter = MultiMetricExporter(metricExporters: [self.metricExporter, metricExporter])
+        }
+    }
+}

Sources/OpenTelemetrySdk/Metrics/MeterSharedState.swift

@@ -43,7 +43,7 @@ public enum SpanExporterResultCode {
 
     /// Merges the current result code with other result code
     /// - Parameter newResultCode: the result code to merge with
-    public mutating func mergeResultCode(newResultCode: SpanExporterResultCode) {
+    mutating func mergeResultCode(newResultCode: SpanExporterResultCode) {
         // If both results are success then return success.
         if self == .success && newResultCode == .success {
             self = .success

Sources/OpenTelemetrySdk/Trace/Export/SpanExporter.swift

@@ -38,7 +38,7 @@ final class PushControllerTests: XCTestCase {
         // Setup 2 meters whose Collect will increment the collect count.
         var meter1CollectCount = 0
         var meter2CollectCount = 0
-        let meterSharedState = MeterSharedState(metricProcessor: testProcessor, metricPushInterval: MeterProviderSdk.defaultPushInterval, resource: Resource())
+        let meterSharedState = MeterSharedState(metricProcessor: testProcessor, metricPushInterval: controllerPushIntervalInSec, metricExporter: testExporter, resource: Resource())
 
         let meterInstrumentationLibrary1 = InstrumentationLibraryInfo(name:"meter1")
 
@@ -60,7 +60,7 @@ final class PushControllerTests: XCTestCase {
 
         let pushInterval = controllerPushIntervalInSec
 
-        let controller = PushMetricController(meterProvider: meterProvider, metricProcessor: testProcessor, metricExporter: testExporter, pushInterval: pushInterval)
+        let controller = PushMetricController(meterProvider: meterProvider, meterSharedState: meterSharedState)
 
         // Validate that collect is called on Meter1, Meter2.
         validateMeterCollect(meterCollectCount: &meter1CollectCount, expectedMeterCollectCount: collectionCountExpectedMin, meterName: "meter1", timeout: maxWaitInSec)
@@ -70,7 +70,7 @@ final class PushControllerTests: XCTestCase {
         lock.withLockVoid {
             XCTAssertTrue(exportCalledCount >= collectionCountExpectedMin)
         }
-        XCTAssertEqual(controller.pushInterval, pushInterval)
+        XCTAssertEqual(controller.meterSharedState.metricPushInterval, pushInterval)
     }
 
     func validateMeterCollect(meterCollectCount: inout Int, expectedMeterCollectCount: Int, meterName: String, timeout: TimeInterval) {