Add job heartbeat as webhook event (closes #99)

Author: jbeemster

samples/concurrent-streaming-output.factfile

@@ -28,7 +28,7 @@
             {
                 "name": "stream-branch-2",
                 "executor": "shell",
-                "command": "for i in {1..5}; do echo \"Branch 2: Line $i\"; sleep 1; done",
+                "command": "for i in {1..100}; do echo \"Branch 2: Line $i\"; sleep 1; done",
                 "arguments": [],
                 "dependsOn": [ "start" ],
                 "onResult": {
@@ -39,7 +39,7 @@
             {
                 "name": "stream-branch-3",
                 "executor": "shell",
-                "command": "for i in {1..5}; do echo \"Branch 3: Line $i\"; sleep 1; done",
+                "command": "for i in {1..20}; do echo \"Branch 3: Line $i\"; sleep 1; done",
                 "arguments": [],
                 "dependsOn": [ "start" ],
                 "onResult": {

samples/test-heartbeat.factfile

@@ -0,0 +1,33 @@
+{
+  "schema": "iglu:com.snowplowanalytics.factotum/factfile/jsonschema/1-0-0",
+  "data": {
+    "name": "heartbeat_test",
+    "tasks": [
+      {
+        "name": "quick_task",
+        "executor": "shell",
+        "command": "echo",
+        "arguments": ["Starting tests..."],
+        "dependsOn": [],
+        "onResult": {
+          "terminateJobWithSuccess": [],
+          "continueJob": [0]
+        }
+      },
+      {
+        "name": "long_task_with_logs",
+        "executor": "shell",
+        "command": "bash",
+        "arguments": [
+          "-c",
+          "for i in 1 2 3 4 5 6 7 8 9 10; do echo Progress: $i/10; sleep 3; done; echo Task completed"
+        ],
+        "dependsOn": ["quick_task"],
+        "onResult": {
+          "terminateJobWithSuccess": [],
+          "continueJob": [0]
+        }
+      }
+    ]
+  }
+}

src/factotum/executor/execution_strategy/mod.rs

@@ -18,6 +18,7 @@ use std::process::{Command, Stdio};
 use std::time::{Instant, Duration};
 use std::io::{BufRead, BufReader};
 use std::thread;
+use std::sync::{Arc, Mutex};
 use colored::*;
 
 #[derive(Clone, PartialEq, Debug)]
@@ -66,6 +67,10 @@ pub fn execute_simulation(name: &str, command: &mut Command) -> RunResult {
 }
 
 pub fn execute_os(name: &str, command: &mut Command) -> RunResult {
+    execute_os_with_buffers(name, command, None, None)
+}
+
+pub fn execute_os_with_buffers(name: &str, command: &mut Command, stdout_buffer: Option<Arc<Mutex<String>>>, stderr_buffer: Option<Arc<Mutex<String>>>) -> RunResult {
     let run_start = Instant::now();
     info!("Executing sh {:?}", command);
 
@@ -84,25 +89,45 @@ pub fn execute_os(name: &str, command: &mut Command) -> RunResult {
             let name_stdout = name.to_string();
             let name_stderr = name.to_string();
 
+            let stdout_buffer_clone = stdout_buffer.clone();
             let stdout_handle = thread::spawn(move || {
                 let reader = BufReader::new(stdout);
                 let mut lines = Vec::new();
                 for line in reader.lines() {
                     if let Ok(line) = line {
                         println!("[{}] {}", name_stdout.cyan(), line);
-                        lines.push(line);
+                        lines.push(line.clone());
+
+                        // Also append to shared buffer if provided
+                        if let Some(ref buffer) = stdout_buffer_clone {
+                            let mut buf = buffer.lock().unwrap_or_else(|p| p.into_inner());
+                            if !buf.is_empty() {
+                                buf.push('\n');
+                            }
+                            buf.push_str(&line);
+                        }
                     }
                 }
                 lines.join("\n")
             });
 
+            let stderr_buffer_clone = stderr_buffer.clone();
             let stderr_handle = thread::spawn(move || {
                 let reader = BufReader::new(stderr);
                 let mut lines = Vec::new();
                 for line in reader.lines() {
                     if let Ok(line) = line {
                         eprintln!("[{}] {}", name_stderr.cyan(), line.red());
-                        lines.push(line);
+                        lines.push(line.clone());
+
+                        // Also append to shared buffer if provided
+                        if let Some(ref buffer) = stderr_buffer_clone {
+                            let mut buf = buffer.lock().unwrap_or_else(|p| p.into_inner());
+                            if !buf.is_empty() {
+                                buf.push('\n');
+                            }
+                            buf.push_str(&line);
+                        }
                     }
                 }
                 lines.join("\n")

src/factotum/executor/heartbeat.rs

@@ -0,0 +1,144 @@
+use std::sync::Mutex;
+use std::collections::HashMap;
+use chrono::{DateTime, UTC};
+use std::sync::Arc;
+use std::time::Duration;
+use std::sync::mpsc::Sender;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread::{self, JoinHandle};
+use super::{ExecutionUpdate, ExecutionState, Transition, HeartbeatTaskData, get_task_snapshot};
+
+// Shared state for streaming logs
+pub struct TaskStreamState {
+    pub start_time: DateTime<UTC>,
+    pub stdout_buffer: Arc<Mutex<String>>,
+    pub stderr_buffer: Arc<Mutex<String>>,
+}
+
+// Global state - tracks running tasks with log buffers
+lazy_static! {
+    static ref RUNNING_TASKS: Mutex<HashMap<String, TaskStreamState>> =
+        Mutex::new(HashMap::new());
+}
+
+// Called by executor when task starts
+// Returns Arc clones to log buffers for the streaming threads
+pub fn register_task_start(name: String, start: DateTime<UTC>)
+    -> (Arc<Mutex<String>>, Arc<Mutex<String>>)
+{
+    let stdout_buffer = Arc::new(Mutex::new(String::new()));
+    let stderr_buffer = Arc::new(Mutex::new(String::new()));
+
+    RUNNING_TASKS.lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .insert(name, TaskStreamState {
+            start_time: start,
+            stdout_buffer: stdout_buffer.clone(),
+            stderr_buffer: stderr_buffer.clone(),
+        });
+
+    (stdout_buffer, stderr_buffer)
+}
+
+// Called by executor when task completes
+pub fn unregister_task_completion(name: &str) {
+    RUNNING_TASKS.lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .remove(name);
+}
+
+// Snapshot with cumulative logs (for heartbeats)
+pub struct TaskHeartbeatData {
+    pub name: String,
+    pub start_time: DateTime<UTC>,
+    pub elapsed: Duration,
+    pub stdout: String,  // Cumulative stdout up to now
+    pub stderr: String,  // Cumulative stderr up to now
+}
+
+// Get snapshot for heartbeat with cumulative logs
+pub fn get_running_tasks_with_logs() -> Vec<TaskHeartbeatData> {
+    let now = UTC::now();
+    let tasks = RUNNING_TASKS.lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+
+    tasks.iter()
+        .map(|(name, state)| {
+            let stdout = state.stdout_buffer.lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .clone();
+            let stderr = state.stderr_buffer.lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .clone();
+
+            let elapsed_chrono = now - state.start_time;
+            let elapsed_std = elapsed_chrono.to_std()
+                .unwrap_or(Duration::from_secs(0));
+
+            TaskHeartbeatData {
+                name: name.clone(),
+                start_time: state.start_time,
+                elapsed: elapsed_std,
+                stdout,
+                stderr,
+            }
+        })
+        .collect()
+}
+
+// Spawn heartbeat timer thread
+pub fn spawn_heartbeat_thread(
+    interval: Duration,
+    progress_channel: Sender<ExecutionUpdate>,
+    shutdown_flag: Arc<AtomicBool>,
+    shared_snapshot: Arc<Mutex<super::TaskSnapshot>>,
+) -> JoinHandle<()> {
+    thread::Builder::new()
+        .name("heartbeat".to_string())
+        .spawn(move || {
+            loop {
+                thread::sleep(interval);
+
+                // Check shutdown first
+                if shutdown_flag.load(Ordering::SeqCst) {
+                    break;
+                }
+
+                let running = get_running_tasks_with_logs();
+                if running.is_empty() {
+                    continue;
+                }
+
+                // Convert to HeartbeatTaskData for the transition
+                let heartbeat_data: Vec<HeartbeatTaskData> = running.iter()
+                    .map(|t| HeartbeatTaskData {
+                        task_name: t.name.clone(),
+                        elapsed_seconds: t.elapsed.as_secs(),
+                        stdout: t.stdout.clone(),
+                        stderr: t.stderr.clone(),
+                    })
+                    .collect();
+
+                // Get live snapshot from shared state
+                let snapshot = shared_snapshot.lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .clone();
+
+                // Create heartbeat ExecutionUpdate
+                let update = ExecutionUpdate::new(
+                    ExecutionState::Running,
+                    snapshot,
+                    Transition::Heartbeat(heartbeat_data)
+                );
+
+                match progress_channel.send(update) {
+                    Ok(_) => {},
+                    Err(e) => {
+                        warn!("Heartbeat send failed: {}, exiting", e);
+                        break;
+                    }
+                }
+            }
+        })
+        .expect("Failed to spawn heartbeat thread")
+}