feat(events): add metadata info to events (#4875)

Author: lsampras

crates/events/src/lib.rs

@@ -49,6 +49,11 @@ pub trait Event: EventInfo {
 
     /// The class/type of the event. This is used to group/categorize events together.
     fn class(&self) -> Self::EventType;
+
+    /// Metadata associated with the event
+    fn metadata(&self) -> HashMap<String, String> {
+        HashMap::new()
+    }
 }
 
 /// Hold the context information for any events
@@ -73,7 +78,8 @@ where
     event: E,
 }
 
-struct RawEvent<T, A: Event<EventType = T>>(HashMap<String, Value>, A);
+/// A flattened event that flattens the context provided to it along with the actual event.
+struct FlatMapEvent<T, A: Event<EventType = T>>(HashMap<String, Value>, A);
 
 impl<T, A, E, D> EventBuilder<T, A, E, D>
 where
@@ -109,12 +115,13 @@ where
     /// Emit the event.
     pub fn try_emit(self) -> Result<(), EventsError> {
         let ts = self.event.timestamp();
+        let metadata = self.event.metadata();
         self.message_sink
-            .send_message(RawEvent(self.metadata, self.event), ts)
+            .send_message(FlatMapEvent(self.metadata, self.event), metadata, ts)
     }
 }
 
-impl<T, A> Serialize for RawEvent<T, A>
+impl<T, A> Serialize for FlatMapEvent<T, A>
 where
     A: Event<EventType = T>,
 {
@@ -236,7 +243,12 @@ pub trait MessagingInterface {
     /// The type of the event used for categorization by the event publisher.
     type MessageClass;
     /// Send a message that follows the defined message class.
-    fn send_message<T>(&self, data: T, timestamp: PrimitiveDateTime) -> Result<(), EventsError>
+    fn send_message<T>(
+        &self,
+        data: T,
+        metadata: HashMap<String, String>,
+        timestamp: PrimitiveDateTime,
+    ) -> Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize;
 }
@@ -252,7 +264,7 @@ pub trait Message {
     fn identifier(&self) -> String;
 }
 
-impl<T, A> Message for RawEvent<T, A>
+impl<T, A> Message for FlatMapEvent<T, A>
 where
     A: Event<EventType = T>,
 {

crates/router/src/events.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use error_stack::ResultExt;
 use events::{EventsError, Message, MessagingInterface};
 use hyperswitch_domain_models::errors::{StorageError, StorageResult};
@@ -94,14 +96,15 @@ impl MessagingInterface for EventsHandler {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize,
     {
         match self {
-            Self::Kafka(a) => a.send_message(data, timestamp),
-            Self::Logs(a) => a.send_message(data, timestamp),
+            Self::Kafka(a) => a.send_message(data, metadata, timestamp),
+            Self::Logs(a) => a.send_message(data, metadata, timestamp),
         }
     }
 }

crates/router/src/events/event_logger.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use events::{EventsError, Message, MessagingInterface};
 use masking::ErasedMaskSerialize;
 use time::PrimitiveDateTime;
@@ -21,12 +23,13 @@ impl MessagingInterface for EventLogger {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         _timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
         T: Message<Class = Self::MessageClass> + ErasedMaskSerialize,
     {
-        logger::info!(event =? data.masked_serialize().unwrap_or_else(|e| serde_json::json!({"error": e.to_string()})), event_type =? data.get_message_class(), event_id =? data.identifier(), log_type =? "event");
+        logger::info!(event =? data.masked_serialize().unwrap_or_else(|e| serde_json::json!({"error": e.to_string()})), event_type =? data.get_message_class(), event_id =? data.identifier(), log_type =? "event", metadata = ?metadata);
         Ok(())
     }
 }

crates/router/src/middleware.rs

@@ -157,7 +157,8 @@ where
                     payment_method = Empty,
                     status_code = Empty,
                     flow = "UNKNOWN",
-                    golden_log_line = Empty
+                    golden_log_line = Empty,
+                    tenant_id = "ta"
                 )
                 .or_current(),
             ),

crates/router/src/services/kafka.rs

@@ -1,11 +1,12 @@
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 use bigdecimal::ToPrimitive;
 use common_utils::errors::CustomResult;
 use error_stack::{report, ResultExt};
 use events::{EventsError, Message, MessagingInterface};
 use rdkafka::{
     config::FromClientConfig,
+    message::{Header, OwnedHeaders},
     producer::{BaseRecord, DefaultProducerContext, Producer, ThreadedProducer},
 };
 #[cfg(feature = "payouts")]
@@ -528,6 +529,7 @@ impl MessagingInterface for KafkaProducer {
     fn send_message<T>(
         &self,
         data: T,
+        metadata: HashMap<String, String>,
         timestamp: PrimitiveDateTime,
     ) -> error_stack::Result<(), EventsError>
     where
@@ -538,12 +540,20 @@ impl MessagingInterface for KafkaProducer {
             .masked_serialize()
             .and_then(|i| serde_json::to_vec(&i))
             .change_context(EventsError::SerializationError)?;
+        let mut headers = OwnedHeaders::new();
+        for (k, v) in metadata.iter() {
+            headers = headers.insert(Header {
+                key: k.as_str(),
+                value: Some(v),
+            });
+        }
         self.producer
             .0
             .send(
                 BaseRecord::to(topic)
                     .key(&data.identifier())
                     .payload(&json_data)
+                    .headers(headers)
                     .timestamp(
                         (timestamp.assume_utc().unix_timestamp_nanos() / 1_000_000)
                             .to_i64()