Feature: expose librdkafka statistics as swift metrics

Package.swift

@@ -48,9 +48,11 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.55.0"),
         .package(url: "https://github.com/swift-server/swift-service-lifecycle.git", from: "2.0.0-beta.1"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
+        .package(url: "https://github.com/apple/swift-metrics", from: "2.4.1"),
         // The zstd Swift package produces warnings that we cannot resolve:
         // https://github.com/facebook/zstd/issues/3328
         .package(url: "https://github.com/facebook/zstd.git", from: "1.5.0"),
+        .package(url: "https://github.com/swift-extras/swift-extras-json.git", .upToNextMajor(from: "0.6.0")),
     ],
     targets: [
         .target(
@@ -80,6 +82,8 @@ let package = Package(
                 .product(name: "NIOCore", package: "swift-nio"),
                 .product(name: "ServiceLifecycle", package: "swift-service-lifecycle"),
                 .product(name: "Logging", package: "swift-log"),
+                .product(name: "Metrics", package: "swift-metrics"),
+                .product(name: "ExtrasJSON", package: "swift-extras-json"),
             ]
         ),
         .target(

Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift

@@ -0,0 +1,84 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-client open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-client project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-client project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Metrics
+
+extension KafkaConfiguration {
+    // MARK: - Metrics
+
+    /// Use to configure metrics.
+    public struct KafkaMetrics: Sendable {
+        internal var someMetricsSet: Bool {
+            self.timestamp != nil ||
+                self.time != nil ||
+                self.age != nil ||
+                self.replyQueue != nil ||
+                self.messageCount != nil ||
+                self.messageSize != nil ||
+                self.messageMax != nil ||
+                self.messageSizeMax != nil ||
+                self.totalRequestsSent != nil ||
+                self.totalBytesSent != nil ||
+                self.totalResponsesRecieved != nil ||
+                self.totalBytesReceived != nil ||
+                self.totalMessagesSent != nil ||
+                self.totalMessagesBytesSent != nil ||
+                self.totalBytesReceived != nil ||
+                self.metadataCacheCount != nil
+        }
+
+        /// librdkafka's internal monotonic clock (microseconds)
+        public var timestamp: Gauge?
+        /// Wall clock time in seconds since the epoch
+        public var time: Timer?
+        /// Time since this client instance was created
+        public var age: Timer?
+        /// Number of ops (callbacks, events, etc) waiting in queue for application to serve
+        public var replyQueue: Gauge?
+        /// Current number of messages in producer queues
+        public var messageCount: Gauge?
+        /// Current total size of messages in producer queues
+        public var messageSize: Gauge?
+        /// Threshold: maximum number of messages allowed allowed on the producer queues
+        public var messageMax: Gauge?
+        /// Threshold: maximum total size of messages allowed on the producer queues
+        public var messageSizeMax: Gauge?
+
+        /// Total number of requests sent to Kafka brokers
+        public var totalRequestsSent: Gauge?
+        /// Total number of bytes transmitted to Kafka brokers
+        public var totalBytesSent: Gauge?
+        /// Total number of responses received from Kafka brokers
+        public var totalResponsesRecieved: Gauge?
+        /// Total number of bytes received from Kafka brokers
+        public var totalBytesReceived: Gauge?
+
+        /// Total number of messages transmitted (produced) to Kafka brokers
+        public var totalMessagesSent: Gauge?
+        /// Total number of message bytes (including framing, such as per-Message framing and MessageSet/batch framing) transmitted to Kafka brokers
+        public var totalMessagesBytesSent: Gauge?
+        /// Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers.
+        public var totalMessagesRecieved: Gauge?
+        /// Total number of message bytes (including framing) received from Kafka brokers
+        public var totalMessagesBytesRecieved: Gauge?
+
+        /// Number of topics in the metadata cache.
+        public var metadataCacheCount: Gauge?
+    }
+
+    public enum Metrics: Sendable {
+        case disabled
+        case enabled(updateInterval: Duration, options: KafkaMetrics)
+    }
+}

Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift

@@ -207,6 +207,22 @@ public struct KafkaConsumerConfiguration {
     /// Reconnect options.
     public var reconnect: KafkaConfiguration.ReconnectOptions = .init()
 
+    /// Options for librdkafka metrics updates
+    public var metrics: KafkaConfiguration.Metrics = .disabled {
+        didSet {
+            if case .enabled(let updateInterval, let options) = metrics {
+                precondition(
+                    updateInterval.canBeRepresentedAsMilliseconds,
+                    "Lowest granularity is milliseconds"
+                )
+                precondition(
+                    options.someMetricsSet,
+                    "No metrics set but enabled"
+                )
+            }
+        }
+    }
+
     /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl).
     /// Default: `.plaintext`
     public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext
@@ -271,6 +287,13 @@ extension KafkaConsumerConfiguration {
         resultDict["reconnect.backoff.ms"] = String(reconnect.backoff.rawValue)
         resultDict["reconnect.backoff.max.ms"] = String(reconnect.maximumBackoff.inMilliseconds)
 
+        switch self.metrics {
+        case .disabled:
+            resultDict["statistics.interval.ms"] = "0"
+        case .enabled(let interval, _):
+            resultDict["statistics.interval.ms"] = String(interval.inMilliseconds)
+        }
+
         // Merge with SecurityProtocol configuration dictionary
         resultDict.merge(securityProtocol.dictionary) { _, _ in
             fatalError("securityProtocol and \(#file) should not have duplicate keys")

Sources/Kafka/Configuration/KafkaProducerConfiguration.swift

@@ -161,6 +161,22 @@ public struct KafkaProducerConfiguration {
     /// Reconnect options.
     public var reconnect: KafkaConfiguration.ReconnectOptions = .init()
 
+    /// Options for librdkafka metrics updates
+    public var metrics: KafkaConfiguration.Metrics = .disabled {
+        didSet {
+            if case .enabled(let updateInterval, let options) = metrics {
+                precondition(
+                    updateInterval.canBeRepresentedAsMilliseconds,
+                    "Lowest granularity is milliseconds"
+                )
+                precondition(
+                    options.someMetricsSet,
+                    "No metrics set but enabled"
+                )
+            }
+        }
+    }
+
     /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl).
     /// Default: `.plaintext`
     public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext
@@ -212,6 +228,13 @@ extension KafkaProducerConfiguration {
         resultDict["reconnect.backoff.ms"] = String(self.reconnect.backoff.rawValue)
         resultDict["reconnect.backoff.max.ms"] = String(self.reconnect.maximumBackoff.inMilliseconds)
 
+        switch self.metrics {
+        case .disabled:
+            resultDict["statistics.interval.ms"] = "0"
+        case .enabled(let interval, _):
+            resultDict["statistics.interval.ms"] = String(interval.inMilliseconds)
+        }
+
         // Merge with SecurityProtocol configuration dictionary
         resultDict.merge(self.securityProtocol.dictionary) { _, _ in
             fatalError("securityProtocol and \(#file) should not have duplicate keys")

Sources/Kafka/KafkaConsumer.swift

@@ -130,6 +130,12 @@ public final class KafkaConsumer: Sendable, Service {
         NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure,
         KafkaConsumerCloseOnTerminate
     >
+    typealias ProducerEvents = NIOAsyncSequenceProducer<
+        KafkaConsumerEvent,
+        NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure,
+        KafkaConsumerCloseOnTerminate
+    >
+
     /// The configuration object of the consumer client.
     private let configuration: KafkaConsumerConfiguration
     /// A logger.
@@ -210,6 +216,9 @@ public final class KafkaConsumer: Sendable, Service {
         if configuration.isAutoCommitEnabled == false {
             subscribedEvents.append(.offsetCommit)
         }
+        if case .enabled = configuration.metrics {
+            subscribedEvents.append(.statistics)
+        }
 
         let client = try RDKafkaClient.makeClient(
             type: .consumer,
@@ -250,6 +259,9 @@ public final class KafkaConsumer: Sendable, Service {
         if configuration.isAutoCommitEnabled == false {
             subscribedEvents.append(.offsetCommit)
         }
+        if case .enabled = configuration.metrics {
+            subscribedEvents.append(.statistics)
+        }
 
         let client = try RDKafkaClient.makeClient(
             type: .consumer,
@@ -341,6 +353,14 @@ public final class KafkaConsumer: Sendable, Service {
                     case .consumerMessages(let result):
                         // We do not support back pressure, we can ignore the yield result
                         _ = source.yield(result)
+                    case .statistics(let statistics):
+                        switch self.configuration.metrics {
+                        case .enabled(_, let options):
+                            assert(options.someMetricsSet, "Unexpected statistics received when no metrics configured")
+                            statistics.fill(options)
+                        case .disabled:
+                            assertionFailure("Unexpected statistics received when metrics disabled")
+                        }
                     default:
                         break // Ignore
                     }

Sources/Kafka/KafkaConsumerEvent.swift

@@ -19,6 +19,8 @@ public enum KafkaConsumerEvent: Sendable, Hashable {
 
     internal init(_ event: RDKafkaClient.KafkaEvent) {
         switch event {
+        case .statistics:
+            fatalError("Cannot cast \(event) to KafkaConsumerEvent")
         case .deliveryReport:
             fatalError("Cannot cast \(event) to KafkaConsumerEvent")
         case .consumerMessages:

Sources/Kafka/KafkaProducer.swift

@@ -116,10 +116,16 @@ public final class KafkaProducer: Service, Sendable {
     ) throws {
         let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
 
+        var subscribedEvents: [RDKafkaEvent] = [.log] // No .deliveryReport here!
+
+        if case .enabled = configuration.metrics {
+            subscribedEvents.append(.statistics)
+        }
+
         let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: configuration.dictionary,
-            events: [.log], // No .deliveryReport here!
+            events: subscribedEvents,
             logger: logger
         )
 
@@ -156,10 +162,16 @@ public final class KafkaProducer: Service, Sendable {
     ) throws -> (KafkaProducer, KafkaProducerEvents) {
         let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
 
+        var subscribedEvents: [RDKafkaEvent] = [.log, .deliveryReport]
+        // Listen to statistics events when statistics enabled
+        if case .enabled = configuration.metrics {
+            subscribedEvents.append(.statistics)
+        }
+
         let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: configuration.dictionary,
-            events: [.log, .deliveryReport],
+            events: subscribedEvents,
             logger: logger
         )
 
@@ -212,9 +224,20 @@ public final class KafkaProducer: Service, Sendable {
             case .pollAndYield(let client, let source):
                 let events = client.eventPoll()
                 for event in events {
-                    let producerEvent = KafkaProducerEvent(event)
-                    // Ignore YieldResult as we don't support back pressure in KafkaProducer
-                    _ = source?.yield(producerEvent)
+                    switch event {
+                    case .statistics(let statistics):
+                        switch self.configuration.metrics {
+                        case .enabled(_, let options):
+                            assert(options.someMetricsSet, "Unexpected statistics received when no metrics configured")
+                            statistics.fill(options)
+                        case .disabled:
+                            assertionFailure("Unexpected statistics received when metrics disabled")
+                        }
+                    case .deliveryReport(let reports):
+                        _ = source?.yield(.deliveryReports(reports))
+                    case .consumerMessages:
+                        fatalError("Unexpected event for producer \(event)")
+                    }
                 }
                 try await Task.sleep(for: self.configuration.pollInterval)
             case .flushFinishSourceAndTerminatePollLoop(let client, let source):

Sources/Kafka/KafkaProducerEvent.swift

@@ -23,6 +23,8 @@ public enum KafkaProducerEvent: Sendable, Hashable {
         switch event {
         case .deliveryReport(results: let results):
             self = .deliveryReports(results)
+        case .statistics:
+            fatalError("Cannot cast \(event) to KafkaProducerEvent")
         case .consumerMessages:
             fatalError("Cannot cast \(event) to KafkaProducerEvent")
         }

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -311,6 +311,7 @@ final class RDKafkaClient: Sendable {
     enum KafkaEvent {
         case deliveryReport(results: [KafkaDeliveryReport])
         case consumerMessages(result: Result<KafkaConsumerMessage, Error>)
+        case statistics(KafkaStatistics)
     }
 
     /// Poll the event `rd_kafka_queue_t` for new events.
@@ -341,6 +342,8 @@ final class RDKafkaClient: Sendable {
                 self.handleLogEvent(event)
             case .offsetCommit:
                 self.handleOffsetCommitEvent(event)
+            case .statistics:
+                events.append(self.handleStatistics(event))
             case .none:
                 // Finished reading events, return early
                 return events
@@ -392,6 +395,11 @@ final class RDKafkaClient: Sendable {
         // The returned message(s) MUST NOT be freed with rd_kafka_message_destroy().
     }
 
+    private func handleStatistics(_ event: OpaquePointer?) -> KafkaEvent {
+        let jsonStr = String(cString: rd_kafka_event_stats(event))
+        return .statistics(KafkaStatistics(jsonString: jsonStr))
+    }
+
     /// Handle event of type `RDKafkaEvent.log`.
     ///
     /// - Parameter event: Pointer to underlying `rd_kafka_event_t`.