statistics.rs

//! Client and broker statistics.
//!
//! These statistics are collected automatically by librdkafka when the client
//! is configured with a non-zero `statistics.interval.ms`. They are made
//! available via the [`ClientContext::stats`] callback.
//!
//! Refer to the [librdkafka statistics documentation][librdkafka-stats] for
//! details.
//!
//! [`ClientContext::stats`]: crate::ClientContext::stats
//! [librdkafka-stats]: https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md

use std::collections::HashMap;

use serde::{Deserialize, Serialize};

/// Overall statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Statistics {
    /// The name of the librdkafka handle.
    pub name: String,
    /// The configured `client.id`.
    pub client_id: String,
    /// The instance type (producer or consumer).
    #[serde(rename = "type")]
    pub client_type: String,
    /// The current value of librdkafka's internal monotonic clock, in
    // microseconds since start.
    pub ts: i64,
    /// Wall clock time, in seconds since the Unix epoch.
    pub time: i64,
    /// Time since this client instance was created, in microseconds.
    pub age: i64,
    /// The number of operations (callbacks, events, etc.) waiting in queue.
    pub replyq: i64,
    /// The current number of messages in producer queues.
    pub msg_cnt: u64,
    /// The current total size of messages in producer queues.
    pub msg_size: u64,
    /// The maximum number of messages allowed in the producer queues.
    pub msg_max: u64,
    /// The maximum total size of messages allowed in the producer queues.
    pub msg_size_max: u64,
    /// The total number of requests sent to brokers.
    pub tx: i64,
    /// The total number of bytes transmitted to brokers.
    pub tx_bytes: i64,
    /// The total number of responses received from brokers.
    pub rx: i64,
    /// The total number of bytes received from brokers.
    pub rx_bytes: i64,
    /// The total number of messages transmitted (produced) to brokers.
    pub txmsgs: i64,
    /// The total number of bytes transmitted (produced) to brokers.
    pub txmsg_bytes: i64,
    /// The total number of messages consumed from brokers, not including
    /// ignored messages.
    pub rxmsgs: i64,
    /// The total number of bytes (including framing) consumed from brokers.
    pub rxmsg_bytes: i64,
    /// Internal tracking of legacy vs. new consumer API state.
    pub simple_cnt: i64,
    /// Number of topics in the metadata cache.
    pub metadata_cache_cnt: i64,
    /// Per-broker statistics.
    pub brokers: HashMap<String, Broker>,
    /// Per-topic statistics.
    pub topics: HashMap<String, Topic>,
    /// Consumer group statistics.
    pub cgrp: Option<ConsumerGroup>,
    /// Exactly-once semantics and idempotent producer statistics.
    pub eos: Option<ExactlyOnceSemantics>,
}

/// Per-broker statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Broker {
    /// The broker hostname, port, and ID, in the form `HOSTNAME:PORT/ID`.
    pub name: String,
    /// The broker ID (-1 for bootstraps).
    pub nodeid: i32,
    /// The broker hostname and port.
    pub nodename: String,
    /// The broker source (learned, configured, internal, or logical).
    pub source: String,
    /// The broker state (INIT, DOWN, CONNECT, AUTH, APIVERSION_QUERY,
    /// AUTH_HANDSHAKE, UP, UPDATE).
    pub state: String,
    /// The time since the last broker state change, in microseconds.
    pub stateage: i64,
    /// The number of requests awaiting transmission to the broker.
    pub outbuf_cnt: i64,
    /// The number of messages awaiting transmission to the broker.
    pub outbuf_msg_cnt: i64,
    /// The number of requests in-flight to the broker that are awaiting a
    /// response.
    pub waitresp_cnt: i64,
    /// The number of messages in-flight to the broker that are awaiting a
    /// response.
    pub waitresp_msg_cnt: i64,
    /// The total number of requests sent to the broker.
    pub tx: u64,
    /// The total number of bytes sent to the broker.
    pub txbytes: u64,
    /// The total number of transmission errors.
    pub txerrs: u64,
    /// The total number of request retries.
    pub txretries: u64,
    /// Microseconds since last socket send, or -1 if no sends yet for the
    /// current connection.
    pub txidle: i64,
    /// The total number of requests that timed out.
    pub req_timeouts: u64,
    /// The total number of responses received from the broker.
    pub rx: u64,
    /// The total number of bytes received from the broker.
    pub rxbytes: u64,
    /// The total number of receive errors.
    pub rxerrs: u64,
    /// The number of unmatched correlation IDs in response, typically for
    /// timed out requests.
    pub rxcorriderrs: u64,
    /// The total number of partial message sets received. The broker may return
    /// partial responses if the full message set could not fit in the remaining
    /// fetch response size.
    pub rxpartial: u64,
    /// Microseconds since last socket receive, or -1 if no receives yet for the
    /// current connection.
    pub rxidle: i64,
    /// Request type counters. The object key is the name of the request type
    /// and the value is the number of requests of that type that have been
    /// sent.
    pub req: HashMap<String, i64>,
    /// The total number of decompression buffer size increases.
    pub zbuf_grow: u64,
    /// The total number of buffer size increases (deprecated and unused).
    pub buf_grow: u64,
    /// The number of broker thread poll wakeups.
    pub wakeups: Option<u64>,
    /// The number of connection attempts, including successful and failed
    /// attempts, and name resolution failures.
    pub connects: Option<i64>,
    /// The number of disconnections, whether triggered by the broker, the
    /// network, the load balancer, or something else.
    pub disconnects: Option<i64>,
    /// Rolling window statistics for the internal producer queue latency, in
    /// microseconds.
    pub int_latency: Option<Window>,
    /// Rolling window statistics for the internal request queue latency, in
    /// microseconds.
    ///
    /// This is the time between when a request is enqueued on the transmit
    /// (outbuf) queue and the time the request is written to the TCP socket.
    /// Additional buffering and latency may be incurred by the TCP stack and
    /// network.
    pub outbuf_latency: Option<Window>,
    /// Rolling window statistics for the broker latency/round-trip time,
    /// in microseconds.
    pub rtt: Option<Window>,
    /// Rolling window statistics for the broker throttling time, in
    /// milliseconds.
    pub throttle: Option<Window>,
    /// The partitions that are handled by this broker handle.
    pub toppars: HashMap<String, TopicPartition>,
}

/// Rolling window statistics.
///
/// These values are not exact; they are sampled estimates maintained by an
/// HDR histogram in librdkafka.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Window {
    /// The smallest value.
    pub min: i64,
    /// The largest value.
    pub max: i64,
    /// The mean value.
    pub avg: i64,
    /// The sum of all values.
    pub sum: i64,
    /// The total number of values.
    pub cnt: i64,
    /// The standard deviation.
    pub stddev: i64,
    /// The memory size of the underlying HDR histogram.
    pub hdrsize: i64,
    /// The 50th percentile.
    pub p50: i64,
    /// The 75th percentile.
    pub p75: i64,
    /// The 90th percentile.
    pub p90: i64,
    /// The 95th percentile.
    pub p95: i64,
    /// The 99th percentile.
    pub p99: i64,
    /// The 99.99th percentile.
    pub p99_99: i64,
    /// The number of values not included in the underlying histogram because
    /// they were out of range.
    pub outofrange: i64,
}

/// A topic and partition specifier.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct TopicPartition {
    /// The name of the topic.
    pub topic: String,
    /// The ID of the partition.
    pub partition: i32,
}

/// Per-topic statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Topic {
    /// The name of the topic.
    pub topic: String,
    /// The age of the client's metadata for this topic, in milliseconds.
    pub metadata_age: i64,
    /// Rolling window statistics for batch sizes, in bytes.
    pub batchsize: Window,
    /// Rolling window statistics for batch message counts.
    pub batchcnt: Window,
    /// Per-partition statistics.
    pub partitions: HashMap<i32, Partition>,
}

/// Per-partition statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct Partition {
    /// The partition ID.
    pub partition: i32,
    /// The ID of the broker from which messages are currently being fetched.
    pub broker: i32,
    /// The broker ID of the leader.
    pub leader: i32,
    /// Whether the partition is explicitly desired by the application.
    pub desired: bool,
    /// Whether the partition is not seen in the topic metadata from the broker.
    pub unknown: bool,
    /// The number of messages waiting to be produced in the first-level queue.
    pub msgq_cnt: i64,
    /// The number of bytes waiting to be produced in the first-level queue.
    pub msgq_bytes: u64,
    /// The number of messages ready to be produced in the transmit queue.
    pub xmit_msgq_cnt: i64,
    /// The number of bytes ready to be produced in the transmit queue.
    pub xmit_msgq_bytes: u64,
    /// The number of prefetched messages in the fetch queue.
    pub fetchq_cnt: i64,
    /// The number of bytes in the fetch queue.
    pub fetchq_size: u64,
    /// The consumer fetch state for this partition (none, stopping, stopped,
    /// offset-query, offset-wait, active).
    pub fetch_state: String,
    /// The current/last logical offset query.
    pub query_offset: i64,
    /// The next offset to fetch.
    pub next_offset: i64,
    /// The offset of the last message passed to the application, plus one.
    pub app_offset: i64,
    /// The offset to be committed.
    pub stored_offset: i64,
    /// The last committed offset.
    pub committed_offset: i64,
    /// The last offset for which partition EOF was signaled.
    pub eof_offset: i64,
    /// The low watermark offset on the broker.
    pub lo_offset: i64,
    /// The high watermark offset on the broker.
    pub hi_offset: i64,
    /// The last stable offset on the broker.
    pub ls_offset: i64,
    /// The difference between `hi_offset` and `committed_offset`.
    pub consumer_lag: i64,
    /// The difference between `hi_offset` and `stored_offset`.
    pub consumer_lag_stored: i64,
    /// The total number of messages transmitted (produced).
    pub txmsgs: u64,
    /// The total number of bytes transmitted (produced).
    pub txbytes: u64,
    /// The total number of messages consumed, not included ignored messages.
    pub rxmsgs: u64,
    /// The total bytes consumed.
    pub rxbytes: u64,
    /// The total number of messages received, for consumers, or the total
    /// number of messages produced, for producers.
    pub msgs: u64,
    /// The number of dropped outdated messages.
    pub rx_ver_drops: u64,
    /// The current number of messages in flight to or from the broker.
    pub msgs_inflight: i64,
    /// The next expected acked sequence number, for idempotent producers.
    pub next_ack_seq: i64,
    /// The next expected errored sequence number, for idempotent producers.
    pub next_err_seq: i64,
    /// The last acked internal message ID, for idempotent producers.
    pub acked_msgid: u64,
}

/// Consumer group manager statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct ConsumerGroup {
    /// The local consumer group handler's state.
    pub state: String,
    /// The time elapsed since the last state change, in milliseconds.
    pub stateage: i64,
    /// The local consumer group hander's join state.
    pub join_state: String,
    /// The time elapsed since the last rebalance (assign or revoke), in
    /// milliseconds.
    pub rebalance_age: i64,
    /// The total number of rebalances (assign or revoke).
    pub rebalance_cnt: i64,
    /// The reason for the last rebalance.
    ///
    /// This string will be empty if no rebalances have occurred.
    pub rebalance_reason: String,
    /// The partition count for the current assignment.
    pub assignment_size: i32,
}

/// Exactly-once semantics statistics.
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
pub struct ExactlyOnceSemantics {
    /// The current idempotent producer state.
    pub idemp_state: String,
    /// THe time elapsed since the last idempotent producer state change, in
    /// milliseconds.
    pub idemp_stateage: i64,
    /// The current transactional producer state.
    pub txn_state: String,
    /// The time elapsed since the last transactional producer state change, in
    /// milliseconds.
    pub txn_stateage: i64,
    /// Whether the transactional state allows enqueing (producing) new
    /// messages.
    pub txn_may_enq: bool,
    /// The currently assigned producer ID, or -1.
    pub producer_id: i64,
    /// The current epoch, or -1.
    pub producer_epoch: i64,
    /// The number of producer ID assignments.
    pub epoch_cnt: i64,
}

#[cfg(test)]
mod tests {
    use maplit::hashmap;

    use super::*;

    #[test]
    fn test_statistics() {
        let stats: Statistics = serde_json::from_str(EXAMPLE).unwrap();

        assert_eq!(stats.name, "rdkafka#producer-1");
        assert_eq!(stats.client_type, "producer");
        assert_eq!(stats.ts, 1163982743268);
        assert_eq!(stats.time, 1589652530);
        assert_eq!(stats.replyq, 0);
        assert_eq!(stats.msg_cnt, 320);
        assert_eq!(stats.msg_size, 9920);
        assert_eq!(stats.msg_max, 500000);
        assert_eq!(stats.msg_size_max, 1073741824);
        assert_eq!(stats.simple_cnt, 0);

        assert_eq!(stats.brokers.len(), 1);

        let broker = stats.brokers.values().collect::<Vec<_>>()[0];

        assert_eq!(
            broker.req,
            hashmap! {
                "Produce".to_string() => 31307,
                "Offset".to_string() => 0,
                "Metadata".to_string() => 2,
                "FindCoordinator".to_string() => 0,
                "SaslHandshake".to_string() => 0,
                "ApiVersion".to_string() => 2,
                "InitProducerId".to_string() => 0,
                "AddPartitionsToTxn".to_string() => 0,
                "AddOffsetsToTxn".to_string() => 0,
                "EndTxn".to_string() => 0,
                "TxnOffsetCommit".to_string() => 0,
                "SaslAuthenticate".to_string() => 0,
            }
        );

        assert_eq!(stats.topics.len(), 1);
    }

    // Example from https://github.com/edenhill/librdkafka/wiki/Statistics
    const EXAMPLE: &str = r#"
      {
        "name": "rdkafka#producer-1",
        "client_id": "rdkafka",
        "type": "producer",
        "ts": 1163982743268,
        "time": 1589652530,
        "age": 5,
        "replyq": 0,
        "msg_cnt": 320,
        "msg_size": 9920,
        "msg_max": 500000,
        "msg_size_max": 1073741824,
        "simple_cnt": 0,
        "metadata_cache_cnt": 1,
        "brokers": {
          "localhost:9092/0": {
            "name": "localhost:9092/0",
            "nodeid": 0,
            "nodename": "localhost:9092",
            "source": "configured",
            "state": "UP",
            "stateage": 8005652,
            "outbuf_cnt": 0,
            "outbuf_msg_cnt": 0,
            "waitresp_cnt": 1,
            "waitresp_msg_cnt": 126,
            "tx": 31311,
            "txbytes": 463869957,
            "txerrs": 0,
            "txretries": 0,
            "txidle": 5,
            "req_timeouts": 0,
            "rx": 31310,
            "rxbytes": 1753668,
            "rxerrs": 0,
            "rxcorriderrs": 0,
            "rxpartial": 0,
            "rxidle": 5,
            "zbuf_grow": 0,
            "buf_grow": 0,
            "wakeups": 131568,
            "connects": 1,
            "disconnects": 0,
            "int_latency": {
              "min": 2,
              "max": 9193,
              "avg": 605,
              "sum": 874202325,
              "stddev": 1080,
              "p50": 319,
              "p75": 481,
              "p90": 1135,
              "p95": 3023,
              "p99": 5919,
              "p99_99": 9087,
              "outofrange": 0,
              "hdrsize": 15472,
              "cnt": 1443154
            },
            "outbuf_latency": {
              "min": 1,
              "max": 308,
              "avg": 22,
              "sum": 107311,
              "stddev": 21,
              "p50": 22,
              "p75": 29,
              "p90": 36,
              "p95": 44,
              "p99": 111,
              "p99_99": 309,
              "outofrange": 0,
              "hdrsize": 11376,
              "cnt": 4740
            },
            "rtt": {
              "min": 94,
              "max": 3279,
              "avg": 237,
              "sum": 1124867,
              "stddev": 198,
              "p50": 193,
              "p75": 245,
              "p90": 329,
              "p95": 393,
              "p99": 1183,
              "p99_99": 3279,
              "outofrange": 0,
              "hdrsize": 13424,
              "cnt": 4739
            },
            "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": 4739
            },
            "req": {
              "Produce": 31307,
              "Offset": 0,
              "Metadata": 2,
              "FindCoordinator": 0,
              "SaslHandshake": 0,
              "ApiVersion": 2,
              "InitProducerId": 0,
              "AddPartitionsToTxn": 0,
              "AddOffsetsToTxn": 0,
              "EndTxn": 0,
              "TxnOffsetCommit": 0,
              "SaslAuthenticate": 0
            },
            "toppars": {
              "test-0": {
                "topic": "test",
                "partition": 0
              },
              "test-1": {
                "topic": "test",
                "partition": 1
              },
              "test-2": {
                "topic": "test",
                "partition": 2
              }
            }
          }
        },
        "topics": {
          "test": {
            "topic": "test",
            "metadata_age": 7014,
            "batchsize": {
              "min": 99,
              "max": 240276,
              "avg": 11871,
              "sum": 56260370,
              "stddev": 13137,
              "p50": 10431,
              "p75": 11583,
              "p90": 12799,
              "p95": 13823,
              "p99": 72191,
              "p99_99": 240639,
              "outofrange": 0,
              "hdrsize": 14448,
              "cnt": 4739
            },
            "batchcnt": {
              "min": 1,
              "max": 6161,
              "avg": 304,
              "sum": 1442353,
              "stddev": 336,
              "p50": 267,
              "p75": 297,
              "p90": 329,
              "p95": 353,
              "p99": 1847,
              "p99_99": 6175,
              "outofrange": 0,
              "hdrsize": 8304,
              "cnt": 4739
            },
            "partitions": {
              "0": {
                "partition": 0,
                "broker": 0,
                "leader": 0,
                "desired": false,
                "unknown": false,
                "msgq_cnt": 845,
                "msgq_bytes": 26195,
                "xmit_msgq_cnt": 0,
                "xmit_msgq_bytes": 0,
                "fetchq_cnt": 0,
                "fetchq_size": 0,
                "fetch_state": "none",
                "query_offset": -1001,
                "next_offset": 0,
                "app_offset": -1001,
                "stored_offset": -1001,
                "commited_offset": -1001,
                "committed_offset": -1001,
                "eof_offset": -1001,
                "lo_offset": -1001,
                "hi_offset": -1001,
                "ls_offset": -1001,
                "consumer_lag": -1,
                "consumer_lag_stored": 0,
                "txmsgs": 3950967,
                "txbytes": 122479977,
                "rxmsgs": 0,
                "rxbytes": 0,
                "msgs": 3951812,
                "rx_ver_drops": 0,
                "msgs_inflight": 1067,
                "next_ack_seq": 0,
                "next_err_seq": 0,
                "acked_msgid": 0
              },
              "1": {
                "partition": 1,
                "broker": 0,
                "leader": 0,
                "desired": false,
                "unknown": false,
                "msgq_cnt": 229,
                "msgq_bytes": 7099,
                "xmit_msgq_cnt": 0,
                "xmit_msgq_bytes": 0,
                "fetchq_cnt": 0,
                "fetchq_size": 0,
                "fetch_state": "none",
                "query_offset": -1001,
                "next_offset": 0,
                "app_offset": -1001,
                "stored_offset": -1001,
                "commited_offset": -1001,
                "committed_offset": -1001,
                "eof_offset": -1001,
                "lo_offset": -1001,
                "hi_offset": -1001,
                "ls_offset": -1001,
                "consumer_lag": -1,
                "consumer_lag_stored": 0,
                "txmsgs": 3950656,
                "txbytes": 122470336,
                "rxmsgs": 0,
                "rxbytes": 0,
                "msgs": 3952618,
                "rx_ver_drops": 0,
                "msgs_inflight": 0,
                "next_ack_seq": 0,
                "next_err_seq": 0,
                "acked_msgid": 0
              },
              "2": {
                "partition": 2,
                "broker": 0,
                "leader": 0,
                "desired": false,
                "unknown": false,
                "msgq_cnt": 1816,
                "msgq_bytes": 56296,
                "xmit_msgq_cnt": 0,
                "xmit_msgq_bytes": 0,
                "fetchq_cnt": 0,
                "fetchq_size": 0,
                "fetch_state": "none",
                "query_offset": -1001,
                "next_offset": 0,
                "app_offset": -1001,
                "stored_offset": -1001,
                "commited_offset": -1001,
                "committed_offset": -1001,
                "eof_offset": -1001,
                "lo_offset": -1001,
                "hi_offset": -1001,
                "ls_offset": -1001,
                "consumer_lag": -1,
                "consumer_lag_stored": 0,
                "txmsgs": 3952027,
                "txbytes": 122512837,
                "rxmsgs": 0,
                "rxbytes": 0,
                "msgs": 3953855,
                "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": -1001,
                "next_offset": 0,
                "app_offset": -1001,
                "stored_offset": -1001,
                "commited_offset": -1001,
                "committed_offset": -1001,
                "eof_offset": -1001,
                "lo_offset": -1001,
                "hi_offset": -1001,
                "ls_offset": -1001,
                "consumer_lag": -1,
                "consumer_lag_stored": 0,
                "txmsgs": 0,
                "txbytes": 0,
                "rxmsgs": 0,
                "rxbytes": 0,
                "msgs": 500000,
                "rx_ver_drops": 0,
                "msgs_inflight": 0,
                "next_ack_seq": 0,
                "next_err_seq": 0,
                "acked_msgid": 0
              }
            }
          }
        },
        "tx": 31311,
        "tx_bytes": 463869957,
        "rx": 31310,
        "rx_bytes": 1753668,
        "txmsgs": 11853650,
        "txmsg_bytes": 367463150,
        "rxmsgs": 0,
        "rxmsg_bytes": 0
      }"#;
}