Add cache and runtime metadata to Process workunits (#12469)

To improve our ability to debug cache hits/misses, and to determine what the upper bound is on "time saved", this change:
* Adds a `source` field to `ProcessResultMetadata` and to `Process` workunits.
* Records the full serialized definition of the `Process` on the workunit.
* Adds metrics for the CPU time of actually running a `Process` (locally or remotely).

Author: stuhood

src/rust/engine/Cargo.lock

@@ -134,6 +134,7 @@ sharded_lmdb = { path = "sharded_lmdb" }
 smallvec = "0.6"
 stdio = { path = "stdio" }
 store = { path = "fs/store" }
+serde_json = "1.0"
 task_executor = { path = "task_executor" }
 tempfile = "3"
 testutil_mock = { package = "mock", path = "testutil/mock" }

src/rust/engine/Cargo.toml

@@ -16,6 +16,7 @@ lazy_static = "1"
 log = "0.4"
 parking_lot = "0.11"
 rlimit = "0.3"
+serde = "1.0.104"
 task_executor = { path = "../task_executor" }
 
 [dev-dependencies]

src/rust/engine/fs/Cargo.toml

@@ -51,6 +51,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use futures::future::{self, TryFutureExt};
 use lazy_static::lazy_static;
+use serde::Serialize;
 
 lazy_static! {
   static ref EMPTY_IGNORE: Arc<GitignoreStyleExcludes> = Arc::new(GitignoreStyleExcludes {
@@ -74,7 +75,7 @@ pub fn default_cache_path() -> PathBuf {
   cache_path.join("pants")
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)]
+#[derive(Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash, Serialize)]
 pub struct RelativePath(PathBuf);
 
 impl RelativePath {

src/rust/engine/fs/src/lib.rs

@@ -17,7 +17,7 @@ use workunit_store::{
 
 use crate::{
   Context, FallibleProcessResultWithPlatform, MultiPlatformProcess, Platform, Process,
-  ProcessCacheScope, ProcessMetadata,
+  ProcessCacheScope, ProcessMetadata, ProcessResultSource,
 };
 
 #[allow(dead_code)]
@@ -182,6 +182,7 @@ impl CommandRunner {
           action_result,
           platform,
           true,
+          ProcessResultSource::HitLocally,
         )
         .await?
       } else {

src/rust/engine/process_execution/src/cache.rs

@@ -142,7 +142,7 @@ impl TryFrom<String> for Platform {
   }
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize)]
 pub enum ProcessCacheScope {
   // Cached in all locations, regardless of success or failure.
   Always,
@@ -173,10 +173,14 @@ impl TryFrom<String> for ProcessCacheScope {
   }
 }
 
+fn serialize_level<S: serde::Serializer>(level: &log::Level, s: S) -> Result<S::Ok, S::Error> {
+  s.serialize_str(&level.to_string())
+}
+
 ///
 /// A process to be executed.
 ///
-#[derive(Derivative, Clone, Debug, Eq)]
+#[derive(Derivative, Clone, Debug, Eq, Serialize)]
 #[derivative(PartialEq, Hash)]
 pub struct Process {
   ///
@@ -216,6 +220,8 @@ pub struct Process {
   #[derivative(PartialEq = "ignore", Hash = "ignore")]
   pub description: String,
 
+  // NB: We serialize with a function to avoid adding a serde dep to the logging crate.
+  #[serde(serialize_with = "serialize_level")]
   pub level: log::Level,
 
   ///
@@ -391,17 +397,40 @@ pub struct FallibleProcessResultWithPlatform {
 
 /// Metadata for a ProcessResult corresponding to the REAPI `ExecutedActionMetadata` proto. This
 /// conversion is lossy, but the interesting parts are preserved.
-#[derive(Clone, Debug, Default, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub struct ProcessResultMetadata {
   /// The time from starting to completion, including preparing the chroot and cleanup.
   /// Corresponds to `worker_start_timestamp` and `worker_completed_timestamp` from
   /// `ExecutedActionMetadata`.
+  ///
+  /// NB: This is optional because the REAPI does not guarantee that it is returned.
   pub total_elapsed: Option<Duration>,
+  /// The source of the result.
+  pub source: ProcessResultSource,
 }
 
 impl ProcessResultMetadata {
-  pub fn new(total_elapsed: Option<Duration>) -> Self {
-    ProcessResultMetadata { total_elapsed }
+  pub fn new(total_elapsed: Option<Duration>, source: ProcessResultSource) -> Self {
+    ProcessResultMetadata {
+      total_elapsed,
+      source,
+    }
+  }
+
+  pub fn new_from_metadata(metadata: ExecutedActionMetadata, source: ProcessResultSource) -> Self {
+    let total_elapsed = match (
+      metadata.worker_start_timestamp,
+      metadata.worker_completed_timestamp,
+    ) {
+      (Some(started), Some(completed)) => TimeSpan::from_start_and_end(&started, &completed, "")
+        .map(|span| span.duration)
+        .ok(),
+      _ => None,
+    };
+    Self {
+      total_elapsed,
+      source,
+    }
   }
 
   /// How much faster a cache hit was than running the process again.
@@ -429,21 +458,6 @@ impl ProcessResultMetadata {
   }
 }
 
-impl From<ExecutedActionMetadata> for ProcessResultMetadata {
-  fn from(metadata: ExecutedActionMetadata) -> Self {
-    let total_elapsed = match (
-      metadata.worker_start_timestamp,
-      metadata.worker_completed_timestamp,
-    ) {
-      (Some(started), Some(completed)) => TimeSpan::from_start_and_end(&started, &completed, "")
-        .map(|span| span.duration)
-        .ok(),
-      _ => None,
-    };
-    Self { total_elapsed }
-  }
-}
-
 impl From<ProcessResultMetadata> for ExecutedActionMetadata {
   fn from(metadata: ProcessResultMetadata) -> ExecutedActionMetadata {
     let (total_start, total_end) = match metadata.total_elapsed {
@@ -470,6 +484,14 @@ impl From<ProcessResultMetadata> for ExecutedActionMetadata {
   }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ProcessResultSource {
+  RanLocally,
+  RanRemotely,
+  HitLocally,
+  HitRemotely,
+}
+
 #[derive(Clone)]
 pub struct Context {
   workunit_store: WorkunitStore,

src/rust/engine/process_execution/src/lib.rs

@@ -33,7 +33,7 @@ use workunit_store::{in_workunit, Metric, RunningWorkunit, WorkunitMetadata};
 
 use crate::{
   Context, FallibleProcessResultWithPlatform, MultiPlatformProcess, NamedCaches, Platform, Process,
-  ProcessResultMetadata,
+  ProcessResultMetadata, ProcessResultSource,
 };
 
 pub const USER_EXECUTABLE_MODE: u32 = 0o100755;
@@ -590,7 +590,8 @@ pub trait CapturedWorkdir {
     }
 
     let elapsed = start_time.elapsed();
-    let result_metadata = ProcessResultMetadata::new(Some(elapsed.into()));
+    let result_metadata =
+      ProcessResultMetadata::new(Some(elapsed.into()), ProcessResultSource::RanLocally);
 
     match child_results_result {
       Ok(child_results) => {

src/rust/engine/process_execution/src/local.rs

@@ -1,11 +1,12 @@
 use std::collections::BTreeMap;
 use std::path::PathBuf;
 
-use fs::default_cache_path;
+use serde::Serialize;
 
 use crate::RelativePath;
+use fs::default_cache_path;
 
-#[derive(Clone, Debug, Eq, PartialEq, Hash, PartialOrd, Ord)]
+#[derive(Clone, Debug, Eq, PartialEq, Hash, PartialOrd, Ord, Serialize)]
 pub struct CacheName(String);
 
 impl CacheName {
@@ -24,7 +25,7 @@ impl CacheName {
   }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Hash)]
+#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize)]
 pub struct CacheDest(String);
 
 impl CacheDest {

src/rust/engine/process_execution/src/named_caches.rs

@@ -43,7 +43,7 @@ use workunit_store::{
 
 use crate::{
   Context, FallibleProcessResultWithPlatform, MultiPlatformProcess, Platform, Process,
-  ProcessCacheScope, ProcessMetadata, ProcessResultMetadata,
+  ProcessCacheScope, ProcessMetadata, ProcessResultMetadata, ProcessResultSource,
 };
 
 // Environment variable which is exclusively used for cache key invalidation.
@@ -491,6 +491,11 @@ impl CommandRunner {
             action_result,
             self.platform,
             false,
+            if execute_response.cached_result {
+              ProcessResultSource::HitRemotely
+            } else {
+              ProcessResultSource::RanRemotely
+            },
           )
           .await
           .map_err(ExecutionError::Fatal);
@@ -1065,7 +1070,7 @@ pub async fn populate_fallible_execution_result_for_timeout(
     exit_code: -libc::SIGTERM,
     output_directory: hashing::EMPTY_DIGEST,
     platform,
-    metadata: ProcessResultMetadata::default(),
+    metadata: ProcessResultMetadata::new(Some(elapsed.into()), ProcessResultSource::RanRemotely),
   })
 }
 
@@ -1081,6 +1086,7 @@ pub fn populate_fallible_execution_result(
   action_result: &remexec::ActionResult,
   platform: Platform,
   treat_tree_digest_as_final_directory_hack: bool,
+  source: ProcessResultSource,
 ) -> BoxFuture<Result<FallibleProcessResultWithPlatform, String>> {
   future::try_join3(
     extract_stdout(&store, action_result),
@@ -1102,7 +1108,9 @@ pub fn populate_fallible_execution_result(
         metadata: action_result
           .execution_metadata
           .clone()
-          .map_or(ProcessResultMetadata::default(), |metadata| metadata.into()),
+          .map_or(ProcessResultMetadata::new(None, source), |metadata| {
+            ProcessResultMetadata::new_from_metadata(metadata, source)
+          }),
       })
     },
   )
@@ -1372,9 +1380,14 @@ pub async fn check_action_cache(
       match action_result_response {
         Ok(action_result) => {
           let action_result = action_result.into_inner();
-          let response =
-            populate_fallible_execution_result(store.clone(), &action_result, platform, false)
-              .await?;
+          let response = populate_fallible_execution_result(
+            store.clone(),
+            &action_result,
+            platform,
+            false,
+            ProcessResultSource::HitRemotely,
+          )
+          .await?;
           // TODO: This should move inside the retry_call above, both in order to be retried, and
           // to ensure that we increment a miss if we fail to eagerly fetch.
           if eager_fetch {

src/rust/engine/process_execution/src/remote.rs

@@ -21,7 +21,7 @@ use crate::remote::{ensure_action_stored_locally, make_execute_request};
 use crate::{
   CommandRunner as CommandRunnerTrait, Context, FallibleProcessResultWithPlatform,
   MultiPlatformProcess, Platform, Process, ProcessMetadata, ProcessResultMetadata,
-  RemoteCacheWarningsBehavior,
+  ProcessResultSource, RemoteCacheWarningsBehavior,
 };
 
 /// A mock of the local runner used for better hermeticity of the tests.
@@ -45,7 +45,7 @@ impl MockLocalCommandRunner {
         exit_code,
         output_directory: EMPTY_DIGEST,
         platform: Platform::current().unwrap(),
-        metadata: ProcessResultMetadata::default(),
+        metadata: ProcessResultMetadata::new(None, ProcessResultSource::RanLocally),
       }),
       call_counter,
       delay: Duration::from_millis(delay_ms),
@@ -612,7 +612,7 @@ async fn make_action_result_basic() {
     output_directory: directory_digest,
     exit_code: 102,
     platform: Platform::Linux_x86_64,
-    metadata: ProcessResultMetadata::default(),
+    metadata: ProcessResultMetadata::new(None, ProcessResultSource::RanLocally),
   };
 
   let (action_result, digests) = runner

src/rust/engine/process_execution/src/remote_cache_tests.rs

@@ -5,7 +5,7 @@ use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 use std::time::Duration;
 
-use crate::{Process, ProcessResultMetadata};
+use crate::{Process, ProcessResultMetadata, ProcessResultSource};
 use bazel_protos::gen::build::bazel::remote::execution::v2 as remexec;
 use prost_types::Timestamp;
 use remexec::ExecutedActionMetadata;
@@ -59,10 +59,14 @@ fn process_result_metadata_to_and_from_executed_action_metadata() {
     ..ExecutedActionMetadata::default()
   };
 
-  let converted_process_result: ProcessResultMetadata = action_metadata.into();
+  let converted_process_result: ProcessResultMetadata =
+    ProcessResultMetadata::new_from_metadata(action_metadata, ProcessResultSource::RanLocally);
   assert_eq!(
     converted_process_result,
-    ProcessResultMetadata::new(Some(concrete_time::Duration::new(20, 30)))
+    ProcessResultMetadata::new(
+      Some(concrete_time::Duration::new(20, 30)),
+      ProcessResultSource::RanLocally
+    )
   );
 
   // The conversion from `ExecutedActionMetadata` to `ProcessResultMetadata` is lossy.
@@ -83,26 +87,37 @@ fn process_result_metadata_to_and_from_executed_action_metadata() {
   );
 
   // The relevant metadata may be missing from either type.
-  let action_metadata_missing: ProcessResultMetadata = ExecutedActionMetadata::default().into();
-  assert_eq!(action_metadata_missing, ProcessResultMetadata::default());
-  let process_result_missing: ExecutedActionMetadata = ProcessResultMetadata::default().into();
+  let empty = ProcessResultMetadata::new(None, ProcessResultSource::RanLocally);
+  let action_metadata_missing: ProcessResultMetadata = ProcessResultMetadata::new_from_metadata(
+    ExecutedActionMetadata::default(),
+    ProcessResultSource::RanLocally,
+  );
+  assert_eq!(action_metadata_missing, empty);
+  let process_result_missing: ExecutedActionMetadata = empty.into();
   assert_eq!(process_result_missing, ExecutedActionMetadata::default());
 }
 
 #[test]
 fn process_result_metadata_time_saved_from_cache() {
-  let metadata = ProcessResultMetadata::new(Some(concrete_time::Duration::new(5, 150)));
+  let metadata = ProcessResultMetadata::new(
+    Some(concrete_time::Duration::new(5, 150)),
+    ProcessResultSource::RanLocally,
+  );
   let time_saved = metadata.time_saved_from_cache(Duration::new(1, 100));
   assert_eq!(time_saved, Some(Duration::new(4, 50)));
 
   // If the cache lookup took more time than the process, we return 0.
-  let metadata = ProcessResultMetadata::new(Some(concrete_time::Duration::new(1, 0)));
+  let metadata = ProcessResultMetadata::new(
+    Some(concrete_time::Duration::new(1, 0)),
+    ProcessResultSource::RanLocally,
+  );
   let time_saved = metadata.time_saved_from_cache(Duration::new(5, 0));
   assert_eq!(time_saved, Some(Duration::new(0, 0)));
 
   // If the original process time wasn't recorded, we can't compute the time saved.
   assert_eq!(
-    ProcessResultMetadata::default().time_saved_from_cache(Duration::new(1, 100)),
+    ProcessResultMetadata::new(None, ProcessResultSource::RanLocally)
+      .time_saved_from_cache(Duration::new(1, 100)),
     None
   );
 }