mz_storage/metrics/sink/
kafka.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Metrics for kafka sinks.
11
12use mz_ore::metric;
13use mz_ore::metrics::{
14    DeregisterOnDropGauge, DeregisterOnDropVec, MetricTag, MetricVisibility, MetricsRegistry, raw,
15};
16use mz_repr::GlobalId;
17use prometheus::core::{AtomicI64, AtomicU64};
18
19/// Definitions for librdkafka produced metrics used in sinks.
20#[derive(Clone, Debug)]
21pub(crate) struct KafkaSinkMetricDefs {
22    /// The current number of messages in producer queues.
23    pub rdkafka_msg_cnt: DeregisterOnDropVec<raw::UIntGaugeVec>,
24    /// The current total size of messages in producer queues.
25    pub rdkafka_msg_size: DeregisterOnDropVec<raw::UIntGaugeVec>,
26    /// The total number of messages transmitted (produced) to brokers.
27    pub rdkafka_txmsgs: DeregisterOnDropVec<raw::IntGaugeVec>,
28    /// The total number of bytes transmitted (produced) to brokers.
29    pub rdkafka_txmsg_bytes: DeregisterOnDropVec<raw::IntGaugeVec>,
30    /// The total number of requests sent to brokers.
31    pub rdkafka_tx: DeregisterOnDropVec<raw::IntGaugeVec>,
32    /// The total number of bytes transmitted to brokers.
33    pub rdkafka_tx_bytes: DeregisterOnDropVec<raw::IntGaugeVec>,
34    /// The number of requests awaiting transmission across all brokers.
35    pub rdkafka_outbuf_cnt: DeregisterOnDropVec<raw::IntGaugeVec>,
36    /// The number of messages awaiting transmission across all brokers.
37    pub rdkafka_outbuf_msg_cnt: DeregisterOnDropVec<raw::IntGaugeVec>,
38    /// The number of requests in-flight across all brokers that are awaiting a response.
39    pub rdkafka_waitresp_cnt: DeregisterOnDropVec<raw::IntGaugeVec>,
40    /// The number of messages in-flight across all brokers that are awaiting a response.
41    pub rdkafka_waitresp_msg_cnt: DeregisterOnDropVec<raw::IntGaugeVec>,
42    /// The total number of transmission errors across all brokers.
43    pub rdkafka_txerrs: DeregisterOnDropVec<raw::UIntGaugeVec>,
44    /// The total number of request retries across all brokers.
45    pub rdkafka_txretries: DeregisterOnDropVec<raw::UIntGaugeVec>,
46    /// The total number of requests that timed out across all brokers.
47    pub rdkafka_req_timeouts: DeregisterOnDropVec<raw::UIntGaugeVec>,
48    /// The number of connection attempts, including successful and failed attempts, and name
49    /// resolution failures across all brokers.
50    pub rdkafka_connects: DeregisterOnDropVec<raw::IntGaugeVec>,
51    /// The number of disconnections, whether triggered by the broker, the network, the load
52    /// balancer, or something else across all brokers.
53    pub rdkafka_disconnects: DeregisterOnDropVec<raw::IntGaugeVec>,
54    /// The number of outstanding progress records that need to be read before the sink can resume.
55    pub outstanding_progress_records: DeregisterOnDropVec<raw::UIntGaugeVec>,
56    /// The number of progress records consumed while resuming the sink.
57    pub consumed_progress_records: DeregisterOnDropVec<raw::UIntGaugeVec>,
58    /// The number of partitions this sink is publishing to.
59    pub partition_count: DeregisterOnDropVec<raw::UIntGaugeVec>,
60}
61
62impl KafkaSinkMetricDefs {
63    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
64        Self {
65            rdkafka_msg_cnt: registry.register(metric!(
66                name: "mz_sink_rdkafka_msg_cnt",
67                help: "The current number of messages in producer queues.",
68                var_labels: ["sink_id"],
69            )),
70            rdkafka_msg_size: registry.register(metric!(
71                name: "mz_sink_rdkafka_msg_size",
72                help: "The current total size of messages in producer queues.",
73                var_labels: ["sink_id"],
74            )),
75            rdkafka_txmsgs: registry.register(metric!(
76                name: "mz_sink_rdkafka_txmsgs",
77                help: "The total number of messages transmitted (produced) to brokers.",
78                var_labels: ["sink_id"],
79            )),
80            rdkafka_txmsg_bytes: registry.register(metric!(
81                name: "mz_sink_rdkafka_txmsg_bytes",
82                help: "The total number of bytes transmitted (produced) to brokers.",
83                var_labels: ["sink_id"],
84            )),
85            rdkafka_tx: registry.register(metric!(
86                name: "mz_sink_rdkafka_tx",
87                help: "The total number of requests sent to brokers.",
88                var_labels: ["sink_id"],
89            )),
90            rdkafka_tx_bytes: registry.register(metric!(
91                name: "mz_sink_rdkafka_tx_bytes",
92                help: "The total number of bytes transmitted to brokers.",
93                var_labels: ["sink_id"],
94            )),
95            rdkafka_outbuf_cnt: registry.register(metric!(
96                name: "mz_sink_rdkafka_outbuf_cnt",
97                help: "The number of requests awaiting transmission across all brokers.",
98                var_labels: ["sink_id"],
99            )),
100            rdkafka_outbuf_msg_cnt: registry.register(metric!(
101                name: "mz_sink_rdkafka_outbuf_msg_cnt",
102                help: "The number of messages awaiting transmission across all brokers.",
103                var_labels: ["sink_id"],
104                visibility: MetricVisibility::Public,
105                tags: [MetricTag::Sink],
106            )),
107            rdkafka_waitresp_cnt: registry.register(metric!(
108                name: "mz_sink_rdkafka_waitresp_cnt",
109                help: "The number of requests in-flight across all brokers that are awaiting a \
110                       response.",
111                var_labels: ["sink_id"],
112            )),
113            rdkafka_waitresp_msg_cnt: registry.register(metric!(
114                name: "mz_sink_rdkafka_waitresp_msg_cnt",
115                help: "The number of messages in-flight across all brokers that are awaiting a \
116                       response.",
117                var_labels: ["sink_id"],
118            )),
119            rdkafka_txerrs: registry.register(metric!(
120                name: "mz_sink_rdkafka_txerrs",
121                help: "The total number of transmission errors across all brokers.",
122                var_labels: ["sink_id"],
123                visibility: MetricVisibility::Public,
124                tags: [MetricTag::Sink],
125            )),
126            rdkafka_txretries: registry.register(metric!(
127                name: "mz_sink_rdkafka_txretries",
128                help: "The total number of request retries across all brokers.",
129                var_labels: ["sink_id"],
130            )),
131            rdkafka_req_timeouts: registry.register(metric!(
132                name: "mz_sink_rdkafka_req_timeouts",
133                help: "The total number of requests that timed out across all brokers.",
134                var_labels: ["sink_id"],
135            )),
136            rdkafka_connects: registry.register(metric!(
137                name: "mz_sink_rdkafka_connects",
138                help: "The number of connection attempts, including successful and failed \
139                       attempts, and name resolution failures across all brokers.",
140                var_labels: ["sink_id"],
141                visibility: MetricVisibility::Public,
142                tags: [MetricTag::Sink],
143            )),
144            rdkafka_disconnects: registry.register(metric!(
145                name: "mz_sink_rdkafka_disconnects",
146                help: "The number of disconnections, whether triggered by the broker, the \
147                      network, the load balancer, or something else across all brokers.",
148                var_labels: ["sink_id"],
149                visibility: MetricVisibility::Public,
150                tags: [MetricTag::Sink],
151            )),
152            outstanding_progress_records: registry.register(metric!(
153                name: "mz_sink_oustanding_progress_records",
154                help: "The number of outstanding progress records that need to be read before the sink can resume.",
155                var_labels: ["sink_id"],
156            )),
157            consumed_progress_records: registry.register(metric!(
158                name: "mz_sink_consumed_progress_records",
159                help: "The number of progress records consumed by the sink.",
160                var_labels: ["sink_id"],
161            )),
162            partition_count: registry.register(metric!(
163                name: "mz_sink_partition_count",
164                help: "The number of partitions this sink is publishing to.",
165                var_labels: ["sink_id"],
166            )),
167        }
168    }
169}
170
171/// Metrics reported by librdkafka
172pub(crate) struct KafkaSinkMetrics {
173    /// The current number of messages in producer queues.
174    pub rdkafka_msg_cnt: DeregisterOnDropGauge<AtomicU64>,
175    /// The current total size of messages in producer queues.
176    pub rdkafka_msg_size: DeregisterOnDropGauge<AtomicU64>,
177    /// The total number of messages transmitted (produced) to brokers.
178    pub rdkafka_txmsgs: DeregisterOnDropGauge<AtomicI64>,
179    /// The total number of bytes transmitted (produced) to brokers.
180    pub rdkafka_txmsg_bytes: DeregisterOnDropGauge<AtomicI64>,
181    /// The total number of requests sent to brokers.
182    pub rdkafka_tx: DeregisterOnDropGauge<AtomicI64>,
183    /// The total number of bytes transmitted to brokers.
184    pub rdkafka_tx_bytes: DeregisterOnDropGauge<AtomicI64>,
185    /// The number of requests awaiting transmission across all brokers.
186    pub rdkafka_outbuf_cnt: DeregisterOnDropGauge<AtomicI64>,
187    /// The number of messages awaiting transmission across all brokers.
188    pub rdkafka_outbuf_msg_cnt: DeregisterOnDropGauge<AtomicI64>,
189    /// The number of requests in-flight across all brokers that are awaiting a response.
190    pub rdkafka_waitresp_cnt: DeregisterOnDropGauge<AtomicI64>,
191    /// The number of messages in-flight across all brokers that are awaiting a response.
192    pub rdkafka_waitresp_msg_cnt: DeregisterOnDropGauge<AtomicI64>,
193    /// The total number of transmission errors across all brokers.
194    pub rdkafka_txerrs: DeregisterOnDropGauge<AtomicU64>,
195    /// The total number of request retries across all brokers.
196    pub rdkafka_txretries: DeregisterOnDropGauge<AtomicU64>,
197    /// The total number of requests that timed out across all brokers.
198    pub rdkafka_req_timeouts: DeregisterOnDropGauge<AtomicU64>,
199    /// The number of connection attempts, including successful and failed attempts, and name
200    /// resolution failures across all brokers.
201    pub rdkafka_connects: DeregisterOnDropGauge<AtomicI64>,
202    /// The number of disconnections, whether triggered by the broker, the network, the load
203    /// balancer, or something else across all brokers.
204    pub rdkafka_disconnects: DeregisterOnDropGauge<AtomicI64>,
205    /// The number of outstanding progress records that need to be read before the sink can resume.
206    pub outstanding_progress_records: DeregisterOnDropGauge<AtomicU64>,
207    /// The number of progress records consumed while resuming the sink.
208    pub consumed_progress_records: DeregisterOnDropGauge<AtomicU64>,
209    /// The number of partitions this sink is publishing to.
210    pub partition_count: DeregisterOnDropGauge<AtomicU64>,
211}
212
213impl KafkaSinkMetrics {
214    /// Initializes source metrics for a given (source_id, worker_id)
215    pub fn new(defs: &KafkaSinkMetricDefs, sink_id: GlobalId) -> Self {
216        let labels = &[sink_id.to_string()];
217        Self {
218            rdkafka_msg_cnt: defs
219                .rdkafka_msg_cnt
220                .get_deregister_on_drop_metric(labels.to_vec()),
221            rdkafka_msg_size: defs
222                .rdkafka_msg_size
223                .get_deregister_on_drop_metric(labels.to_vec()),
224            rdkafka_txmsgs: defs
225                .rdkafka_txmsgs
226                .get_deregister_on_drop_metric(labels.to_vec()),
227            rdkafka_txmsg_bytes: defs
228                .rdkafka_txmsg_bytes
229                .get_deregister_on_drop_metric(labels.to_vec()),
230            rdkafka_tx: defs
231                .rdkafka_tx
232                .get_deregister_on_drop_metric(labels.to_vec()),
233            rdkafka_tx_bytes: defs
234                .rdkafka_tx_bytes
235                .get_deregister_on_drop_metric(labels.to_vec()),
236            rdkafka_outbuf_cnt: defs
237                .rdkafka_outbuf_cnt
238                .get_deregister_on_drop_metric(labels.to_vec()),
239            rdkafka_outbuf_msg_cnt: defs
240                .rdkafka_outbuf_msg_cnt
241                .get_deregister_on_drop_metric(labels.to_vec()),
242            rdkafka_waitresp_cnt: defs
243                .rdkafka_waitresp_cnt
244                .get_deregister_on_drop_metric(labels.to_vec()),
245            rdkafka_waitresp_msg_cnt: defs
246                .rdkafka_waitresp_msg_cnt
247                .get_deregister_on_drop_metric(labels.to_vec()),
248            rdkafka_txerrs: defs
249                .rdkafka_txerrs
250                .get_deregister_on_drop_metric(labels.to_vec()),
251            rdkafka_txretries: defs
252                .rdkafka_txretries
253                .get_deregister_on_drop_metric(labels.to_vec()),
254            rdkafka_req_timeouts: defs
255                .rdkafka_req_timeouts
256                .get_deregister_on_drop_metric(labels.to_vec()),
257            rdkafka_connects: defs
258                .rdkafka_connects
259                .get_deregister_on_drop_metric(labels.to_vec()),
260            rdkafka_disconnects: defs
261                .rdkafka_disconnects
262                .get_deregister_on_drop_metric(labels.to_vec()),
263            outstanding_progress_records: defs
264                .outstanding_progress_records
265                .get_deregister_on_drop_metric(labels.to_vec()),
266            consumed_progress_records: defs
267                .consumed_progress_records
268                .get_deregister_on_drop_metric(labels.to_vec()),
269            partition_count: defs
270                .partition_count
271                .get_deregister_on_drop_metric(labels.to_vec()),
272        }
273    }
274}
mz_storage/metrics/sink/kafka.rs

mz_storage/metrics/sink/
kafka.rs
