RUM-8042 Add count of batches blocked

Author: maxep

DatadogCore/Sources/Core/Upload/DataUploadWorker.swift

@@ -73,21 +73,23 @@ internal class DataUploadWorker: DataUploadWorkerType {
             let context = contextProvider.read()
             let blockersForUpload = uploadConditions.blockersForUpload(with: context)
             let isSystemReady = blockersForUpload.isEmpty
-            let files = isSystemReady ? fileReader.readFiles(limit: maxBatchesPerUpload) : nil
-            if let files = files, !files.isEmpty {
+            let files = fileReader.readFiles(limit: maxBatchesPerUpload)
+
+            if !files.isEmpty && isSystemReady {
                 DD.logger.debug("⏳ (\(self.featureName)) Uploading batches...")
                 self.backgroundTaskCoordinator?.beginBackgroundTask()
                 self.uploadFile(from: files.reversed(), context: context)
                 sendUploadCycleMetric()
             } else {
-                let batchLabel = files?.isEmpty == false ? "YES" : (isSystemReady ? "NO" : "NOT CHECKED")
+                let batchLabel = files.isEmpty ? "NO" : "YES"
                 DD.logger.debug("💡 (\(self.featureName)) No upload. Batch to upload: \(batchLabel), System conditions: \(blockersForUpload.description)")
                 self.delay.increase()
                 self.backgroundTaskCoordinator?.endBackgroundTask()
                 self.scheduleNextCycle()
-                sendBatchBlockedMetric(blockers: blockersForUpload)
+                sendBatchBlockedMetric(blockers: blockersForUpload, batchCount: files.count)
             }
         }
+
         self.readWork = readWorkItem
 
         // Start sending batches immediately after initialization:
@@ -107,6 +109,7 @@ internal class DataUploadWorker: DataUploadWorkerType {
                 return
             }
 
+            let filesCount = files.count
             var files = files
             guard let file = files.popLast() else {
                 self.scheduleNextCycle()
@@ -127,6 +130,7 @@ internal class DataUploadWorker: DataUploadWorkerType {
                         DD.logger.debug("   → (\(self.featureName)) not delivered, will be retransmitted: \(uploadStatus.userDebugDescription)")
                         self.delay.increase()
                         self.scheduleNextCycle()
+                        sendBatchBlockedMetric(status: uploadStatus, batchCount: filesCount)
                         return
                     }
 
@@ -147,7 +151,6 @@ internal class DataUploadWorker: DataUploadWorkerType {
                     previousUploadStatus = nil
 
                     if let error = uploadStatus.error {
-                        sendBatchBlockedMetric(error: error)
                         throw error // Throw to report the request error accordingly
                     }
                 } catch DataUploadError.httpError(statusCode: .unauthorized), DataUploadError.httpError(statusCode: .forbidden) {
@@ -241,7 +244,7 @@ internal class DataUploadWorker: DataUploadWorkerType {
         )
     }
 
-    private func sendBatchBlockedMetric(blockers: [DataUploadConditions.Blocker]) {
+    private func sendBatchBlockedMetric(blockers: [DataUploadConditions.Blocker], batchCount: Int) {
         guard !blockers.isEmpty else {
             return
         }
@@ -252,6 +255,7 @@ internal class DataUploadWorker: DataUploadWorkerType {
                 SDKMetricFields.typeKey: BatchBlockedMetric.typeValue,
                 BatchMetric.trackKey: featureName,
                 BatchBlockedMetric.uploaderDelayKey: delay.current,
+                BatchBlockedMetric.batchCount: batchCount,
                 BatchBlockedMetric.blockers: blockers.map {
                     switch $0 {
                     case .battery: return "low_battery"
@@ -264,13 +268,18 @@ internal class DataUploadWorker: DataUploadWorkerType {
         )
     }
 
-    private func sendBatchBlockedMetric(error: DataUploadError) {
+    private func sendBatchBlockedMetric(status: DataUploadStatus, batchCount: Int) {
+        guard let error = status.error else {
+            return
+        }
+
         telemetry.metric(
             name: BatchBlockedMetric.name,
             attributes: [
                 SDKMetricFields.typeKey: BatchBlockedMetric.typeValue,
                 BatchMetric.trackKey: featureName,
                 BatchBlockedMetric.uploaderDelayKey: delay.current,
+                BatchBlockedMetric.batchCount: batchCount,
                 BatchBlockedMetric.failure: {
                     switch error {
                     case let .httpError(code): return "intake-code-\(code.rawValue)"

DatadogCore/Sources/SDKMetrics/BatchMetrics.swift

@@ -142,7 +142,9 @@ internal enum BatchBlockedMetric {
     /// It is applied in addition to the telemetry sample rate (20% by default).
     static let sampleRate: Float = 1.5 // 1.5%
     /// The key for uploader's current delay.
-    static let uploaderDelayKey = "uploader_delay"
+    static let uploaderDelayKey = "uploader_delay.current"
+    /// The key for count of bacthes being blocked.
+    static let batchCount = "batch_count"
 
     /// List of upload blockers
     static let blockers = "blockers"

DatadogCore/Tests/Datadog/Core/Upload/DataUploadWorkerTests.swift

@@ -609,13 +609,8 @@ class DataUploadWorkerTests: XCTestCase {
         worker.cancelSynchronously()
 
         // Then
-        XCTAssertEqual(telemetry.messages.count, 2)
+        XCTAssertEqual(telemetry.messages.count, 1)
         XCTAssertNotNil(telemetry.messages.firstMetric(named: "upload_cycle"), "An upload cycle metric should be send to `telemetry`.")
-
-        let metric = try XCTUnwrap(telemetry.messages.firstMetric(named: "Batch Blocked"), "A Batch Blocked metric should be send to `telemetry`.")
-        XCTAssertEqual(metric.attributes["failure"] as? String, "intake-code-\(randomStatusCode.rawValue)")
-        XCTAssertNil(metric.attributes["blockers"])
-        XCTAssertEqual(metric.attributes["track"] as? String, featureName)
     }
 
     func testWhenDataIsUploadedWithAlertingStatusCode_itSendsErrorTelemetry() throws {
@@ -659,17 +654,9 @@ class DataUploadWorkerTests: XCTestCase {
         worker.cancelSynchronously()
 
         // Then
-        XCTAssertEqual(telemetry.messages.count, 3)
-
-        XCTAssertNotNil(telemetry.messages.firstMetric(named: "upload_cycle"), "An upload cycle metric should be send to `telemetry`.")
 
         let error = try XCTUnwrap(telemetry.messages.firstError(), "An error should be send to `telemetry`.")
         XCTAssertEqual(error.message,"Data upload finished with status code: \(randomStatusCode.rawValue)")
-
-        let metric = try XCTUnwrap(telemetry.messages.firstMetric(named: "Batch Blocked"), "A Batch Blocked metric should be send to `telemetry`.")
-        XCTAssertEqual(metric.attributes["failure"] as? String, "intake-code-\(randomStatusCode.rawValue)")
-        XCTAssertNil(metric.attributes["blockers"])
-        XCTAssertEqual(metric.attributes["track"] as? String, featureName)
     }
 
     func testWhenDataCannotBeUploadedDueToNetworkError_itSendsErrorTelemetry() throws {
@@ -707,15 +694,52 @@ class DataUploadWorkerTests: XCTestCase {
         worker.cancelSynchronously()
 
         // Then
-        XCTAssertEqual(telemetry.messages.count, 3)
-
-        XCTAssertNotNil(telemetry.messages.firstMetric(named: "upload_cycle"), "An upload cycle metric should be send to `telemetry`.")
-
         let error = try XCTUnwrap(telemetry.messages.firstError(), "An error should be send to `telemetry`.")
         XCTAssertEqual(error.message, #"Data upload finished with error - Error Domain=abc Code=0 "(null)""#)
+    }
+
+    func testWhenDataIsUploadedWithRetryableStatusCode_itSendsBatchBlockedTelemetry() throws {
+        // Given
+        let telemetry = TelemetryMock()
+
+        writer.write(value: ["key": "value"])
+        let randomStatusCode: HTTPResponseStatusCode = [
+            .requestTimeout,
+            .tooManyRequests,
+            .internalServerError,
+            .serviceUnavailable
+        ].randomElement()!
+
+        // When
+        let startUploadExpectation = self.expectation(description: "Upload has started")
+        let mockDataUploader = DataUploaderMock(
+            uploadStatus: .mockWith(needsRetry: true, error: .httpError(statusCode: randomStatusCode))
+        )
 
+        mockDataUploader.onUpload = { previousUploadStatus in
+            XCTAssertNil(previousUploadStatus)
+            startUploadExpectation.fulfill()
+        }
+
+        let featureName: String = .mockRandom()
+        let worker = DataUploadWorker(
+            queue: uploaderQueue,
+            fileReader: reader,
+            dataUploader: mockDataUploader,
+            contextProvider: .mockAny(),
+            uploadConditions: .alwaysUpload(),
+            delay: DataUploadDelay(performance: UploadPerformanceMock.veryQuickInitialUpload),
+            featureName: featureName,
+            telemetry: telemetry,
+            maxBatchesPerUpload: .mockRandom(min: 1, max: 100)
+        )
+
+        wait(for: [startUploadExpectation], timeout: 0.5)
+        worker.cancelSynchronously()
+
+        // Then
         let metric = try XCTUnwrap(telemetry.messages.firstMetric(named: "Batch Blocked"), "A Batch Blocked metric should be send to `telemetry`.")
-        XCTAssertEqual(metric.attributes["failure"] as? String, "network-code-\(nserror.code)")
+        XCTAssertEqual(metric.attributes["failure"] as? String, "intake-code-\(randomStatusCode.rawValue)")
         XCTAssertNil(metric.attributes["blockers"])
         XCTAssertEqual(metric.attributes["track"] as? String, featureName)
     }