Merge pull request #199 from bryce-b/bryce/metric-processor-improvements

This PR fixes a retention loop between MetricProviderSdk & PushMetricController, and also fixes the resulting crash by using a DispatchSourceTimer to clean up the dispatched thread when the PushMetricController is released.

fixes #198

Author: Ignacio Bonafonte

Sources/OpenTelemetrySdk/Metrics/Export/PushMetricController.swift

@@ -6,35 +6,46 @@ import Foundation
 
 class PushMetricController {
     var meterSharedState: MeterSharedState
-    var meterProvider: MeterProviderSdk
+    weak var meterProvider: MeterProviderSdk?
 
     let pushMetricQueue = DispatchQueue(label: "org.opentelemetry.PushMetricController.pushMetricQueue")
+    let metricPushTimer: DispatchSourceTimer
 
     init(meterProvider: MeterProviderSdk, meterSharedState: MeterSharedState, shouldCancel: (() -> Bool)? = nil) {
         self.meterProvider = meterProvider
         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
-                    values.forEach {
-                        $0.collect()
-                    }
+        metricPushTimer = DispatchSource.makeTimerSource(flags: DispatchSource.TimerFlags(), queue: pushMetricQueue)
+        metricPushTimer.setEventHandler { [weak self] in
+            autoreleasepool {
+                guard let self = self,
+                      let meterProvider = self.meterProvider else {
+                    return
+                }
+                if shouldCancel?() ?? false {
+                    self.metricPushTimer.cancel()
+                    return
+                }
+                let start = Date()
+                let values = meterProvider.getMeters().values
+                values.forEach {
+                    $0.collect()
+                }
 
-                    let metricToExport = self.meterSharedState.metricProcessor.finishCollectionCycle()
+                let metricToExport = self.meterSharedState.metricProcessor.finishCollectionCycle()
 
-                    _ = meterSharedState.metricExporter.export(metrics: metricToExport, shouldCancel: shouldCancel)
-                    let timeInterval = Date().timeIntervalSince(start)
-                    let remainingWait = meterSharedState.metricPushInterval - timeInterval
-                    if remainingWait > 0 {
-                        usleep(UInt32(remainingWait * 1000000))
-                    }
-                }
+                _ = meterSharedState.metricExporter.export(metrics: metricToExport, shouldCancel: shouldCancel)
             }
         }
+
+        metricPushTimer.schedule(deadline: .now() + meterSharedState.metricPushInterval, repeating: meterSharedState.metricPushInterval)
+        metricPushTimer.activate()
+    }
+
+    deinit {
+        metricPushTimer.suspend() // suspending the timer prior to checking `isCancelled()` prevents a race condition between the check and actually calling `cancel()`
+        if !metricPushTimer.isCancelled {
+            metricPushTimer.cancel()
+            metricPushTimer.resume()
+        }
     }
 }

Sources/OpenTelemetrySdk/Metrics/MeterProviderSdk.swift

@@ -26,14 +26,15 @@ public class MeterProviderSdk: MeterProvider {
                 metricPushInterval: TimeInterval = MeterProviderSdk.defaultPushInterval,
                 resource: Resource = EnvVarResource.resource)
     {
-        self.meterSharedState = MeterSharedState(metricProcessor: metricProcessor, metricPushInterval: metricPushInterval, metricExporter: metricExporter, resource: resource)
+        meterSharedState = MeterSharedState(metricProcessor: metricProcessor, metricPushInterval: metricPushInterval, metricExporter: metricExporter, resource: resource)
 
-        defaultMeter = MeterSdk(meterSharedState: self.meterSharedState, instrumentationLibraryInfo: InstrumentationLibraryInfo())
+        defaultMeter = MeterSdk(meterSharedState: meterSharedState, instrumentationLibraryInfo: InstrumentationLibraryInfo())
 
         pushMetricController = PushMetricController(
             meterProvider: self,
-            meterSharedState: self.meterSharedState) {
-                false
+            meterSharedState: meterSharedState
+        ) {
+            false
         }
     }
 
@@ -49,7 +50,7 @@ public class MeterProviderSdk: MeterProvider {
         let instrumentationLibraryInfo = InstrumentationLibraryInfo(name: instrumentationName, version: instrumentationVersion)
         var meter: MeterSdk! = meterRegistry[instrumentationLibraryInfo]
         if meter == nil {
-            meter = MeterSdk(meterSharedState: self.meterSharedState, instrumentationLibraryInfo: instrumentationLibraryInfo)
+            meter = MeterSdk(meterSharedState: meterSharedState, instrumentationLibraryInfo: instrumentationLibraryInfo)
             meterRegistry[instrumentationLibraryInfo] = meter!
         }
         return meter!

Tests/ExportersTests/DatadogExporter/DatadogExporterTests.swift

@@ -120,10 +120,11 @@ class DatadogExporterTests: XCTestCase {
 
         let datadogExporter = try! DatadogExporter(config: exporterConfiguration)
 
+        let provider = MeterProviderSdk(metricProcessor: UngroupedBatcher(),
+                              metricExporter: datadogExporter,
+                              metricPushInterval: 0.1)
 
-        let meter = MeterProviderSdk(metricProcessor: UngroupedBatcher(),
-                                     metricExporter: datadogExporter,
-                                     metricPushInterval: 0.1).get(instrumentationName: "MyMeter")
+        let meter = provider.get(instrumentationName: "MyMeter")
 
         let testCounter = meter.createIntCounter(name: "MyCounter")
 

Tests/ExportersTests/Prometheus/PrometheusExporterTests.swift

@@ -12,6 +12,7 @@ class PrometheusExporterTests: XCTestCase {
     let metricPushIntervalSec = 0.05
     let waitDuration = 0.1 + 0.1
 
+    
     func testMetricsHttpServerAsync() {
         let promOptions = PrometheusExporterOptions(url: "http://localhost:9184/metrics/")
         let promExporter = PrometheusExporter(options: promOptions)
@@ -29,7 +30,7 @@ class PrometheusExporterTests: XCTestCase {
             }
         }
 
-        collectMetrics(simpleProcessor: simpleProcessor, exporter: promExporter)
+        let retain_me = collectMetrics(simpleProcessor: simpleProcessor, exporter: promExporter)
         usleep(useconds_t(waitDuration * 1000000))
         let url = URL(string: "http://localhost:9184/metrics/")!
         let task = URLSession.shared.dataTask(with: url) { data, response, error in
@@ -59,10 +60,10 @@ class PrometheusExporterTests: XCTestCase {
         metricsHttpServer.stop()
     }
 
-    private func collectMetrics(simpleProcessor: UngroupedBatcher, exporter: MetricExporter) {
+    private func collectMetrics(simpleProcessor: UngroupedBatcher, exporter: MetricExporter) -> MeterProviderSdk {
 
         let meterProvider = MeterProviderSdk(metricProcessor: simpleProcessor, metricExporter: exporter, metricPushInterval: metricPushIntervalSec)
-
+        
         let meter = meterProvider.get(instrumentationName: "library1")
 
         let testCounter = meter.createIntCounter(name: "testCounter")
@@ -81,6 +82,7 @@ class PrometheusExporterTests: XCTestCase {
             testMeasure.record(value: 5, labelset: meter.getLabelSet(labels: labels1))
             testMeasure.record(value: 500, labelset: meter.getLabelSet(labels: labels1))
         }
+        return meterProvider
     }
 
     private func validateResponse(responseText: String) {