Emit structured logs from the agent to help isolate which steps do what

This is stupidly dumping to stdout at the moment, but the raw structure is
there.

Cargo.lock

@@ -567,6 +567,15 @@ dependencies = [
  "vcpkg 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "os_pipe"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "libc 0.2.71 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "osp"
 version = "0.1.0"
@@ -583,6 +592,7 @@ name = "otto-agent"
 version = "0.1.0"
 dependencies = [
  "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "os_pipe 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "osp 0.1.0",
  "serde 1.0.117 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_yaml 0.8.13 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1254,6 +1264,7 @@ dependencies = [
 "checksum openssl 0.10.30 (registry+https://github.com/rust-lang/crates.io-index)" = "8d575eff3665419f9b83678ff2815858ad9d11567e082f5ac1814baba4e2bcb4"
 "checksum openssl-probe 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "77af24da69f9d9341038eba93a073b1fdaaa1b788221b00a69bce9e762cb32de"
 "checksum openssl-sys 0.9.58 (registry+https://github.com/rust-lang/crates.io-index)" = "a842db4709b604f0fe5d1170ae3565899be2ad3d9cbc72dedc789ac0511f78de"
+"checksum os_pipe 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)" = "fb233f06c2307e1f5ce2ecad9f8121cffbbee2c95428f44ea85222e460d0d213"
 "checksum percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
 "checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677"
 "checksum ppv-lite86 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)" = "237a5ed80e274dbc66f86bd59c1e25edc039660be53194b5fe0a482e0f2612ea"

agent/Cargo.toml

@@ -17,5 +17,6 @@ log = "~0.4.8"
 serde_yaml = "~0.8.13"
 serde = {version = "~1.0.117", features = ["rc", "derive"]}
 osp = { path = "../osp" }
+os_pipe = "~0.9.2"
 tempfile = "~3.1.0"
 uuid = { version = "~0.8.1", features = ["v4", "serde"]}

agent/src/lib.rs

@@ -1,12 +1,12 @@
 
 use log::*;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use serde_yaml::Value;
 use std::collections::HashMap;
 use std::fs::File;
 use std::io::{stdout, stderr, Write};
 use std::path::{Path, PathBuf};
-use std::process::Command;
+use std::process::{Command, Stdio};
 use tempfile::NamedTempFile;
 use uuid::Uuid;
 
@@ -46,6 +46,35 @@ pub struct Step {
     pub parameters: Value,
 }
 
+/**
+ * Log is a data structure which captures the necessary metadata for logging a single line
+ */
+#[derive(Clone, Debug, Serialize)]
+pub enum Log {
+    StepStart {
+        symbol: String,
+        uuid: Uuid,
+    },
+    StepOutput {
+        symbol: String,
+        uuid: Uuid,
+        buffer: String,
+        stream: LogStream,
+    },
+    StepEnd {
+        symbol: String,
+        uuid: Uuid,
+    },
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub enum LogStream {
+    Stdout,
+    Stderr,
+}
+
+
+
 /**
  * Generate a UUID v4 for use in structs, etc
  */
@@ -96,12 +125,47 @@ pub fn run(steps_dir: &str, steps: &Vec<Step>) -> std::io::Result<()> {
             serde_yaml::to_writer(&mut file, &step_args)
                 .expect("Failed to write temporary file for script");
 
-            let output = Command::new(entrypoint)
-                .arg(file.path())
-                .output()
-                .expect("Failed to invoke the script");
-            stdout().write_all(&output.stdout).unwrap();
-            stderr().write_all(&output.stderr).unwrap();
+            use os_pipe::pipe;
+            use std::io::{BufReader, BufRead};
+            let mut cmd = Command::new(entrypoint);
+            cmd.arg(file.path());
+            let (mut reader, writer) = pipe().unwrap();
+            let writer_clone = writer.try_clone().unwrap();
+            cmd.stdout(writer);
+            cmd.stderr(writer_clone);
+
+
+            let log = Log::StepStart { symbol: step.symbol.clone(), uuid: step.uuid };
+            println!("{:?}", log);
+
+            let mut handle = cmd.spawn()?;
+            drop(cmd);
+
+            let bufr = BufReader::new(reader);
+            for line in bufr.lines() {
+                if let Ok(buffer) = line {
+                    if "dir" == step.symbol {
+                        println!("{}", buffer);
+                    }
+                    else {
+                        let log = Log::StepOutput {
+                            // TODO: Remove this allocation
+                            symbol: step.symbol.clone(),
+                            uuid: step.uuid,
+                            stream: LogStream::Stdout,
+                            buffer,
+                        };
+                        // TODO: send this to a log service
+                        println!("{:?}", log);
+                    }
+                }
+            }
+
+            handle.wait()?;
+
+            let log = Log::StepEnd { symbol: step.symbol.clone(), uuid: step.uuid };
+            println!("{:?}", log);
+
         }
     }
 
@@ -109,6 +173,7 @@ pub fn run(steps_dir: &str, steps: &Vec<Step>) -> std::io::Result<()> {
 }
 
 
+
 #[cfg(test)]
 mod tests {
 }