Structured Logs: Buffering & flushing of logs

SentryTestUtils/TestClient.swift

@@ -150,8 +150,8 @@ public class TestClient: SentryClient {
         flushInvocations.record(timeout)
     }
 
-    public var captureLogsDataInvocations = Invocations<Data>()
-    public override func captureLogsData(_ data: Data) {
-        captureLogsDataInvocations.record(data)
+    public var captureLogsDataInvocations = Invocations<(data: Data, count: NSNumber)>()
+    public override func captureLogsData(_ data: Data, with count: NSNumber) {
+        captureLogsDataInvocations.record((data, count))
     }
 }

SentryTestUtils/TestSentryDispatchQueueWrapper.swift

@@ -62,6 +62,20 @@ import Foundation
         dispatchAfterInvocations.invocations.last?.block()
     }
 
+    public var dispatchAfterWorkItemInvocations = Invocations<(interval: TimeInterval, workItem: DispatchWorkItem)>()
+    public override func dispatch(after interval: TimeInterval, workItem: DispatchWorkItem) {
+        dispatchAfterWorkItemInvocations.record((interval, workItem))
+        if blockBeforeMainBlock() {
+            if dispatchAfterExecutesBlock {
+                workItem.perform()
+            }
+        }
+    }
+
+    public func invokeLastDispatchAfterWorkItem() {
+        dispatchAfterWorkItemInvocations.invocations.last?.workItem.perform()
+    }
+
     public var dispatchCancelInvocations = 0
     public override var shouldDispatchCancel: Bool {
         dispatchCancelInvocations += 1

Sources/Sentry/SentryClient.m

@@ -1128,12 +1128,13 @@ - (void)removeAttachmentProcessor:(id<SentryClientAttachmentProcessor>)attachmen
     return processedAttachments;
 }
 
-- (void)captureLogsData:(NSData *)data
+- (void)captureLogsData:(NSData *)data with:(NSNumber *)itemCount;
 {
     SentryEnvelopeItemHeader *header =
         [[SentryEnvelopeItemHeader alloc] initWithType:SentryEnvelopeItemTypeLog
                                                 length:data.length
-                                           contentType:@"application/vnd.sentry.items.log+json"];
+                                           contentType:@"application/vnd.sentry.items.log+json"
+                                             itemCount:itemCount];
 
     SentryEnvelopeItem *envelopeItem = [[SentryEnvelopeItem alloc] initWithHeader:header data:data];
     SentryEnvelope *envelope = [[SentryEnvelope alloc] initWithHeader:[SentryEnvelopeHeader empty]

Sources/Sentry/SentrySDK.m

@@ -118,14 +118,19 @@ + (SentryReplayApi *)replay
 
 + (SentryLogger *)logger
 {
+
     @synchronized(currentLoggerLock) {
         if (currentLogger == nil) {
+
             SentryLogBatcher *batcher;
             if (nil != currentHub.client && currentHub.client.options.experimental.enableLogs) {
-                batcher = [[SentryLogBatcher alloc] initWithClient:currentHub.client];
+                batcher = [[SentryLogBatcher alloc]
+                    initWithClient:currentHub.client
+                     dispatchQueue:SentryDependencyContainer.sharedInstance.dispatchQueueWrapper];
             } else {
                 batcher = nil;
             }
+
             currentLogger = [[SentryLogger alloc]
                  initWithHub:currentHub
                 dateProvider:SentryDependencyContainer.sharedInstance.dateProvider

Sources/Sentry/include/SentryClient+Logs.h

@@ -7,7 +7,7 @@ NS_ASSUME_NONNULL_BEGIN
 /**
  * Helper to capture encoded logs, as SentryEnvelope can't be used in the Swift SDK.
  */
-- (void)captureLogsData:(NSData *)data;
+- (void)captureLogsData:(NSData *)data with:(NSNumber *)itemCount;
 
 @end
 

Sources/Swift/Helper/SentryDispatchQueueWrapper.swift

@@ -46,6 +46,10 @@
         internalWrapper.dispatch(after: interval, block: block)
     }
 
+    public func dispatch(after interval: TimeInterval, workItem: DispatchWorkItem) {
+        internalWrapper.queue.asyncAfter(deadline: .now() + interval, execute: workItem)
+    }
+
     public func dispatchOnce(_ predicate: UnsafeMutablePointer<CLong>, block: @escaping () -> Void) {
         internalWrapper.dispatchOnce(predicate, block: block)
     }

Sources/Swift/Tools/SentryLogBatcher.swift

@@ -6,27 +6,135 @@ import Foundation
 @_spi(Private) public class SentryLogBatcher: NSObject {
 
     private let client: SentryClient
+    private let flushTimeout: TimeInterval
+    private let maxBufferSizeBytes: Int
+    private let dispatchQueue: SentryDispatchQueueWrapper
 
-    let options: Options
+    internal let options: Options
+
+    // All mutable state is accessed from the same serial dispatch queue.
 
-    @_spi(Private) public init(client: SentryClient) {
+    // Every logs data is added sepratley. They are flushed together in an envelope.
+    private var encodedLogs: [Data] = []
+    private var encodedLogsSize: Int = 0
+    private var timerWorkItem: DispatchWorkItem?
+
+    /// Initializes a new SentryLogBatcher.
+    /// - Parameters:
+    ///   - client: The SentryClient to use for sending logs
+    ///   - flushTimeout: The timeout interval after which buffered logs will be flushed
+    ///   - maxBufferSizeBytes: The maximum buffer size in bytes before triggering an immediate flush
+    ///   - dispatchQueue: A **serial** dispatch queue wrapper for thread-safe access to mutable state
+    ///
+    /// - Important: The `dispatchQueue` parameter MUST be a serial queue to ensure thread safety.
+    ///              Passing a concurrent queue will result in undefined behavior and potential data races.
+    @_spi(Private) public init(
+        client: SentryClient,
+        flushTimeout: TimeInterval,
+        maxBufferSizeBytes: Int,
+        dispatchQueue: SentryDispatchQueueWrapper
+    ) {
         self.client = client
         self.options = client.options
+        self.flushTimeout = flushTimeout
+        self.maxBufferSizeBytes = maxBufferSizeBytes
+        self.dispatchQueue = dispatchQueue
         super.init()
     }
 
-    func add(_ log: SentryLog) {
-        dispatch(logs: [log])
+    /// Convenience initializer with default flush timeout and buffer size.
+    /// - Parameters:
+    ///   - client: The SentryClient to use for sending logs
+    ///   - dispatchQueue: A **serial** dispatch queue wrapper for thread-safe access to mutable state
+    ///
+    /// - Important: The `dispatchQueue` parameter MUST be a serial queue to ensure thread safety.
+    ///              Passing a concurrent queue will result in undefined behavior and potential data races.
+    @_spi(Private) public convenience init(client: SentryClient, dispatchQueue: SentryDispatchQueueWrapper) {
+        self.init(
+            client: client,
+            flushTimeout: 5,
+            maxBufferSizeBytes: 1_024 * 1_024, // 1MB
+            dispatchQueue: dispatchQueue
+        )
+    }
+
+    @_spi(Private) func add(_ log: SentryLog) {
+        dispatchQueue.dispatchAsync { [weak self] in
+            self?.encodeAndBuffer(log: log)
+        }
     }
 
-    private func dispatch(logs: [SentryLog]) {
+    @objc
+    @_spi(Private) func flush() {
+        dispatchQueue.dispatchAsync { [weak self] in
+            self?.performFlush()
+        }
+    }
+
+    // Helper
+
+    // Only ever call this from the serial dispatch queue.
+    private func encodeAndBuffer(log: SentryLog) {
         do {
-            let payload = ["items": logs]
-            let data = try encodeToJSONData(data: payload)
+            let encodedLog = try encodeToJSONData(data: log)
+
+            let encodedLogsWereEmpty = encodedLogs.isEmpty
 
-            client.captureLogsData(data)
+            encodedLogs.append(encodedLog)
+            encodedLogsSize += encodedLog.count
+
+            if encodedLogsSize >= maxBufferSizeBytes {
+                performFlush()
+            } else if encodedLogsWereEmpty && timerWorkItem == nil {
+                startTimer()
+            }
         } catch {
-            SentrySDKLog.error("Failed to create logs envelope.")
+            SentrySDKLog.error("Failed to encode log: \(error)")
+        }
+    }
+
+    // Only ever call this from the serial dispatch queue.
+    private func startTimer() {
+        let timerWorkItem = DispatchWorkItem { [weak self] in
+            SentrySDKLog.debug("SentryLogBatcher: Timer fired, calling performFlush().")
+            self?.performFlush()
+        }
+        self.timerWorkItem = timerWorkItem
+        dispatchQueue.dispatch(after: flushTimeout, workItem: timerWorkItem)
+    }
+
+    // Only ever call this from the serial dispatch queue.
+    private func performFlush() {
+        // Reset logs on function exit
+        defer {
+            encodedLogs.removeAll()
+            encodedLogsSize = 0
+        }
+
+        // Reset timer state
+        timerWorkItem?.cancel()
+        timerWorkItem = nil
+
+        guard encodedLogs.count > 0 else {
+            SentrySDKLog.debug("SentryLogBatcher: No logs to flush.")
+            return
+        }
+
+        // Create the payload.
+
+        var payloadData = Data()
+        payloadData.append(Data("{\"items\":[".utf8))
+        let separator = Data(",".utf8)
+        for (index, encodedLog) in encodedLogs.enumerated() {
+            if index > 0 {
+                payloadData.append(separator)
+            }
+            payloadData.append(encodedLog)
         }
+        payloadData.append(Data("]}".utf8))
+
+        // Send the payload.
+
+        client.captureLogsData(payloadData, with: NSNumber(value: encodedLogs.count))
     }
 }

Tests/SentryTests/SentryClientTests.swift

@@ -2098,7 +2098,7 @@ class SentryClientTest: XCTestCase {
         let sut = fixture.getSut()
         let logData = Data("{\"items\":[{\"timestamp\":1627846801,\"level\":\"info\",\"body\":\"Test log message\"}]}".utf8)
 
-        sut.captureLogsData(logData)
+        sut.captureLogsData(logData, with: NSNumber(value: 1))
 
         // Verify that an envelope was sent
         XCTAssertEqual(1, fixture.transport.sentEnvelopes.count)
@@ -2114,6 +2114,7 @@ class SentryClientTest: XCTestCase {
         XCTAssertEqual("log", item.header.type)
         XCTAssertEqual(UInt(logData.count), item.header.length)
         XCTAssertEqual("application/vnd.sentry.items.log+json", item.header.contentType)
+        XCTAssertEqual(NSNumber(value: 1), item.header.itemCount)
 
         // Verify the envelope item data
         XCTAssertEqual(logData, item.data)
@@ -2123,7 +2124,7 @@ class SentryClientTest: XCTestCase {
         let sut = fixture.getSutDisabledSdk()
         let logData = Data("{\"items\":[{\"timestamp\":1627846801,\"level\":\"info\",\"body\":\"Test log message\"}]}".utf8)
 
-        sut.captureLogsData(logData)
+        sut.captureLogsData(logData, with: NSNumber(value: 1))
 
         // Verify that no envelope was sent when client is disabled
         XCTAssertEqual(0, fixture.transport.sentEnvelopes.count)