Bug: [REFACTOR] Limiting `Tracing` to only Payments API



## Description
We are tracing all the APIs, which isn't effective due to the variance of the traffic. This change will include the tracing layer to a specific scope instead of the entire application
diff --git a/config/config.example.toml b/config/config.example.toml
index e980ea4fffd..b69519b6ac4 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -28,6 +28,7 @@ port = 5432                 # DB Port
 dbname = "hyperswitch_db"   # Name of Database
 pool_size = 5               # Number of connections to keep open
 connection_timeout = 10     # Timeout for database connection in seconds
+queue_strategy = "Fifo"     # Add the queue strategy used by the database bb8 client
 
 # Replica SQL data store credentials
 [replica_database]
@@ -38,6 +39,7 @@ port = 5432                 # DB Port
 dbname = "hyperswitch_db"   # Name of Database
 pool_size = 5               # Number of connections to keep open
 connection_timeout = 10     # Timeout for database connection in seconds
+queue_strategy = "Fifo"     # Add the queue strategy used by the database bb8 client
 
 # Redis credentials
 [redis]
@@ -93,6 +95,7 @@ sampling_rate = 0.1                                   # decimal rate between 0.0
 otel_exporter_otlp_endpoint = "http://localhost:4317" # endpoint to send metrics and traces to, can include port number
 otel_exporter_otlp_timeout = 5000                     # timeout (in milliseconds) for sending metrics and traces
 use_xray_generator = false                            # Set this to true for AWS X-ray compatible traces
+route_to_trace = [ "*/confirm" ]
 
 # This section provides some secret values.
 [secrets]
@@ -422,4 +425,4 @@ supported_connectors = "braintree"
 apple_pay_ppc = "APPLE_PAY_PAYMENT_PROCESSING_CERTIFICATE"              #Payment Processing Certificate provided by Apple Pay (https://developer.apple.com/) Certificates, Identifiers & Profiles > Apple Pay Payment Processing Certificate
 apple_pay_ppc_key = "APPLE_PAY_PAYMENT_PROCESSING_CERTIFICATE_KEY"      #Private key generate by Elliptic-curve prime256v1 curve
 apple_pay_merchant_cert = "APPLE_PAY_MERCHNAT_CERTIFICATE"              #Merchant Certificate provided by Apple Pay (https://developer.apple.com/) Certificates, Identifiers & Profiles > Apple Pay Merchant Identity Certificate
-apple_pay_merchant_cert_key = "APPLE_PAY_MERCHNAT_CERTIFICATE_KEY"      #Private key generate by RSA:2048 algorithm
\ No newline at end of file
+apple_pay_merchant_cert_key = "APPLE_PAY_MERCHNAT_CERTIFICATE_KEY"      #Private key generate by RSA:2048 algorithm
diff --git a/crates/router/src/configs/defaults.rs b/crates/router/src/configs/defaults.rs
index ec06a6d7078..3bb1c31d180 100644
--- a/crates/router/src/configs/defaults.rs
+++ b/crates/router/src/configs/defaults.rs
@@ -29,6 +29,7 @@ impl Default for super::settings::Database {
             dbname: String::new(),
             pool_size: 5,
             connection_timeout: 10,
+            queue_strategy: Default::default(),
         }
     }
 }
diff --git a/crates/router/src/configs/kms.rs b/crates/router/src/configs/kms.rs
index 0491fb61fc6..317ad0608b4 100644
--- a/crates/router/src/configs/kms.rs
+++ b/crates/router/src/configs/kms.rs
@@ -61,6 +61,7 @@ impl KmsDecrypt for settings::Database {
             password: self.password.decrypt_inner(kms_client).await?.into(),
             pool_size: self.pool_size,
             connection_timeout: self.connection_timeout,
+            queue_strategy: self.queue_strategy.into(),
         })
     }
 }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index fd4a398fd56..70d05b79f96 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -464,6 +464,24 @@ pub struct Database {
     pub dbname: String,
     pub pool_size: u32,
     pub connection_timeout: u64,
+    pub queue_strategy: QueueStrategy,
+}
+
+#[derive(Debug, Deserialize, Clone, Default)]
+#[serde(rename_all = "PascalCase")]
+pub enum QueueStrategy {
+    #[default]
+    Fifo,
+    Lifo,
+}
+
+impl From<QueueStrategy> for bb8::QueueStrategy {
+    fn from(value: QueueStrategy) -> Self {
+        match value {
+            QueueStrategy::Fifo => Self::Fifo,
+            QueueStrategy::Lifo => Self::Lifo,
+        }
+    }
 }
 
 #[cfg(not(feature = "kms"))]
@@ -477,6 +495,7 @@ impl Into<storage_impl::config::Database> for Database {
             dbname: self.dbname,
             pool_size: self.pool_size,
             connection_timeout: self.connection_timeout,
+            queue_strategy: self.queue_strategy.into(),
         }
     }
 }
@@ -704,9 +723,11 @@ impl Settings {
                     .try_parsing(true)
                     .separator("__")
                     .list_separator(",")
+                    .with_list_parse_key("log.telemetry.route_to_trace")
                     .with_list_parse_key("redis.cluster_urls")
                     .with_list_parse_key("connectors.supported.wallets")
                     .with_list_parse_key("connector_request_reference_id_config.merchant_ids_send_payment_id_as_connector_request_id"),
+
             )
             .build()?;
 
diff --git a/crates/router_env/src/logger/config.rs b/crates/router_env/src/logger/config.rs
index 269759ed44f..664c0d508f5 100644
--- a/crates/router_env/src/logger/config.rs
+++ b/crates/router_env/src/logger/config.rs
@@ -101,6 +101,8 @@ pub struct LogTelemetry {
     pub otel_exporter_otlp_timeout: Option<u64>,
     /// Whether to use xray ID generator, (enable this if you plan to use AWS-XRAY)
     pub use_xray_generator: bool,
+    /// Route Based Tracing
+    pub route_to_trace: Option<Vec<String>>,
 }
 
 /// Telemetry / tracing.
diff --git a/crates/router_env/src/logger/setup.rs b/crates/router_env/src/logger/setup.rs
index 78cbe9f342f..992de3e747e 100644
--- a/crates/router_env/src/logger/setup.rs
+++ b/crates/router_env/src/logger/setup.rs
@@ -12,6 +12,7 @@ use opentelemetry::{
         trace::BatchConfig,
         Resource,
     },
+    trace::{TraceContextExt, TraceState},
     KeyValue,
 };
 use opentelemetry_otlp::{TonicExporterBuilder, WithExportConfig};
@@ -132,6 +133,92 @@ fn get_opentelemetry_exporter(config: &config::LogTelemetry) -> TonicExporterBui
     exporter_builder
 }
 
+#[derive(Debug, Clone)]
+enum TraceUrlAssert {
+    Match(String),
+    EndsWith(String),
+}
+
+impl TraceUrlAssert {
+    fn compare_url(&self, url: &str) -> bool {
+        match self {
+            Self::Match(value) => url == value,
+            Self::EndsWith(end) => url.ends_with(end),
+        }
+    }
+}
+
+impl From<String> for TraceUrlAssert {
+    fn from(value: String) -> Self {
+        match value {
+            url if url.starts_with('*') => Self::EndsWith(url.trim_start_matches('*').to_string()),
+            url => Self::Match(url),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct TraceAssertion {
+    clauses: Option<Vec<TraceUrlAssert>>,
+    /// default behaviour for tracing if no condition is provided
+    default: bool,
+}
+
+impl TraceAssertion {
+    ///
+    /// Should the provided url be traced
+    ///
+    fn should_trace_url(&self, url: &str) -> bool {
+        match &self.clauses {
+            Some(clauses) => clauses.iter().all(|cur| cur.compare_url(url)),
+            None => self.default,
+        }
+    }
+}
+
+///
+/// Conditional Sampler for providing control on url based tracing
+///
+#[derive(Clone, Debug)]
+struct ConditionalSampler<T: trace::ShouldSample + Clone + 'static>(TraceAssertion, T);
+
+impl<T: trace::ShouldSample + Clone + 'static> trace::ShouldSample for ConditionalSampler<T> {
+    fn should_sample(
+        &self,
+        parent_context: Option<&opentelemetry::Context>,
+        trace_id: opentelemetry::trace::TraceId,
+        name: &str,
+        span_kind: &opentelemetry::trace::SpanKind,
+        attributes: &opentelemetry::trace::OrderMap<opentelemetry::Key, opentelemetry::Value>,
+        links: &[opentelemetry::trace::Link],
+        instrumentation_library: &opentelemetry::InstrumentationLibrary,
+    ) -> opentelemetry::trace::SamplingResult {
+        match attributes
+            .get(&opentelemetry::Key::new("http.route"))
+            .map_or(self.0.default, |inner| {
+                self.0.should_trace_url(&inner.as_str())
+            }) {
+            true => self.1.should_sample(
+                parent_context,
+                trace_id,
+                name,
+                span_kind,
+                attributes,
+                links,
+                instrumentation_library,
+            ),
+            false => opentelemetry::trace::SamplingResult {
+                decision: opentelemetry::trace::SamplingDecision::Drop,
+                attributes: Vec::new(),
+                trace_state: match parent_context {
+                    Some(ctx) => ctx.span().span_context().trace_state().clone(),
+                    None => TraceState::default(),
+                },
+            },
+        }
+    }
+}
+
 fn setup_tracing_pipeline(
     config: &config::LogTelemetry,
     service_name: &str,
@@ -140,9 +227,16 @@ fn setup_tracing_pipeline(
     global::set_text_map_propagator(TraceContextPropagator::new());
 
     let mut trace_config = trace::config()
-        .with_sampler(trace::Sampler::TraceIdRatioBased(
-            config.sampling_rate.unwrap_or(1.0),
-        ))
+        .with_sampler(trace::Sampler::ParentBased(Box::new(ConditionalSampler(
+            TraceAssertion {
+                clauses: config
+                    .route_to_trace
+                    .clone()
+                    .map(|inner| inner.into_iter().map(Into::into).collect()),
+                default: false,
+            },
+            trace::Sampler::TraceIdRatioBased(config.sampling_rate.unwrap_or(1.0)),
+        ))))
         .with_resource(Resource::new(vec![KeyValue::new(
             "service.name",
             service_name.to_owned(),
diff --git a/crates/storage_impl/src/config.rs b/crates/storage_impl/src/config.rs
index ad543b1942a..ceed3da81b3 100644
--- a/crates/storage_impl/src/config.rs
+++ b/crates/storage_impl/src/config.rs
@@ -9,4 +9,5 @@ pub struct Database {
     pub dbname: String,
     pub pool_size: u32,
     pub connection_timeout: u64,
+    pub queue_strategy: bb8::QueueStrategy,
 }
diff --git a/crates/storage_impl/src/database/store.rs b/crates/storage_impl/src/database/store.rs
index f9a7450b1cd..a09f1b75256 100644
--- a/crates/storage_impl/src/database/store.rs
+++ b/crates/storage_impl/src/database/store.rs
@@ -88,6 +88,7 @@ pub async fn diesel_make_pg_pool(
     let manager = async_bb8_diesel::ConnectionManager::<PgConnection>::new(database_url);
     let mut pool = bb8::Pool::builder()
         .max_size(database.pool_size)
+        .queue_strategy(database.queue_strategy)
         .connection_timeout(std::time::Duration::from_secs(database.connection_timeout));
 
     if test_transaction {

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [ ] New feature
- [x] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description

This change introduces a new middleware for tracing requests with the specified url. Providing a filtration mechanism.
This change also includes changing the bb8 queue strategy from FIFO to LIFO.


### 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).
-->


## 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)?
-->


## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
