feat(kafka): Report producer errors (#677)

relay-server/src/actors/outcome.rs

@@ -338,7 +338,7 @@ mod processing {
 
     use failure::{Fail, ResultExt};
     use rdkafka::error::KafkaError;
-    use rdkafka::producer::{BaseRecord, DefaultProducerContext};
+    use rdkafka::producer::BaseRecord;
     use rdkafka::ClientConfig;
     use serde_json::Error as SerdeSerializationError;
 
@@ -347,8 +347,7 @@ mod processing {
 
     use crate::metrics::RelayCounters;
     use crate::service::ServerErrorKind;
-
-    type ThreadedProducer = rdkafka::producer::ThreadedProducer<DefaultProducerContext>;
+    use crate::utils::{CaptureErrorContext, ThreadedProducer};
 
     #[derive(Fail, Debug)]
     pub enum OutcomeError {
@@ -388,7 +387,7 @@ mod processing {
                     client_config.set(config_p.name.as_str(), config_p.value.as_str());
                 }
                 let future_producer = client_config
-                    .create()
+                    .create_with_context(CaptureErrorContext)
                     .context(ServerErrorKind::KafkaError)?;
                 (Some(future_producer), None)
             } else {

relay-server/src/actors/store.rs

@@ -8,7 +8,7 @@ use actix::prelude::*;
 use bytes::Bytes;
 use failure::{Fail, ResultExt};
 use rdkafka::error::KafkaError;
-use rdkafka::producer::{BaseRecord, DefaultProducerContext};
+use rdkafka::producer::BaseRecord;
 use rdkafka::ClientConfig;
 use rmp_serde::encode::Error as RmpError;
 use serde::{ser::Error, Serialize};
@@ -22,8 +22,7 @@ use relay_quotas::Scoping;
 use crate::envelope::{AttachmentType, Envelope, Item, ItemType};
 use crate::metrics::RelayCounters;
 use crate::service::{ServerError, ServerErrorKind};
-
-type ThreadedProducer = rdkafka::producer::ThreadedProducer<DefaultProducerContext>;
+use crate::utils::{CaptureErrorContext, ThreadedProducer};
 
 lazy_static::lazy_static! {
     static ref NAMESPACE_DID: Uuid =
@@ -61,7 +60,7 @@ impl StoreForwarder {
         }
 
         let producer = client_config
-            .create()
+            .create_with_context(CaptureErrorContext)
             .context(ServerErrorKind::KafkaError)?;
 
         Ok(Self {

relay-server/src/metrics.rs

@@ -215,6 +215,12 @@ pub enum RelayCounters {
     /// either `event` or `attachment` representing the type of message produced on the Kafka queue.
     #[cfg(feature = "processing")]
     ProcessingMessageProduced,
+    /// Counts the number of producer errors occurred after an event was already enqueued for
+    /// sending to Kafka. These errors might include e.g. MessageTooLarge errors when the broker
+    /// does not accept the requests over a certain size, which is usually due to invalic or
+    /// inconsistent broker/producer configurations.
+    #[cfg(feature = "processing")]
+    ProcessingProduceError,
     /// Counts the number of events that hit any of the Store like endpoints (Store, Security,
     /// MiniDump, Unreal). The events are counted before they are rate limited , filtered or
     /// processed in any way. The counter has a `version` tag that tracks the message event
@@ -265,6 +271,8 @@ impl CounterMetric for RelayCounters {
             RelayCounters::ServerStarting => "server.starting",
             #[cfg(feature = "processing")]
             RelayCounters::ProcessingMessageProduced => "processing.event.produced",
+            #[cfg(feature = "processing")]
+            RelayCounters::ProcessingProduceError => "processing.produce.error",
             RelayCounters::EventProtocol => "event.protocol",
             RelayCounters::Requests => "requests",
             RelayCounters::ResponsesStatusCodes => "responses.status_codes",

relay-server/src/utils/kafka.rs

@@ -0,0 +1,30 @@
+use rdkafka::producer::{DeliveryResult, ProducerContext};
+use rdkafka::ClientContext;
+
+use relay_common::{metric, LogError};
+
+use crate::metrics::RelayCounters;
+
+/// Kafka producer context that logs producer errors
+pub struct CaptureErrorContext;
+
+impl ClientContext for CaptureErrorContext {}
+
+impl ProducerContext for CaptureErrorContext {
+    type DeliveryOpaque = ();
+
+    /// This method is called after attempting to send a message to Kafka.
+    /// It's called asynchronously for every message, so we want to handle errors explicitly here.
+    fn delivery(&self, result: &DeliveryResult, _delivery_opaque: Self::DeliveryOpaque) {
+        if let Err((error, _message)) = result {
+            log::error!(
+                "failed to produce message to Kafka (delivery callback): {}",
+                LogError(error)
+            );
+
+            metric!(counter(RelayCounters::ProcessingProduceError) += 1);
+        }
+    }
+}
+
+pub type ThreadedProducer = rdkafka::producer::ThreadedProducer<CaptureErrorContext>;

relay-server/src/utils/mod.rs

@@ -8,6 +8,9 @@ mod request;
 mod shutdown;
 mod timer;
 
+#[cfg(feature = "processing")]
+mod kafka;
+
 #[cfg(feature = "processing")]
 mod unreal;
 
@@ -21,5 +24,8 @@ pub use self::request::*;
 pub use self::shutdown::*;
 pub use self::timer::*;
 
+#[cfg(feature = "processing")]
+pub use self::kafka::*;
+
 #[cfg(feature = "processing")]
 pub use self::unreal::*;