Bug: [BUG] Add metrics in router scheduler flow as it is being ignored now



### Bug Description

Router is not maintaining any metrics about the task added in scheduler. this has to be fixed
File: crates/router/src/core/payments/helpers.rs
Method: add_domain_task_to_pt

### Expected Behavior

Metrics has to be added to know the TASKS_ADDED_COUNT and TASKS_RESET_COUNT

### Actual Behavior

Currently those are not available

### Steps To Reproduce

NA

### Context For The Bug

_No response_

### Environment

NA

### Have you spent some time checking if this bug has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

None
diff --git a/crates/router/src/core/api_keys.rs b/crates/router/src/core/api_keys.rs
index c1ddc43cd65..78d4e801e8f 100644
--- a/crates/router/src/core/api_keys.rs
+++ b/crates/router/src/core/api_keys.rs
@@ -270,6 +270,11 @@ pub async fn add_api_key_expiry_task(
                 api_key_expiry_tracker.key_id
             )
         })?;
+    metrics::TASKS_ADDED_COUNT.add(
+        &metrics::CONTEXT,
+        1,
+        &[metrics::request::add_attributes("flow", "ApiKeyExpiry")],
+    );
 
     Ok(())
 }
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index 39bc54fa157..f005920c24f 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -2855,6 +2855,14 @@ impl TempLockerCardSupport {
         )
         .await?;
         metrics::TOKENIZED_DATA_COUNT.add(&metrics::CONTEXT, 1, &[]);
+        metrics::TASKS_ADDED_COUNT.add(
+            &metrics::CONTEXT,
+            1,
+            &[metrics::request::add_attributes(
+                "flow",
+                "DeleteTokenizeData",
+            )],
+        );
         Ok(card)
     }
 }
diff --git a/crates/router/src/core/payment_methods/vault.rs b/crates/router/src/core/payment_methods/vault.rs
index 070bca234c8..59be4087eaa 100644
--- a/crates/router/src/core/payment_methods/vault.rs
+++ b/crates/router/src/core/payment_methods/vault.rs
@@ -1026,7 +1026,18 @@ pub async fn retry_delete_tokenize(
     let schedule_time = get_delete_tokenize_schedule_time(db, pm, pt.retry_count).await;
 
     match schedule_time {
-        Some(s_time) => pt.retry(db.as_scheduler(), s_time).await,
+        Some(s_time) => {
+            let retry_schedule = pt.retry(db.as_scheduler(), s_time).await;
+            metrics::TASKS_RESET_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[metrics::request::add_attributes(
+                    "flow",
+                    "DeleteTokenizeData",
+                )],
+            );
+            retry_schedule
+        }
         None => {
             pt.finish_with_status(db.as_scheduler(), "RETRIES_EXCEEDED".to_string())
                 .await
diff --git a/crates/router/src/core/payments/helpers.rs b/crates/router/src/core/payments/helpers.rs
index 7230d74e9a9..e15fed0327f 100644
--- a/crates/router/src/core/payments/helpers.rs
+++ b/crates/router/src/core/payments/helpers.rs
@@ -986,14 +986,30 @@ where
         match schedule_time {
             Some(stime) => {
                 if !requeue {
-                    // scheduler_metrics::TASKS_ADDED_COUNT.add(&metrics::CONTEXT, 1, &[]); // Metrics
+                    // Here, increment the count of added tasks every time a payment has been confirmed or PSync has been called
+                    metrics::TASKS_ADDED_COUNT.add(
+                        &metrics::CONTEXT,
+                        1,
+                        &[metrics::request::add_attributes(
+                            "flow",
+                            format!("{:#?}", operation),
+                        )],
+                    );
                     super::add_process_sync_task(&*state.store, payment_attempt, stime)
                         .await
                         .into_report()
                         .change_context(errors::ApiErrorResponse::InternalServerError)
                         .attach_printable("Failed while adding task to process tracker")
                 } else {
-                    // scheduler_metrics::TASKS_RESET_COUNT.add(&metrics::CONTEXT, 1, &[]); // Metrics
+                    // When the requeue is true, we reset the tasks count as we reset the task every time it is requeued
+                    metrics::TASKS_RESET_COUNT.add(
+                        &metrics::CONTEXT,
+                        1,
+                        &[metrics::request::add_attributes(
+                            "flow",
+                            format!("{:#?}", operation),
+                        )],
+                    );
                     super::reset_process_sync_task(&*state.store, payment_attempt, stime)
                         .await
                         .into_report()
diff --git a/crates/router/src/core/refunds.rs b/crates/router/src/core/refunds.rs
index e60c341dedc..4b1c33296e6 100644
--- a/crates/router/src/core/refunds.rs
+++ b/crates/router/src/core/refunds.rs
@@ -1088,6 +1088,12 @@ pub async fn add_refund_sync_task(
                 refund.refund_id
             )
         })?;
+    metrics::TASKS_ADDED_COUNT.add(
+        &metrics::CONTEXT,
+        1,
+        &[metrics::request::add_attributes("flow", "Refund")],
+    );
+
     Ok(response)
 }
 
@@ -1170,7 +1176,15 @@ pub async fn retry_refund_sync_task(
         get_refund_sync_process_schedule_time(db, &connector, &merchant_id, pt.retry_count).await?;
 
     match schedule_time {
-        Some(s_time) => pt.retry(db.as_scheduler(), s_time).await,
+        Some(s_time) => {
+            let retry_schedule = pt.retry(db.as_scheduler(), s_time).await;
+            metrics::TASKS_RESET_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[metrics::request::add_attributes("flow", "Refund")],
+            );
+            retry_schedule
+        }
         None => {
             pt.finish_with_status(db.as_scheduler(), "RETRIES_EXCEEDED".to_string())
                 .await
diff --git a/crates/router/src/routes/metrics.rs b/crates/router/src/routes/metrics.rs
index b3629ab7d52..6c3293dba9d 100644
--- a/crates/router/src/routes/metrics.rs
+++ b/crates/router/src/routes/metrics.rs
@@ -113,5 +113,8 @@ counter_metric!(AUTO_RETRY_GSM_MATCH_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_EXHAUSTED_COUNT, GLOBAL_METER);
 counter_metric!(AUTO_RETRY_PAYMENT_COUNT, GLOBAL_METER);
 
+counter_metric!(TASKS_ADDED_COUNT, GLOBAL_METER); // Tasks added to process tracker
+counter_metric!(TASKS_RESET_COUNT, GLOBAL_METER); // Tasks reset in process tracker for requeue flow
+
 pub mod request;
 pub mod utils;
diff --git a/crates/router/src/workflows/api_key_expiry.rs b/crates/router/src/workflows/api_key_expiry.rs
index 62d8a54c402..eb3c1d9c1ce 100644
--- a/crates/router/src/workflows/api_key_expiry.rs
+++ b/crates/router/src/workflows/api_key_expiry.rs
@@ -115,6 +115,12 @@ Team Hyperswitch"),
             let task_ids = vec![task_id];
             db.process_tracker_update_process_status_by_ids(task_ids, updated_process_tracker_data)
                 .await?;
+            // Remaining tasks are re-scheduled, so will be resetting the added count
+            metrics::TASKS_RESET_COUNT.add(
+                &metrics::CONTEXT,
+                1,
+                &[metrics::request::add_attributes("flow", "ApiKeyExpiry")],
+            );
         }
 
         Ok(())
diff --git a/crates/scheduler/src/metrics.rs b/crates/scheduler/src/metrics.rs
index 134f5599b31..ca4fb9ec242 100644
--- a/crates/scheduler/src/metrics.rs
+++ b/crates/scheduler/src/metrics.rs
@@ -6,8 +6,6 @@ global_meter!(PT_METER, "PROCESS_TRACKER");
 histogram_metric!(CONSUMER_STATS, PT_METER, "CONSUMER_OPS");
 
 counter_metric!(PAYMENT_COUNT, PT_METER); // No. of payments created
-counter_metric!(TASKS_ADDED_COUNT, PT_METER); // Tasks added to process tracker
-counter_metric!(TASKS_RESET_COUNT, PT_METER); // Tasks reset in process tracker for requeue flow
 counter_metric!(TASKS_PICKED_COUNT, PT_METER); // Tasks picked by
 counter_metric!(BATCHES_CREATED, PT_METER); // Batches added to stream
 counter_metric!(BATCHES_CONSUMED, PT_METER); // Batches consumed by consumer

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

- [x] Bugfix

## Description
<!-- Describe your changes in detail -->
Currently `TASKS_ADDED_COUNT` and `TASKS_RESET_COUNT` metrics does not exist in router to track the tasks added in scheduler. This PR aims at adding that.  
Closes #2090

## 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).
-->
We need to track the tasks added in scheduler.

## 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)?
-->
Did a few payments (Ran Stripe and Volt collection a few times):
<img width="1728" alt="image" src="https://github.com/juspay/hyperswitch/assets/69745008/1e2dca12-c7e8-453c-97c9-bc95e8dbf742">



### Local setup:

Modify 3 files:
- `docker-compose.yml`: Expose the `otel-collector` port by replacing `"4317"` with `"4317:4317"`
- `development.toml`: Enable logging by setting `metrics_enabled` to `true` and `otel_exporter_otlp_endpoint` to `http://localhost:4317`
- `otel-collector.yml`: Add logging in exporters of pipeline service: `[prometheus, logging]` instead of `[prometheus]`
- Open a terminal window, start server by executing: `cargo r`
- Create another terminal window for psql, execute:
	```
	psql hyperswitch_db
	\x
	\dt
	table process_tracker;
	``` 
- Create another 3 terminal windows. One for consumer, another for producer and another for Grafana. Execute:
	```
	# In consumer terminal window
	SCHEDULER_FLOW=consumer cargo r --bin scheduler 
	# In producer terminal window
	SCHEDULER_FLOW=producer cargo r --bin scheduler
	# In Grafana terminal window
	docker compose up -d grafana prometheus otel-collector
	```
- Execute `curl telnet://localhost:4317` in yet another terminal window to see if  this URL can be accessed or not
- Grafana username/password - admin/admin 
  > incase reset is required, execute in terminal:
  > ` grafana-cli admin reset-admin-password <new-password>`

In env:
- Open Grafana > VM Single > `Metric:router_TASKS_ADDED_COUNT_total`, `label:job=router` > `Run Query`

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