From e22f9e01a6a856eb30879bd701a022ac3882a32b Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 13:24:29 +0000 Subject: [PATCH 1/6] Add a mapping for most of librdkafka stats in `RDKafkaStatistics` Adds broker, topic, partition and consumer-group level properties for parsing `librdkafka` statistics output. This covers most of the fields exported in `librdkafka` statistics events, skipping some internal ones (e.g. internal timestamp) that are unlikely to be useful as metrics. Renames existing properties to have a more consistent naming convention: using `total` as a suffix to closer match the metric names recommended by Prometheus docs and dropping repeated `...KafkaBroker...`. --- Sources/Kafka/RDKafka/RDKafkaStatistics.swift | 385 ++++++++++++++++-- 1 file changed, 361 insertions(+), 24 deletions(-) diff --git a/Sources/Kafka/RDKafka/RDKafkaStatistics.swift b/Sources/Kafka/RDKafka/RDKafkaStatistics.swift index 96ceb4b2..1d2a5b7b 100644 --- a/Sources/Kafka/RDKafka/RDKafkaStatistics.swift +++ b/Sources/Kafka/RDKafka/RDKafkaStatistics.swift @@ -15,31 +15,368 @@ // MARK: - RDKafkaStatistics struct RDKafkaStatistics: Hashable, Codable { - let queuedOperation: Int? - let queuedProducerMessages: Int? - let queuedProducerMessagesSize: Int? - let topicsInMetadataCache: Int? - let totalKafkaBrokerRequests: Int? - let totalKafkaBrokerBytesSent: Int? - let totalKafkaBrokerResponses: Int? - let totalKafkaBrokerResponsesSize: Int? - let totalKafkaBrokerMessagesSent: Int? - let totalKafkaBrokerMessagesBytesSent: Int? - let totalKafkaBrokerMessagesRecieved: Int? - let totalKafkaBrokerMessagesBytesRecieved: Int? + /// Handle instance name. + let name: String + /// Client identifier. + let clientId: String + /// Instance type (consumer, producer). + let type: String + + /// Number of operations (callbacks, events, etc.) waiting in queue for application to serve with `poll()` + let queueOperations: Int + /// Current number of messages in producer queues. + let queueMessages: Int + /// Current total size of messages in producer queues (bytes). + let queueMessagesSize: Int + /// Threshold: maximum number of messages allowed in the producer queues. + let queueMessagesMax: Int + /// Threshold: maximum total size of messages allowed in the producer queues (bytes). + let queueMessagesSizeMax: Int + + /// Total number of requests sent to Kafka brokers. + let requestsSentTotal: Int + /// Total number of bytes sent to Kafka brokers. + let bytesSentTotal: Int + /// Total number of responses received from Kafka brokers. + let responsesReceivedTotal: Int + /// Total number of bytes received from Kafka brokers. + let bytesReceivedTotal: Int + /// Total number of messages transmitted (produced) to Kafka brokers. + let messagesSentTotal: Int + /// Total number of message bytes (including framing) transmitted to Kafka brokers. + let messageBytesSentTotal: Int + /// Total number of messages consumed not including ignored messages (due to offset, etc) from Kafka brokers. + let messagesReceivedTotal: Int + /// Total number of message bytes (including framing) received from Kafka brokers. + let messageBytesReceivedTotal: Int + + /// Per-broker statistics, keyed by broker name (e.g. "localhost:9092/0"). + let brokers: [String: BrokerStats] + /// Per-topic statistics, keyed by topic name. + let topics: [String: TopicStats] + /// Consumer group statistics (consumer instances only). + let consumerGroup: ConsumerGroupStats? enum CodingKeys: String, CodingKey { - case queuedOperation = "replyq" - case queuedProducerMessages = "msg_cnt" - case queuedProducerMessagesSize = "msg_size" - case topicsInMetadataCache = "metadata_cache_cnt" - case totalKafkaBrokerRequests = "tx" - case totalKafkaBrokerBytesSent = "tx_bytes" - case totalKafkaBrokerResponses = "rx" - case totalKafkaBrokerResponsesSize = "rx_bytes" - case totalKafkaBrokerMessagesSent = "txmsgs" - case totalKafkaBrokerMessagesBytesSent = "txmsg_bytes" - case totalKafkaBrokerMessagesRecieved = "rxmsgs" - case totalKafkaBrokerMessagesBytesRecieved = "rxmsg_bytes" + case name + case clientId = "client_id" + case type + case queueOperations = "replyq" + case queueMessages = "msg_cnt" + case queueMessagesSize = "msg_size" + case queueMessagesMax = "msg_max" + case queueMessagesSizeMax = "msg_size_max" + case requestsSentTotal = "tx" + case bytesSentTotal = "tx_bytes" + case responsesReceivedTotal = "rx" + case bytesReceivedTotal = "rx_bytes" + case messagesSentTotal = "txmsgs" + case messageBytesSentTotal = "txmsg_bytes" + case messagesReceivedTotal = "rxmsgs" + case messageBytesReceivedTotal = "rxmsg_bytes" + case brokers + case topics + case consumerGroup = "cgrp" + } +} + +extension RDKafkaStatistics { + /// Rolling window statistics (latency, batch sizes, etc.). + struct WindowStats: Hashable, Codable { + let min: Int + let max: Int + let avg: Int + let sum: Int + let count: Int + let stddev: Int + let hdrsize: Int + let p50: Int + let p75: Int + let p90: Int + let p95: Int + let p99: Int + let p99_99: Int + let outOfRange: Int + + enum CodingKeys: String, CodingKey { + case min, max, avg, sum + case count = "cnt" + case stddev, hdrsize + case p50, p75, p90, p95, p99 + case p99_99 + case outOfRange = "outofrange" + } + } +} + +extension RDKafkaStatistics { + /// Per-broker statistics. + struct BrokerStats: Hashable, Codable { + /// Broker hostname, port, and broker id. + let name: String + /// Broker id (-1 for bootstraps). + let nodeId: Int + /// Broker hostname. + let nodeName: String + /// Broker source (learned, configured, internal, logical). + let source: String + /// Broker state (INIT, DOWN, CONNECT, AUTH, APIVERSION, AUTH_HANDSHAKE, UP, UPDATE). + let state: String + /// Time since last broker state change (microseconds). + let stateAge: Int + + /// Number of requests awaiting transmission to the broker. + let outbufRequests: Int + /// Number of messages awaiting transmission to the broker. + let outbufMessages: Int + /// Number of requests in-flight to the broker awaiting response. + let waitrespRequests: Int + /// Number of messages in-flight to the broker awaiting response. + let waitrespMessages: Int + + /// Total number of requests sent. + let requestsSentTotal: Int + /// Total number of bytes sent. + let bytesSentTotal: Int + /// Total number of transmission errors. + let transmitErrorsTotal: Int + /// Total number of request retries. + let requestRetriesTotal: Int + /// Microseconds since last socket send (or -1 if no sends yet). + let transmitIdleTime: Int + /// Total number of requests timed out. + let requestTimeoutsTotal: Int + /// Total number of responses received. + let responsesReceivedTotal: Int + /// Total number of bytes received. + let bytesReceivedTotal: Int + /// Total number of receive errors. + let receiveErrorsTotal: Int + /// Total number of unmatched correlation ids in response. + let correlationIdErrorsTotal: Int + /// Total number of partial MessageSets received. + let partialResponsesTotal: Int + + /// Microseconds since last socket receive (or -1 if no receives yet). + let receiveIdleTime: Int + /// Total number of decompression buffer size increases. + let decompressBufferGrowTotal: Int + /// Broker thread poll wakeups. + let wakeupsTotal: Int + /// Number of connection attempts, including successful and failed. + let connectsTotal: Int + /// Number of disconnects (triggered by broker, network, load-balancer, etc.). + let disconnectsTotal: Int + /// Internal producer queue latency (microseconds). + let internalLatency: WindowStats + /// Request queue latency (microseconds). + let outbufLatency: WindowStats + /// Broker latency / round-trip time (microseconds). + let roundTripTime: WindowStats + /// Broker throttle time (milliseconds). + let throttleTime: WindowStats + + enum CodingKeys: String, CodingKey { + case name + case nodeId = "nodeid" + case nodeName = "nodename" + case source, state + case stateAge = "stateage" + case outbufRequests = "outbuf_cnt" + case outbufMessages = "outbuf_msg_cnt" + case waitrespRequests = "waitresp_cnt" + case waitrespMessages = "waitresp_msg_cnt" + case requestsSentTotal = "tx" + case bytesSentTotal = "txbytes" + case transmitErrorsTotal = "txerrs" + case requestRetriesTotal = "txretries" + case transmitIdleTime = "txidle" + case requestTimeoutsTotal = "req_timeouts" + case responsesReceivedTotal = "rx" + case bytesReceivedTotal = "rxbytes" + case receiveErrorsTotal = "rxerrs" + case correlationIdErrorsTotal = "rxcorriderrs" + case partialResponsesTotal = "rxpartial" + case receiveIdleTime = "rxidle" + case decompressBufferGrowTotal = "zbuf_grow" + case wakeupsTotal = "wakeups" + case connectsTotal = "connects" + case disconnectsTotal = "disconnects" + case internalLatency = "int_latency" + case outbufLatency = "outbuf_latency" + case roundTripTime = "rtt" + case throttleTime = "throttle" + } + } +} + +extension RDKafkaStatistics { + /// Per-topic statistics. + struct TopicStats: Hashable, Codable { + /// Topic name. + let topic: String? + /// Age of the client's topic object (microseconds). + let age: Int + /// Age of metadata from broker for this topic (milliseconds). + let metadataAge: Int + /// Batch sizes in bytes. + let batchBytes: WindowStats? + /// Batch message counts. + let batchMessages: WindowStats? + /// Per-partition statistics, keyed by partition id as string. + let partitions: [String: PartitionStats]? + + enum CodingKeys: String, CodingKey { + case topic, age + case metadataAge = "metadata_age" + case batchBytes = "batchsize" + case batchMessages = "batchcnt" + case partitions + } + } +} + +extension RDKafkaStatistics { + /// Per-partition statistics. + struct PartitionStats: Hashable, Codable { + /// Partition id. + let partition: Int + /// The id of the broker this partition is served by. + let broker: Int + /// Current leader broker id. + let leader: Int + /// Partition is explicitly desired by the application. + let desired: Bool + /// Partition is not seen in topic metadata from broker. + let unknown: Bool + /// Number of messages waiting to be produced in first-level queue. + let queueMessages: Int + /// Number of bytes in messages waiting to be produced in first-level queue. + let queueBytes: Int + /// Number of messages ready to be produced in transmit queue. + let transmitQueueMessages: Int + /// Number of bytes ready to be produced in transmit queue. + let transmitQueueBytes: Int + /// Number of pre-fetched messages in fetch queue. + let fetchQueueMessages: Int + /// Bytes of pre-fetched messages in fetch queue. + let fetchQueueSize: Int + + /// Consumer fetch state for this partition (e.g. "active", "none"). + let fetchState: String? + /// Current/Last logical offset query. + let queryOffset: Int + /// Next offset to fetch. + let nextOffset: Int + /// Offset of last message passed to the application + 1. + let appOffset: Int + /// Offset to be committed. + let storedOffset: Int + /// Partition leader epoch of stored offset + let storedLeaderEpoch: Int + /// Last committed offset. + let committedOffset: Int + /// Partition leader epoch of committed offset + let committedLeaderEpoch: Int + /// Last PARTITION_EOF signaled offset. + let eofOffset: Int + /// Partition's low watermark offset on the broker. + let lowWatermarkOffset: Int + /// Partition's high watermark offset on the broker. + let highWatermarkOffset: Int + /// Partition's last stable offset on the broker (or same as highWatermarkOffset). + let lastStableOffset: Int + /// Difference between highWatermarkOffset and committedOffset (consumer lag). + let consumerLag: Int + /// Difference between highWatermarkOffset and storedOffset (consumer lag including not yet committed). + let consumerLagStored: Int + /// Last known partition leader epoch, or -1 if unknown. + let leaderEpoch: Int + + /// Total number of messages transmitted (produced). + let messagesSentTotal: Int + /// Total number of bytes transmitted. + let bytesSentTotal: Int + /// Total number of messages consumed. + let messagesReceivedTotal: Int + /// Total number of bytes received. + let bytesReceivedTotal: Int + /// Total number of messages received (consumer) or produced (producer). + let messagesTotal: Int + /// Dropped outdated messages. + let receivedVersionDropsTotal: Int + /// Current number of messages in-flight to/from broker. + let messagesInFlight: Int + /// Next expected acknowledged sequence (idempotent producer). + let nextAckSequence: Int + /// Next expected error sequence (idempotent producer). + let nextErrorSequence: Int + /// Last acknowledged internal message id. + let ackedMessageId: Int + + enum CodingKeys: String, CodingKey { + case partition, broker, leader, desired, unknown + case queueMessages = "msgq_cnt" + case queueBytes = "msgq_bytes" + case transmitQueueMessages = "xmit_msgq_cnt" + case transmitQueueBytes = "xmit_msgq_bytes" + case fetchQueueMessages = "fetchq_cnt" + case fetchQueueSize = "fetchq_size" + case fetchState = "fetch_state" + case queryOffset = "query_offset" + case nextOffset = "next_offset" + case appOffset = "app_offset" + case storedOffset = "stored_offset" + case storedLeaderEpoch = "stored_leader_epoch" + case committedOffset = "committed_offset" + case committedLeaderEpoch = "committed_leader_epoch" + case eofOffset = "eof_offset" + case lowWatermarkOffset = "lo_offset" + case highWatermarkOffset = "hi_offset" + case lastStableOffset = "ls_offset" + case consumerLag = "consumer_lag" + case consumerLagStored = "consumer_lag_stored" + case leaderEpoch = "leader_epoch" + case messagesSentTotal = "txmsgs" + case bytesSentTotal = "txbytes" + case messagesReceivedTotal = "rxmsgs" + case bytesReceivedTotal = "rxbytes" + case messagesTotal = "msgs" + case receivedVersionDropsTotal = "rx_ver_drops" + case messagesInFlight = "msgs_inflight" + case nextAckSequence = "next_ack_seq" + case nextErrorSequence = "next_err_seq" + case ackedMessageId = "acked_msgid" + } + } +} + +extension RDKafkaStatistics { + /// Consumer group statistics. + struct ConsumerGroupStats: Hashable, Codable { + /// Local consumer group handler's state. + let state: String? + /// Time elapsed since last state change (milliseconds). + let stateAge: Int + /// Local consumer group handler's join state. + let joinState: String? + /// Time elapsed since last rebalance (assign or revoke) (milliseconds). + let rebalanceAge: Int + /// Total number of rebalances (assign or revoke). + let rebalancesTotal: Int + /// Reason for the last rebalance. + let rebalanceReason: String? + /// Current assignment's partition count. + let assignmentSize: Int + + enum CodingKeys: String, CodingKey { + case state + case stateAge = "stateage" + case joinState = "join_state" + case rebalanceAge = "rebalance_age" + case rebalancesTotal = "rebalance_cnt" + case rebalanceReason = "rebalance_reason" + case assignmentSize = "assignment_size" + } } } From a0f461157ea0982117be3e6025d027a40b1d3b03 Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 13:40:08 +0000 Subject: [PATCH 2/6] Add `KafkaMetrics` to convert librdkafka statistics to swift-metrics `KafkaMetrics` is going to replace metrics structs from `KafkaConfiguration`. Instead of expecting users to initialize metric instances it creates them on demand using the provided configuration and label prefix. Note: all metric instances are created on-demand, we're not holding a cache of metric handlers as the list of broker and topic statistics might change between updates. Statistics interval can only be configured in increments of 1 second, but this might still have a performance impact when there are many brokers/partitions involved. --- Sources/Kafka/KafkaMetrics.swift | 239 +++++++++++++++++++++++++++++++ 1 file changed, 239 insertions(+) create mode 100644 Sources/Kafka/KafkaMetrics.swift diff --git a/Sources/Kafka/KafkaMetrics.swift b/Sources/Kafka/KafkaMetrics.swift new file mode 100644 index 00000000..b33025c2 --- /dev/null +++ b/Sources/Kafka/KafkaMetrics.swift @@ -0,0 +1,239 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the swift-kafka-client open source project +// +// Copyright (c) 2026 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 + +public struct KafkaMetrics: Sendable { + /// Prefix used for all metric labels. + public let prefix: String + + /// Enable export of per-broker metrics. + public var enableBrokerMetrics: Bool + + /// Enable export of per-topic and per-partition metrics. + public var enableTopicMetrics: Bool + + /// Configure which window statistics percentiles/aggregations to export. + public var windowMetrics: Set + + /// List of metrics to export for rolling window statistics (e.g. broker latency or batch sizes). + public enum WindowMetric: Sendable, Hashable { + case min + case max + case avg + case sum + case count + case stddev + case hdrsize + case p50 + case p75 + case p90 + case p95 + case p99 + case p99_99 + case outOfRange + } + + private let factory: any MetricsFactory + + /// Initialize a ``KafkaMetrics`` instance for a Kafka client. + /// + /// If `.statisticsIntervalMs` is set on the client configuration `librdkafka` statistics + /// will be exported as metrics. + /// + /// - Parameters: + /// - prefix: Prefix for all metric labels. Default: `"kafka"`. + /// - enableBrokerMetrics: Export per-broker statistics as metrics. Default: `true`. + /// - enableTopicMetrics: Export per-topic and per-partition statistics as metrics. Default: `true`. + /// - windowMetrics: Which window statistics to export. Default: min, max, avg, count, p50, p95, p99. + /// - factory: An optional custom `MetricsFactory`. If `nil`, uses `MetricsSystem.factory`. + public init( + prefix: String = "kafka", + enableBrokerMetrics: Bool = true, + enableTopicMetrics: Bool = true, + windowMetrics: Set = [.min, .max, .avg, .count, .p95, .p99], + factory: (any MetricsFactory)? = nil + ) { + self.prefix = prefix + self.enableBrokerMetrics = enableBrokerMetrics + self.enableTopicMetrics = enableTopicMetrics + self.windowMetrics = windowMetrics + self.factory = factory ?? MetricsSystem.factory + } + + internal func update(with stats: RDKafkaStatistics) { + let clientDimension: [(String, String)] = [("client", stats.clientId)] + + self.recordTopLevel(stats: stats, dimensions: clientDimension) + + if self.enableBrokerMetrics { + for (brokerName, broker) in stats.brokers { + let brokerDimensions = clientDimension + [("broker", brokerName)] + self.recordBroker(broker: broker, dimensions: brokerDimensions) + } + } + + if self.enableTopicMetrics { + for (topicName, topic) in stats.topics { + let topicDimensions = clientDimension + [("topic", topicName)] + self.recordTopic(topic: topic, dimensions: topicDimensions) + + if let partitions = topic.partitions { + for (partId, partition) in partitions { + let partitionDimensions = clientDimension + [("topic", topicName), ("partition", partId)] + self.recordPartition(partition: partition, dimensions: partitionDimensions) + } + } + } + } + + // Consumer group metrics + if let cgrp = stats.consumerGroup { + self.recordConsumerGroup(cgrp: cgrp, dimensions: clientDimension) + } + } + + private func recordTopLevel(stats: RDKafkaStatistics, dimensions: [(String, String)]) { + record("queue_operations", stats.queueOperations, dimensions) + record("queue_messages", stats.queueMessages, dimensions) + record("queue_messages_size", stats.queueMessagesSize, dimensions) + record("queue_messages_max", stats.queueMessagesMax, dimensions) + record("queue_messages_size_max", stats.queueMessagesSizeMax, dimensions) + record("requests_sent_total", stats.requestsSentTotal, dimensions) + record("bytes_sent_total", stats.bytesSentTotal, dimensions) + record("responses_received_total", stats.responsesReceivedTotal, dimensions) + record("bytes_received_total", stats.bytesReceivedTotal, dimensions) + record("messages_sent_total", stats.messagesSentTotal, dimensions) + record("message_bytes_sent_total", stats.messageBytesSentTotal, dimensions) + record("messages_received_total", stats.messagesReceivedTotal, dimensions) + record("message_bytes_received_total", stats.messageBytesReceivedTotal, dimensions) + } + + private func recordBroker(broker: RDKafkaStatistics.BrokerStats, dimensions: [(String, String)]) { + record("broker_state_age", broker.stateAge, dimensions) + record("broker_outbuf_requests", broker.outbufRequests, dimensions) + record("broker_outbuf_messages", broker.outbufMessages, dimensions) + record("broker_waitresp_requests", broker.waitrespRequests, dimensions) + record("broker_waitresp_messages", broker.waitrespMessages, dimensions) + record("broker_requests_sent_total", broker.requestsSentTotal, dimensions) + record("broker_bytes_sent_total", broker.bytesSentTotal, dimensions) + record("broker_transmit_errors_total", broker.transmitErrorsTotal, dimensions) + record("broker_request_retries_total", broker.requestRetriesTotal, dimensions) + record("broker_transmit_idle_time", broker.transmitIdleTime, dimensions) + record("broker_request_timeouts_total", broker.requestTimeoutsTotal, dimensions) + record("broker_responses_received_total", broker.responsesReceivedTotal, dimensions) + record("broker_bytes_received_total", broker.bytesReceivedTotal, dimensions) + record("broker_receive_errors_total", broker.receiveErrorsTotal, dimensions) + record("broker_correlation_id_errors_total", broker.correlationIdErrorsTotal, dimensions) + record("broker_partial_responses_total", broker.partialResponsesTotal, dimensions) + record("broker_receive_idle_time", broker.receiveIdleTime, dimensions) + record("broker_decompress_buffer_grow_total", broker.decompressBufferGrowTotal, dimensions) + record("broker_wakeups_total", broker.wakeupsTotal, dimensions) + record("broker_connects_total", broker.connectsTotal, dimensions) + record("broker_disconnects_total", broker.disconnectsTotal, dimensions) + + recordWindowStats("broker_internal_latency", broker.internalLatency, dimensions) + recordWindowStats("broker_outbuf_latency", broker.outbufLatency, dimensions) + recordWindowStats("broker_round_trip_time", broker.roundTripTime, dimensions) + recordWindowStats("broker_throttle_time", broker.throttleTime, dimensions) + } + + private func recordTopic(topic: RDKafkaStatistics.TopicStats, dimensions: [(String, String)]) { + record("topic_age", topic.age, dimensions) + record("topic_metadata_age", topic.metadataAge, dimensions) + + recordWindowStats("topic_batch_bytes", topic.batchBytes, dimensions) + recordWindowStats("topic_batch_messages", topic.batchMessages, dimensions) + } + + private func recordPartition(partition: RDKafkaStatistics.PartitionStats, dimensions: [(String, String)]) { + record("partition_queue_messages", partition.queueMessages, dimensions) + record("partition_queue_bytes", partition.queueBytes, dimensions) + record("partition_transmit_queue_messages", partition.transmitQueueMessages, dimensions) + record("partition_transmit_queue_bytes", partition.transmitQueueBytes, dimensions) + record("partition_fetch_queue_messages", partition.fetchQueueMessages, dimensions) + record("partition_fetch_queue_size", partition.fetchQueueSize, dimensions) + record("partition_query_offset", partition.queryOffset, dimensions) + record("partition_next_offset", partition.nextOffset, dimensions) + record("partition_app_offset", partition.appOffset, dimensions) + record("partition_stored_offset", partition.storedOffset, dimensions) + record("partition_committed_offset", partition.committedOffset, dimensions) + record("partition_eof_offset", partition.eofOffset, dimensions) + record("partition_low_watermark_offset", partition.lowWatermarkOffset, dimensions) + record("partition_high_watermark_offset", partition.highWatermarkOffset, dimensions) + record("partition_last_stable_offset", partition.lastStableOffset, dimensions) + record("partition_consumer_lag", partition.consumerLag, dimensions) + record("partition_consumer_lag_stored", partition.consumerLagStored, dimensions) + record("partition_messages_sent_total", partition.messagesSentTotal, dimensions) + record("partition_bytes_sent_total", partition.bytesSentTotal, dimensions) + record("partition_messages_received_total", partition.messagesReceivedTotal, dimensions) + record("partition_bytes_received_total", partition.bytesReceivedTotal, dimensions) + record("partition_messages_total", partition.messagesTotal, dimensions) + record("partition_received_version_drops_total", partition.receivedVersionDropsTotal, dimensions) + record("partition_messages_in_flight", partition.messagesInFlight, dimensions) + record("partition_next_ack_sequence", partition.nextAckSequence, dimensions) + record("partition_next_error_sequence", partition.nextErrorSequence, dimensions) + } + + private func recordConsumerGroup(cgrp: RDKafkaStatistics.ConsumerGroupStats, dimensions: [(String, String)]) { + record("cgrp_state_age", cgrp.stateAge, dimensions) + record("cgrp_rebalance_age", cgrp.rebalanceAge, dimensions) + record("cgrp_rebalances_total", cgrp.rebalancesTotal, dimensions) + record("cgrp_assignment_size", cgrp.assignmentSize, dimensions) + } + + private func record(_ label: String, _ value: Int, _ dimensions: [(String, String)]) { + Gauge(label: "\(prefix)_\(label)", dimensions: dimensions, factory: self.factory).record(value) + } + + private func recordWindowStats( + _ base: String, + _ stats: RDKafkaStatistics.WindowStats?, + _ dimensions: [(String, String)] + ) { + guard let stats else { return } + for ws in self.windowMetrics { + switch ws { + case .min: + record("\(base)_min", stats.min, dimensions) + case .max: + record("\(base)_max", stats.max, dimensions) + case .avg: + record("\(base)_avg", stats.avg, dimensions) + case .sum: + record("\(base)_sum", stats.sum, dimensions) + case .count: + record("\(base)_count", stats.count, dimensions) + case .stddev: + record("\(base)_stddev", stats.stddev, dimensions) + case .hdrsize: + record("\(base)_hdrsize", stats.hdrsize, dimensions) + case .p50: + record("\(base)_p50", stats.p50, dimensions) + case .p75: + record("\(base)_p75", stats.p75, dimensions) + case .p90: + record("\(base)_p90", stats.p90, dimensions) + case .p95: + record("\(base)_p95", stats.p95, dimensions) + case .p99: + record("\(base)_p99", stats.p99, dimensions) + case .p99_99: + record("\(base)_p99_99", stats.p99_99, dimensions) + case .outOfRange: + record("\(base)_outofrange", stats.outOfRange, dimensions) + } + } + } +} From 15ad7bdbb93363558f080a0f1f18fb120d923477 Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 14:00:07 +0000 Subject: [PATCH 3/6] Replace `KafkaConfiguration.*Metrics` with `KafkaMetrics` in configs --- .../Configuration/KafkaConsumerConfig.swift | 17 ++++++++--------- .../Configuration/KafkaConsumerConfig.swift.gyb | 17 ++++++++--------- .../Configuration/KafkaProducerConfig.swift | 17 ++++++++--------- .../Configuration/KafkaProducerConfig.swift.gyb | 17 ++++++++--------- 4 files changed, 32 insertions(+), 36 deletions(-) diff --git a/Sources/Kafka/Configuration/KafkaConsumerConfig.swift b/Sources/Kafka/Configuration/KafkaConsumerConfig.swift index 4a23a964..42687b45 100644 --- a/Sources/Kafka/Configuration/KafkaConsumerConfig.swift +++ b/Sources/Kafka/Configuration/KafkaConsumerConfig.swift @@ -31,8 +31,14 @@ public struct KafkaConsumerConfig: Sendable { /// The strategy used for consuming messages. public var consumptionStrategy: KafkaConsumerConfiguration.ConsumptionStrategy? - /// Consumer metrics configuration. - public var metrics: KafkaConfiguration.ConsumerMetrics = .init() + /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable + /// automatic publishing of librdkafka statistics as swift-metrics gauges. + /// Requires ``statisticsIntervalMs`` to be configured. + public var metrics: KafkaMetrics? + + internal var metricsEnabled: Bool { + ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil) + } // MARK: - Properties generated from librdkafka config list @@ -1178,13 +1184,6 @@ public struct KafkaConsumerConfig: Sendable { } } - if config["statistics.interval.ms"] == nil, - metrics.enabled, - let updateInterval = metrics.updateInterval - { - config["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - return config } diff --git a/Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb b/Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb index 037daf92..ee528c8d 100644 --- a/Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb +++ b/Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb @@ -29,8 +29,14 @@ public struct KafkaConsumerConfig: Sendable { /// The strategy used for consuming messages. public var consumptionStrategy: KafkaConsumerConfiguration.ConsumptionStrategy? - /// Consumer metrics configuration. - public var metrics: KafkaConfiguration.ConsumerMetrics = .init() + /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable + /// automatic publishing of librdkafka statistics as swift-metrics gauges. + /// Requires ``statisticsIntervalMs`` to be configured. + public var metrics: KafkaMetrics? + + internal var metricsEnabled: Bool { + ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil) + } // MARK: - Properties generated from librdkafka config list @@ -91,13 +97,6 @@ public struct KafkaConsumerConfig: Sendable { } } - if config["statistics.interval.ms"] == nil, - metrics.enabled, - let updateInterval = metrics.updateInterval - { - config["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - return config } diff --git a/Sources/Kafka/Configuration/KafkaProducerConfig.swift b/Sources/Kafka/Configuration/KafkaProducerConfig.swift index 2f657ff3..c19cd2bb 100644 --- a/Sources/Kafka/Configuration/KafkaProducerConfig.swift +++ b/Sources/Kafka/Configuration/KafkaProducerConfig.swift @@ -36,8 +36,14 @@ public struct KafkaProducerConfig: Sendable { /// Default: `10000` public var shutdownFlushTimeoutMs: Int = 10000 - /// Producer metrics configuration. - public var metrics: KafkaConfiguration.ProducerMetrics = .init() + /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable + /// automatic publishing of librdkafka statistics as swift-metrics gauges. + /// Requires ``statisticsIntervalMs`` to be configured. + public var metrics: KafkaMetrics? + + internal var metricsEnabled: Bool { + ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil) + } // MARK: - Properties generated from librdkafka config list @@ -1147,13 +1153,6 @@ public struct KafkaProducerConfig: Sendable { config["partitioner"] = self.partitioner?.description config["compression.level"] = self.compressionLevel?.description - if config["statistics.interval.ms"] == nil, - metrics.enabled, - let updateInterval = metrics.updateInterval - { - config["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - return config } diff --git a/Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb b/Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb index bd15bd73..55f13656 100644 --- a/Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb +++ b/Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb @@ -34,8 +34,14 @@ public struct KafkaProducerConfig: Sendable { /// Default: `10000` public var shutdownFlushTimeoutMs: Int = 10000 - /// Producer metrics configuration. - public var metrics: KafkaConfiguration.ProducerMetrics = .init() + /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable + /// automatic publishing of librdkafka statistics as swift-metrics gauges. + /// Requires ``statisticsIntervalMs`` to be configured. + public var metrics: KafkaMetrics? + + internal var metricsEnabled: Bool { + ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil) + } // MARK: - Properties generated from librdkafka config list @@ -79,13 +85,6 @@ public struct KafkaProducerConfig: Sendable { % end % end - if config["statistics.interval.ms"] == nil, - metrics.enabled, - let updateInterval = metrics.updateInterval - { - config["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - return config } From 46d71e22c19235438234150dfa7ac93c0fbdbffe Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 14:06:50 +0000 Subject: [PATCH 4/6] Update clients to use `KafkaMetrics` --- Sources/Kafka/KafkaConsumer.swift | 6 +++--- Sources/Kafka/KafkaProducer.swift | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Sources/Kafka/KafkaConsumer.swift b/Sources/Kafka/KafkaConsumer.swift index 99a65659..f8d1d770 100644 --- a/Sources/Kafka/KafkaConsumer.swift +++ b/Sources/Kafka/KafkaConsumer.swift @@ -202,7 +202,7 @@ public final class KafkaConsumer: Sendable, Service { if !isAutoCommitEnabled { subscribedEvents.append(.offsetCommit) } - if config.metrics.enabled { + if config.metricsEnabled { subscribedEvents.append(.statistics) } @@ -245,7 +245,7 @@ public final class KafkaConsumer: Sendable, Service { if !isAutoCommitEnabled { subscribedEvents.append(.offsetCommit) } - if config.metrics.enabled { + if config.metricsEnabled { subscribedEvents.append(.statistics) } @@ -378,7 +378,7 @@ public final class KafkaConsumer: Sendable, Service { for event in events { switch event { case .statistics(let statistics): - self.config.metrics.update(with: statistics) + self.config.metrics?.update(with: statistics) default: break } diff --git a/Sources/Kafka/KafkaProducer.swift b/Sources/Kafka/KafkaProducer.swift index 4464028a..35610c45 100644 --- a/Sources/Kafka/KafkaProducer.swift +++ b/Sources/Kafka/KafkaProducer.swift @@ -113,7 +113,7 @@ public final class KafkaProducer: Service, Sendable { var subscribedEvents: [RDKafkaEvent] = [.log] // No .deliveryReport here! - if config.metrics.enabled { + if config.metricsEnabled { subscribedEvents.append(.statistics) } @@ -158,7 +158,7 @@ public final class KafkaProducer: Service, Sendable { var subscribedEvents: [RDKafkaEvent] = [.log, .deliveryReport] // Listen to statistics events when statistics enabled - if config.metrics.enabled { + if config.metricsEnabled { subscribedEvents.append(.statistics) } @@ -242,7 +242,7 @@ public final class KafkaProducer: Service, Sendable { for event in events { switch event { case .statistics(let statistics): - self.config.metrics.update(with: statistics) + self.config.metrics?.update(with: statistics) case .deliveryReport(let reports): _ = source?.yield(.deliveryReports(reports)) } From 874fecbaaedbd9372c9e98e10df3bb469ed30d1b Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 14:19:53 +0000 Subject: [PATCH 5/6] Drop `KafkaConfiguration` metrics property and structs Supporting both old and new implementations at the same time would require an additional public protocol. With the goal of removing old configuration structs before 1.0 we drop the `.metrics` field from them instead. --- .../KafkaConfiguration+Metrics.swift | 143 ------------------ .../KafkaConsumerConfiguration.swift | 10 -- .../KafkaProducerConfiguration.swift | 10 -- Tests/KafkaTests/KafkaConfigTests.swift | 2 - 4 files changed, 165 deletions(-) delete mode 100644 Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift diff --git a/Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift b/Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift deleted file mode 100644 index 839f8fc8..00000000 --- a/Sources/Kafka/Configuration/KafkaConfiguration+Metrics.swift +++ /dev/null @@ -1,143 +0,0 @@ -//===----------------------------------------------------------------------===// -// -// 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 - - /// Configuration for the consumer metrics emitted by `SwiftKafka`. - public struct ConsumerMetrics: Sendable { - internal var enabled: Bool { - self.updateInterval != nil - && (self.queuedOperation != nil || self.totalKafkaBrokerRequests != nil - || self.totalKafkaBrokerBytesSent != nil || self.totalKafkaBrokerResponses != nil - || self.totalKafkaBrokerResponsesSize != nil || self.totalKafkaBrokerMessagesBytesRecieved != nil - || self.topicsInMetadataCache != nil) - } - - /// Update interval for statistics. - public var updateInterval: Duration? - - /// Number of operations (callbacks, events, etc) waiting in the queue. - public var queuedOperation: Gauge? - - /// Total number of requests sent to Kafka brokers. - public var totalKafkaBrokerRequests: Gauge? - /// Total number of bytes transmitted to Kafka brokers. - public var totalKafkaBrokerBytesSent: Gauge? - /// Total number of responses received from Kafka brokers. - public var totalKafkaBrokerResponses: Gauge? - /// Total number of bytes received from Kafka brokers. - public var totalKafkaBrokerResponsesSize: Gauge? - - /// Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers. - public var totalKafkaBrokerMessagesRecieved: Gauge? - /// Total number of message bytes (including framing) received from Kafka brokers. - public var totalKafkaBrokerMessagesBytesRecieved: Gauge? - - /// Number of topics in the metadata cache. - public var topicsInMetadataCache: Gauge? - - private static func record(_ value: T?, to: Gauge?) { - guard let value, - let to - else { - return - } - to.record(value) - } - - internal func update(with rdKafkaStatistics: RDKafkaStatistics) { - Self.record(rdKafkaStatistics.queuedOperation, to: self.queuedOperation) - - Self.record(rdKafkaStatistics.totalKafkaBrokerRequests, to: self.totalKafkaBrokerRequests) - Self.record(rdKafkaStatistics.totalKafkaBrokerBytesSent, to: self.totalKafkaBrokerBytesSent) - Self.record(rdKafkaStatistics.totalKafkaBrokerResponses, to: self.totalKafkaBrokerResponses) - Self.record(rdKafkaStatistics.totalKafkaBrokerResponsesSize, to: self.totalKafkaBrokerResponsesSize) - - Self.record(rdKafkaStatistics.totalKafkaBrokerMessagesRecieved, to: self.totalKafkaBrokerMessagesRecieved) - Self.record( - rdKafkaStatistics.totalKafkaBrokerMessagesBytesRecieved, - to: self.totalKafkaBrokerMessagesBytesRecieved - ) - - Self.record(rdKafkaStatistics.topicsInMetadataCache, to: self.topicsInMetadataCache) - } - } - - /// Configuration for the producer metrics emitted by `SwiftKafka`. - public struct ProducerMetrics: Sendable { - internal var enabled: Bool { - self.updateInterval != nil - && (self.queuedOperation != nil || self.queuedProducerMessages != nil - || self.queuedProducerMessagesSize != nil || self.totalKafkaBrokerRequests != nil - || self.totalKafkaBrokerBytesSent != nil || self.totalKafkaBrokerResponses != nil - || self.totalKafkaBrokerResponsesSize != nil || self.totalKafkaBrokerMessagesSent != nil - || self.totalKafkaBrokerMessagesBytesSent != nil || self.topicsInMetadataCache != nil) - } - - /// Update interval for statistics. - public var updateInterval: Duration? - - /// Number of operations (callbacks, events, etc) waiting in the queue. - public var queuedOperation: Gauge? - /// Current number of queued producer messages. - public var queuedProducerMessages: Gauge? - /// Current total size in bytes of queued producer messages. - public var queuedProducerMessagesSize: Gauge? - - /// Total number of requests sent to Kafka brokers. - public var totalKafkaBrokerRequests: Gauge? - /// Total number of bytes transmitted to Kafka brokers. - public var totalKafkaBrokerBytesSent: Gauge? - /// Total number of responses received from Kafka brokers. - public var totalKafkaBrokerResponses: Gauge? - /// Total number of bytes received from Kafka brokers. - public var totalKafkaBrokerResponsesSize: Gauge? - - /// Total number of messages transmitted (produced) to Kafka brokers. - public var totalKafkaBrokerMessagesSent: Gauge? - /// Total number of message bytes (including framing, such as per-Message framing and MessageSet/batch framing) transmitted to Kafka brokers. - public var totalKafkaBrokerMessagesBytesSent: Gauge? - - /// Number of topics in the metadata cache. - public var topicsInMetadataCache: Gauge? - - private static func record(_ value: T?, to: Gauge?) { - guard let value, - let to - else { - return - } - to.record(value) - } - - internal func update(with rdKafkaStatistics: RDKafkaStatistics) { - Self.record(rdKafkaStatistics.queuedOperation, to: self.queuedOperation) - Self.record(rdKafkaStatistics.queuedProducerMessages, to: self.queuedProducerMessages) - Self.record(rdKafkaStatistics.queuedProducerMessagesSize, to: self.queuedProducerMessagesSize) - - Self.record(rdKafkaStatistics.totalKafkaBrokerRequests, to: self.totalKafkaBrokerRequests) - Self.record(rdKafkaStatistics.totalKafkaBrokerBytesSent, to: self.totalKafkaBrokerBytesSent) - Self.record(rdKafkaStatistics.totalKafkaBrokerResponses, to: self.totalKafkaBrokerResponses) - Self.record(rdKafkaStatistics.totalKafkaBrokerResponsesSize, to: self.totalKafkaBrokerResponsesSize) - - Self.record(rdKafkaStatistics.totalKafkaBrokerMessagesSent, to: self.totalKafkaBrokerMessagesSent) - Self.record(rdKafkaStatistics.totalKafkaBrokerMessagesBytesSent, to: self.totalKafkaBrokerMessagesBytesSent) - - Self.record(rdKafkaStatistics.topicsInMetadataCache, to: self.topicsInMetadataCache) - } - } -} diff --git a/Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift b/Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift index 8b17a6eb..e68db755 100644 --- a/Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift +++ b/Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift @@ -222,9 +222,6 @@ public struct KafkaConsumerConfiguration { /// Reconnect options. public var reconnect: KafkaConfiguration.ReconnectOptions = .init() - /// Options for librdkafka metrics updates - public var metrics: KafkaConfiguration.ConsumerMetrics = .init() - /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl). /// Default: `.plaintext` public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext @@ -296,12 +293,6 @@ extension KafkaConsumerConfiguration { resultDict["reconnect.backoff.ms"] = String(reconnect.backoff.rawValue) resultDict["reconnect.backoff.max.ms"] = String(reconnect.maximumBackoff.inMilliseconds) - if self.metrics.enabled, - let updateInterval = self.metrics.updateInterval - { - resultDict["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - // Merge with SecurityProtocol configuration dictionary resultDict.merge(securityProtocol.dictionary) { _, _ in fatalError("securityProtocol and \(#file) should not have duplicate keys") @@ -321,7 +312,6 @@ extension KafkaConsumerConfiguration { // Non-rdkafka properties config.pollInterval = self.pollInterval config.consumptionStrategy = self.consumptionStrategy - config.metrics = self.metrics return config } diff --git a/Sources/Kafka/Configuration/KafkaProducerConfiguration.swift b/Sources/Kafka/Configuration/KafkaProducerConfiguration.swift index a1623293..dadb77cb 100644 --- a/Sources/Kafka/Configuration/KafkaProducerConfiguration.swift +++ b/Sources/Kafka/Configuration/KafkaProducerConfiguration.swift @@ -161,9 +161,6 @@ public struct KafkaProducerConfiguration { /// Reconnect options. public var reconnect: KafkaConfiguration.ReconnectOptions = .init() - /// Options for librdkafka metrics updates - public var metrics: KafkaConfiguration.ProducerMetrics = .init() - /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl). /// Default: `.plaintext` public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext @@ -218,12 +215,6 @@ extension KafkaProducerConfiguration { resultDict["reconnect.backoff.ms"] = String(self.reconnect.backoff.rawValue) resultDict["reconnect.backoff.max.ms"] = String(self.reconnect.maximumBackoff.inMilliseconds) - if self.metrics.enabled, - let updateInterval = self.metrics.updateInterval - { - resultDict["statistics.interval.ms"] = String(updateInterval.inMilliseconds) - } - // Merge with SecurityProtocol configuration dictionary resultDict.merge(self.securityProtocol.dictionary) { _, _ in fatalError("securityProtocol and \(#file) should not have duplicate keys") @@ -247,7 +238,6 @@ extension KafkaProducerConfiguration { // Non-rdkafka properties config.pollInterval = self.pollInterval config.shutdownFlushTimeoutMs = self.flushTimeoutMilliseconds - config.metrics = self.metrics return config } diff --git a/Tests/KafkaTests/KafkaConfigTests.swift b/Tests/KafkaTests/KafkaConfigTests.swift index 21bb2785..8cc500dd 100644 --- a/Tests/KafkaTests/KafkaConfigTests.swift +++ b/Tests/KafkaTests/KafkaConfigTests.swift @@ -68,7 +68,6 @@ import Testing config.config == configuration.dictionary ) #expect(config.pollInterval == configuration.pollInterval) - #expect(config.metrics.enabled == configuration.metrics.enabled) #expect(config.consumptionStrategy == configuration.consumptionStrategy) } @@ -106,6 +105,5 @@ import Testing #expect(config.config == configDict) #expect(config.pollInterval == configuration.pollInterval) #expect(config.shutdownFlushTimeoutMs == configuration.flushTimeoutMilliseconds) - #expect(config.metrics.enabled == configuration.metrics.enabled) } } From ba54a8f472f0be1d17299ca23274d05efcd1214a Mon Sep 17 00:00:00 2001 From: Alexey Bezhan Date: Wed, 11 Mar 2026 14:40:07 +0000 Subject: [PATCH 6/6] Update `KafkaMetricsTests` Add tests for per-broker and per-partition metrics and make the tests run concurrently by specifying the metrics factory. --- Tests/KafkaTests/KafkaMetricsTests.swift | 652 +++++++++++++++++++++-- 1 file changed, 621 insertions(+), 31 deletions(-) diff --git a/Tests/KafkaTests/KafkaMetricsTests.swift b/Tests/KafkaTests/KafkaMetricsTests.swift index 3d92d904..db63e42a 100644 --- a/Tests/KafkaTests/KafkaMetricsTests.swift +++ b/Tests/KafkaTests/KafkaMetricsTests.swift @@ -20,7 +20,6 @@ import Testing import struct Foundation.UUID -@testable import CoreMetrics // for MetricsSystem.bootstrapInternal @testable import Kafka #if canImport(FoundationEssentials) @@ -29,28 +28,627 @@ import FoundationEssentials import Foundation #endif -@Suite(.serialized) final class KafkaMetricsTests { - var metrics: TestMetrics = TestMetrics() let kafkaHost: String = ProcessInfo.processInfo.environment["KAFKA_HOST"] ?? "localhost" let kafkaPort: Int = .init(ProcessInfo.processInfo.environment["KAFKA_PORT"] ?? "9092")! + let statisticsJson = """ + { + "name": "rdkafka#producer-1", + "client_id": "rdkafka", + "type": "producer", + "ts": 5016483227792, + "time": 1527060869, + "replyq": 0, + "msg_cnt": 22710, + "msg_size": 704010, + "msg_max": 500000, + "msg_size_max": 1073741824, + "simple_cnt": 0, + "metadata_cache_cnt": 1, + "brokers": { + "localhost:9092/2": { + "name": "localhost:9092/2", + "nodeid": 2, + "nodename": "localhost:9092", + "source": "learned", + "state": "UP", + "stateage": 9057234, + "outbuf_cnt": 0, + "outbuf_msg_cnt": 0, + "waitresp_cnt": 0, + "waitresp_msg_cnt": 0, + "tx": 320, + "txbytes": 84283332, + "txerrs": 0, + "txretries": 0, + "req_timeouts": 0, + "rx": 320, + "rxbytes": 15708, + "rxerrs": 0, + "rxcorriderrs": 0, + "rxpartial": 0, + "zbuf_grow": 0, + "buf_grow": 0, + "wakeups": 591067, + "connects": 1, + "disconnects": 0, + "txidle": 10000, + "rxidle": 10000, + "int_latency": { + "min": 86, + "max": 59375, + "avg": 23726, + "sum": 5694616664, + "stddev": 13982, + "p50": 28031, + "p75": 36095, + "p90": 39679, + "p95": 43263, + "p99": 48639, + "p99_99": 59391, + "outofrange": 0, + "hdrsize": 11376, + "cnt": 240012 + }, + "outbuf_latency": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 0, + "cnt": 0 + }, + "rtt": { + "min": 1580, + "max": 3389, + "avg": 2349, + "sum": 79868, + "stddev": 474, + "p50": 2319, + "p75": 2543, + "p90": 3183, + "p95": 3199, + "p99": 3391, + "p99_99": 3391, + "outofrange": 0, + "hdrsize": 13424, + "cnt": 34 + }, + "throttle": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 17520, + "cnt": 34 + }, + "toppars": { + "test-1": { + "topic": "test", + "partition": 1 + } + } + }, + "localhost:9093/3": { + "name": "localhost:9093/3", + "nodeid": 3, + "nodename": "localhost:9093", + "source": "learned", + "state": "UP", + "stateage": 9057209, + "outbuf_cnt": 0, + "outbuf_msg_cnt": 0, + "waitresp_cnt": 0, + "waitresp_msg_cnt": 0, + "tx": 310, + "txbytes": 84301122, + "txerrs": 0, + "txretries": 0, + "req_timeouts": 0, + "rx": 310, + "rxbytes": 15104, + "rxerrs": 0, + "rxcorriderrs": 0, + "rxpartial": 0, + "zbuf_grow": 0, + "buf_grow": 0, + "wakeups": 607956, + "connects": 1, + "disconnects": 0, + "txidle": 10000, + "rxidle": 10000, + "int_latency": { + "min": 82, + "max": 58069, + "avg": 23404, + "sum": 5617432101, + "stddev": 14021, + "p50": 27391, + "p75": 35839, + "p90": 39679, + "p95": 42751, + "p99": 48639, + "p99_99": 58111, + "outofrange": 0, + "hdrsize": 11376, + "cnt": 240016 + }, + "outbuf_latency": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 0, + "cnt": 0 + }, + "rtt": { + "min": 1704, + "max": 3572, + "avg": 2493, + "sum": 87289, + "stddev": 559, + "p50": 2447, + "p75": 2895, + "p90": 3375, + "p95": 3407, + "p99": 3583, + "p99_99": 3583, + "outofrange": 0, + "hdrsize": 13424, + "cnt": 35 + }, + "throttle": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 17520, + "cnt": 35 + }, + "toppars": { + "test-0": { + "topic": "test", + "partition": 0 + } + } + }, + "localhost:9094/4": { + "name": "localhost:9094/4", + "nodeid": 4, + "nodename": "localhost:9094", + "source": "learned", + "state": "UP", + "stateage": 9057207, + "outbuf_cnt": 0, + "outbuf_msg_cnt": 0, + "waitresp_cnt": 0, + "waitresp_msg_cnt": 0, + "tx": 1, + "txbytes": 25, + "txerrs": 0, + "txretries": 0, + "req_timeouts": 0, + "rx": 1, + "rxbytes": 272, + "rxerrs": 0, + "rxcorriderrs": 0, + "rxpartial": 0, + "zbuf_grow": 0, + "buf_grow": 0, + "wakeups": 4, + "connects": 1, + "disconnects": 0, + "txidle": 0, + "rxidle": 0, + "int_latency": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 11376, + "cnt": 0 + }, + "outbuf_latency": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 0, + "cnt": 0 + }, + "rtt": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 13424, + "cnt": 0 + }, + "throttle": { + "min": 0, + "max": 0, + "avg": 0, + "sum": 0, + "stddev": 0, + "p50": 0, + "p75": 0, + "p90": 0, + "p95": 0, + "p99": 0, + "p99_99": 0, + "outofrange": 0, + "hdrsize": 17520, + "cnt": 0 + }, + "toppars": {} + } + }, + "topics": { + "test": { + "topic": "test", + "age": 9060000, + "metadata_age": 9060, + "batchsize": { + "min": 99, + "max": 391805, + "avg": 272593, + "sum": 18808985, + "stddev": 180408, + "p50": 393215, + "p75": 393215, + "p90": 393215, + "p95": 393215, + "p99": 393215, + "p99_99": 393215, + "outofrange": 0, + "hdrsize": 14448, + "cnt": 69 + }, + "batchcnt": { + "min": 1, + "max": 10000, + "avg": 6956, + "sum": 480028, + "stddev": 4608, + "p50": 10047, + "p75": 10047, + "p90": 10047, + "p95": 10047, + "p99": 10047, + "p99_99": 10047, + "outofrange": 0, + "hdrsize": 8304, + "cnt": 69 + }, + "partitions": { + "0": { + "partition": 0, + "broker": 3, + "leader": 3, + "desired": false, + "unknown": false, + "msgq_cnt": 1, + "msgq_bytes": 31, + "xmit_msgq_cnt": 0, + "xmit_msgq_bytes": 0, + "fetchq_cnt": 0, + "fetchq_size": 0, + "fetch_state": "none", + "query_offset": 0, + "next_offset": 0, + "app_offset": -1001, + "stored_offset": -1001, + "stored_leader_epoch": -1, + "commited_offset": -1001, + "committed_offset": -1001, + "committed_leader_epoch": -1, + "eof_offset": -1001, + "lo_offset": -1001, + "hi_offset": -1001, + "ls_offset": -1001, + "consumer_lag": -1, + "consumer_lag_stored": -1, + "leader_epoch": -1, + "txmsgs": 2150617, + "txbytes": 66669127, + "rxmsgs": 0, + "rxbytes": 0, + "msgs": 2160510, + "rx_ver_drops": 0, + "msgs_inflight": 0, + "next_ack_seq": 0, + "next_err_seq": 0, + "acked_msgid": 0 + }, + "1": { + "partition": 1, + "broker": 2, + "leader": 2, + "desired": false, + "unknown": false, + "msgq_cnt": 0, + "msgq_bytes": 0, + "xmit_msgq_cnt": 0, + "xmit_msgq_bytes": 0, + "fetchq_cnt": 0, + "fetchq_size": 0, + "fetch_state": "none", + "query_offset": 0, + "next_offset": 0, + "app_offset": -1001, + "stored_offset": -1001, + "stored_leader_epoch": -1, + "commited_offset": -1001, + "committed_offset": -1001, + "committed_leader_epoch": -1, + "eof_offset": -1001, + "lo_offset": -1001, + "hi_offset": -1001, + "ls_offset": -1001, + "consumer_lag": -1, + "consumer_lag_stored": -1, + "leader_epoch": -1, + "txmsgs": 2150136, + "txbytes": 66654216, + "rxmsgs": 0, + "rxbytes": 0, + "msgs": 2159735, + "rx_ver_drops": 0, + "msgs_inflight": 0, + "next_ack_seq": 0, + "next_err_seq": 0, + "acked_msgid": 0 + }, + "-1": { + "partition": -1, + "broker": -1, + "leader": -1, + "desired": false, + "unknown": false, + "msgq_cnt": 0, + "msgq_bytes": 0, + "xmit_msgq_cnt": 0, + "xmit_msgq_bytes": 0, + "fetchq_cnt": 0, + "fetchq_size": 0, + "fetch_state": "none", + "query_offset": 0, + "next_offset": 0, + "app_offset": -1001, + "stored_offset": -1001, + "stored_leader_epoch": -1, + "commited_offset": -1001, + "committed_offset": -1001, + "committed_leader_epoch": -1, + "eof_offset": -1001, + "lo_offset": -1001, + "hi_offset": -1001, + "ls_offset": -1001, + "consumer_lag": -1, + "consumer_lag_stored": -1, + "leader_epoch": -1, + "txmsgs": 0, + "txbytes": 0, + "rxmsgs": 0, + "rxbytes": 0, + "msgs": 1177, + "rx_ver_drops": 0, + "msgs_inflight": 0, + "next_ack_seq": 0, + "next_err_seq": 0, + "acked_msgid": 0 + } + } + } + }, + "tx": 631, + "tx_bytes": 168584479, + "rx": 631, + "rx_bytes": 31084, + "txmsgs": 4300753, + "txmsg_bytes": 133323343, + "rxmsgs": 0, + "rxmsg_bytes": 0, + "cgrp": { + "state": "up", + "stateage": 0, + "join_state": "steady", + "rebalance_age": 0, + "rebalance_cnt": 2, + "assignment_size": 0 + } + } + """.data(using: .utf8)! + + private var testMetrics: TestMetrics + private var kafkaMetrics: KafkaMetrics + private var stats: RDKafkaStatistics + + init() throws { + testMetrics = TestMetrics() + kafkaMetrics = KafkaMetrics(prefix: "kafka", factory: testMetrics) + stats = try JSONDecoder().decode(RDKafkaStatistics.self, from: statisticsJson) + } + + @Test func topLevelStatistics() throws { + kafkaMetrics.update(with: stats) + + let metric = try testMetrics.expectGauge("kafka_bytes_sent_total", [("client", "rdkafka")]) + #expect(metric.lastValue == 168584479.0) + } + + @Test func brokerStatistics() throws { + kafkaMetrics.update(with: stats) + + let dimensions = [("client", "rdkafka"), ("broker", "localhost:9092/2")] + + let metric = try testMetrics.expectGauge("kafka_broker_requests_sent_total", dimensions) + #expect(metric.lastValue == 320.0) + } + + @Test func topicStatistics() throws { + kafkaMetrics.update(with: stats) + + let dimensions = [("client", "rdkafka"), ("topic", "test")] + + let metric = try testMetrics.expectGauge("kafka_topic_batch_bytes_min", dimensions) + #expect(metric.lastValue == 99.0) + } + + @Test func partitionStatistics() throws { + kafkaMetrics.update(with: stats) + + let dimensions = [("client", "rdkafka"), ("topic", "test"), ("partition", "1")] + + let metric = try testMetrics.expectGauge("kafka_partition_messages_sent_total", dimensions) + #expect(metric.lastValue == 2150136.0) + } + + @Test func consumerGroupStatistics() throws { + kafkaMetrics.update(with: stats) + + let metric = try testMetrics.expectGauge("kafka_cgrp_rebalances_total", [("client", "rdkafka")]) + #expect(metric.lastValue == 2.0) + } + + @Test func customPrefix() throws { + let kafkaMetrics = KafkaMetrics(prefix: "myapp_consumer", factory: testMetrics) + kafkaMetrics.update(with: stats) - init() async throws { - MetricsSystem.bootstrapInternal(self.metrics) + let metric = try testMetrics.expectGauge("myapp_consumer_bytes_sent_total", [("client", "rdkafka")]) + #expect(metric.lastValue == 168584479.0) } - deinit { - MetricsSystem.bootstrapInternal(NOOPMetricsHandler.instance) + @Test func brokerMetricsDisabled() throws { + let kafkaMetrics = KafkaMetrics( + prefix: "kafka", + enableBrokerMetrics: false, + factory: testMetrics + ) + kafkaMetrics.update(with: stats) + + let metric = try testMetrics.expectGauge("kafka_queue_operations", [("client", "rdkafka")]) + #expect(metric.lastValue == 0.0) + + let skippedMetric = [("client", "rdkafka"), ("broker", "localhost:9092/2")] + #expect(throws: TestMetricsError.self) { + try self.testMetrics.expectGauge("kafka_broker_requests_sent_total", skippedMetric) + } } - @Test func consumerStatistics() async throws { - let uniqueGroupID = UUID().uuidString + + @Test func topicMetricsDisabled() throws { + let kafkaMetrics = KafkaMetrics( + prefix: "kafka", + enableTopicMetrics: false, + factory: testMetrics + ) + kafkaMetrics.update(with: stats) + + let metric = try testMetrics.expectGauge("kafka_queue_operations", [("client", "rdkafka")]) + #expect(metric.lastValue == 0.0) + + #expect(throws: TestMetricsError.self) { + try self.testMetrics.expectGauge("kafka_topic_metadata_age", [("client", "rdkafka"), ("topic", "test")]) + } + + #expect(throws: TestMetricsError.self) { + try self.testMetrics.expectGauge( + "kafka_partition_consumer_lag", + [("client", "rdkafka"), ("topic", "test"), ("partition", "0")] + ) + } + } + + @Test func customWindowStatistics() throws { + let kafkaMetrics = KafkaMetrics( + prefix: "kafka", + windowMetrics: [.p99, .p99_99], + factory: testMetrics + ) + + let stats = try JSONDecoder().decode(RDKafkaStatistics.self, from: statisticsJson) + kafkaMetrics.update(with: stats) + + let dimensions = [("client", "rdkafka"), ("broker", "localhost:9092/2")] + + let rttP99 = try testMetrics.expectGauge("kafka_broker_round_trip_time_p99", dimensions) + #expect(rttP99.lastValue == 3391.0) + + let rttP9999 = try testMetrics.expectGauge("kafka_broker_round_trip_time_p99_99", dimensions) + #expect(rttP9999.lastValue == 3391.0) + + #expect(throws: TestMetricsError.self) { + try self.testMetrics.expectGauge("kafka_broker_round_trip_time_min", dimensions) + } + } + + @Test func consumerIntegration() async throws { var config = KafkaConsumerConfig() config.consumptionStrategy = .group( - id: uniqueGroupID, + id: UUID().uuidString, topics: ["this-topic-does-not-exist"] ) - config.metrics.updateInterval = .milliseconds(100) - config.metrics.queuedOperation = .init(label: "operations") + config.bootstrapServers = ["\(kafkaHost):\(kafkaPort)"] + config.clientId = "consumer-1" + config.brokerAddressFamily = .v4 + config.statisticsIntervalMs = 1000 + config.metrics = kafkaMetrics let consumer = try KafkaConsumer(config: config, logger: .kafkaTest) @@ -58,32 +656,26 @@ final class KafkaMetricsTests { let serviceGroup = ServiceGroup(configuration: svcGroupConfig) try await withThrowingTaskGroup(of: Void.self) { group in - // Run Task group.addTask { try await serviceGroup.run() } - try await Task.sleep(for: .seconds(1)) + try await Task.sleep(for: .milliseconds(1100)) - // Shutdown the serviceGroup await serviceGroup.triggerGracefulShutdown() } - let value = try metrics.expectGauge("operations").lastValue - #expect(value != nil) + let replyq = try testMetrics.expectGauge("kafka_bytes_sent_total", [("client", "consumer-1")]) + #expect(replyq.lastValue! > 0.0) } - @Test func producerStatistics() async throws { - let bootstrapBrokerAddress = KafkaConfiguration.BrokerAddress( - host: self.kafkaHost, - port: self.kafkaPort - ) - + @Test func producerIntegration() async throws { var config = KafkaProducerConfig() - config.bootstrapServers = ["\(bootstrapBrokerAddress)"] + config.bootstrapServers = ["\(kafkaHost):\(kafkaPort)"] + config.clientId = "producer-1" config.brokerAddressFamily = .v4 - config.metrics.updateInterval = .milliseconds(100) - config.metrics.queuedOperation = .init(label: "operations") + config.statisticsIntervalMs = 1000 + config.metrics = kafkaMetrics let producer = try KafkaProducer( config: config, @@ -94,18 +686,16 @@ final class KafkaMetricsTests { let serviceGroup = ServiceGroup(configuration: svcGroupConfig) try await withThrowingTaskGroup(of: Void.self) { group in - // Run Task group.addTask { try await serviceGroup.run() } - try await Task.sleep(for: .seconds(1)) + try await Task.sleep(for: .milliseconds(1100)) - // Shutdown the serviceGroup await serviceGroup.triggerGracefulShutdown() } - let value = try metrics.expectGauge("operations").lastValue - #expect(value != nil) + let replyq = try testMetrics.expectGauge("kafka_bytes_sent_total", [("client", "producer-1")]) + #expect(replyq.lastValue! > 0.0) } }