re_grpc: wire up OSS RequestMetadata

app/buck2_action_impl/src/actions/impls/run.rs

@@ -381,6 +381,7 @@ enum ExecuteResult {
         executor_preference: ExecutorPreference,
         action_and_blobs: ActionDigestAndBlobs,
         input_files_bytes: u64,
+        request: CommandExecutionRequest,
     },
 }
 
@@ -1131,6 +1132,7 @@ impl RunAction {
             executor_preference: req.executor_preference,
             action_and_blobs,
             input_files_bytes: req.paths().input_files_bytes(),
+            request: req,
         })
     }
 
@@ -1446,6 +1448,7 @@ impl Action for RunAction {
             executor_preference,
             action_and_blobs,
             input_files_bytes,
+            request,
         ) = match self.execute_inner(ctx).await? {
             ExecuteResult::LocalDepFileHit(outputs, metadata) => {
                 return Ok((outputs, metadata));
@@ -1456,12 +1459,14 @@ impl Action for RunAction {
                 executor_preference,
                 action_and_blobs,
                 input_files_bytes,
+                request,
             } => (
                 result,
                 dep_file_bundle,
                 executor_preference,
                 action_and_blobs,
                 input_files_bytes,
+                request,
             ),
         };
 
@@ -1489,6 +1494,7 @@ impl Action for RunAction {
             let re_result = result.action_result.take();
             let upload_result = ctx
                 .cache_upload(
+                    &request,
                     &action_and_blobs,
                     &result,
                     re_result,

app/buck2_build_api/src/actions.rs

@@ -262,6 +262,7 @@ pub trait ActionExecutionCtx: Send + Sync {
 
     async fn cache_upload(
         &mut self,
+        request: &CommandExecutionRequest,
         action: &ActionDigestAndBlobs,
         execution_result: &CommandExecutionResult,
         re_result: Option<TActionResult2>,

app/buck2_build_api/src/actions/execute/action_execution_target.rs

@@ -86,4 +86,22 @@ impl CommandExecutionTarget for ActionExecutionTarget<'_> {
             identifier: self.action.identifier().unwrap_or("").to_owned(),
         }
     }
+
+    fn action_mnemonic(&self) -> Option<String> {
+        Some(self.action.category().as_str().to_owned())
+    }
+
+    fn target_label(&self) -> Option<String> {
+        self.action
+            .owner()
+            .unpack_target_label()
+            .map(ToString::to_string)
+    }
+
+    fn configuration_hash(&self) -> Option<String> {
+        self.action
+            .owner()
+            .unpack_target_label()
+            .map(|label| label.cfg().output_hash().as_str().to_owned())
+    }
 }

app/buck2_build_api/src/actions/execute/action_executor.rs

@@ -593,6 +593,7 @@ impl ActionExecutionCtx for BuckActionExecutionContext<'_> {
 
     async fn cache_upload(
         &mut self,
+        request: &CommandExecutionRequest,
         action_digest_and_blobs: &ActionDigestAndBlobs,
         execution_result: &CommandExecutionResult,
         re_result: Option<TActionResult2>,
@@ -608,6 +609,7 @@ impl ActionExecutionCtx for BuckActionExecutionContext<'_> {
                     digest_config: self.digest_config(),
                     mergebase: self.mergebase().0.as_ref(),
                     re_platform: self.re_platform(),
+                    paths: request.paths(),
                 },
                 execution_result,
                 re_result,

app/buck2_execute/src/execute/cache_uploader.rs

@@ -17,6 +17,7 @@ use remote_execution::TActionResult2;
 use crate::digest_config::DigestConfig;
 use crate::execute::action_digest_and_blobs::ActionDigestAndBlobs;
 use crate::execute::dep_file_digest::DepFileDigest;
+use crate::execute::request::CommandExecutionPaths;
 use crate::execute::result::CommandExecutionResult;
 use crate::execute::target::CommandExecutionTarget;
 use crate::materialize::materializer::Materializer;
@@ -26,6 +27,7 @@ pub struct CacheUploadInfo<'a> {
     pub digest_config: DigestConfig,
     pub mergebase: &'a Option<String>,
     pub re_platform: &'a remote_execution::Platform,
+    pub paths: &'a CommandExecutionPaths,
 }
 
 #[async_trait]

app/buck2_execute/src/execute/target.rs

@@ -18,4 +18,19 @@ pub trait CommandExecutionTarget: Send + Sync + Debug {
     fn as_proto_action_key(&self) -> buck2_data::ActionKey;
 
     fn as_proto_action_name(&self) -> buck2_data::ActionName;
+
+    /// Optional mnemonic describing the action kind (e.g. `CxxCompile`).
+    fn action_mnemonic(&self) -> Option<String> {
+        None
+    }
+
+    /// Optional configured target label this action belongs to.
+    fn target_label(&self) -> Option<String> {
+        None
+    }
+
+    /// Optional hash identifying the build configuration of the target.
+    fn configuration_hash(&self) -> Option<String> {
+        None
+    }
 }

app/buck2_execute/src/re/action_identity.rs

@@ -28,6 +28,14 @@ pub struct ReActionIdentity<'a> {
     /// Details about the action collected while uploading
     pub paths: &'a CommandExecutionPaths,
 
+    /// Optional action id (usually the action digest hash) used for request metadata.
+    pub action_id: Option<String>,
+
+    /// Optional action mnemonic, target label and configuration hash used for OSS RE metadata.
+    pub action_mnemonic: Option<String>,
+    pub target_label: Option<String>,
+    pub configuration_hash: Option<String>,
+
     //// Trace ID which started the execution of this action, to be added on the RE side
     pub trace_id: TraceId,
 }
@@ -37,19 +45,27 @@ impl<'a> ReActionIdentity<'a> {
         target: &'a dyn CommandExecutionTarget,
         executor_action_key: Option<&str>,
         paths: &'a CommandExecutionPaths,
+        action_id: Option<String>,
     ) -> Self {
         let mut action_key = target.re_action_key();
         if let Some(executor_action_key) = executor_action_key {
             action_key = format!("{executor_action_key} {action_key}");
         }
 
         let trace_id = get_dispatcher().trace_id().to_owned();
+        let action_mnemonic = target.action_mnemonic();
+        let target_label = target.target_label();
+        let configuration_hash = target.configuration_hash();
 
         Self {
             _target: target,
             action_key,
             affinity_key: target.re_affinity_key(),
             paths,
+            action_id,
+            action_mnemonic,
+            target_label,
+            configuration_hash,
             trace_id,
         }
     }

app/buck2_execute/src/re/client.rs

@@ -923,13 +923,16 @@ impl RemoteExecutionClientImpl {
         action_digest: ActionDigest,
         use_case: RemoteExecutorUseCase,
     ) -> buck2_error::Result<Option<ActionResultResponse>> {
+        let mut metadata = use_case.metadata(None);
+        metadata.action_id = Some(action_digest.raw_digest().to_string());
+
         let res = with_error_handler(
             "action_cache",
             self.get_session_id(),
             self.client()
                 .get_action_cache_client()
                 .get_action_result(
-                    use_case.metadata(None),
+                    metadata,
                     ActionResultRequest {
                         digest: action_digest.to_re(),
                         ..Default::default()
@@ -1298,6 +1301,7 @@ impl RemoteExecutionClientImpl {
                 ..Default::default()
             }),
             respect_file_symlinks: Some(self.respect_file_symlinks),
+            action_id: Some(action_digest.raw_digest().to_string()),
             ..use_case.metadata(Some(identity))
         };
 
@@ -1376,13 +1380,17 @@ impl RemoteExecutionClientImpl {
             return Ok((Vec::new(), TLocalCacheStats::default()));
         }
         let expected_blobs = digests.len();
+        let mut metadata = use_case.metadata(identity);
+        metadata.action_id = identity
+            .and_then(|id| id.action_id.clone())
+            .or(metadata.action_id);
         let response = with_error_handler(
             "download_typed_blobs",
             self.get_session_id(),
             self.client()
                 .get_cas_client()
                 .download(
-                    use_case.metadata(identity),
+                    metadata,
                     DownloadRequest {
                         inlined_digests: Some(digests),
                         ..Default::default()
@@ -1420,13 +1428,15 @@ impl RemoteExecutionClientImpl {
         use_case: RemoteExecutorUseCase,
     ) -> buck2_error::Result<(Vec<u8>, TLocalCacheStats)> {
         let re_action = format!("download_blob for digest {digest}");
+        let mut metadata = use_case.metadata(None);
+        metadata.action_id = Some(digest.hash.clone());
         let response = with_error_handler(
             re_action.as_str(),
             self.get_session_id(),
             self.client()
                 .get_cas_client()
                 .download(
-                    use_case.metadata(None),
+                    metadata,
                     DownloadRequest {
                         inlined_digests: Some(vec![digest.clone()]),
                         ..Default::default()
@@ -1601,6 +1611,7 @@ impl RemoteExecutionClientImpl {
                             attributes,
                             ..Default::default()
                         }),
+                        action_id: Some(digest.raw_digest().to_string()),
                         ..use_case.metadata(None)
                     },
                     WriteActionResultRequest {

app/buck2_execute/src/re/metadata.rs

@@ -30,14 +30,18 @@ impl RemoteExecutionMetadataExt for RemoteExecutorUseCase {
         RemoteExecutionMetadata {
             use_case_id: self.as_str().to_owned(),
             buck_info: Some(BuckInfo {
-                build_id: trace_id,
+                build_id: trace_id.clone(),
                 ..Default::default()
             }),
+            correlated_invocations_id: Some(trace_id),
             action_history_info: identity.map(|identity| ActionHistoryInfo {
                 action_key: identity.action_key.clone(),
                 disable_retry_on_oom: false,
                 ..Default::default()
             }),
+            action_mnemonic: identity.and_then(|identity| identity.action_mnemonic.clone()),
+            target_id: identity.and_then(|identity| identity.target_label.clone()),
+            configuration_id: identity.and_then(|identity| identity.configuration_hash.clone()),
             ..Default::default()
         }
     }

app/buck2_execute/src/re/uploader.rs

@@ -175,7 +175,10 @@ impl Uploader {
             })
         } else {
             let client = client.clone();
-            let metadata = use_case.metadata(identity);
+            let mut metadata = use_case.metadata(identity);
+            if let Some(id) = identity.and_then(|id| id.action_id.clone()) {
+                metadata.action_id = Some(id);
+            }
             let request = GetDigestsTtlRequest {
                 digests: input_digests.iter().map(|d| d.to_re()).collect(),
                 ..Default::default()
@@ -432,12 +435,17 @@ impl Uploader {
 
         // Upload
         if !upload_files.is_empty() || !upload_blobs.is_empty() {
+            let mut metadata = use_case.metadata(identity);
+            if let Some(id) = identity.and_then(|id| id.action_id.clone()) {
+                metadata.action_id = Some(id);
+            }
             with_error_handler(
                 "upload",
                 client.get_session_id(),
-                client.get_raw_re_client()
+                client
+                    .get_raw_re_client()
                     .upload(
-                        use_case.metadata(identity),
+                        metadata,
                         UploadRequest {
                             files_with_digest: Some(upload_files),
                             inlined_blobs_with_digest: Some(upload_blobs),
@@ -666,7 +674,10 @@ fn query_digest_ttls<'s>(
     input_digests: Vec<TrackedFileDigest>,
 ) -> BoxFuture<'s, buck2_error::Result<HashMap<TrackedFileDigest, i64>>> {
     let client = client.dupe();
-    let metadata = use_case.metadata(identity);
+    let mut metadata = use_case.metadata(identity);
+    if let Some(id) = identity.and_then(|id| id.action_id.clone()) {
+        metadata.action_id = Some(id);
+    }
     let request = GetDigestsTtlRequest {
         digests: input_digests.iter().map(|d| d.to_re()).collect(),
         ..Default::default()

app/buck2_execute_impl/src/executors/action_cache.rs

@@ -107,7 +107,12 @@ async fn query_action_cache_and_download_result(
     )
     .await;
 
-    let identity = None; // TODO(#503): implement this
+    let identity = Some(ReActionIdentity::new(
+        command.target,
+        re_action_key.as_deref(),
+        command.request.paths(),
+        Some(action_digest.raw_digest().to_string()),
+    ));
     if upload_all_actions {
         match re_client
             .upload(
@@ -116,7 +121,7 @@ async fn query_action_cache_and_download_result(
                 action_blobs,
                 ProjectRelativePath::empty(),
                 request.paths().input_directory(),
-                identity,
+                identity.as_ref(),
                 digest_config,
                 deduplicate_get_digests_ttl_calls,
             )
@@ -167,6 +172,7 @@ async fn query_action_cache_and_download_result(
         command.target,
         re_action_key.as_deref(),
         command.request.paths(),
+        Some(action_digest.raw_digest().to_string()),
     );
 
     let response = ActionCacheResult(response, cache_type.to_proto());