Handle metrics for benign application failures (#905)

* updated api

* don't track failures for benign application errors

* remove closure

* linting/formatting

* add workflow rule rpcs to client

Author: THardy98

client/src/raw.rs

@@ -847,6 +847,51 @@ proxier! {
             r.extensions_mut().insert(labels);
         }
     );
+    (
+        create_workflow_rule,
+        CreateWorkflowRuleRequest,
+        CreateWorkflowRuleResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        describe_workflow_rule,
+        DescribeWorkflowRuleRequest,
+        DescribeWorkflowRuleResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        delete_workflow_rule,
+        DeleteWorkflowRuleRequest,
+        DeleteWorkflowRuleResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        list_workflow_rules,
+        ListWorkflowRulesRequest,
+        ListWorkflowRulesResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
+    (
+        trigger_workflow_rule,
+        TriggerWorkflowRuleRequest,
+        TriggerWorkflowRuleResponse,
+        |r| {
+            let labels = namespaced_request!(r);
+            r.extensions_mut().insert(labels);
+        }
+    );
     (
         get_search_attributes,
         GetSearchAttributesRequest,

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,13 @@ 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";
 
+/// Track a failure metric if the failure is not a benign application failure.
+pub(crate) fn should_record_failure_metric(failure: &Option<Failure>) -> bool {
+    !failure
+        .as_ref()
+        .is_some_and(|f| f.is_benign_application_failure())
+}
+
 /// 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,9 @@ 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, should_record_failure_metric, workflow_type,
+    },
     worker::{
         activities::activity_heartbeat_manager::ActivityHeartbeatError, client::WorkerClient,
     },
@@ -349,7 +351,9 @@ impl WorkerActivityTasks {
                             .err()
                     }
                     aer::Status::Failed(ar::Failure { failure }) => {
-                        act_metrics.act_execution_failed();
+                        if should_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, should_record_failure_metric, workflow_type},
     worker::workflow::HeartbeatTimeoutMsg,
 };
 use futures_util::{
@@ -583,7 +583,9 @@ impl LocalActivityManager {
             la_metrics.la_exec_latency(runtime);
             let outcome = match &status {
                 LocalActivityExecutionResult::Failed(fail) => {
-                    la_metrics.la_execution_failed();
+                    if should_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,10 @@ impl ManagedRun {
                     Some(CmdAttribs::CompleteWorkflowExecutionCommandAttributes(_)) => {
                         self.metrics.wf_completed();
                     }
-                    Some(CmdAttribs::FailWorkflowExecutionCommandAttributes(_)) => {
-                        self.metrics.wf_failed();
+                    Some(CmdAttribs::FailWorkflowExecutionCommandAttributes(attrs)) => {
+                        if metrics::should_record_failure_metric(&attrs.failure) {
+                            self.metrics.wf_failed();
+                        }
                     }
                     Some(CmdAttribs::ContinueAsNewWorkflowExecutionCommandAttributes(_)) => {
                         self.metrics.wf_continued_as_new();