Datadog exporter: Fully synchronize on flush (#70)

Explain in the comments how the synchronisation works on the flush involved methods

Author: nachoBonafonte

README.md

@@ -1,6 +1,4 @@
 
-[![Scope](https://app.scope.dev/api/badge/aac7d72a-28e3-4c66-ac04-ad816166cd41/default)](https://app.scope.dev/external/v1/inspect/54b1fab6-1a7f-4d03-9fb4-26fafd169131/aac7d72a-28e3-4c66-ac04-ad816166cd41/default)
-
 # opentelemetry-swift
 
 A swift [OpenTelemetry](https://opentelemetry.io/) client
@@ -35,17 +33,17 @@ or
 
 ## Current status
 
-Currently Tracing, metrics and Correlation Context and  API's and SDK are implemented, also OpenTracing shims, for compatibility with existing Opentracing code.
+Currently Tracing, Metrics and Correlation Context API's and SDK are implemented, also OpenTracing shims, for compatibility with existing Opentracing code.
 
-Implemented a simple stdout, Jaeger, Zipkin  and OpenTelemetry collector for traces;
+Implemented traces exporters: simple stdout, Jaeger, Zipkin, Datadog and OpenTelemetry collector
 
-Implemented a Prometheus exporter for metrics.
+Implemented metrics exporters: Prometheus
 
 ## Examples
 
 The package includes some example projects with basic functionality:
 
 - `Logging Tracer` -  Simple api implementation of a Tracer that logs every api call
-- `Simple Exporter` - Shows the Jaeger an Stdout exporters in action using a MultiSpanExporter
+- `Simple Exporter` - Shows the Jaeger an Stdout exporters in action using a MultiSpanExporter. Can be easily modified for other exporters
 - `Prometheus Sample` - Shows the Prometheus exporter reporting metrics to a Prometheus instance
 

Sources/Exporters/DatadogExporter/DatadogExporter.swift

@@ -40,11 +40,12 @@ public class DatadogExporter: SpanExporter {
     }
 
     public func flush() -> SpanExporterResultCode {
-        if let writer = spansExporter.tracesStorage.writer as? FileWriter {
-            writer.queue.sync {}
-        }
+        spansExporter.tracesStorage.writer.queue.sync {}
+        logsExporter.logsStorage.writer.queue.sync {}
+
         _ = logsExporter.logsUpload.uploader.flush()
-        return spansExporter.tracesUpload.uploader.flush()
+        _ = spansExporter.tracesUpload.uploader.flush()
+        return .success
     }
 
     public func shutdown() {

Sources/Exporters/DatadogExporter/Files/File.swift

@@ -24,7 +24,7 @@ internal protocol WritableFile {
     func size() throws -> UInt64
 
     /// Synchronously appends given data at the end of this file.
-    func append(data: Data) throws
+    func append(data: Data, synchronized: Bool ) throws
 }
 
 /// Provides convenient interface for reading contents and metadata of the file.
@@ -51,7 +51,7 @@ internal struct File: WritableFile, ReadableFile {
     }
 
     /// Appends given data at the end of this file.
-    func append(data: Data) throws {
+    func append(data: Data, synchronized: Bool = false) throws {
         let fileHandle = try FileHandle(forWritingTo: url)
 
         // NOTE: RUMM-669
@@ -72,7 +72,12 @@ internal struct File: WritableFile, ReadableFile {
           This is fixed in iOS 14/Xcode 12
          */
         if #available(OSX 10.15, iOS 13.4, watchOS 6.0, tvOS 13.0, *) {
-            defer { try? fileHandle.close() }
+            defer {
+                try? fileHandle.close()
+                if synchronized {
+                    try? fileHandle.synchronize()
+                }
+            }
             try fileHandle.seekToEnd()
             try fileHandle.write(contentsOf: data)
         } else {

Sources/Exporters/DatadogExporter/Persistence/FileReader.swift

@@ -62,6 +62,23 @@ internal final class FileReader {
         return nil
     }
 
+    /// This method  gets remaining files at once, and process each file after with the block passed.
+    /// Being on a queue assures that no other previous batches are uploaded while these are being handled
+    internal func onRemainingBatches(process: (Batch)->()) -> Bool {
+        queue.sync {
+            do {
+                try orchestrator.getAllFiles(excludingFilesNamed: Set(filesRead.map { $0.name }))?.forEach {
+                    let fileData = try $0.read()
+                    let batchData = dataFormat.prefixData + fileData + dataFormat.suffixData
+                    process(Batch(data: batchData, file: $0))
+                }
+            } catch {
+                return false
+            }
+            return true
+        }
+    }
+
     // MARK: - Accepting batches
 
     func markBatchAsRead(_ batch: Batch) {
@@ -70,7 +87,7 @@ internal final class FileReader {
         }
     }
 
-    private func synchronizedMarkBatchAsRead(_ batch: Batch) {
+    func synchronizedMarkBatchAsRead(_ batch: Batch) {
         orchestrator.delete(readableFile: batch.file)
         filesRead.append(batch.file)
     }

Sources/Exporters/DatadogExporter/Persistence/FileWriter.swift

@@ -45,20 +45,20 @@ internal final class FileWriter {
 
     func writeSync<T: Encodable>(value: T) {
         queue.sync { [weak self] in
-            self?.synchronizedWrite(value: value)
+            self?.synchronizedWrite(value: value, syncOnEnd: true)
         }
     }
 
-    private func synchronizedWrite<T: Encodable>(value: T) {
+    private func synchronizedWrite<T: Encodable>(value: T, syncOnEnd: Bool = false) {
         do {
             let data = try jsonEncoder.encode(value)
             let file = try orchestrator.getWritableFile(writeSize: UInt64(data.count))
 
             if try file.size() == 0 {
-                try file.append(data: data)
+                try file.append(data: data, synchronized: syncOnEnd)
             } else {
                 let atomicData = dataFormat.separatorData + data
-                try file.append(data: atomicData)
+                try file.append(data: atomicData, synchronized: syncOnEnd)
             }
         } catch {
             print("🔥 Failed to write file: \(error)")

Sources/Exporters/DatadogExporter/Persistence/FilesOrchestrator.swift

@@ -113,6 +113,16 @@ internal class FilesOrchestrator {
         }
     }
 
+    func getAllFiles(excludingFilesNamed excludedFileNames: Set<String> = []) -> [ReadableFile]? {
+        do {
+            return try directory.files()
+                .filter({ excludedFileNames.contains($0.name) == false })
+        } catch {
+            print("🔥 Failed to obtain readable files: \(error)")
+            return nil
+        }
+    }
+
     func delete(readableFile: ReadableFile) {
         do {
             try readableFile.delete()

Sources/Exporters/DatadogExporter/Upload/DataUploadWorker.swift

@@ -86,20 +86,18 @@ internal class DataUploadWorker: DataUploadWorkerType {
         }
     }
 
+    /// This method  gets remaining files at once, and uploads them
+    /// It assures that periodic uploader cannot read or upload the files while the flush is being processed
     internal func flush() -> SpanExporterResultCode {
-        var resultCode: SpanExporterResultCode = .success
-        queue.sync {
-            let nextBatch = self.fileReader.readNextBatch()
-            if let batch = nextBatch {
-                let uploadStatus = self.dataUploader.upload(data: batch.data)
+        let success = queue.sync {
+            self.fileReader.onRemainingBatches {
+                let uploadStatus = self.dataUploader.upload(data: $0.data)
                 let shouldBeAccepted = self.acceptableUploadStatuses.contains(uploadStatus)
                 if shouldBeAccepted {
-                    self.fileReader.markBatchAsRead(batch)
-                } else {
-                    resultCode = .failure
+                    self.fileReader.synchronizedMarkBatchAsRead($0)
                 }
             }
         }
-        return resultCode
+        return success ? .success : .failure
     }
 }

Sources/OpenTelemetrySdk/Trace/Export/SimpleSpanProcessor.swift

@@ -41,6 +41,7 @@ public struct SimpleSpanProcessor: SpanProcessor {
     }
 
     public func forceFlush() {
+        spanExporter.flush()
     }
 
     /// Returns a new SimpleSpansProcessor that converts spans to proto and forwards them to

Tests/ExportersTests/DatadogExporter/Persistence/FilesOrchestratorTests.swift

@@ -83,7 +83,7 @@ class FilesOrchestratorTests: XCTestCase {
         )
 
         let file1 = try orchestrator.getWritableFile(writeSize: performance.maxObjectSize)
-        try chunkedData.forEach { chunk in try file1.append(data: chunk) }
+        try chunkedData.forEach { chunk in try file1.append(data: chunk, synchronized: false) }
         let file2 = try orchestrator.getWritableFile(writeSize: 1)
 
         XCTAssertNotEqual(file1.name, file2.name)
@@ -144,15 +144,15 @@ class FilesOrchestratorTests: XCTestCase {
 
         // write 1MB to first file (1MB of directory size in total)
         let file1 = try orchestrator.getWritableFile(writeSize: oneMB)
-        try file1.append(data: .mock(ofSize: oneMB))
+        try file1.append(data: .mock(ofSize: oneMB), synchronized: false)
 
         // write 1MB to second file (2MB of directory size in total)
         let file2 = try orchestrator.getWritableFile(writeSize: oneMB)
-        try file2.append(data: .mock(ofSize: oneMB))
+        try file2.append(data: .mock(ofSize: oneMB), synchronized: true)
 
         // write 1MB to third file (3MB of directory size in total)
         let file3 = try orchestrator.getWritableFile(writeSize: oneMB + 1) // +1 byte to exceed the limit
-        try file3.append(data: .mock(ofSize: oneMB + 1))
+        try file3.append(data: .mock(ofSize: oneMB + 1), synchronized: false)
 
         XCTAssertEqual(try temporaryDirectory.files().count, 3)
 
@@ -161,7 +161,7 @@ class FilesOrchestratorTests: XCTestCase {
         let file4 = try orchestrator.getWritableFile(writeSize: oneMB)
         XCTAssertEqual(try temporaryDirectory.files().count, 3)
         XCTAssertNil(try? temporaryDirectory.file(named: file1.name))
-        try file4.append(data: .mock(ofSize: oneMB + 1))
+        try file4.append(data: .mock(ofSize: oneMB + 1), synchronized: true)
 
         _ = try orchestrator.getWritableFile(writeSize: oneMB)
         XCTAssertEqual(try temporaryDirectory.files().count, 3)