Wait for target to be running in capture manager (#998)

* Add target running signal, wait on it in capture manager

* Support no-target mode

Author: GeorgeHahn

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 - Warmup period is now respected when container targeting is in use.
+- Capture manager waits for the target to start running before recording data.
 
 ## [0.23.1]
 ### Fixed

lading/src/bin/lading.rs

@@ -347,6 +347,7 @@ async fn inner_main(
 ) -> Result<(), Error> {
     let (shutdown_watcher, shutdown_broadcast) = lading_signal::signal();
     let (experiment_started_watcher, experiment_started_broadcast) = lading_signal::signal();
+    let (target_running_watcher, target_running_broadcast) = lading_signal::signal();
 
     // Set up the telemetry sub-system.
     //
@@ -390,6 +391,7 @@ async fn inner_main(
                 path,
                 shutdown_watcher.register()?,
                 experiment_started_watcher.clone(),
+                target_running_watcher.clone(),
             )
             .await?;
             for (k, v) in global_labels {
@@ -474,8 +476,6 @@ async fn inner_main(
         }
     }
 
-    let mut sequence_tgt_recv = tgt_snd.subscribe();
-
     let mut tsrv_joinset = tokio::task::JoinSet::new();
     let mut osrv_joinset = tokio::task::JoinSet::new();
     //
@@ -491,20 +491,20 @@ async fn inner_main(
         // TARGET
         //
         let target_server = target::Server::new(target, shutdown_watcher.clone());
-        tsrv_joinset.spawn(target_server.run(tgt_snd));
+        tsrv_joinset.spawn(target_server.run(tgt_snd, target_running_broadcast));
     } else {
-        // Many lading servers synchronize on target startup.
+        // Many lading servers synchronize on target startup using the PID sender. Some by necessity, others by legacy.
         tgt_snd.send(None)?;
+        // Newer usage prefers the `target_running` signal where the PID isn't needed.
+        target_running_broadcast.signal();
     };
 
     let (timer_watcher, timer_broadcast) = lading_signal::signal();
 
     tokio::spawn(
         async move {
             info!("waiting for target startup");
-            if let Err(e) = sequence_tgt_recv.recv().await {
-                warn!("failed to wait: {:?}", e);
-            }
+            target_running_watcher.recv().await;
             info!("target is running, now sleeping for warmup");
             sleep(warmup_duration).await;
             experiment_started_broadcast.signal();

lading/src/captures.rs

@@ -59,6 +59,7 @@ pub struct CaptureManager {
     capture_path: PathBuf,
     shutdown: lading_signal::Watcher,
     _experiment_started: lading_signal::Watcher,
+    target_running: lading_signal::Watcher,
     inner: Arc<Inner>,
     global_labels: FxHashMap<String, String>,
 }
@@ -73,6 +74,7 @@ impl CaptureManager {
         capture_path: PathBuf,
         shutdown: lading_signal::Watcher,
         experiment_started: lading_signal::Watcher,
+        target_running: lading_signal::Watcher,
     ) -> Result<Self, io::Error> {
         let fp = tokio::fs::File::create(&capture_path).await?;
         let fp = fp.into_std().await;
@@ -83,6 +85,7 @@ impl CaptureManager {
             capture_path,
             shutdown,
             _experiment_started: experiment_started,
+            target_running,
             inner: Arc::new(Inner {
                 registry: Registry::atomic(),
             }),
@@ -190,6 +193,9 @@ impl CaptureManager {
         std::thread::Builder::new()
             .name("capture-manager".into())
             .spawn(move || {
+                while let Ok(false) = self.target_running.try_recv() {
+                    std::thread::sleep(Duration::from_millis(100));
+                }
                 loop {
                     if self.shutdown.try_recv().expect("polled after signal") {
                         info!("shutdown signal received");

lading/src/target.rs

@@ -25,6 +25,7 @@ use std::{
 };
 
 use bollard::Docker;
+use lading_signal::Broadcaster;
 use metrics::gauge;
 use nix::{
     errno::Errno,
@@ -221,7 +222,11 @@ impl Server {
     /// # Panics
     ///
     /// None are known.
-    pub async fn run(self, pid_snd: TargetPidSender) -> Result<(), Error> {
+    pub async fn run(
+        self,
+        pid_snd: TargetPidSender,
+        target_running: Broadcaster,
+    ) -> Result<(), Error> {
         let config = self.config;
 
         // Note that each target mode has different expectations around target
@@ -230,13 +235,13 @@ impl Server {
         // signalled to exit.
         match config {
             Config::Pid(config) => {
-                Self::watch(config, pid_snd, self.shutdown).await?;
+                Self::watch(config, pid_snd, target_running, self.shutdown).await?;
             }
             Config::Docker(config) => {
-                Self::watch_container(config, pid_snd, self.shutdown).await?;
+                Self::watch_container(config, pid_snd, target_running, self.shutdown).await?;
             }
             Config::Binary(config) => {
-                Self::execute_binary(config, pid_snd, self.shutdown).await?;
+                Self::execute_binary(config, pid_snd, target_running, self.shutdown).await?;
             }
         }
 
@@ -248,6 +253,7 @@ impl Server {
     async fn watch_container(
         config: DockerConfig,
         pid_snd: TargetPidSender,
+        target_running: Broadcaster,
         shutdown: lading_signal::Watcher,
     ) -> Result<(), Error> {
         let docker = Docker::connect_with_socket_defaults()?;
@@ -269,6 +275,7 @@ impl Server {
             pid.try_into().expect("cannot convert pid to 32 bit type"),
         ))?;
         drop(pid_snd);
+        target_running.signal();
 
         // Use PIDfd to watch the target process (linux kernel 5.3 and up)
         #[cfg(target_os = "linux")]
@@ -329,6 +336,7 @@ impl Server {
     async fn watch(
         config: PidConfig,
         pid_snd: TargetPidSender,
+        target_running: Broadcaster,
         shutdown: lading_signal::Watcher,
     ) -> Result<(), Error> {
         // Convert pid config value to a plain i32 (no truncation concerns;
@@ -348,6 +356,7 @@ impl Server {
 
         pid_snd.send(Some(config.pid.get()))?;
         drop(pid_snd);
+        target_running.signal();
 
         // Use PIDfd to watch the target process (linux kernel 5.3 and up)
         #[cfg(target_os = "linux")]
@@ -406,6 +415,7 @@ impl Server {
     async fn execute_binary(
         config: BinaryConfig,
         pid_snd: TargetPidSender,
+        target_running: Broadcaster,
         shutdown: lading_signal::Watcher,
     ) -> Result<ExitStatus, Error> {
         let mut target_cmd = Command::new(config.command);
@@ -424,6 +434,7 @@ impl Server {
         let target_id = target_child.id().ok_or(Error::ProcessFinished)?;
         pid_snd.send(Some(target_id))?;
         drop(pid_snd);
+        target_running.signal();
 
         let mut interval = time::interval(Duration::from_secs(400));
         let shutdown_wait = shutdown.recv();