Merge pull request #227 from open-telemetry/Datadog-Exporter-Migrate-to-Intake-v2

Datadog exporter: migrate to intake v2

Author: Ignacio Bonafonte

Examples/Datadog Sample/main.swift

@@ -11,8 +11,7 @@ import OpenTelemetrySdk
 import UIKit
 #endif
 
-let clientToken = ""
-let apikey = ""
+let apikeyOrClientToken = ""
 
 let sampleKey = "sampleKey"
 let sampleValue = "sampleValue"
@@ -36,9 +35,8 @@ let exporterConfiguration = ExporterConfiguration(
     applicationName: "SwiftDatadogSample",
     applicationVersion: "1.0.0",
     environment: "test",
-    clientToken: clientToken,
-    apiKey: apikey,
-    endpoint: Endpoint.us,
+    apiKey: apikeyOrClientToken,
+    endpoint: Endpoint.us1,
     uploadCondition: { true },
     performancePreset: .instantDataDelivery,
     hostName: hostName

Package.resolved

@@ -51,7 +51,7 @@ let package = Package(
         .package(name: "swift-atomics", url: "https://github.com/apple/swift-atomics.git", from: "0.0.1"),
         .package(name: "swift-metrics", url: "https://github.com/apple/swift-metrics.git", from: "2.1.1"),
         .package(name: "Reachability", url: "https://github.com/ashleymills/Reachability.swift", from: "5.1.0"),
-
+        .package(name: "DataCompression", url: "https://github.com/mw99/DataCompression.git",  from: "3.5.0")
     ],
     targets: [
         .target(name: "OpenTelemetryApi",
@@ -109,7 +109,7 @@ let package = Package(
                 dependencies: ["OpenTelemetrySdk"],
                 path: "Sources/Exporters/InMemory"),
         .target(name: "DatadogExporter",
-                dependencies: ["OpenTelemetrySdk"],
+                dependencies: ["OpenTelemetrySdk", "DataCompression"],
                 path: "Sources/Exporters/DatadogExporter",
                 exclude: ["NOTICE", "README.md"]),
         .testTarget(name: "NetworkStatusTests",

Package.swift

@@ -13,18 +13,10 @@ public class DatadogExporter: SpanExporter, MetricExporter {
     var metricsExporter: MetricsExporter?
 
     public init(config: ExporterConfiguration) throws {
-        guard config.clientToken != nil || config.apiKey != nil else {
-            throw ExporterError(description: "Traces exporter needs a client token and Metrics Exporter need an api key")
-        }
-
         self.configuration = config
-        if configuration.clientToken != nil {
-            spansExporter = try SpansExporter(config: configuration)
-            logsExporter = try LogsExporter(config: configuration)
-        }
-        if configuration.apiKey != nil {
-            metricsExporter = try MetricsExporter(config: configuration)
-        }
+        spansExporter = try SpansExporter(config: configuration)
+        logsExporter = try LogsExporter(config: configuration)
+        metricsExporter = try MetricsExporter(config: configuration)
     }
 
     public func export(spans: [SpanData]) -> SpanExporterResultCode {

Sources/Exporters/DatadogExporter/DatadogExporter.swift

@@ -13,16 +13,17 @@ public struct ExporterConfiguration {
     /// The name of the service, resource, version,... that will be reported to the backend.
     var serviceName: String
     var resource: String
-    var applicationName: String 
+    var applicationName: String
     var version: String
     var environment: String
 
-    /// Either the RUM client token (which supports RUM, Logging and APM) or regular client token, only for Logging and APM.
-    var clientToken: String?
-    /// The api key, needed for metrics reporting
-    var apiKey: String?
+    /// Either the API key or a regular client token
+    /// For metrics reporting API key is needed
+    var apiKey: String
     /// Endpoint that will be used for reporting.
     var endpoint: Endpoint
+
+    var source: String
     /// This conditon will be evaluated before trying to upload data
     /// Can be used to avoid reporting when no connection
     var uploadCondition: () -> Bool
@@ -37,15 +38,15 @@ public struct ExporterConfiguration {
     /// Option to add a custom prefix to all the metrics sent by the exporter
     var metricsNamePrefix: String?
 
-    public init(serviceName: String, resource: String, applicationName: String, applicationVersion: String, environment: String, clientToken: String?, apiKey: String?, endpoint: Endpoint, uploadCondition: @escaping () -> Bool, performancePreset: PerformancePreset = .default, exportUnsampledSpans: Bool = true, exportUnsampledLogs: Bool = true, hostName: String? = nil, metricsNamePrefix: String? = "otel") {
+    public init(serviceName: String, resource: String, applicationName: String, applicationVersion: String, environment: String, apiKey: String, endpoint: Endpoint, source: String = "ios", uploadCondition: @escaping () -> Bool, performancePreset: PerformancePreset = .default, exportUnsampledSpans: Bool = true, exportUnsampledLogs: Bool = true, hostName: String? = nil, metricsNamePrefix: String? = "otel") {
         self.serviceName = serviceName
         self.resource = resource
         self.applicationName = applicationName
         self.version = applicationVersion
         self.environment = environment
-        self.clientToken = clientToken
         self.apiKey = apiKey
         self.endpoint = endpoint
+        self.source = source
         self.uploadCondition = uploadCondition
         self.performancePreset = performancePreset
         self.exportUnsampledSpans = exportUnsampledSpans

Sources/Exporters/DatadogExporter/ExporterConfiguration.swift

@@ -27,10 +27,6 @@ internal class LogsExporter {
     let logsUploadQueue = DispatchQueue(label: "com.otel.datadog.logsupload", target: .global(qos: .userInteractive))
 
     init(config: ExporterConfiguration) throws {
-        guard let clientToken = config.clientToken else {
-            throw ExporterError(description: "Logs Exporter need a client token")
-        }
-
         self.configuration = config
 
         let filesOrchestrator = FilesOrchestrator(
@@ -39,7 +35,7 @@ internal class LogsExporter {
             dateProvider: SystemDateProvider()
         )
 
-        let dataFormat = DataFormat(prefix: "", suffix: "", separator: "\n")
+        let dataFormat = DataFormat(prefix: "[", suffix: "]", separator: ",")
 
         let logsFileWriter = FileWriter(
             dataFormat: dataFormat,
@@ -55,27 +51,29 @@ internal class LogsExporter {
 
         logsStorage = FeatureStorage(writer: logsFileWriter, reader: logsFileReader)
 
-        let urlProvider = UploadURLProvider(
-            urlWithClientToken: try configuration.endpoint.logsUrlWithClientToken(clientToken: clientToken),
-            queryItemProviders: [
-                .ddsource(),
-                .batchTime(using: SystemDateProvider())
+        let requestBuilder = RequestBuilder(
+            url: configuration.endpoint.logsURL,
+            queryItems: [
+                .ddsource(source: configuration.source)
+            ],
+            headers: [
+                .contentTypeHeader(contentType: .applicationJSON),
+                .contentEncodingHeader(contentEncoding: .deflate),
+                .userAgentHeader(
+                    appName: configuration.applicationName,
+                    appVersion: configuration.version,
+                    device: Device.current
+                ),
+                .ddAPIKeyHeader(apiKey: configuration.apiKey),
+                .ddEVPOriginHeader(source: configuration.source),
+                .ddEVPOriginVersionHeader(version: configuration.version),
+                .ddRequestIDHeader()
             ]
         )
 
-        let httpHeaders = HTTPHeaders(headers: [
-            .contentTypeHeader(contentType: .textPlainUTF8),
-            .userAgentHeader(
-                appName: configuration.applicationName,
-                appVersion: configuration.version,
-                device: Device.current
-            )
-        ])
-
         logsUpload = FeatureUpload(featureName: "logsUpload",
                                    storage: logsStorage,
-                                   uploadHTTPHeaders: httpHeaders,
-                                   uploadURLProvider: urlProvider,
+                                   requestBuilder: requestBuilder,
                                    performance: configuration.performancePreset,
                                    uploadCondition: configuration.uploadCondition)
     }

Sources/Exporters/DatadogExporter/Logs/LogsExporter.swift

@@ -18,10 +18,6 @@ internal class MetricsExporter {
     let metricsUploadQueue = DispatchQueue(label: "com.otel.datadog.metricsupload", target: .global(qos: .userInteractive))
 
     init(config: ExporterConfiguration) throws {
-        guard let apiKey = config.apiKey else {
-            throw ExporterError(description: "Metrics Exporter need an api key")
-        }
-
         configuration = config
 
         let filesOrchestrator = FilesOrchestrator(
@@ -46,26 +42,26 @@ internal class MetricsExporter {
 
         metricsStorage = FeatureStorage(writer: spanFileWriter, reader: spanFileReader)
 
-        let urlProvider = UploadURLProvider(
-            urlWithClientToken: configuration.endpoint.metricsURL,
-            queryItemProviders: [
-                .apiKey(apiKey: apiKey),
+        let requestBuilder = RequestBuilder(
+            url: configuration.endpoint.metricsURL,
+            queryItems: [],
+            headers: [
+                .contentTypeHeader(contentType: .textPlainUTF8),
+                .userAgentHeader(
+                    appName: configuration.applicationName,
+                    appVersion: configuration.version,
+                    device: Device.current
+                ),
+                .ddAPIKeyHeader(apiKey: configuration.apiKey),
+                .ddEVPOriginHeader(source: configuration.source),
+                .ddEVPOriginVersionHeader(version: configuration.version),
+                .ddRequestIDHeader(),
             ]
         )
 
-        let httpHeaders = HTTPHeaders(headers: [
-            .contentTypeHeader(contentType: .textPlainUTF8),
-            .userAgentHeader(
-                appName: configuration.applicationName,
-                appVersion: configuration.version,
-                device: Device.current
-            ),
-        ])
-
         metricsUpload = FeatureUpload(featureName: "metricsUpload",
                                       storage: metricsStorage,
-                                      uploadHTTPHeaders: httpHeaders,
-                                      uploadURLProvider: urlProvider,
+                                      requestBuilder: requestBuilder,
                                       performance: configuration.performancePreset,
                                       uploadCondition: configuration.uploadCondition)
     }

Sources/Exporters/DatadogExporter/Metrics/MetricsExporter.swift

@@ -32,19 +32,13 @@ internal final class FileReader {
     // MARK: - Reading batches
 
     func readNextBatch() -> Batch? {
-        queue.sync {
-            synchronizedReadNextBatch()
-        }
-    }
-
-    private func synchronizedReadNextBatch() -> Batch? {
         if let file = orchestrator.getReadableFile(excludingFilesNamed: Set(filesRead.map { $0.name })) {
             do {
                 let fileData = try file.read()
                 let batchData = dataFormat.prefixData + fileData + dataFormat.suffixData
                 return Batch(data: batchData, file: file)
             } catch {
-                print("🔥 Failed to read file: \(error)")
+                print("Failed to read data from file")
                 return nil
             }
         }
@@ -53,31 +47,23 @@ internal final class FileReader {
     }
 
     /// 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
+    /// Currently called from flush method
+    func onRemainingBatches(process: (Batch) -> ()) -> Bool {
+        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))
             }
-            return true
+        } catch {
+            return false
         }
+        return true
     }
 
     // MARK: - Accepting batches
 
     func markBatchAsRead(_ batch: Batch) {
-        queue.sync { [weak self] in
-            self?.synchronizedMarkBatchAsRead(batch)
-        }
-    }
-
-    func synchronizedMarkBatchAsRead(_ batch: Batch) {
         orchestrator.delete(readableFile: batch.file)
         filesRead.append(batch.file)
     }

Sources/Exporters/DatadogExporter/Persistence/FileReader.swift

@@ -18,10 +18,6 @@ internal class SpansExporter {
     let tracesUploadQueue = DispatchQueue(label: "com.otel.datadog.tracesupload", target: .global(qos: .userInteractive))
 
     init(config: ExporterConfiguration) throws {
-        guard let clientToken = config.clientToken else {
-            throw ExporterError(description: "Span Exporter need a client token")
-        }
-
         self.configuration = config
 
         let filesOrchestrator = FilesOrchestrator(
@@ -46,26 +42,27 @@ internal class SpansExporter {
 
         tracesStorage = FeatureStorage(writer: spanFileWriter, reader: spanFileReader)
 
-        let urlProvider = UploadURLProvider(
-            urlWithClientToken: try configuration.endpoint.tracesUrlWithClientToken(clientToken: clientToken),
-            queryItemProviders: [
-                .batchTime(using: SystemDateProvider())
+        let requestBuilder = RequestBuilder(
+            url: configuration.endpoint.tracesURL,
+            queryItems: [],
+            headers: [
+                .contentTypeHeader(contentType: .textPlainUTF8),
+                .contentEncodingHeader(contentEncoding: .deflate),
+                .userAgentHeader(
+                    appName: configuration.applicationName,
+                    appVersion: configuration.version,
+                    device: Device.current
+                ),
+                .ddAPIKeyHeader(apiKey: config.apiKey),
+                .ddEVPOriginHeader(source: configuration.source),
+                .ddEVPOriginVersionHeader(version: configuration.version),
+                .ddRequestIDHeader(),
             ]
         )
 
-        let httpHeaders = HTTPHeaders(headers: [
-            .contentTypeHeader(contentType: .textPlainUTF8),
-            .userAgentHeader(
-                appName: configuration.applicationName,
-                appVersion: configuration.version,
-                device: Device.current
-            )
-        ])
-
         tracesUpload = FeatureUpload(featureName: "tracesUpload",
                                      storage: tracesStorage,
-                                     uploadHTTPHeaders: httpHeaders,
-                                     uploadURLProvider: urlProvider,
+                                     requestBuilder: requestBuilder,
                                      performance: configuration.performancePreset,
                                      uploadCondition: configuration.uploadCondition)
     }