Issue:
Bug: Logging framework - Add a endrequest log line which should be a common line for accumulating all parameters



Currently we use the `EndRequest` filter by actix to rely on this, but we need a log line that represents a request uniquely & contains all the key metadata about the request

the list of fields that we need

- Flow (already present)
- payment_id
- connector_name
- payment_method

Pull request description:
## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
- Create a request middleware to create a top level span consisting of all the metadata fields
- Update the subscriber layer to propogate the metadata fields to the top layer

### Additional Changes

- [ ] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [ ] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
- Currently we don't have guaranteed keys/metadata present at the top level
- finding the relavant info (payment_id/connector etc) requires context about the lower level functions where this info is extracted
- instead this change provides a single log line that is guaranteed to be present for all API calls
- And this log line would have all the relevant fields needed for analysis/debugging


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->

run the application locally
all api requests should have the following log line at the end of request
- the fields should be populated when available
- These log line will be generated even for health & non-existent paths (e.g `/` or `/favicon.ico`)

```json
{
    "message": "[GOLDEN_LOG_LINE - EVENT] REQUEST MIDDLEWARE END",
    "hostname": "<MY_MACHINE>.local",
    "pid": 62901,
    "env": "development",
    "level": "INFO",
    "target": "router::middleware",
    "service": "router",
    "line": 61,
    "file": "crates/router/src/middleware.rs",
    "fn": "golden_log_line",
    "full_name": "router::middleware::golden_log_line",
    "time": "2024-01-24T06:35:06.771569000Z",
    "merchant_id": "merchant_1706060159",
    "request_id": "018d3a2d-e583-7581-8992-8596b3052995",
    "extra": {
        "trace_id": "00000000000000000000000000000000",
        "http.method": "POST",
        "http.route": "/payments",
        "http.flavor": "1.1",
        "http.user_agent": "PostmanRuntime/7.33.0",
        "payment_id": "payment_intent_id = \"pay_y8XHx5whOaWTdoGrgAAf\"",
        "otel.name": "HTTP POST /payments",
        "http.client_ip": "::1",
        "http.host": "localhost:8080",
        "http.target": "/payments",
        "http.scheme": "http",
        "otel.kind": "server"
    }
}
```

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [ ] I formatted the code `cargo +nightly fmt --all`
- [ ] I addressed lints thrown by `cargo clippy`
- [ ] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable


Testing:


run the application locally
all api requests should have the following log line at the end of request
- the fields should be populated when available
- These log line will be generated even for health & non-existent paths (e.g `/` or `/favicon.ico`)

```json
{
    "message": "[GOLDEN_LOG_LINE - EVENT] REQUEST MIDDLEWARE END",
    "hostname": "<MY_MACHINE>.local",
    "pid": 62901,
    "env": "development",
    "level": "INFO",
    "target": "router::middleware",
    "service": "router",
    "line": 61,
    "file": "crates/router/src/middleware.rs",
    "fn": "golden_log_line",
    "full_name": "router::middleware::golden_log_line",
    "time": "2024-01-24T06:35:06.771569000Z",
    "merchant_id": "merchant_1706060159",
    "request_id": "018d3a2d-e583-7581-8992-8596b3052995",
    "extra": {
        "trace_id": "00000000000000000000000000000000",
        "http.method": "POST",
        "http.route": "/payments",
        "http.flavor": "1.1",
        "http.user_agent": "PostmanRuntime/7.33.0",
        "payment_id": "payment_intent_id = \"pay_y8XHx5whOaWTdoGrgAAf\"",
        "otel.name": "HTTP POST /payments",
        "http.client_ip": "::1",
        "http.host": "localhost:8080",
        "http.target": "/payments",
        "http.scheme": "http",
        "otel.kind": "server"
    }
}
```

Patch:
diff --git a/crates/drainer/src/query.rs b/crates/drainer/src/query.rs
index f79291f3eae..a1e04fb6d0f 100644
--- a/crates/drainer/src/query.rs
+++ b/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);
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 712ae9e4035..968e6cb07d6 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/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
diff --git a/crates/router/src/core/payments/tokenization.rs b/crates/router/src/core/payments/tokenization.rs
index 15d88c94660..f52bce46d8e 100644
--- a/crates/router/src/core/payments/tokenization.rs
+++ b/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?
                 };
 
diff --git a/crates/router/src/lib.rs b/crates/router/src/lib.rs
index c38a4dc85b5..7b1aba14106 100644
--- a/crates/router/src/lib.rs
+++ b/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())
 }
diff --git a/crates/router/src/middleware.rs b/crates/router/src/middleware.rs
index c6c94d3a78e..1feba66a34f 100644
--- a/crates/router/src/middleware.rs
+++ b/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 {
-            let request_id = request_id_fut.await?;
-            let request_id = request_id.as_hyphenated().to_string();
-            if let Some(upstream_request_id) = old_x_request_id {
-                router_env::logger::info!(?request_id, ?upstream_request_id);
+        Box::pin(
+            async move {
+                let request_id = request_id_fut.await?;
+                let request_id = request_id.as_hyphenated().to_string();
+                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(
+                    http::header::HeaderName::from_static("x-request-id"),
+                    http::HeaderValue::from_str(&request_id)?,
+                );
+
+                Ok(response)
             }
-            let mut response = response_fut.await?;
-            response.headers_mut().append(
-                http::header::HeaderName::from_static("x-request-id"),
-                http::HeaderValue::from_str(&request_id)?,
-            );
-
-            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(),
+            ),
+        )
+    }
+}
diff --git a/crates/router/src/routes/admin.rs b/crates/router/src/routes/admin.rs
index ab404125a38..f5a6a49b9c9 100644
--- a/crates/router/src/routes/admin.rs
+++ b/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))]
diff --git a/crates/router/src/routes/mandates.rs b/crates/router/src/routes/mandates.rs
index 468d202b94c..342fe23c38a 100644
--- a/crates/router/src/routes/mandates.rs
+++ b/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
diff --git a/crates/router/src/routes/payment_methods.rs b/crates/router/src/routes/payment_methods.rs
index 6ef5de886be..55564a6386f 100644
--- a/crates/router/src/routes/payment_methods.rs
+++ b/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,
diff --git a/crates/router/src/services/api.rs b/crates/router/src/services/api.rs
index ad463fcf2b9..307dff55071 100644
--- a/crates/router/src/services/api.rs
+++ b/crates/router/src/services/api.rs
@@ -1131,7 +1131,6 @@ where
         status_code = response_code,
         time_taken_ms = request_duration.as_millis(),
     );
-
     res
 }
 
diff --git a/crates/router_env/src/logger/storage.rs b/crates/router_env/src/logger/storage.rs
index 51e701213b9..961a77c65aa 100644
--- a/crates/router_env/src/logger/storage.rs
+++ b/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
diff --git a/docker-compose-development.yml b/docker-compose-development.yml
index 5a3eca4cdf3..665bdf2f05d 100644
--- a/docker-compose-development.yml
+++ b/docker-compose-development.yml
@@ -31,7 +31,7 @@ services:
     networks:
       - router_net
     ports:
-      - "6379"
+      - "6379:6379"
 
   migration_runner:
     image: rust:latest
diff --git a/docker-compose.yml b/docker-compose.yml
index 9f8e7bb4efb..3839269a522 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -27,7 +27,7 @@ services:
     networks:
       - router_net
     ports:
-      - "6379"
+      - "6379:6379"
 
   migration_runner:
     image: rust:latest