fix(metrics): Add TASKS_ADDED_COUNT and TASKS_RESET_COUNT metrics in router scheduler flow

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(())
 }

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)
     }
 }

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

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()

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

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;

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(())

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