Merge pull request #273 from nachoBonafonte/Implement-ForceFlush-spec

Implement ForceFlush spec adding a timeout parameter.
This parameter takes effect only in BatchSpanProcessor built-in processors

Author: Ignacio Bonafonte

Sources/Instrumentation/SignPostIntegration/SignPostIntegration.swift

@@ -28,5 +28,5 @@ public class SignPostIntegration: SpanProcessor {
     }
 
     public func shutdown() {}
-    public func forceFlush() {}
+    public func forceFlush(timeout: TimeInterval? = nil) {}
 }

Sources/OpenTelemetrySdk/Trace/SpanProcessor.swift

@@ -18,6 +18,7 @@ public protocol SpanProcessor {
     /// Called when a Span is started, if the Span.isRecording is true.
     /// This method is called synchronously on the execution thread, should not throw or block the
     /// execution thread.
+    /// - Parameter parentContext: the context of the span parent, if exists
     /// - Parameter span: the ReadableSpan that just started
     func onStart(parentContext: SpanContext?, span: ReadableSpan)
 
@@ -33,5 +34,13 @@ public protocol SpanProcessor {
 
     /// Processes all span events that have not yet been processed.
     /// This method is executed synchronously on the calling thread
-    func forceFlush()
+    /// - Parameter timeout: Maximum time the flush complete or abort. If nil, it will wait indefinitely
+    func forceFlush(timeout: TimeInterval?)
 }
+
+extension SpanProcessor {
+    func forceFlush() {
+        return forceFlush(timeout: nil)
+    }
+}
+

Sources/OpenTelemetrySdk/Trace/SpanProcessors/BatchSpanProcessor.swift

@@ -18,7 +18,7 @@ public struct BatchSpanProcessor: SpanProcessor {
     fileprivate var worker: BatchWorker
 
     public init(spanExporter: SpanExporter, scheduleDelay: TimeInterval = 5, exportTimeout: TimeInterval = 30,
-                maxQueueSize: Int = 2048, maxExportBatchSize: Int = 512, willExportCallback:((inout [SpanData]) -> Void)? = nil)
+                maxQueueSize: Int = 2048, maxExportBatchSize: Int = 512, willExportCallback: ((inout [SpanData]) -> Void)? = nil)
     {
         worker = BatchWorker(spanExporter: spanExporter,
                              scheduleDelay: scheduleDelay,
@@ -46,8 +46,8 @@ public struct BatchSpanProcessor: SpanProcessor {
         worker.shutdown()
     }
 
-    public func forceFlush() {
-        worker.forceFlush()
+    public func forceFlush(timeout: TimeInterval? = nil) {
+        worker.forceFlush(explicitTimeout: timeout)
     }
 }
 
@@ -60,13 +60,13 @@ private class BatchWorker: Thread {
     let maxQueueSize: Int
     let exportTimeout: TimeInterval
     let maxExportBatchSize: Int
-    let willExportCallback:((inout [SpanData]) -> Void)?
+    let willExportCallback: ((inout [SpanData]) -> Void)?
     let halfMaxQueueSize: Int
     private let cond = NSCondition()
     var spanList = [ReadableSpan]()
     var queue: OperationQueue
 
-    init(spanExporter: SpanExporter, scheduleDelay: TimeInterval, exportTimeout:TimeInterval, maxQueueSize: Int, maxExportBatchSize: Int, willExportCallback:((inout [SpanData]) -> Void)?) {
+    init(spanExporter: SpanExporter, scheduleDelay: TimeInterval, exportTimeout: TimeInterval, maxQueueSize: Int, maxExportBatchSize: Int, willExportCallback: ((inout [SpanData]) -> Void)?) {
         self.spanExporter = spanExporter
         self.scheduleDelay = scheduleDelay
         self.exportTimeout = exportTimeout
@@ -109,33 +109,34 @@ private class BatchWorker: Thread {
                 spansCopy = spanList
                 spanList.removeAll()
                 cond.unlock()
-                self.exportBatch(spanList: spansCopy)
+                self.exportBatch(spanList: spansCopy, explicitTimeout: nil)
             }
         } while true
     }
 
     func shutdown() {
-        forceFlush()
+        forceFlush(explicitTimeout: nil)
         spanExporter.shutdown()
     }
 
-    public func forceFlush() {
+    public func forceFlush(explicitTimeout: TimeInterval?) {
         var spansCopy: [ReadableSpan]
         cond.lock()
         spansCopy = spanList
         spanList.removeAll()
         cond.unlock()
         // Execute the batch export outside the synchronized to not block all producers.
-        exportBatch(spanList: spansCopy)
+        exportBatch(spanList: spansCopy, explicitTimeout: explicitTimeout)
     }
 
-    private func exportBatch(spanList: [ReadableSpan]) {
+    private func exportBatch(spanList: [ReadableSpan], explicitTimeout: TimeInterval?) {
         let exportOperation = BlockOperation { [weak self] in
             self?.exportAction(spanList: spanList)
         }
         let timeoutTimer = DispatchSource.makeTimerSource(queue: DispatchQueue.global())
-        timeoutTimer.setEventHandler{ exportOperation.cancel() }
-        timeoutTimer.schedule(deadline: .now() + .milliseconds(Int(exportTimeout.toMilliseconds)), leeway: .milliseconds(1))
+        timeoutTimer.setEventHandler { exportOperation.cancel() }
+        let maxTimeOut = min(explicitTimeout ?? TimeInterval.greatestFiniteMagnitude, exportTimeout)
+        timeoutTimer.schedule(deadline: .now() + .milliseconds(Int(maxTimeOut.toMilliseconds)), leeway: .milliseconds(1))
         timeoutTimer.activate()
         queue.addOperation(exportOperation)
         queue.waitUntilAllOperationsAreFinished()

Sources/OpenTelemetrySdk/Trace/SpanProcessors/MultiSpanProcessor.swift

@@ -51,9 +51,9 @@ public struct MultiSpanProcessor: SpanProcessor {
         }
     }
 
-    public func forceFlush() {
+    public func forceFlush(timeout: TimeInterval? = nil) {
         spanProcessorsAll.forEach {
-            $0.forceFlush()
+            $0.forceFlush(timeout: timeout)
         }
     }
 }

Sources/OpenTelemetrySdk/Trace/SpanProcessors/NoopSpanProcessor.swift

@@ -18,5 +18,5 @@ struct NoopSpanProcessor: SpanProcessor {
 
     func shutdown() {}
 
-    func forceFlush() {}
+    func forceFlush(timeout: TimeInterval? = nil) {}
 }

Sources/OpenTelemetrySdk/Trace/SpanProcessors/SimpleSpanProcessor.swift

@@ -6,8 +6,9 @@
 import Foundation
 import OpenTelemetryApi
 
-/// An implementation of the SpanProcessor that converts the ReadableSpan SpanData
-///  and passes it to the configured exporter.
+/// A really simple implementation of the SpanProcessor that converts the ReadableSpan SpanData
+/// and passes it to the configured exporter.
+/// For production environment BatchSpanProcessor is configurable and is preferred.
 public struct SimpleSpanProcessor: SpanProcessor {
     private let spanExporter: SpanExporter
     private var sampled: Bool = true
@@ -35,7 +36,9 @@ public struct SimpleSpanProcessor: SpanProcessor {
         }
     }
 
-    public func forceFlush() {
+    /// Forces the processing of the remaining spans
+    /// - Parameter timeout: unused in this processor
+    public func forceFlush(timeout: TimeInterval? = nil) {
         processorQueue.sync {
             _ = spanExporter.flush()
         }

Sources/OpenTelemetrySdk/Trace/TracerProviderSdk.swift

@@ -140,7 +140,7 @@ public class TracerProviderSdk: TracerProvider {
     }
 
     /// Requests the active span processor to process all span events that have not yet been processed.
-    public func forceFlush() {
-        sharedState.activeSpanProcessor.forceFlush()
+    public func forceFlush(timeout: TimeInterval? = nil) {
+        sharedState.activeSpanProcessor.forceFlush(timeout: timeout)
     }
 }

Tests/OpenTelemetrySdkTests/Trace/Mocks/SpanProcessorMock.swift

@@ -36,7 +36,7 @@ class SpanProcessorMock: SpanProcessor {
         shutdownCalledTimes += 1
     }
 
-    func forceFlush() {
+    func forceFlush(timeout: TimeInterval? = nil) {
         forceFlushCalledTimes += 1
     }
 }