feat(logging): add a logging middleware to log all api requests (#3437)

crates/drainer/src/query.rs

@@ -37,7 +37,7 @@ impl ExecuteQuery for kv::DBOperation {
         ];
 
         let (result, execution_time) =
-            common_utils::date_time::time_it(|| self.execute(&conn)).await;
+            Box::pin(common_utils::date_time::time_it(|| self.execute(&conn))).await;
 
         push_drainer_delay(pushed_at, operation, table, tags);
         metrics::QUERY_EXECUTION_TIME.record(&metrics::CONTEXT, execution_time, tags);

crates/router/src/core/payment_methods/cards.rs

@@ -245,7 +245,13 @@ pub async fn update_customer_payment_method(
             .as_ref()
             .map(|card_network| card_network.to_string()),
     };
-    add_payment_method(state, new_pm, &merchant_account, &key_store).await
+    Box::pin(add_payment_method(
+        state,
+        new_pm,
+        &merchant_account,
+        &key_store,
+    ))
+    .await
 }
 
 // Wrapper function to switch lockers

crates/router/src/core/payments/tokenization.rs

@@ -81,11 +81,11 @@ where
                     )
                     .await?
                 } else {
-                    save_in_locker(
+                    Box::pin(save_in_locker(
                         state,
                         merchant_account,
                         payment_method_create_request.to_owned(),
-                    )
+                    ))
                     .await?
                 };
 

crates/router/src/lib.rs

@@ -267,5 +267,6 @@ pub fn get_application_builder(
         .wrap(middleware::default_response_headers())
         .wrap(middleware::RequestId)
         .wrap(cors::cors())
+        .wrap(middleware::LogSpanInitializer)
         .wrap(router_env::tracing_actix_web::TracingLogger::default())
 }

crates/router/src/middleware.rs

@@ -1,3 +1,4 @@
+use router_env::tracing::{field::Empty, Instrument};
 /// Middleware to include request ID in response header.
 pub struct RequestId;
 
@@ -48,20 +49,23 @@ where
         let request_id_fut = req.extract::<router_env::tracing_actix_web::RequestId>();
         let response_fut = self.service.call(req);
 
-        Box::pin(async move {
+        Box::pin(
-            let request_id = request_id_fut.await?;
+            async move {
-            let request_id = request_id.as_hyphenated().to_string();
+                let request_id = request_id_fut.await?;
-            if let Some(upstream_request_id) = old_x_request_id {
+                let request_id = request_id.as_hyphenated().to_string();
-                router_env::logger::info!(?request_id, ?upstream_request_id);
+                if let Some(upstream_request_id) = old_x_request_id {
-            }
+                    router_env::logger::info!(?request_id, ?upstream_request_id);
-            let mut response = response_fut.await?;
+                }
-            response.headers_mut().append(
+                let mut response = response_fut.await?;
-                http::header::HeaderName::from_static("x-request-id"),
+                response.headers_mut().append(
-                http::HeaderValue::from_str(&request_id)?,
+                    http::header::HeaderName::from_static("x-request-id"),
-            );
+                    http::HeaderValue::from_str(&request_id)?,
+                );
 
-            Ok(response)
+                Ok(response)
-        })
+            }
+            .in_current_span(),
+        )
     }
 }
 
@@ -81,3 +85,67 @@ pub fn default_response_headers() -> actix_web::middleware::DefaultHeaders {
         .add((header::STRICT_TRANSPORT_SECURITY, "max-age=31536000"))
         .add((header::VIA, "HyperSwitch"))
 }
+
+/// Middleware to build a TOP level domain span for each request.
+pub struct LogSpanInitializer;
+
+impl<S, B> actix_web::dev::Transform<S, actix_web::dev::ServiceRequest> for LogSpanInitializer
+where
+    S: actix_web::dev::Service<
+        actix_web::dev::ServiceRequest,
+        Response = actix_web::dev::ServiceResponse<B>,
+        Error = actix_web::Error,
+    >,
+    S::Future: 'static,
+    B: 'static,
+{
+    type Response = actix_web::dev::ServiceResponse<B>;
+    type Error = actix_web::Error;
+    type Transform = LogSpanInitializerMiddleware<S>;
+    type InitError = ();
+    type Future = std::future::Ready<Result<Self::Transform, Self::InitError>>;
+
+    fn new_transform(&self, service: S) -> Self::Future {
+        std::future::ready(Ok(LogSpanInitializerMiddleware { service }))
+    }
+}
+
+pub struct LogSpanInitializerMiddleware<S> {
+    service: S,
+}
+
+impl<S, B> actix_web::dev::Service<actix_web::dev::ServiceRequest>
+    for LogSpanInitializerMiddleware<S>
+where
+    S: actix_web::dev::Service<
+        actix_web::dev::ServiceRequest,
+        Response = actix_web::dev::ServiceResponse<B>,
+        Error = actix_web::Error,
+    >,
+    S::Future: 'static,
+    B: 'static,
+{
+    type Response = actix_web::dev::ServiceResponse<B>;
+    type Error = actix_web::Error;
+    type Future = futures::future::LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    actix_web::dev::forward_ready!(service);
+
+    // TODO: have a common source of truth for the list of top level fields
+    // /crates/router_env/src/logger/storage.rs also has a list of fields  called PERSISTENT_KEYS
+    fn call(&self, req: actix_web::dev::ServiceRequest) -> Self::Future {
+        let response_fut = self.service.call(req);
+
+        Box::pin(
+            response_fut.instrument(
+                router_env::tracing::info_span!(
+                    "golden_log_line",
+                    payment_id = Empty,
+                    merchant_id = Empty,
+                    connector_name = Empty
+                )
+                .or_current(),
+            ),
+        )
+    }
+}

crates/router/src/routes/admin.rs

@@ -520,7 +520,7 @@ pub async fn business_profile_update(
     let flow = Flow::BusinessProfileUpdate;
     let (merchant_id, profile_id) = path.into_inner();
 
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -535,7 +535,7 @@ pub async fn business_profile_update(
             req.headers(),
         ),
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 #[instrument(skip_all, fields(flow = ?Flow::BusinessProfileDelete))]

crates/router/src/routes/mandates.rs

@@ -75,7 +75,7 @@ pub async fn revoke_mandate(
     let mandate_id = mandates::MandateId {
         mandate_id: path.into_inner(),
     };
-    api::server_wrap(
+    Box::pin(api::server_wrap(
         flow,
         state,
         &req,
@@ -85,7 +85,7 @@ pub async fn revoke_mandate(
         },
         &auth::ApiKeyAuth,
         api_locking::LockAction::NotApplicable,
-    )
+    ))
     .await
 }
 /// Mandates - List Mandates

crates/router/src/routes/payment_methods.rs

@@ -44,7 +44,13 @@ pub async fn create_payment_method_api(
         &req,
         json_payload.into_inner(),
         |state, auth, req| async move {
-            cards::add_payment_method(state, req, &auth.merchant_account, &auth.key_store).await
+            Box::pin(cards::add_payment_method(
+                state,
+                req,
+                &auth.merchant_account,
+                &auth.key_store,
+            ))
+            .await
         },
         &auth::ApiKeyAuth,
         api_locking::LockAction::NotApplicable,

crates/router/src/services/api.rs

@@ -1131,7 +1131,6 @@ where
         status_code = response_code,
         time_taken_ms = request_duration.as_millis(),
     );
-
     res
 }
 

crates/router_env/src/logger/storage.rs

@@ -92,6 +92,8 @@ impl Visit for Storage<'_> {
     }
 }
 
+const PERSISTENT_KEYS: [&str; 3] = ["payment_id", "connector_name", "merchant_id"];
+
 impl<S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>> Layer<S>
     for StorageSubscription
 {
@@ -99,6 +101,7 @@ impl<S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>> Layer
     fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
         #[allow(clippy::expect_used)]
         let span = ctx.span(id).expect("No span");
+        let mut extensions = span.extensions_mut();
 
         let mut visitor = if let Some(parent_span) = span.parent() {
             let mut extensions = parent_span.extensions_mut();
@@ -110,7 +113,6 @@ impl<S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>> Layer
             Storage::default()
         };
 
-        let mut extensions = span.extensions_mut();
         attrs.record(&mut visitor);
         extensions.insert(visitor);
     }
@@ -150,6 +152,18 @@ impl<S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>> Layer
                 .unwrap_or(0)
         };
 
+        if let Some(s) = span.extensions().get::<Storage<'_>>() {
+            s.values.iter().for_each(|(k, v)| {
+                if PERSISTENT_KEYS.contains(k) {
+                    span.parent().and_then(|p| {
+                        p.extensions_mut()
+                            .get_mut::<Storage<'_>>()
+                            .map(|s| s.values.insert(k, v.to_owned()))
+                    });
+                }
+            })
+        };
+
         let mut extensions_mut = span.extensions_mut();
         #[allow(clippy::expect_used)]
         let visitor = extensions_mut

docker-compose-development.yml

@@ -31,7 +31,7 @@ services:
     networks:
       - router_net
     ports:
-      - "6379"
+      - "6379:6379"
 
   migration_runner:
     image: rust:latest

docker-compose.yml

@@ -27,7 +27,7 @@ services:
     networks:
       - router_net
     ports:
-      - "6379"
+      - "6379:6379"
 
   migration_runner:
     image: rust:latest