Add remote workunits for Zipkin trace

src/rust/engine/Cargo.toml

@@ -97,6 +97,7 @@ rule_graph = { path = "rule_graph" }
 smallvec = "0.6"
 store = { path = "fs/store" }
 tempfile = "3"
+time = "0.1.40"
 ui = { path = "ui" }
 url = "1.7.1"
 task_executor = { path = "task_executor" }

src/rust/engine/engine_cffi/Cargo.toml

@@ -17,6 +17,7 @@ logging = { path = "../logging" }
 rule_graph = { path = "../rule_graph" }
 store = { path = "../fs/store" }
 tar_api = { path = "../tar_api" }
+time = "0.1.40"
 workunit_store = { path = "../workunit_store" }
 
 [build-dependencies]

src/rust/engine/engine_cffi/src/lib.rs

@@ -60,7 +60,7 @@ use std::os::raw;
 use std::panic;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
-use workunit_store::WorkUnitStore;
+use time::Timespec;
 
 // TODO: Consider renaming and making generic for collections of PyResults.
 #[repr(C)]
@@ -331,20 +331,25 @@ pub extern "C" fn scheduler_metrics(
         .collect::<Vec<_>>();
       if session.should_record_zipkin_spans() {
         let workunits = session
+          .workunit_store()
           .get_workunits()
           .lock()
           .iter()
           .map(|workunit| {
-            let workunit_zipkin_trace_info = vec![
+            let mut workunit_zipkin_trace_info = vec![
               externs::store_utf8("name"),
               externs::store_utf8(&workunit.name),
               externs::store_utf8("start_timestamp"),
-              externs::store_f64(workunit.start_timestamp),
+              externs::store_f64(timespec_as_float_secs(&workunit.start_timestamp)),
               externs::store_utf8("end_timestamp"),
-              externs::store_f64(workunit.end_timestamp),
+              externs::store_f64(timespec_as_float_secs(&workunit.end_timestamp)),
               externs::store_utf8("span_id"),
               externs::store_utf8(&workunit.span_id),
             ];
+            if let Some(parent_id) = &workunit.parent_id {
+              workunit_zipkin_trace_info.push(externs::store_utf8("parent_id"));
+              workunit_zipkin_trace_info.push(externs::store_utf8(parent_id));
+            }
             externs::store_dict(&workunit_zipkin_trace_info)
           })
           .collect::<Vec<_>>();
@@ -356,6 +361,13 @@ pub extern "C" fn scheduler_metrics(
   })
 }
 
+fn timespec_as_float_secs(timespec: &Timespec) -> f64 {
+  //  Reverting time from Timespec to f64 decreases precision.
+  let whole_secs = timespec.sec as f64;
+  let fract_part_in_nanos = f64::from(timespec.nsec);
+  whole_secs + fract_part_in_nanos / 1_000_000_000.0
+}
+
 ///
 /// Prepares to fork by shutting down any background threads used for execution, and then
 /// calling the given callback function (which should execute the fork) while holding exclusive

src/rust/engine/process_execution/Cargo.toml

@@ -26,8 +26,10 @@ time = "0.1.40"
 tokio-codec = "0.1"
 tokio-process = "0.2.1"
 tokio-timer = "0.2"
+workunit_store = { path = "../workunit_store" }
 
 [dev-dependencies]
+maplit = "1.0.1"
 mock = { path = "../testutil/mock" }
 tempfile = "3"
 testutil = { path = "../testutil" }

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

@@ -37,6 +37,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 use store::UploadSummary;
+use workunit_store::WorkUnitStore;
 
 use async_semaphore::AsyncSemaphore;
 
@@ -133,7 +134,11 @@ impl AddAssign<UploadSummary> for ExecutionStats {
 }
 
 pub trait CommandRunner: Send + Sync {
-  fn run(&self, req: ExecuteProcessRequest) -> BoxFuture<FallibleExecuteProcessResult, String>;
+  fn run(
+    &self,
+    req: ExecuteProcessRequest,
+    workunit_store: WorkUnitStore,
+  ) -> BoxFuture<FallibleExecuteProcessResult, String>;
 }
 
 ///
@@ -153,9 +158,16 @@ impl BoundedCommandRunner {
 }
 
 impl CommandRunner for BoundedCommandRunner {
-  fn run(&self, req: ExecuteProcessRequest) -> BoxFuture<FallibleExecuteProcessResult, String> {
+  fn run(
+    &self,
+    req: ExecuteProcessRequest,
+    workunit_store: WorkUnitStore,
+  ) -> BoxFuture<FallibleExecuteProcessResult, String> {
     let inner = self.inner.clone();
-    self.inner.1.with_acquired(move || inner.0.run(req))
+    self
+      .inner
+      .1
+      .with_acquired(move || inner.0.run(req, workunit_store))
   }
 }
 

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

@@ -21,6 +21,7 @@ use tokio_process::CommandExt;
 use super::{ExecuteProcessRequest, FallibleExecuteProcessResult};
 
 use bytes::{Bytes, BytesMut};
+use workunit_store::WorkUnitStore;
 
 pub struct CommandRunner {
   store: Store,
@@ -212,7 +213,13 @@ impl super::CommandRunner for CommandRunner {
   ///
   /// Runs a command on this machine in the passed working directory.
   ///
-  fn run(&self, req: ExecuteProcessRequest) -> BoxFuture<FallibleExecuteProcessResult, String> {
+  /// TODO: start to create workunits for local process execution
+  ///
+  fn run(
+    &self,
+    req: ExecuteProcessRequest,
+    _workunit_store: WorkUnitStore,
+  ) -> BoxFuture<FallibleExecuteProcessResult, String> {
     let workdir = try_future!(tempfile::Builder::new()
       .prefix("process-execution")
       .tempdir_in(&self.work_dir)
@@ -348,6 +355,7 @@ mod tests {
   use testutil::data::{TestData, TestDirectory};
   use testutil::path::find_bash;
   use testutil::{as_bytes, owned_string_vec};
+  use workunit_store::WorkUnitStore;
 
   #[test]
   #[cfg(unix)]
@@ -905,6 +913,6 @@ mod tests {
       work_dir: dir,
       cleanup_local_dirs: cleanup,
     };
-    executor.block_on(runner.run(req))
+    executor.block_on(runner.run(req, WorkUnitStore::new()))
   }
 }