Handle metrics for benign application failures

core/src/telemetry/metrics.rs

@@ -13,6 +13,7 @@ use temporal_sdk_core_api::telemetry::metrics::{
     NoOpCoreMeter,
 };
 use temporal_sdk_core_protos::temporal::api::enums::v1::WorkflowTaskFailedCause;
+use temporal_sdk_core_protos::temporal::api::failure::v1::Failure;
 
 /// Used to track context associated with metrics, and record/update them
 ///
@@ -592,6 +593,19 @@ pub(super) const TASK_SLOTS_AVAILABLE_NAME: &str = "worker_task_slots_available"
 pub(super) const TASK_SLOTS_USED_NAME: &str = "worker_task_slots_used";
 pub(super) const STICKY_CACHE_SIZE_NAME: &str = "sticky_cache_size";
 
+/// Calls the provided metric function only if the failure is not a benign application failure.
+pub(crate) fn record_failure_metric(
+    failure: &Option<Failure>,
+    metric_fn: impl FnOnce(),
+) {
+    let is_benign = failure
+        .as_ref()
+        .map_or(false, |f| f.is_benign_application_failure());
+    if !is_benign {
+        metric_fn();
+    }
+}
+
 /// Helps define buckets once in terms of millis, but also generates a seconds version
 macro_rules! define_latency_buckets {
     ($(($metric_name:pat, $name:ident, $sec_name:ident, [$($bucket:expr),*])),*) => {

core/src/worker/activities.rs

@@ -13,7 +13,7 @@ use crate::{
         UsedMeteredSemPermit,
     },
     pollers::{BoxedActPoller, PermittedTqResp, TrackedPermittedTqResp, new_activity_task_poller},
-    telemetry::metrics::{MetricsContext, activity_type, eager, workflow_type},
+    telemetry::metrics::{MetricsContext, activity_type, eager, workflow_type, record_failure_metric},
     worker::{
         activities::activity_heartbeat_manager::ActivityHeartbeatError, client::WorkerClient,
     },
@@ -349,7 +349,7 @@ impl WorkerActivityTasks {
                             .err()
                     }
                     aer::Status::Failed(ar::Failure { failure }) => {
-                        act_metrics.act_execution_failed();
+                        record_failure_metric(&failure, || act_metrics.act_execution_failed());
                         client
                             .fail_activity_task(task_token.clone(), failure)
                             .await

core/src/worker/activities/local_activities.rs

@@ -3,7 +3,7 @@ use crate::{
     abstractions::{MeteredPermitDealer, OwnedMeteredSemPermit, UsedMeteredSemPermit, dbg_panic},
     protosext::ValidScheduleLA,
     retry_logic::RetryPolicyExt,
-    telemetry::metrics::{activity_type, workflow_type},
+    telemetry::metrics::{activity_type, workflow_type, record_failure_metric},
     worker::workflow::HeartbeatTimeoutMsg,
 };
 use futures_util::{
@@ -583,7 +583,7 @@ impl LocalActivityManager {
             la_metrics.la_exec_latency(runtime);
             let outcome = match &status {
                 LocalActivityExecutionResult::Failed(fail) => {
-                    la_metrics.la_execution_failed();
+                    record_failure_metric(&fail.failure, || la_metrics.la_execution_failed());
                     Outcome::FailurePath {
                         backoff: calc_backoff!(fail),
                     }

core/src/worker/workflow/managed_run.rs

@@ -1125,8 +1125,8 @@ impl ManagedRun {
                     Some(CmdAttribs::CompleteWorkflowExecutionCommandAttributes(_)) => {
                         self.metrics.wf_completed();
                     }
-                    Some(CmdAttribs::FailWorkflowExecutionCommandAttributes(_)) => {
-                        self.metrics.wf_failed();
+                    Some(CmdAttribs::FailWorkflowExecutionCommandAttributes(attrs)) => {
+                        metrics::record_failure_metric(&attrs.failure, || self.metrics.wf_failed());
                     }
                     Some(CmdAttribs::ContinueAsNewWorkflowExecutionCommandAttributes(_)) => {
                         self.metrics.wf_continued_as_new();