feat(kafka source): Add topic and partition labels to consumed messages metrics

changelog.d/21330_kafka_topic_partition_labels.enhancement.md

@@ -0,0 +1,9 @@
+`kafka` sink/source: The `kafka_consumed_messages_total`, `kafka_consumed_messages_bytes_total`,
+`kafka_produced_messages_total`, and `kafka_produced_messages_bytes_total` metrics now include
+`topic` and `partition` labels, allowing users to monitor both consumption and production metrics
+per topic and partition when multiple topics are configured.
+
+The labels on the non-default `kafka_consumer_lag` metric has also had its labels updated for
+consistency.
+
+authors: jpds

src/internal_events/kafka.rs

@@ -129,20 +129,40 @@ impl InternalEvent for KafkaStatisticsReceived<'_> {
         counter!("kafka_requests_bytes_total").absolute(self.statistics.tx_bytes as u64);
         counter!("kafka_responses_total").absolute(self.statistics.rx as u64);
         counter!("kafka_responses_bytes_total").absolute(self.statistics.rx_bytes as u64);
-        counter!("kafka_produced_messages_total").absolute(self.statistics.txmsgs as u64);
-        counter!("kafka_produced_messages_bytes_total")
-            .absolute(self.statistics.txmsg_bytes as u64);
-        counter!("kafka_consumed_messages_total").absolute(self.statistics.rxmsgs as u64);
-        counter!("kafka_consumed_messages_bytes_total")
-            .absolute(self.statistics.rxmsg_bytes as u64);
 
-        if self.expose_lag_metrics {
-            for (topic_id, topic) in &self.statistics.topics {
-                for (partition_id, partition) in &topic.partitions {
+        // Emit per-topic, per-partition message metrics
+        for (topic_id, topic) in &self.statistics.topics {
+            for (partition_id, partition) in &topic.partitions {
+                counter!(
+                    "kafka_produced_messages_total",
+                    "topic" => topic_id.clone(),
+                    "partition" => partition_id.to_string(),
+                )
+                .absolute(partition.txmsgs);
+                counter!(
+                    "kafka_produced_messages_bytes_total",
+                    "topic" => topic_id.clone(),
+                    "partition" => partition_id.to_string(),
+                )
+                .absolute(partition.txbytes);
+                counter!(
+                    "kafka_consumed_messages_total",
+                    "topic" => topic_id.clone(),
+                    "partition" => partition_id.to_string(),
+                )
+                .absolute(partition.rxmsgs);
+                counter!(
+                    "kafka_consumed_messages_bytes_total",
+                    "topic" => topic_id.clone(),
+                    "partition" => partition_id.to_string(),
+                )
+                .absolute(partition.rxbytes);
+
+                if self.expose_lag_metrics {
                     gauge!(
                         "kafka_consumer_lag",
-                        "topic_id" => topic_id.clone(),
-                        "partition_id" => partition_id.to_string(),
+                        "topic" => topic_id.clone(),
+                        "partition" => partition_id.to_string(),
                     )
                     .set(partition.consumer_lag as f64);
                 }

website/cue/reference/components/sources/internal_metrics.cue

@@ -559,37 +559,73 @@ components: sources: internal_metrics: {
 			description:       "Total number of messages transmitted (produced) to Kafka brokers."
 			type:              "counter"
 			default_namespace: "vector"
-			tags:              _component_tags
+			tags: _component_tags & {
+				topic: {
+					description: "The Kafka topic."
+					required:    true
+				}
+				partition: {
+					description: "The Kafka partition."
+					required:    true
+				}
+			}
 		}
 		kafka_produced_messages_bytes_total: {
 			description:       "Total number of message bytes (including framing, such as per-Message framing and MessageSet/batch framing) transmitted to Kafka brokers."
 			type:              "counter"
 			default_namespace: "vector"
-			tags:              _component_tags
+			tags: _component_tags & {
+				topic: {
+					description: "The Kafka topic."
+					required:    true
+				}
+				partition: {
+					description: "The Kafka partition."
+					required:    true
+				}
+			}
 		}
 		kafka_consumed_messages_total: {
 			description:       "Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers."
 			type:              "counter"
 			default_namespace: "vector"
-			tags:              _component_tags
+			tags: _component_tags & {
+				topic: {
+					description: "The Kafka topic."
+					required:    true
+				}
+				partition: {
+					description: "The Kafka partition."
+					required:    true
+				}
+			}
 		}
 		kafka_consumed_messages_bytes_total: {
 			description:       "Total number of message bytes (including framing) received from Kafka brokers."
 			type:              "counter"
 			default_namespace: "vector"
-			tags:              _component_tags
+			tags: _component_tags & {
+				topic: {
+					description: "The Kafka topic."
+					required:    true
+				}
+				partition: {
+					description: "The Kafka partition."
+					required:    true
+				}
+			}
 		}
 		kafka_consumer_lag: {
 			description:       "The Kafka consumer lag."
 			type:              "gauge"
 			default_namespace: "vector"
 			tags: _component_tags & {
-				topic_id: {
-					description: "The Kafka topic id."
+				topic: {
+					description: "The Kafka topic."
 					required:    true
 				}
-				partition_id: {
-					description: "The Kafka partition id."
+				partition: {
+					description: "The Kafka partition."
 					required:    true
 				}
 			}