Adding metrics to the consumer_groups module

Cargo.toml

@@ -27,22 +27,22 @@ exclude = [
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-trait = "0.1.73"
+async-trait = "0.1.74"
 axum = { version = "0.6.20", features = ["http2"] }
 chrono = "0.4.31"
-clap = { version = "4.4.6", features = ["derive", "deprecated", "env", "wrap_help"] }
-const_format = "0.2.31"
+clap = { version = "4.4.7", features = ["derive", "deprecated", "env", "wrap_help"] }
+const_format = "0.2.32"
 ctrlc = { version = "3.4.1", features = ["termination"] }
 env_logger = "0.10.0"
 exit-code = "1.0.0"
 hyper = { version = "0.14.27", features = ["http1", "http2", "server", "runtime", "tcp"] }
 konsumer_offsets = { version = "0.3.0", default-features = false, features = ["ts_chrono"] }
 log = "0.4.20"
 prometheus = "0.13.3"
-regex = "1.9.6"
-thiserror = "1.0.49"
+regex = "1.10.2"
+thiserror = "1.0.50"
 tokio = { version = "1.33.0", features = ["rt", "rt-multi-thread", "time", "sync", "macros"] }
-tokio-util = "0.7.9"
+tokio-util = "0.7.10"
 
 [target.'cfg(unix)'.dependencies]
 rdkafka = { version = "0.34.0", features = ["ssl-vendored", "gssapi-vendored", "libz-static"] }

METRICS.md

@@ -87,6 +87,54 @@ Below is the list of the current Metrics exposed by Kommitted.
   </dd>
 </dl>
 
+### Cluster Metrics
+
+Those are metrics specific to the component in Kommitted that fetches the set of Consumer Groups and Members from
+the Kafka cluster, and sends it to the rest of the system for further processing (i.e. to know which consumer groups
+to produce the lag information for).
+
+<dl>
+  <dt><code>kmtd_consumer_groups_total</code></dt>
+  <dd>
+    <b>Description:</b> <i>"Consumer groups currently in the cluster.</i><br/>
+    <b>Labels:</b> <code>cluster_id</code><br/>
+    <b>Type:</b> <code>gauge</code><br/>
+    <b>Timestamped:</b> <code>true</code>
+  </dd>
+</dl>
+
+<dl>
+  <dt><code>kmtd_consumer_groups_members_total</code></dt>
+  <dd>
+    <b>Description:</b> <i>Members of consumer groups currently in the cluster.</i><br/>
+    <b>Labels:</b> <code>cluster_id, group</code><br/>
+    <b>Type:</b> <code>gauge</code><br/>
+    <b>Timestamped:</b> <code>true</code>
+  </dd>
+</dl>
+
+### Kommitted (internal) Metrics
+
+<dl>
+  <dt><code>kmtd_consumer_groups_emitter_fetch_time_milliseconds</code></dt>
+  <dd>
+    <b>Description:</b> <i>Time (in milliseconds) taken to fetch information about all consumer groups in cluster.</i><br/>
+    <b>Labels:</b> <code>cluster_id</code><br/>
+    <b>Type:</b> <code>histogram</code><br/>
+    <b>Timestamped:</b> <code>true</code>
+  </dd>
+</dl>
+
+<dl>
+  <dt><code>kmtd_consumer_groups_emitter_channel_capacity</code></dt>
+  <dd>
+    <b>Description:</b> <i>Capacity of internal channel used to send consumer groups metadata to rest of the service.</i><br/>
+    <b>Labels:</b> <code>cluster_id</code><br/>
+    <b>Type:</b> <code>gauge</code><br/>
+    <b>Timestamped:</b> <code>true</code>
+  </dd>
+</dl>
+
 ## Labels
 
 Each metrics has some or all of the following labels applied; what labels applies

src/consumer_groups/emitter.rs

@@ -1,7 +1,12 @@
-use std::collections::{HashMap, HashSet};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 use async_trait::async_trait;
+use const_format::formatcp;
 use konsumer_offsets::ConsumerProtocolAssignment;
+use prometheus::{Histogram, HistogramOpts, IntGauge, IntGaugeVec, Opts, Registry};
 use rdkafka::{admin::AdminClient, client::DefaultClientContext, groups::GroupList, ClientConfig};
 use tokio::{
     sync::mpsc,
@@ -13,12 +18,25 @@ use tokio_util::sync::CancellationToken;
 use crate::constants::KOMMITTED_CONSUMER_OFFSETS_CONSUMER;
 use crate::internals::Emitter;
 use crate::kafka_types::{Group, GroupWithMembers, Member, MemberWithAssignment, TopicPartition};
+use crate::prometheus_metrics::{LABEL_GROUP, NAMESPACE};
 
 const CHANNEL_SIZE: usize = 5;
 
 const FETCH_TIMEOUT: Duration = Duration::from_secs(10);
 const FETCH_INTERVAL: Duration = Duration::from_secs(60);
 
+const M_CG_NAME: &str = formatcp!("{NAMESPACE}_consumer_groups_total");
+const M_CG_HELP: &str = "Consumer groups currently in the cluster";
+const M_CG_MEMBERS_NAME: &str = formatcp!("{NAMESPACE}_consumer_groups_members_total");
+const M_CG_MEMBERS_HELP: &str = "Members of consumer groups currently in the cluster";
+const M_CG_FETCH_NAME: &str =
+    formatcp!("{NAMESPACE}_consumer_groups_emitter_fetch_time_milliseconds");
+const M_CG_FETCH_HELP: &str =
+    "Time (in milliseconds) taken to fetch information about all consumer groups in cluster";
+const M_CG_CH_CAP_NAME: &str = formatcp!("{NAMESPACE}_consumer_groups_emitter_channel_capacity");
+const M_CG_CH_CAP_HELP: &str =
+    "Capacity of internal channel used to send consumer groups metadata to rest of the service";
+
 /// A map of all the known Consumer Groups, at a given point in time.
 ///
 /// This reflects the internal state of Kafka and it's active Consumer Groups.
@@ -95,6 +113,12 @@ impl From<GroupList> for ConsumerGroups {
 /// It shuts down when the provided [`CancellationToken`] is cancelled.
 pub struct ConsumerGroupsEmitter {
     admin_client_config: ClientConfig,
+
+    // Prometheus Metrics
+    metric_cg: IntGauge,
+    metric_cg_members: IntGaugeVec,
+    metric_cg_fetch: Histogram,
+    metric_cg_ch_cap: IntGauge,
 }
 
 impl ConsumerGroupsEmitter {
@@ -103,9 +127,39 @@ impl ConsumerGroupsEmitter {
     /// # Arguments
     ///
     /// * `admin_client_config` - Kafka admin client configuration, used to fetch Consumer Groups
-    pub fn new(admin_client_config: ClientConfig) -> Self {
+    pub fn new(admin_client_config: ClientConfig, metrics: Arc<Registry>) -> Self {
+        // Create metrics
+        let metric_cg = IntGauge::new(M_CG_NAME, M_CG_HELP)
+            .unwrap_or_else(|_| panic!("Failed to create metric: {M_CG_NAME}"));
+        let metric_cg_members =
+            IntGaugeVec::new(Opts::new(M_CG_MEMBERS_NAME, M_CG_MEMBERS_HELP), &[LABEL_GROUP])
+                .unwrap_or_else(|_| panic!("Failed to create metric: {M_CG_MEMBERS_NAME}"));
+        let metric_cg_fetch =
+            Histogram::with_opts(HistogramOpts::new(M_CG_FETCH_NAME, M_CG_FETCH_HELP))
+                .unwrap_or_else(|_| panic!("Failed to create metric: {M_CG_FETCH_NAME}"));
+        let metric_cg_ch_cap = IntGauge::new(M_CG_CH_CAP_NAME, M_CG_CH_CAP_HELP)
+            .unwrap_or_else(|_| panic!("Failed to create metric: {M_CG_CH_CAP_NAME}"));
+
+        // Register metrics
+        metrics
+            .register(Box::new(metric_cg.clone()))
+            .unwrap_or_else(|_| panic!("Failed to register metric: {M_CG_NAME}"));
+        metrics
+            .register(Box::new(metric_cg_members.clone()))
+            .unwrap_or_else(|_| panic!("Failed to register metric: {M_CG_MEMBERS_NAME}"));
+        metrics
+            .register(Box::new(metric_cg_fetch.clone()))
+            .unwrap_or_else(|_| panic!("Failed to register metric: {M_CG_FETCH_NAME}"));
+        metrics
+            .register(Box::new(metric_cg_ch_cap.clone()))
+            .unwrap_or_else(|_| panic!("Failed to register metric: {M_CG_CH_CAP_NAME}"));
+
         Self {
             admin_client_config,
+            metric_cg,
+            metric_cg_members,
+            metric_cg_fetch,
+            metric_cg_ch_cap,
         }
     }
 }
@@ -133,19 +187,37 @@ impl Emitter for ConsumerGroupsEmitter {
 
         let (sx, rx) = mpsc::channel::<Self::Emitted>(CHANNEL_SIZE);
 
+        // Clone metrics so they can be used in the spawned future
+        let metric_cg = self.metric_cg.clone();
+        let metric_cg_members = self.metric_cg_members.clone();
+        let metric_cg_fetch = self.metric_cg_fetch.clone();
+        let metric_cg_ch_cap = self.metric_cg_ch_cap.clone();
+
         let join_handle = tokio::spawn(async move {
             let mut interval = interval(FETCH_INTERVAL);
 
             loop {
+                let timer = metric_cg_fetch.start_timer();
                 let res_groups = admin_client
                     .inner()
                     .fetch_group_list(None, FETCH_TIMEOUT)
                     .map(Self::Emitted::from);
 
+                // Update fetching time metric
+                timer.observe_duration();
+
                 match res_groups {
-                    Ok(groups) => {
+                    Ok(cg) => {
+                        // Update group and group member metrics
+                        metric_cg.set(cg.groups.len() as i64);
+                        for (g, gm) in cg.groups.iter() {
+                            metric_cg_members.with_label_values(&[&g]).set(gm.members.len() as i64);
+                        }
+                        // Update channel capacity metric
+                        metric_cg_ch_cap.set(sx.capacity() as i64);
+
                         tokio::select! {
-                            res = Self::emit_with_interval(&sx, groups, &mut interval) => {
+                            res = Self::emit_with_interval(&sx, cg, &mut interval) => {
                                 if let Err(e) = res {
                                     error!("Failed to emit {}: {e}", std::any::type_name::<ConsumerGroups>());
                                 }

src/consumer_groups/mod.rs

@@ -1,6 +1,9 @@
 // Inner module
 mod emitter;
 
+use std::sync::Arc;
+
+use prometheus::Registry;
 use rdkafka::ClientConfig;
 use tokio::sync::mpsc::Receiver;
 use tokio::task::JoinHandle;
@@ -13,8 +16,9 @@ pub use emitter::{ConsumerGroups, ConsumerGroupsEmitter};
 pub fn init(
     admin_client_config: ClientConfig,
     shutdown_token: CancellationToken,
+    metrics: Arc<Registry>,
 ) -> (Receiver<ConsumerGroups>, JoinHandle<()>) {
-    let consumer_groups_emitter = ConsumerGroupsEmitter::new(admin_client_config);
+    let consumer_groups_emitter = ConsumerGroupsEmitter::new(admin_client_config, metrics);
     let (cg_rx, cg_join) = consumer_groups_emitter.spawn(shutdown_token);
 
     debug!("Initialized");

src/http/mod.rs

@@ -100,7 +100,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
     consumer_partition_offset::append_headers(&mut body);
     iter_lag_reg(&state.lag_reg, &mut body, &cluster_id, consumer_partition_offset::append_metric)
         .await;
-    body.push(String::new());
 
     // ------------------------------------------------------- METRIC: consumer_partition_lag_offset
     consumer_partition_lag_offset::append_headers(&mut body);
@@ -111,7 +110,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
         consumer_partition_lag_offset::append_metric,
     )
     .await;
-    body.push(String::new());
 
     // ------------------------------------------------- METRIC: consumer_partition_lag_milliseconds
     consumer_partition_lag_milliseconds::append_headers(&mut body);
@@ -122,7 +120,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
         consumer_partition_lag_milliseconds::append_metric,
     )
     .await;
-    body.push(String::new());
 
     // ------------------------------------------------- METRIC: partition_earliest_available_offset
     partition_earliest_available_offset::append_headers(&mut body);
@@ -142,7 +139,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
             },
         }
     }
-    body.push(String::new());
 
     // ------------------------------------------------- METRIC: partition_latest_available_offset
     partition_latest_available_offset::append_headers(&mut body);
@@ -162,7 +158,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
             },
         }
     }
-    body.push(String::new());
 
     // ------------------------------------------------- METRIC: partition_earliest_tracked_offset
     partition_earliest_tracked_offset::append_headers(&mut body);
@@ -183,7 +178,6 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
             },
         }
     }
-    body.push(String::new());
 
     // ------------------------------------------------- METRIC: partition_latest_tracked_offset
     partition_latest_tracked_offset::append_headers(&mut body);
@@ -204,24 +198,20 @@ async fn prometheus_metrics(State(state): State<HttpServiceState>) -> impl IntoR
             },
         }
     }
-    body.push(String::new());
 
-    //
     // --- CLUSTER METRICS ---
     //
-    // TODO https://github.com/kafkesc/kommitted/issues/53
     // TODO https://github.com/kafkesc/kommitted/issues/54
-    //
+
     // --- KOMMITTED INTERNAL METRICS ---
     //
-    // TODO https://github.com/kafkesc/kommitted/issues/55
     // TODO https://github.com/kafkesc/kommitted/issues/56
     // TODO https://github.com/kafkesc/kommitted/issues/57
 
     // Turn the bespoke metrics created so far, into a String
     let mut body = body.join("\n");
 
-    // Append the the bespoke metrics, internal (normal?) Prometheus Metrics
+    // Append to the bespoke metrics, classic Prometheus Metrics
     let metrics_family = state.metrics.gather();
     if let Err(e) = TextEncoder.encode_utf8(&metrics_family, &mut body) {
         status = StatusCode::INTERNAL_SERVER_ERROR;

src/internals/emitter.rs

@@ -56,7 +56,6 @@ pub trait Emitter {
 
         // TODO Each `Emitter` implementation should report a metric about
         //   the current saturation of its emitting channel.
-        //   See https://github.com/kafkesc/kommitted/issues/55
         //   See https://github.com/kafkesc/kommitted/issues/56
         //   See https://github.com/kafkesc/kommitted/issues/57
 

src/main.rs

@@ -37,6 +37,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
     cs_reg.await_ready(shutdown_token.clone()).await?;
     let cs_reg_arc = Arc::new(cs_reg);
 
+    // Init `prometheus_metrics` module
+    let prom_reg = prometheus_metrics::init(cs_reg_arc.get_cluster_id().await);
+    let prom_reg_arc = Arc::new(prom_reg);
+
     // Init `partition_offsets` module, and await registry to be ready
     let (po_reg, po_join) = partition_offsets::init(
         admin_client_config.clone(),
@@ -53,25 +57,25 @@ async fn main() -> Result<(), Box<dyn Error>> {
         konsumer_offsets_data::init(admin_client_config.clone(), shutdown_token.clone());
 
     // Init `consumer_groups` module
-    let (cg_rx, cg_join) =
-        consumer_groups::init(admin_client_config.clone(), shutdown_token.clone());
+    let (cg_rx, cg_join) = consumer_groups::init(
+        admin_client_config.clone(),
+        shutdown_token.clone(),
+        prom_reg_arc.clone(),
+    );
 
     // Init `lag_register` module, and await registry to be ready
     let lag_reg = lag_register::init(cg_rx, kod_rx, po_reg_arc.clone());
     lag_reg.await_ready(shutdown_token.clone()).await?;
     let lag_reg_arc = Arc::new(lag_reg);
 
-    // Init `prometheus_metrics` module
-    let prom_reg = prometheus_metrics::init(cs_reg_arc.get_cluster_id().await);
-
     // Init `http` module
     let http_fut = http::init(
         cli.listen_on(),
         cs_reg_arc.clone(),
         po_reg_arc.clone(),
         lag_reg_arc.clone(),
         shutdown_token.clone(),
-        Arc::new(prom_reg),
+        prom_reg_arc.clone(),
     );
 
     // Join all the async tasks, then let it terminate

src/prometheus_metrics/mod.rs

@@ -18,6 +18,7 @@ pub const UNKNOWN_VAL: &str = "UNKNOWN";
 
 pub fn init(cluster_id: String) -> Registry {
     let prom_def_labels = HashMap::from([(LABEL_CLUSTER_ID.to_string(), cluster_id)]);
+
     Registry::new_custom(Some(NAMESPACE.to_string()), Some(prom_def_labels))
         .expect("Unable to create a Prometheus Metrics Registry")
 }