ref(kafka): Add producer_name tag, and more rdkafka stats (#5031)

ref STREAM-324

* The producer_name tag identifies a particular producer instance in
  Relay. It's almost identical to broker_name, but broker_name is not
  generally available because in theory one producer can talk to many
  brokers (though relay doesn't do that), and some statistics are
  related to the producer's internal datastructures and don't have that
  tag.

  Also, we're going to add this tag to other producers in other
  services (and we have _many_ producers in getsentry talking to a
  single broker there), so it pays off to be consistent. With a bit
  of luck we might even have a proper generic dashboard for all of
  our producers in our entire infrastructure..
  someday.

* Add a few metrics that we think might be useful to see if Relay is
  impacted by a broker being unhealthy.

Aside, it would be really nice if Relay moved to rust-arroyo for
producing. I'm implementing the same metrics there. Or maybe not its
entire producer wrapper but at least reusing its stats callback.

Author: untitaker

CHANGELOG.md

@@ -32,6 +32,7 @@
 - Add Jwm to the supported image types. ([#4975](https://github.com/getsentry/relay/pull/4975))
 - Process logs in all non-proxy Relays. ([#4973](https://github.com/getsentry/relay/pull/4973))
 - Add support for pre-hashed signatures. ([#5012](https://github.com/getsentry/relay/pull/5012))
+- Add producer_name tag, and more rdkafka stats. ([#5031](https://github.com/getsentry/relay/pull/5031))
 
 ## 25.7.0
 

relay-kafka/src/producer/mod.rs

@@ -214,10 +214,13 @@ impl Producer {
             return Err(ClientError::MissingTopic);
         };
 
+        let producer_name = producer.context().producer_name();
+
         metric!(
             histogram(KafkaHistograms::KafkaMessageSize) = payload.len() as u64,
             variant = variant,
             topic = topic_name,
+            producer_name = producer_name
         );
 
         let mut headers = headers
@@ -240,6 +243,7 @@ impl Producer {
                     counter(KafkaCounters::ProducerPartitionKeyRateLimit) += 1,
                     variant = variant,
                     topic = topic_name,
+                    producer_name = producer_name
                 );
 
                 headers.insert(Header {
@@ -262,7 +266,8 @@ impl Producer {
             metric!(
                 gauge(KafkaGauges::InFlightCount) = producer.in_flight_count() as u64,
                 variant = variant,
-                topic = topic_name
+                topic = topic_name,
+                producer_name = producer_name
             );
         });
 
@@ -276,7 +281,8 @@ impl Producer {
             metric!(
                 counter(KafkaCounters::ProducerEnqueueError) += 1,
                 variant = variant,
-                topic = topic_name
+                topic = topic_name,
+                producer_name = producer_name
             );
             ClientError::SendFailed(error)
         })?;
@@ -433,9 +439,17 @@ impl KafkaClientBuilder {
                     client_config.set(config_p.name.as_str(), config_p.value.as_str());
                 }
 
+                // Extract producer name from client.id, fallback to config name, then "unknown"
+                let producer_name = config_params
+                    .iter()
+                    .find(|p| p.name == "client.id")
+                    .map(|p| p.value.clone())
+                    .or_else(|| config_name.clone())
+                    .unwrap_or_else(|| "unknown".to_owned());
+
                 let producer = Arc::new(
                     client_config
-                        .create_with_context(Context)
+                        .create_with_context(Context::new(producer_name))
                         .map_err(ClientError::InvalidConfig)?,
                 );
 

relay-kafka/src/producer/utils.rs

@@ -74,51 +74,120 @@ where
 
 /// Kafka client and producer context that logs statistics and producer errors.
 #[derive(Debug)]
-pub struct Context;
+pub struct Context {
+    /// Producer name for deployment identification
+    producer_name: String,
+}
+
+impl Context {
+    pub fn new(producer_name: String) -> Self {
+        Self { producer_name }
+    }
+
+    pub fn producer_name(&self) -> &str {
+        &self.producer_name
+    }
+}
 
 impl ClientContext for Context {
     /// Report client statistics as statsd metrics.
     ///
     /// This method is only called if `statistics.interval.ms` is configured.
     fn stats(&self, statistics: rdkafka::Statistics) {
-        relay_statsd::metric!(gauge(KafkaGauges::MessageCount) = statistics.msg_cnt);
-        relay_statsd::metric!(gauge(KafkaGauges::MessageCountMax) = statistics.msg_max);
-        relay_statsd::metric!(gauge(KafkaGauges::MessageSize) = statistics.msg_size);
-        relay_statsd::metric!(gauge(KafkaGauges::MessageSizeMax) = statistics.msg_size_max);
+        let producer_name = &self.producer_name;
+
+        relay_statsd::metric!(
+            gauge(KafkaGauges::MessageCount) = statistics.msg_cnt,
+            producer_name = producer_name
+        );
+        relay_statsd::metric!(
+            gauge(KafkaGauges::MessageCountMax) = statistics.msg_max,
+            producer_name = producer_name
+        );
+        relay_statsd::metric!(
+            gauge(KafkaGauges::MessageSize) = statistics.msg_size,
+            producer_name = producer_name
+        );
+        relay_statsd::metric!(
+            gauge(KafkaGauges::MessageSizeMax) = statistics.msg_size_max,
+            producer_name = producer_name
+        );
+        relay_statsd::metric!(
+            gauge(KafkaGauges::TxMsgs) = statistics.txmsgs as u64,
+            producer_name = producer_name
+        );
 
         for (_, broker) in statistics.brokers {
             relay_statsd::metric!(
                 gauge(KafkaGauges::OutboundBufferRequests) = broker.outbuf_cnt as u64,
-                broker_name = &broker.name
+                broker_name = &broker.name,
+                producer_name = producer_name
             );
             relay_statsd::metric!(
                 gauge(KafkaGauges::OutboundBufferMessages) = broker.outbuf_msg_cnt as u64,
-                broker_name = &broker.name
+                broker_name = &broker.name,
+                producer_name = producer_name
             );
             if let Some(connects) = broker.connects {
                 relay_statsd::metric!(
                     gauge(KafkaGauges::Connects) = connects as u64,
-                    broker_name = &broker.name
+                    broker_name = &broker.name,
+                    producer_name = producer_name
                 );
             }
             if let Some(disconnects) = broker.disconnects {
                 relay_statsd::metric!(
                     gauge(KafkaGauges::Disconnects) = disconnects as u64,
-                    broker_name = &broker.name
+                    broker_name = &broker.name,
+                    producer_name = producer_name
                 );
             }
             if let Some(int_latency) = broker.int_latency {
                 relay_statsd::metric!(
-                    gauge(KafkaGauges::ProducerQueueLatency) = int_latency.max as u64,
-                    broker_name = &broker.name
+                    gauge(KafkaGauges::BrokerIntLatencyAvg) = (int_latency.avg / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
+                );
+                relay_statsd::metric!(
+                    gauge(KafkaGauges::BrokerIntLatencyP99) = (int_latency.p99 / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
                 );
             }
             if let Some(outbuf_latency) = broker.outbuf_latency {
                 relay_statsd::metric!(
-                    gauge(KafkaGauges::RequestQueueLatency) = outbuf_latency.max as u64,
-                    broker_name = &broker.name
+                    gauge(KafkaGauges::BrokerOutbufLatencyAvg) = (outbuf_latency.avg / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
+                );
+                relay_statsd::metric!(
+                    gauge(KafkaGauges::BrokerOutbufLatencyP99) = (outbuf_latency.p99 / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
+                );
+            }
+            if let Some(rtt) = broker.rtt {
+                relay_statsd::metric!(
+                    gauge(KafkaGauges::BrokerRttAvg) = (rtt.avg / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
+                );
+                relay_statsd::metric!(
+                    gauge(KafkaGauges::BrokerRttP99) = (rtt.p99 / 1000) as u64,
+                    broker_name = &broker.name,
+                    producer_name = producer_name
                 );
             }
+            relay_statsd::metric!(
+                gauge(KafkaGauges::BrokerTx) = broker.tx,
+                broker_name = &broker.name,
+                producer_name = producer_name
+            );
+            relay_statsd::metric!(
+                gauge(KafkaGauges::BrokerTxBytes) = broker.txbytes,
+                broker_name = &broker.name,
+                producer_name = producer_name
+            );
         }
     }
 }
@@ -132,18 +201,28 @@ impl ProducerContext for Context {
         // TODO: any `Accepted` outcomes (e.g. spans) should be logged here instead of on the caller side,
         // such that we do not over-report in the error case.
 
-        if let Err((error, message)) = result {
-            relay_log::error!(
-                error = error as &dyn Error,
-                payload_len = message.payload_len(),
-                tags.topic = message.topic(),
-                "failed to produce message to Kafka (delivery callback)",
-            );
+        match result {
+            Ok(message) => {
+                metric!(
+                    counter(KafkaCounters::ProduceStatusSuccess) += 1,
+                    topic = message.topic(),
+                    producer_name = &self.producer_name
+                );
+            }
+            Err((error, message)) => {
+                relay_log::error!(
+                    error = error as &dyn Error,
+                    payload_len = message.payload_len(),
+                    tags.topic = message.topic(),
+                    "failed to produce message to Kafka (delivery callback)",
+                );
 
-            metric!(
-                counter(KafkaCounters::ProcessingProduceError) += 1,
-                topic = message.topic()
-            );
+                metric!(
+                    counter(KafkaCounters::ProduceStatusError) += 1,
+                    topic = message.topic(),
+                    producer_name = &self.producer_name
+                );
+            }
         }
     }
 }