krabka_client_core/telemetry/
metrics.rs

1//! The metrics that a client pushes, and their conversion to OTLP.
2//!
3//! [`ClientMetrics`] is the registry of one client, as Kafka's `Metrics` is.
4//! [`Collector`] turns a snapshot of it into the `MetricsData` of one push, as
5//! Kafka's `KafkaMetricsCollector` and `ClientTelemetryEmitter` do.
6
7use std::{
8    collections::{BTreeMap, HashMap},
9    sync::{
10        Arc, Mutex, MutexGuard, PoisonError,
11        atomic::{AtomicI64, AtomicU64, Ordering},
12    },
13    time::{SystemTime, UNIX_EPOCH},
14};
15
16use super::otlp::{
17    AggregationTemporality, KeyValue, Metric, MetricData, MetricsData, NumberDataPoint,
18    NumberValue, Resource, ResourceMetrics, ScopeMetrics,
19};
20
21/// The prefix of every client metric name
22/// (`ClientTelemetryProvider.DOMAIN`).
23pub const TELEMETRY_METRIC_DOMAIN: &str = "org.apache.kafka";
24
25/// The tag that the broker already knows from the request header, so Kafka's
26/// `ClientTelemetryReporter` leaves it out of the data point attributes.
27const EXCLUDED_ATTRIBUTE: &str = "client_id";
28
29/// The name and attributes of one metric, as KIP-714 names them.
30#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
31pub struct MetricKey {
32    /// For example `org.apache.kafka.producer.request.total`.
33    pub name: String,
34    /// The data point attributes, with snake case keys.
35    pub tags: BTreeMap<String, String>,
36}
37
38impl MetricKey {
39    /// The key of the Kafka metric `name` in `group` with `tags`, as
40    /// `TelemetryMetricNamingConvention` maps it: the group and name are lower
41    /// case and dot separated, the group loses its `-metrics` part, and each
42    /// tag key is snake case.
43    ///
44    /// ```
45    /// use assert2::assert;
46    /// use krabka_client_core::telemetry::MetricKey;
47    ///
48    /// let key = MetricKey::kafka("producer-metrics", "record-send-rate", []);
49    /// assert!(key.name == "org.apache.kafka.producer.record.send.rate");
50    /// ```
51    #[must_use]
52    pub fn kafka<'a>(
53        group: &str,
54        name: &str,
55        tags: impl IntoIterator<Item = (&'a str, &'a str)>,
56    ) -> Self {
57        let group = clean(group, ".").replace(".metrics", "");
58        Self {
59            name: format!("{TELEMETRY_METRIC_DOMAIN}.{group}.{}", clean(name, ".")),
60            tags: tags
61                .into_iter()
62                .map(|(key, value)| (clean(key, "_"), value.to_owned()))
63                .collect(),
64        }
65    }
66}
67
68fn clean(raw: &str, joiner: &str) -> String {
69    raw.to_lowercase().replace('-', joiner)
70}
71
72/// A monotonic counter, pushed as an OTLP sum (Kafka's `CumulativeSum` and
73/// `WindowedCount`).
74#[derive(Clone, Debug, Default)]
75pub struct Counter(Arc<AtomicU64>);
76
77impl Counter {
78    /// Add `amount` to the counter.
79    pub fn add(&self, amount: u64) {
80        self.0.fetch_add(amount, Ordering::Relaxed);
81    }
82
83    /// The current total.
84    #[must_use]
85    pub fn value(&self) -> u64 {
86        self.0.load(Ordering::Relaxed)
87    }
88}
89
90/// A value that goes up and down, pushed as an OTLP gauge.
91#[derive(Clone, Debug, Default)]
92pub struct Gauge(Arc<AtomicI64>);
93
94impl Gauge {
95    /// Add `delta`, which may be negative.
96    pub fn add(&self, delta: i64) {
97        self.0.fetch_add(delta, Ordering::Relaxed);
98    }
99
100    /// Set the value.
101    pub fn set(&self, value: i64) {
102        self.0.store(value, Ordering::Relaxed);
103    }
104
105    /// The current value.
106    #[must_use]
107    pub fn value(&self) -> i64 {
108        self.0.load(Ordering::Relaxed)
109    }
110}
111
112#[derive(Clone, Debug)]
113enum Cell {
114    Counter(Counter),
115    Gauge(Gauge),
116}
117
118#[derive(Clone, Debug)]
119struct Registered {
120    cell: Cell,
121    /// When the metric was registered, in nanoseconds since the Unix epoch.
122    added_unix_nanos: u64,
123}
124
125/// The metric registry of one client. A clone shares the registry.
126#[derive(Clone, Debug, Default)]
127pub struct ClientMetrics {
128    registered: Arc<Mutex<BTreeMap<MetricKey, Registered>>>,
129}
130
131impl ClientMetrics {
132    /// An empty registry.
133    #[must_use]
134    pub fn new() -> Self {
135        Self::default()
136    }
137
138    /// The counter of `key`, registered now if the registry has no counter of
139    /// that key. A registration replaces a gauge of the same key, as Kafka's
140    /// `metricChange` replaces the metric.
141    #[must_use]
142    pub fn counter(&self, key: MetricKey) -> Counter {
143        let mut registered = self.lock();
144        if let Some(Registered {
145            cell: Cell::Counter(counter),
146            ..
147        }) = registered.get(&key)
148        {
149            return counter.clone();
150        }
151        let counter = Counter::default();
152        registered.insert(
153            key,
154            Registered {
155                cell: Cell::Counter(counter.clone()),
156                added_unix_nanos: unix_nanos_now(),
157            },
158        );
159        counter
160    }
161
162    /// The gauge of `key`, registered now if the registry has no gauge of
163    /// that key.
164    #[must_use]
165    pub fn gauge(&self, key: MetricKey) -> Gauge {
166        let mut registered = self.lock();
167        if let Some(Registered {
168            cell: Cell::Gauge(gauge),
169            ..
170        }) = registered.get(&key)
171        {
172            return gauge.clone();
173        }
174        let gauge = Gauge::default();
175        registered.insert(
176            key,
177            Registered {
178                cell: Cell::Gauge(gauge.clone()),
179                added_unix_nanos: unix_nanos_now(),
180            },
181        );
182        gauge
183    }
184
185    /// Remove the metric of `key` (Kafka's `metricRemoval`).
186    pub fn remove(&self, key: &MetricKey) {
187        self.lock().remove(key);
188    }
189
190    fn lock(&self) -> MutexGuard<'_, BTreeMap<MetricKey, Registered>> {
191        self.registered
192            .lock()
193            .unwrap_or_else(PoisonError::into_inner)
194    }
195
196    fn snapshot(&self) -> Vec<(MetricKey, Registered)> {
197        self.lock()
198            .iter()
199            .map(|(key, registered)| (key.clone(), registered.clone()))
200            .collect()
201    }
202}
203
204/// Nanoseconds since the Unix epoch.
205pub fn unix_nanos_now() -> u64 {
206    SystemTime::now()
207        .duration_since(UNIX_EPOCH)
208        .ok()
209        .and_then(|elapsed| u64::try_from(elapsed.as_nanos()).ok())
210        .unwrap_or(0)
211}
212
213/// The network metrics of a client's connections: the `-total` metrics of
214/// Kafka's `Selector` in the client's metric group, and its
215/// `connection-count`.
216#[derive(Clone, Debug)]
217pub struct NetworkMetrics {
218    connection_creation_total: Counter,
219    connection_close_total: Counter,
220    connection_count: Gauge,
221    request_total: Counter,
222    outgoing_byte_total: Counter,
223    response_total: Counter,
224    incoming_byte_total: Counter,
225}
226
227impl NetworkMetrics {
228    /// Register the network metrics of `group`, for example
229    /// `producer-metrics`, in `metrics`.
230    #[must_use]
231    pub fn register(metrics: &ClientMetrics, group: &str) -> Self {
232        let counter = |name| metrics.counter(MetricKey::kafka(group, name, []));
233        Self {
234            connection_creation_total: counter("connection-creation-total"),
235            connection_close_total: counter("connection-close-total"),
236            connection_count: metrics.gauge(MetricKey::kafka(group, "connection-count", [])),
237            request_total: counter("request-total"),
238            outgoing_byte_total: counter("outgoing-byte-total"),
239            response_total: counter("response-total"),
240            incoming_byte_total: counter("incoming-byte-total"),
241        }
242    }
243
244    /// A connection opened. The returned guard counts its close when it
245    /// drops.
246    pub(crate) fn opened(&self) -> ConnectionMetrics {
247        self.connection_creation_total.add(1);
248        self.connection_count.add(1);
249        ConnectionMetrics(self.clone())
250    }
251}
252
253/// The network metrics of one open connection.
254#[derive(Debug)]
255pub struct ConnectionMetrics(NetworkMetrics);
256
257impl ConnectionMetrics {
258    /// A request of `frame_len` bytes went out. Kafka counts the size prefix
259    /// too.
260    pub(crate) fn request(&self, frame_len: usize) {
261        self.0.request_total.add(1);
262        self.0.outgoing_byte_total.add(wire_len(frame_len));
263    }
264
265    /// A response came in. `body_len` excludes the size prefix and the
266    /// correlation id, which Kafka counts.
267    pub(crate) fn response(&self, body_len: usize) {
268        self.0.response_total.add(1);
269        self.0
270            .incoming_byte_total
271            .add(wire_len(body_len).saturating_add(4));
272    }
273}
274
275fn wire_len(len: usize) -> u64 {
276    u64::try_from(len).unwrap_or(u64::MAX).saturating_add(4)
277}
278
279impl Drop for ConnectionMetrics {
280    fn drop(&mut self) {
281        self.0.connection_close_total.add(1);
282        self.0.connection_count.add(-1);
283    }
284}
285
286/// Which metrics a subscription asks for
287/// (`ClientTelemetryUtils.getSelectorFromRequestedMetrics`).
288#[derive(Clone, Debug, PartialEq, Eq)]
289pub enum MetricSelector {
290    /// An empty `requested_metrics`.
291    NoMetrics,
292    /// A `requested_metrics` of one `*`.
293    AllMetrics,
294    /// The metrics whose name starts with one of these prefixes.
295    Prefixes(Vec<String>),
296}
297
298impl MetricSelector {
299    pub(crate) fn from_requested(requested: &[String]) -> Self {
300        match requested {
301            [] => Self::NoMetrics,
302            [all] if all == "*" => Self::AllMetrics,
303            prefixes => Self::Prefixes(prefixes.to_vec()),
304        }
305    }
306
307    pub(crate) fn matches(&self, name: &str) -> bool {
308        match self {
309            Self::NoMetrics => false,
310            Self::AllMetrics => true,
311            Self::Prefixes(prefixes) => prefixes.iter().any(|prefix| name.starts_with(prefix)),
312        }
313    }
314}
315
316/// The metrics of one push in OTLP form, with the state that delta
317/// temporality needs (Kafka's `KafkaMetricsCollector` and its ledger).
318#[derive(Debug)]
319pub struct Collector {
320    metrics: ClientMetrics,
321    resource: Resource,
322    /// The time and value of each sum at the last delta push.
323    last_values: HashMap<MetricKey, (u64, f64)>,
324    /// The start time of each cumulative sum.
325    added: HashMap<MetricKey, u64>,
326    /// Set by a temporality change: a sum starts at its next collection.
327    reset: bool,
328}
329
330impl Collector {
331    pub(crate) fn new(metrics: ClientMetrics, resource: Resource) -> Self {
332        Self {
333            metrics,
334            resource,
335            last_values: HashMap::new(),
336            added: HashMap::new(),
337            reset: false,
338        }
339    }
340
341    /// Forget the pushed values after a temporality change
342    /// (`metricsReset`).
343    pub(crate) fn reset(&mut self) {
344        self.last_values.clear();
345        self.added.clear();
346        self.reset = true;
347    }
348
349    /// The selected metrics at `now_unix_nanos`, one resource metrics entry
350    /// per metric as Kafka's `createPayload` builds them.
351    pub(crate) fn collect(
352        &mut self,
353        selector: &MetricSelector,
354        delta_temporality: bool,
355        now_unix_nanos: u64,
356    ) -> MetricsData {
357        let mut resource_metrics = Vec::new();
358        for (key, registered) in self.metrics.snapshot() {
359            if !selector.matches(&key.name) {
360                continue;
361            }
362            let attributes = key
363                .tags
364                .iter()
365                .filter(|(tag, _)| tag.as_str() != EXCLUDED_ATTRIBUTE)
366                .map(|(tag, value)| KeyValue::new(tag, value))
367                .collect::<Vec<_>>();
368            let data = match registered.cell {
369                Cell::Gauge(gauge) => MetricData::Gauge {
370                    data_points: vec![NumberDataPoint {
371                        attributes,
372                        start_time_unix_nano: 0,
373                        time_unix_nano: now_unix_nanos,
374                        value: Some(NumberValue::Double(i64_to_f64(gauge.value()))),
375                    }],
376                },
377                Cell::Counter(counter) => {
378                    let value = u64_to_f64(counter.value());
379                    let added = *self.added.entry(key.clone()).or_insert(if self.reset {
380                        now_unix_nanos
381                    } else {
382                        registered.added_unix_nanos
383                    });
384                    let (start, value, aggregation_temporality) = if delta_temporality {
385                        match self
386                            .last_values
387                            .insert(key.clone(), (now_unix_nanos, value))
388                        {
389                            Some((last_time, last_value)) => {
390                                (last_time, value - last_value, AggregationTemporality::Delta)
391                            }
392                            None => (added, value, AggregationTemporality::Delta),
393                        }
394                    } else {
395                        (added, value, AggregationTemporality::Cumulative)
396                    };
397                    MetricData::Sum {
398                        data_points: vec![NumberDataPoint {
399                            attributes,
400                            start_time_unix_nano: start,
401                            time_unix_nano: now_unix_nanos,
402                            value: Some(NumberValue::Double(value)),
403                        }],
404                        aggregation_temporality,
405                        is_monotonic: true,
406                    }
407                }
408            };
409            resource_metrics.push(ResourceMetrics {
410                resource: Some(self.resource.clone()),
411                scope_metrics: vec![ScopeMetrics {
412                    metrics: vec![Metric {
413                        name: key.name,
414                        data: Some(data),
415                    }],
416                }],
417            });
418        }
419        MetricsData { resource_metrics }
420    }
421}
422
423const TWO_POW_32: f64 = 4_294_967_296.0;
424
425/// The nearest `f64`, as Java's `long` to `double` conversion gives it.
426fn u64_to_f64(value: u64) -> f64 {
427    let high = u32::try_from(value >> 32).unwrap_or(u32::MAX);
428    let low = u32::try_from(value & 0xffff_ffff).unwrap_or(u32::MAX);
429    f64::from(high).mul_add(TWO_POW_32, f64::from(low))
430}
431
432fn i64_to_f64(value: i64) -> f64 {
433    let high = i32::try_from(value >> 32).unwrap_or(0);
434    let low = u32::try_from(value & 0xffff_ffff).unwrap_or(0);
435    f64::from(high).mul_add(TWO_POW_32, f64::from(low))
436}
437
438#[cfg(test)]
439mod tests {
440    use assert2::assert;
441
442    use super::*;
443
444    #[test]
445    fn metric_keys_follow_kafkas_naming_convention() {
446        let cases = [
447            (
448                ("producer-metrics", "record-send-rate", vec![]),
449                "org.apache.kafka.producer.record.send.rate",
450                vec![],
451            ),
452            (
453                ("admin-client-metrics", "request-total", vec![]),
454                "org.apache.kafka.admin.client.request.total",
455                vec![],
456            ),
457            (
458                (
459                    "consumer-fetch-manager-metrics",
460                    "Records-Lag-Max",
461                    vec![("client-id", "c"), ("topic", "t")],
462                ),
463                "org.apache.kafka.consumer.fetch.manager.records.lag.max",
464                vec![("client_id", "c"), ("topic", "t")],
465            ),
466            (
467                (
468                    "producer-node-metrics",
469                    "request-latency-avg",
470                    vec![("node-id", "node-1")],
471                ),
472                "org.apache.kafka.producer.node.request.latency.avg",
473                vec![("node_id", "node-1")],
474            ),
475        ];
476        for ((group, name, tags), expected_name, expected_tags) in cases {
477            let expected = MetricKey {
478                name: expected_name.to_owned(),
479                tags: expected_tags
480                    .into_iter()
481                    .map(|(key, value)| (key.to_owned(), value.to_owned()))
482                    .collect(),
483            };
484            assert!(MetricKey::kafka(group, name, tags) == expected);
485        }
486    }
487
488    #[test]
489    fn selector_follows_the_requested_metrics() {
490        let name = "org.apache.kafka.producer.request.total";
491        let cases: [(&[&str], MetricSelector, bool); 5] = [
492            (&[], MetricSelector::NoMetrics, false),
493            (&["*"], MetricSelector::AllMetrics, true),
494            (
495                &["org.apache.kafka.producer"],
496                MetricSelector::Prefixes(vec!["org.apache.kafka.producer".into()]),
497                true,
498            ),
499            (
500                &["org.apache.kafka.consumer", "*"],
501                MetricSelector::Prefixes(vec!["org.apache.kafka.consumer".into(), "*".into()]),
502                false,
503            ),
504            (&[""], MetricSelector::Prefixes(vec![String::new()]), true),
505        ];
506        for (requested, expected, matches) in cases {
507            let requested = requested
508                .iter()
509                .map(|s| (*s).to_owned())
510                .collect::<Vec<_>>();
511            let selector = MetricSelector::from_requested(&requested);
512            assert!(selector == expected, "{requested:?}");
513            assert!(selector.matches(name) == matches, "{requested:?}");
514        }
515    }
516
517    #[test]
518    fn numbers_convert_to_the_nearest_double() {
519        for value in [0u64, 1, 42, u64::from(u32::MAX) + 1, 1 << 53, u64::MAX] {
520            assert!(
521                u64_to_f64(value).to_bits() == format!("{value}").parse::<f64>().unwrap().to_bits()
522            );
523        }
524        for value in [0i64, 1, -1, i64::from(i32::MIN) - 5, i64::MAX, i64::MIN] {
525            assert!(
526                i64_to_f64(value).to_bits() == format!("{value}").parse::<f64>().unwrap().to_bits()
527            );
528        }
529    }
530
531    fn sum_point(
532        name: &str,
533        start: u64,
534        time: u64,
535        value: f64,
536        aggregation_temporality: AggregationTemporality,
537    ) -> ResourceMetrics {
538        ResourceMetrics {
539            resource: Some(Resource {
540                attributes: vec![KeyValue::new("transactional_id", "tx")],
541            }),
542            scope_metrics: vec![ScopeMetrics {
543                metrics: vec![Metric {
544                    name: name.to_owned(),
545                    data: Some(MetricData::Sum {
546                        data_points: vec![NumberDataPoint {
547                            attributes: vec![KeyValue::new("topic", "t")],
548                            start_time_unix_nano: start,
549                            time_unix_nano: time,
550                            value: Some(NumberValue::Double(value)),
551                        }],
552                        aggregation_temporality,
553                        is_monotonic: true,
554                    }),
555                }],
556            }],
557        }
558    }
559
560    /// A counter at 5 then 12: cumulative pushes carry the total from the
561    /// registration, delta pushes the change since the last push, and a
562    /// temporality change starts the sums again.
563    #[test]
564    fn collector_pushes_sums_with_the_subscribed_temporality() {
565        let metrics = ClientMetrics::new();
566        let key = MetricKey::kafka(
567            "producer-topic-metrics",
568            "record-send-total",
569            [("client-id", "p"), ("topic", "t")],
570        );
571        let name = key.name.clone();
572        let counter = metrics.counter(key.clone());
573        let added = metrics.lock()[&key].added_unix_nanos;
574        let mut collector = Collector::new(
575            metrics,
576            Resource {
577                attributes: vec![KeyValue::new("transactional_id", "tx")],
578            },
579        );
580        let all = MetricSelector::AllMetrics;
581        let mut pushes = Vec::new();
582        counter.add(5);
583        pushes.push(collector.collect(&all, false, 100));
584        counter.add(7);
585        pushes.push(collector.collect(&all, false, 200));
586        pushes.push(collector.collect(&all, true, 300));
587        counter.add(3);
588        pushes.push(collector.collect(&all, true, 400));
589        collector.reset();
590        pushes.push(collector.collect(&all, false, 500));
591        pushes.push(collector.collect(&MetricSelector::NoMetrics, false, 600));
592
593        let one = |point| MetricsData {
594            resource_metrics: vec![point],
595        };
596        let cumulative = AggregationTemporality::Cumulative;
597        let delta = AggregationTemporality::Delta;
598        assert!(
599            pushes
600                == [
601                    one(sum_point(&name, added, 100, 5.0, cumulative)),
602                    one(sum_point(&name, added, 200, 12.0, cumulative)),
603                    one(sum_point(&name, added, 300, 12.0, delta)),
604                    one(sum_point(&name, 300, 400, 3.0, delta)),
605                    one(sum_point(&name, 500, 500, 15.0, cumulative)),
606                    MetricsData::default(),
607                ]
608        );
609    }
610
611    #[test]
612    fn network_metrics_count_connections_requests_and_bytes() {
613        let metrics = ClientMetrics::new();
614        let network = NetworkMetrics::register(&metrics, "consumer-metrics");
615        let first = network.opened();
616        let second = network.opened();
617        first.request(96);
618        first.response(20);
619        second.request(10);
620        drop(first);
621
622        let mut collector = Collector::new(metrics, Resource::default());
623        let values = collector
624            .collect(&MetricSelector::AllMetrics, false, 1)
625            .resource_metrics
626            .into_iter()
627            .flat_map(|resource| resource.scope_metrics)
628            .flat_map(|scope| scope.metrics)
629            .map(|metric| {
630                let value = match metric.data {
631                    Some(
632                        MetricData::Gauge { data_points } | MetricData::Sum { data_points, .. },
633                    ) => data_points[0].value,
634                    None => None,
635                };
636                (metric.name, value)
637            })
638            .collect::<Vec<_>>();
639        let expected = [
640            ("connection.close.total", 1.0),
641            ("connection.count", 1.0),
642            ("connection.creation.total", 2.0),
643            ("incoming.byte.total", 28.0),
644            ("outgoing.byte.total", 114.0),
645            ("request.total", 2.0),
646            ("response.total", 1.0),
647        ]
648        .map(|(name, value)| {
649            (
650                format!("org.apache.kafka.consumer.{name}"),
651                Some(NumberValue::Double(value)),
652            )
653        });
654        assert!(values == expected);
655    }
656}
krabka_client_core/telemetry/metrics.rs

krabka_client_core/telemetry/
metrics.rs
