La metrics

core/src/ephemeral_server/mod.rs

@@ -490,7 +490,7 @@ async fn lazy_download_exe(
             info!("Downloading {} to {}", uri, dest.display());
             download_and_extract(client, uri, file_to_extract, &mut temp_file)
                 .await
-                .map_err(|err| {
+                .inspect_err(|_| {
                     // Failed to download, just remove file
                     if let Err(err) = std::fs::remove_file(temp_dest) {
                         warn!(
@@ -499,7 +499,6 @@ async fn lazy_download_exe(
                             err
                         );
                     }
-                    err
                 })
         }
     }?;

core/src/telemetry/metrics.rs

@@ -42,6 +42,12 @@ struct Instruments {
     act_execution_failed: Arc<dyn Counter>,
     act_sched_to_start_latency: Arc<dyn HistogramDuration>,
     act_exec_latency: Arc<dyn HistogramDuration>,
+    act_exec_succeeded_latency: Arc<dyn HistogramDuration>,
+    la_execution_cancelled: Arc<dyn Counter>,
+    la_execution_failed: Arc<dyn Counter>,
+    la_exec_latency: Arc<dyn HistogramDuration>,
+    la_exec_succeeded_latency: Arc<dyn HistogramDuration>,
+    la_total: Arc<dyn Counter>,
     worker_registered: Arc<dyn Counter>,
     num_pollers: Arc<dyn Gauge>,
     task_slots_available: Arc<dyn Gauge>,
@@ -177,6 +183,13 @@ impl MetricsContext {
         self.instruments.act_execution_failed.add(1, &self.kvs);
     }
 
+    /// Record end-to-end (sched-to-complete) time for successful activity executions
+    pub(crate) fn act_execution_succeeded(&self, dur: Duration) {
+        self.instruments
+            .act_exec_succeeded_latency
+            .record(dur, &self.kvs);
+    }
+
     /// Record activity task schedule to start time in millis
     pub(crate) fn act_sched_to_start_latency(&self, dur: Duration) {
         self.instruments
@@ -190,6 +203,28 @@ impl MetricsContext {
         self.instruments.act_exec_latency.record(dur, &self.kvs);
     }
 
+    pub(crate) fn la_execution_cancelled(&self) {
+        self.instruments.la_execution_cancelled.add(1, &self.kvs);
+    }
+
+    pub(crate) fn la_execution_failed(&self) {
+        self.instruments.la_execution_failed.add(1, &self.kvs);
+    }
+
+    pub(crate) fn la_exec_latency(&self, dur: Duration) {
+        self.instruments.la_exec_latency.record(dur, &self.kvs);
+    }
+
+    pub(crate) fn la_exec_succeeded_latency(&self, dur: Duration) {
+        self.instruments
+            .la_exec_succeeded_latency
+            .record(dur, &self.kvs);
+    }
+
+    pub(crate) fn la_executed(&self) {
+        self.instruments.la_total.add(1, &self.kvs);
+    }
+
     /// A worker was registered
     pub(crate) fn worker_registered(&self) {
         self.instruments.worker_registered.add(1, &self.kvs);
@@ -316,6 +351,39 @@ impl Instruments {
                 unit: "duration".into(),
                 description: "Histogram of activity execution latencies".into(),
             }),
+            act_exec_succeeded_latency: meter.histogram_duration(MetricParameters {
+                name: "activity_succeed_endtoend_latency".into(),
+                unit: "duration".into(),
+                description: "Histogram of activity execution latencies for successful activities"
+                    .into(),
+            }),
+            la_execution_cancelled: meter.counter(MetricParameters {
+                name: "local_activity_execution_cancelled".into(),
+                description: "Count of local activity executions that were cancelled".into(),
+                unit: "".into(),
+            }),
+            la_execution_failed: meter.counter(MetricParameters {
+                name: "local_activity_execution_failed".into(),
+                description: "Count of local activity executions that failed".into(),
+                unit: "".into(),
+            }),
+            la_exec_latency: meter.histogram_duration(MetricParameters {
+                name: "local_activity_execution_latency".into(),
+                unit: "duration".into(),
+                description: "Histogram of local activity execution latencies".into(),
+            }),
+            la_exec_succeeded_latency: meter.histogram_duration(MetricParameters {
+                name: "local_activity_succeed_endtoend_latency".into(),
+                unit: "duration".into(),
+                description:
+                    "Histogram of local activity execution latencies for successful local activities"
+                        .into(),
+            }),
+            la_total: meter.counter(MetricParameters {
+                name: "local_activity_total".into(),
+                description: "Count of local activities executed".into(),
+                unit: "".into(),
+            }),
             // name kept as worker start for compat with old sdk / what users expect
             worker_registered: meter.counter(MetricParameters {
                 name: "worker_start".into(),
@@ -807,7 +875,7 @@ mod tests {
         a1.set(Arc::new(DummyCustomAttrs(1))).unwrap();
         // Verify all metrics are created. This number will need to get updated any time a metric
         // is added.
-        let num_metrics = 24;
+        let num_metrics = 30;
         #[allow(clippy::needless_range_loop)] // Sorry clippy, this reads easier.
         for metric_num in 1..=num_metrics {
             let hole = assert_matches!(&events[metric_num],

core/src/worker/activities.rs

@@ -37,7 +37,7 @@ use std::{
         atomic::{AtomicBool, Ordering},
         Arc,
     },
-    time::{Duration, Instant},
+    time::{Duration, Instant, SystemTime},
 };
 use temporal_sdk_core_api::worker::{ActivitySlotInfo, ActivitySlotKind};
 use temporal_sdk_core_protos::{
@@ -93,6 +93,7 @@ struct InFlightActInfo {
     /// Only kept for logging reasons
     workflow_run_id: String,
     start_time: Instant,
+    scheduled_time: Option<SystemTime>,
 }
 
 /// Augments [InFlightActInfo] with details specific to remote activities
@@ -127,6 +128,7 @@ impl RemoteInFlightActInfo {
                 workflow_id: wec.workflow_id,
                 workflow_run_id: wec.run_id,
                 start_time: Instant::now(),
+                scheduled_time: poll_resp.scheduled_time.and_then(|i| i.try_into().ok()),
             },
             heartbeat_timeout: poll_resp.heartbeat_timeout,
             issued_cancel_to_lang: None,
@@ -336,10 +338,19 @@ impl WorkerActivityTasks {
                 let _flushing_guard = self.completers_lock.read().await;
                 let maybe_net_err = match status {
                     aer::Status::WillCompleteAsync(_) => None,
-                    aer::Status::Completed(ar::Success { result }) => client
-                        .complete_activity_task(task_token.clone(), result.map(Into::into))
-                        .await
-                        .err(),
+                    aer::Status::Completed(ar::Success { result }) => {
+                        if let Some(sched_time) = act_info
+                            .base
+                            .scheduled_time
+                            .and_then(|st| st.elapsed().ok())
+                        {
+                            act_metrics.act_execution_succeeded(sched_time);
+                        }
+                        client
+                            .complete_activity_task(task_token.clone(), result.map(Into::into))
+                            .await
+                            .err()
+                    }
                     aer::Status::Failed(ar::Failure { failure }) => {
                         act_metrics.act_execution_failed();
                         client

core/src/worker/activities/local_activities.rs

@@ -2,6 +2,7 @@ use crate::{
     abstractions::{dbg_panic, MeteredPermitDealer, OwnedMeteredSemPermit, UsedMeteredSemPermit},
     protosext::ValidScheduleLA,
     retry_logic::RetryPolicyExt,
+    telemetry::metrics::{activity_type, local_activity_worker_type, workflow_type},
     worker::workflow::HeartbeatTimeoutMsg,
     MetricsContext, TaskToken,
 };
@@ -173,6 +174,9 @@ pub(crate) struct LocalActivityManager {
     rcvs: tokio::sync::Mutex<RcvChans>,
     shutdown_complete_tok: CancellationToken,
     dat: Mutex<LAMData>,
+    /// Note that these metrics do *not* include the `worker_type` label, as every metric
+    /// emitted here is already specific to local activities via the metric name.
+    metrics: MetricsContext,
 }
 
 struct LocalActivityInfo {
@@ -213,7 +217,11 @@ impl LocalActivityManager {
         let (act_req_tx, act_req_rx) = unbounded_channel();
         let (cancels_req_tx, cancels_req_rx) = unbounded_channel();
         let shutdown_complete_tok = CancellationToken::new();
-        let semaphore = MeteredPermitDealer::new(slot_supplier, metrics_context, None);
+        let semaphore = MeteredPermitDealer::new(
+            slot_supplier,
+            metrics_context.with_new_attrs([local_activity_worker_type()]),
+            None,
+        );
         Self {
             namespace,
             rcvs: tokio::sync::Mutex::new(RcvChans::new(
@@ -233,6 +241,7 @@ impl LocalActivityManager {
                 next_tt_num: 0,
             }),
             workflows_have_shut_down: Default::default(),
+            metrics: metrics_context,
         }
     }
 
@@ -479,6 +488,12 @@ impl LocalActivityManager {
         );
 
         let (schedule_to_close, start_to_close) = sa.close_timeouts.into_sched_and_start();
+        self.metrics
+            .with_new_attrs([
+                activity_type(sa.activity_type.clone()),
+                workflow_type(new_la.workflow_type.clone()),
+            ])
+            .la_executed();
         Some(NextPendingLAAction::Dispatch(ActivityTask {
             task_token: tt.0,
             variant: Some(activity_task::Variant::Start(Start {
@@ -500,9 +515,7 @@ impl LocalActivityManager {
                     .ok(),
                 start_to_close_timeout: start_to_close
                     .or(schedule_to_close)
-                    .unwrap()
-                    .try_into()
-                    .ok(),
+                    .and_then(|t| t.try_into().ok()),
                 heartbeat_timeout: None,
                 retry_policy: Some(sa.retry_policy),
                 is_local: true,
@@ -536,6 +549,11 @@ impl LocalActivityManager {
                 }
             }
 
+            let la_metrics = self.metrics.with_new_attrs([
+                activity_type(info.la_info.schedule_cmd.activity_type.clone()),
+                workflow_type(info.la_info.workflow_type.clone()),
+            ]);
+
             enum Outcome {
                 FailurePath { backoff: Option<Duration> },
                 JustReport,
@@ -553,31 +571,48 @@ impl LocalActivityManager {
             }
 
             let mut is_timeout = false;
+            let runtime = info.dispatch_time.elapsed();
+            la_metrics.la_exec_latency(runtime);
             let outcome = match &status {
-                LocalActivityExecutionResult::Failed(fail) => Outcome::FailurePath {
-                    backoff: calc_backoff!(fail),
-                },
-                LocalActivityExecutionResult::TimedOut(fail)
-                    if matches!(status.get_timeout_type(), Some(TimeoutType::StartToClose)) =>
-                {
-                    // Start to close timeouts are retryable, other timeout types aren't.
-                    is_timeout = true;
+                LocalActivityExecutionResult::Failed(fail) => {
+                    la_metrics.la_execution_failed();
                     Outcome::FailurePath {
                         backoff: calc_backoff!(fail),
                     }
                 }
-                LocalActivityExecutionResult::TimedOut(_) => {
+                LocalActivityExecutionResult::TimedOut(fail) => {
+                    la_metrics.la_execution_failed();
                     is_timeout = true;
+                    // Start to close timeouts are retryable, other timeout types aren't.
+                    if matches!(status.get_timeout_type(), Some(TimeoutType::StartToClose)) {
+                        Outcome::FailurePath {
+                            backoff: calc_backoff!(fail),
+                        }
+                    } else {
+                        Outcome::JustReport
+                    }
+                }
+                LocalActivityExecutionResult::Completed(_) => {
+                    if let Some(rt) = info
+                        .la_info
+                        .schedule_cmd
+                        .original_schedule_time
+                        .and_then(|t| t.elapsed().ok())
+                    {
+                        la_metrics.la_exec_succeeded_latency(rt);
+                    }
+                    Outcome::JustReport
+                }
+                LocalActivityExecutionResult::Cancelled { .. } => {
+                    la_metrics.la_execution_cancelled();
                     Outcome::JustReport
                 }
-                LocalActivityExecutionResult::Completed(_)
-                | LocalActivityExecutionResult::Cancelled { .. } => Outcome::JustReport,
             };
 
             let mut resolution = LocalActivityResolution {
                 seq: info.la_info.schedule_cmd.seq,
                 result: status,
-                runtime: info.dispatch_time.elapsed(),
+                runtime,
                 attempt: info.attempt,
                 backoff: None,
                 original_schedule_time: info.la_info.schedule_cmd.original_schedule_time,

core/src/worker/mod.rs

@@ -26,8 +26,8 @@ use crate::{
     protosext::validate_activity_completion,
     telemetry::{
         metrics::{
-            activity_poller, activity_worker_type, local_activity_worker_type, workflow_poller,
-            workflow_sticky_poller, workflow_worker_type, MetricsContext,
+            activity_poller, activity_worker_type, workflow_poller, workflow_sticky_poller,
+            workflow_worker_type, MetricsContext,
         },
         TelemetryInstance,
     },
@@ -385,7 +385,7 @@ impl Worker {
             tuner.local_activity_slot_supplier(),
             config.namespace.clone(),
             hb_tx,
-            metrics.with_new_attrs([local_activity_worker_type()]),
+            metrics.clone(),
         ));
         let at_task_mgr = act_poller.map(|ap| {
             WorkerActivityTasks::new(