Add webhooks (closes #59)

Using the option --webhook <url> Factotum will post status updates
to the given url. These include the current run-state of the job,
failures, skipped tasks etc.

.travis.yml

@@ -9,9 +9,12 @@ os:
 matrix:
   allow_failures:
   - rust: nightly
+script: 
+  - $TRAVIS_BUILD_DIR/.travis/build.sh
 deploy:
+  skip_cleanup: true
   provider: script
-  script: cargo run --quiet -- run deploy/deploy.factfile
+  script: ./target/debug/factotum run .travis/deploy/deploy.factfile
   on:
     tags: true
 env:

.travis/build.sh

@@ -0,0 +1,12 @@
+#!/bin/bash -e
+
+if [ "${TRAVIS_OS_NAME}" == "osx" ]; then
+	echo "Configuring openssl libs for OSX..."
+	#brew install openssl
+	export OPENSSL_INCLUDE_DIR=`brew --prefix openssl`/include
+	export OPENSSL_LIB_DIR=`brew --prefix openssl`/lib
+	echo "...done!"
+fi
+
+cargo build --verbose
+cargo test --verbose

deploy/deploy.factfile → .travis/deploy/deploy.factfile

@@ -6,7 +6,7 @@
             {
                 "name": "is-candidate-arch",
                 "executor": "shell",
-                "command": "deploy/tasks/check_release.rb",
+                "command": ".travis/deploy/tasks/check_release.rb",
                 "arguments": [],
                 "dependsOn": [],
                 "onResult": {
@@ -17,7 +17,7 @@
             {
                 "name": "provision-release-manager",
                 "executor": "shell",
-                "command": "deploy/tasks/provision.sh",
+                "command": ".travis/deploy/tasks/provision.sh",
                 "arguments": [],
                 "dependsOn": [ "is-candidate-arch" ],
                 "onResult": {
@@ -39,7 +39,7 @@
             {
                 "name": "ship-release-artifact",
                 "executor": "shell",
-                "command": "deploy/tasks/ship_release.sh",
+                "command": ".travis/deploy/tasks/ship_release.sh",
                 "arguments": [],
                 "dependsOn": [ "build-release-artifact" ],
                 "onResult": {

deploy/tasks/provision.sh → .travis/deploy/tasks/provision.sh

@@ -1,4 +1,4 @@
 #!/bin/bash -e
 
 pip install --user --upgrade pip
-pip install --user release-manager==0.1.0rc3
+pip install --user release-manager==0.1.0rc3

deploy/tasks/ship_release.sh → .travis/deploy/tasks/ship_release.sh

@@ -9,4 +9,4 @@ else
     SUFFIX="_linux_x86_64"
 fi
 
-env RM_SUFFIX=${SUFFIX} release-manager --config deploy/tasks/release_config.yaml --make-version  --make-artifact --upload-artifact
+env RM_SUFFIX=${SUFFIX} release-manager --config .travis/deploy/tasks/release_config.yaml --make-version  --make-artifact --upload-artifact

Cargo.toml

@@ -13,3 +13,7 @@ valico = "0.8.2"
 chrono = "0.2"
 colored = "1.2"
 mustache = "*"
+rand = "0.3"
+rust-crypto = "^0.2"
+uuid = { version = "0.2", features = ["v4"] }
+hyper = "0.9"

src/factotum/executor/execution_strategy/mod.rs

@@ -20,6 +20,7 @@ use chrono::UTC;
 use std::time::{Instant, Duration};
 use chrono::DateTime;
 
+#[derive(Clone, PartialEq, Debug)]
 pub struct RunResult {
    pub run_started: DateTime<UTC>,
    pub duration: Duration,

src/factotum/executor/mod.rs

@@ -65,61 +65,102 @@ pub fn get_task_execution_list(factfile:&Factfile, start_from:Option<String>) ->
 }
 
 
-pub fn execute_factfile<'a, F>(factfile:&'a Factfile, start_from:Option<String>, strategy:F) -> TaskList<&'a FactfileTask>
+pub type TaskSnapshot = Vec<Task<FactfileTask>>;
+
+#[derive(Debug, PartialEq, Clone)]
+pub enum ExecutionState {
+    STARTED(TaskSnapshot),
+    RUNNING(TaskSnapshot),
+    FINISHED(TaskSnapshot)    
+}
+
+pub fn get_task_snapshot(tasklist: &TaskList<&FactfileTask>) -> TaskSnapshot {
+    tasklist.tasks
+            .iter()
+            .flat_map(|task_grp| task_grp.iter().map(|task| Task { 
+                                                                   name: task.name.clone(), 
+                                                                   task_spec: task.task_spec.clone(), 
+                                                                   state: task.state.clone(),
+                                                                   run_result: task.run_result.clone() 
+                                                                  } ))
+            .collect()
+}
+
+pub fn execute_factfile<'a, F>(factfile:&'a Factfile, start_from:Option<String>, strategy:F, progress_channel:Option<mpsc::Sender<ExecutionState>>) -> TaskList<&'a FactfileTask>
     where F : Fn(&str, &mut Command) -> RunResult + Send + Sync + 'static + Copy {
 
     let mut tasklist = get_task_execution_list(factfile, start_from);
+
+   // notify the progress channel
+   if let Some(ref send) = progress_channel {
+       send.send(ExecutionState::STARTED(get_task_snapshot(&tasklist))).unwrap();
+   }
 
-    for mut task_group in tasklist.tasks.iter_mut() {
+    for task_grp_idx in 0..tasklist.tasks.len() {
         // everything in a task "group" gets run together
         let (tx, rx) = mpsc::channel::<(usize, RunResult)>();
 
-        for (idx,task) in task_group.iter().enumerate() {
-            info!("Running task '{}'!", task.name);
-            {
-                let tx = tx.clone();
-                let args = format_args(&task.task_spec.command, &task.task_spec.arguments);
-                let task_name = task.name.to_string();
-
-                thread::spawn(move || {
-                    let mut command = Command::new("sh");
-                    command.arg("-c");
-                    command.arg(args);
-                    let task_result = strategy(&task_name, &mut command);
-                    tx.send((idx, task_result)).unwrap();
-                });  
-            }          
-        }        
+        {
+            let ref mut task_group = tasklist.tasks[task_grp_idx];
+            for (idx,task) in task_group.into_iter().enumerate() {
+                info!("Running task '{}'!", task.name);
+                task.state = State::RUNNING;
+                {
+                    let tx = tx.clone();
+                    let args = format_args(&task.task_spec.command, &task.task_spec.arguments);
+                    let task_name = task.name.to_string();
+
+                    thread::spawn(move || {
+                        let mut command = Command::new("sh");
+                        command.arg("-c");
+                        command.arg(args);
+                        let task_result = strategy(&task_name, &mut command);
+                        tx.send((idx, task_result)).unwrap();
+                    });  
+                }          
+            }    
+        }
+
+        if let Some(ref send) = progress_channel {
+            send.send(ExecutionState::RUNNING(get_task_snapshot(&tasklist))).unwrap();
+        }            
 
         let mut terminate_job_please = false; 
         let mut task_failed = false;
 
-        for _ in 0..task_group.len() {                    
-            let (idx, task_result) = rx.recv().unwrap();
-
-            info!("'{}' returned {} in {:?}", task_group[idx].name, task_result.return_code, task_result.duration); 
+        for _ in 0..tasklist.tasks[task_grp_idx].len() {                    
+            let (idx, task_result) = rx.recv().unwrap();                              
+            info!("'{}' returned {} in {:?}", tasklist.tasks[task_grp_idx][idx].name, task_result.return_code, task_result.duration); 
 
-            if task_group[idx].task_spec.on_result.terminate_job.contains(&task_result.return_code) {
-                // if the return code is in the terminate early list, prune the sub-tree (set to skipped) return early term
-                task_group[idx].state = State::SUCCESS_NOOP;
-                terminate_job_please = true;
-            } else if task_group[idx].task_spec.on_result.continue_job.contains(&task_result.return_code) {
-                // if the return code is in the continue list, return success
-                task_group[idx].state = State::SUCCESS;
-            } else {
-                // if the return code is not in either list, prune the sub-tree (set to skipped) and return error
-                let expected_codes = task_group[idx].task_spec.on_result.continue_job.iter()
-                                                                         .map(|code| code.to_string())
-                                                                         .collect::<Vec<String>>()
-                                                                         .join(",");
-                let err_msg = format!("the task exited with a value not specified in continue_job - {} (task expects one of the following return codes to continue [{}])", 
-                                      task_result.return_code,
-                                      expected_codes);
-                task_group[idx].state = State::FAILED(err_msg);
-                task_failed = true;
-             }
-
-             task_group[idx].run_result = Some(task_result);
+            {
+                let ref mut task_group = tasklist.tasks[task_grp_idx];
+                if task_group[idx].task_spec.on_result.terminate_job.contains(&task_result.return_code) {
+                    // if the return code is in the terminate early list, prune the sub-tree (set to skipped) return early term
+                    task_group[idx].state = State::SUCCESS_NOOP;
+                    terminate_job_please = true;
+                } else if task_group[idx].task_spec.on_result.continue_job.contains(&task_result.return_code) {
+                    // if the return code is in the continue list, return success
+                    task_group[idx].state = State::SUCCESS;
+                } else {
+                    // if the return code is not in either list, prune the sub-tree (set to skipped) and return error
+                    let expected_codes = task_group[idx].task_spec.on_result.continue_job.iter()
+                                                                            .map(|code| code.to_string())
+                                                                            .collect::<Vec<String>>()
+                                                                            .join(",");
+                    let err_msg = format!("the task exited with a value not specified in continue_job - {} (task expects one of the following return codes to continue [{}])", 
+                                        task_result.return_code,
+                                        expected_codes);
+                    task_group[idx].state = State::FAILED(err_msg);
+                    task_failed = true;
+                }
+
+                task_group[idx].run_result = Some(task_result);
+            }
+
+            if let Some(ref send) = progress_channel {
+                send.send(ExecutionState::RUNNING(get_task_snapshot(&tasklist))).unwrap();
+            }    
+
         }
 
         if terminate_job_please || task_failed {
@@ -128,6 +169,16 @@ pub fn execute_factfile<'a, F>(factfile:&'a Factfile, start_from:Option<String>,
 
     }
 
+    if let Some(ref send) = progress_channel {
+        let mut snapshot = get_task_snapshot(&tasklist);
+        for mut task in snapshot.iter_mut() {
+            if State::WAITING == task.state {
+                 task.state = State::SKIPPED("A prior failure or terminate request caused this task to be skipped".to_string());
+            } 
+        }
+        send.send(ExecutionState::FINISHED(snapshot)).unwrap();
+    }    
+
     tasklist 
 }
 

src/factotum/executor/task_list/mod.rs

@@ -21,16 +21,17 @@ use factotum::executor::execution_strategy::RunResult;
 #[derive(Clone, PartialEq, Debug)]
 pub enum State {
     WAITING,
+    RUNNING,
     SUCCESS,
     SUCCESS_NOOP,
     FAILED(String),
     SKIPPED(String)
 }
 
+#[derive(Clone, PartialEq, Debug)]
 pub struct Task<T> {
     pub name: String,
     pub state: State,
-    pub children: Vec<String>,
     pub task_spec: T,
     pub run_result: Option<RunResult>    
 }
@@ -40,7 +41,6 @@ impl<T> Task<T> {
         Task {
             name: name.into(),
             state: State::WAITING,
-            children: vec![],
             task_spec: task_spec,
             run_result: None 
         }
@@ -49,6 +49,7 @@ impl<T> Task<T> {
 
 pub type TaskGroup<T> = Vec<Task<T>>;
 
+#[derive(Clone, Debug)]
 pub struct TaskList<T> {
     pub tasks: Vec<TaskGroup<T>>,
     edges: HashMap<String,Vec<String>>

src/factotum/executor/task_list/tests.rs

@@ -20,7 +20,7 @@ fn task_new_defaults_good() {
     let task = Task::<String>::new("hello", "world".to_string());
     assert_eq!(task.name, "hello");
     assert_eq!(task.state, State::WAITING);
-    assert_eq!(task.children.len(), 0);
+    // assert_eq!(task.children.len(), 0);
     assert_eq!(task.task_spec, "world".to_string());
     assert!(task.run_result.is_some()==false)
 }