Merge pull request #108 from NixOS/record-build-status

Record build status - 0.1.3

nix/ofborg-carnix.nix

@@ -45,7 +45,7 @@ let kernel = buildPlatform.parsed.kernel.name;
     ) [] (builtins.attrNames feat);
 in
 rec {
-  ofborg = f: ofborg_0_1_2 { features = ofborg_0_1_2_features { ofborg_0_1_2 = f; }; };
+  ofborg = f: ofborg_0_1_3 { features = ofborg_0_1_3_features { ofborg_0_1_3 = f; }; };
   aho_corasick_0_5_3_ = { dependencies?[], buildDependencies?[], features?[] }: buildRustCrate {
     crateName = "aho-corasick";
     version = "0.5.3";

ofborg/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ofborg"
-version = "0.1.2"
+version = "0.1.3"
 authors = ["Graham Christensen <graham@grahamc.com>"]
 include = ["Cargo.toml", "Cargo.lock", "src", "test-srcs", "build.rs"]
 build = "build.rs"

ofborg/src/tasks/build.rs

@@ -189,12 +189,20 @@ impl<'a, 'b> JobActions<'a, 'b> {
 
         let result_exchange = self.result_exchange.clone();
         let result_routing_key = self.result_routing_key.clone();
-
         self.tell(worker::publish_serde_action(
             result_exchange,
             result_routing_key,
             &msg,
         ));
+
+        let log_exchange = self.log_exchange.clone();
+        let log_routing_key = self.log_routing_key.clone();
+        self.tell(worker::publish_serde_action(
+            log_exchange,
+            log_routing_key,
+            &msg,
+        ));
+
         self.tell(worker::Action::Ack);
     }
 
@@ -216,12 +224,20 @@ impl<'a, 'b> JobActions<'a, 'b> {
 
         let result_exchange = self.result_exchange.clone();
         let result_routing_key = self.result_routing_key.clone();
-
         self.tell(worker::publish_serde_action(
             result_exchange,
             result_routing_key,
             &msg,
         ));
+
+        let log_exchange = self.log_exchange.clone();
+        let log_routing_key = self.log_routing_key.clone();
+        self.tell(worker::publish_serde_action(
+            log_exchange,
+            log_routing_key,
+            &msg,
+        ));
+
         self.tell(worker::Action::Ack);
     }
 
@@ -476,7 +492,8 @@ mod tests {
         assert_contains_job(&mut actions, "output\":\"2");
         assert_contains_job(&mut actions, "output\":\"3");
         assert_contains_job(&mut actions, "output\":\"4");
-        assert_contains_job(&mut actions, "success\":true");
+        assert_contains_job(&mut actions, "success\":true"); // First one to the github poster
+        assert_contains_job(&mut actions, "success\":true"); // This one to the logs
         assert_eq!(actions.next(), Some(worker::Action::Ack));
     }
 
@@ -517,7 +534,8 @@ mod tests {
 
         println!("Total actions: {:?}", dummyreceiver.actions.len());
         let mut actions = dummyreceiver.actions.into_iter();
-        assert_contains_job(&mut actions, "skipped_attrs\":[\"not-real");
+        assert_contains_job(&mut actions, "skipped_attrs\":[\"not-real"); // First one to the github poster
+        assert_contains_job(&mut actions, "skipped_attrs\":[\"not-real"); // This one to the logs
         assert_eq!(actions.next(), Some(worker::Action::Ack));
     }
 }

ofborg/src/tasks/log_message_collector.rs

@@ -1,7 +1,6 @@
 extern crate amqp;
 extern crate env_logger;
 
-use either::{Either, Left, Right};
 use lru_cache::LruCache;
 use serde_json;
 use std::fs;
@@ -11,6 +10,7 @@ use std::io::Write;
 
 use ofborg::writetoline::LineWriter;
 use ofborg::message::buildlogmsg::{BuildLogStart, BuildLogMsg};
+use ofborg::message::buildresult::BuildResult;
 use ofborg::worker;
 use amqp::protocol::basic::{Deliver, BasicProperties};
 
@@ -25,10 +25,17 @@ pub struct LogMessageCollector {
     log_root: PathBuf,
 }
 
+#[derive(Debug)]
+enum MsgType {
+    Start(BuildLogStart),
+    Msg(BuildLogMsg),
+    Finish(BuildResult),
+}
+
 #[derive(Debug)]
 pub struct LogMessage {
     from: LogFrom,
-    message: Either<BuildLogStart, BuildLogMsg>,
+    message: MsgType
 }
 
 fn validate_path_segment(segment: &PathBuf) -> Result<(), String> {
@@ -158,20 +165,26 @@ impl worker::SimpleWorker for LogMessageCollector {
         body: &Vec<u8>,
     ) -> Result<Self::J, String> {
 
-        let message: Either<BuildLogStart, BuildLogMsg>;
+        let message: MsgType;
         let attempt_id: String;
 
         let decode_msg: Result<BuildLogMsg, _> = serde_json::from_slice(body);
         if let Ok(msg) = decode_msg {
             attempt_id = msg.attempt_id.clone();
-            message = Right(msg);
+            message = MsgType::Msg(msg);
         } else {
             let decode_msg: Result<BuildLogStart, _> = serde_json::from_slice(body);
             if let Ok(msg) = decode_msg {
                 attempt_id = msg.attempt_id.clone();
-                message = Left(msg);
+                message = MsgType::Start(msg);
             } else {
-                return Err(format!("failed to decode job: {:?}", decode_msg));
+                let decode_msg: Result<BuildResult, _> = serde_json::from_slice(body);
+                if let Ok(msg) = decode_msg {
+                    attempt_id = msg.attempt_id.clone();
+                    message = MsgType::Finish(msg);
+                } else {
+                    return Err(format!("failed to decode job: {:?}", decode_msg));
+                }
             }
         }
 
@@ -186,15 +199,17 @@ impl worker::SimpleWorker for LogMessageCollector {
 
     fn consumer(&mut self, job: &LogMessage) -> worker::Actions {
         match job.message {
-            Left(ref start) => {
+            MsgType::Start(ref start) => {
                 self.write_metadata(&job.from, &start).expect("failed to write metadata");
             },
-            Right(ref message) => {
+            MsgType::Msg(ref message) => {
                 let handle = self.handle_for(&job.from).unwrap();
 
                 handle.write_to_line((message.line_number - 1) as usize,
                                      &message.output);
-            }
+            },
+            MsgType::Finish(ref _finish) => {
+            },
         }
 
         return vec![worker::Action::Ack];
@@ -330,7 +345,7 @@ mod tests {
         };
         let mut job = LogMessage {
             from: make_from("foo"),
-            message: Right(logmsg.clone()),
+            message: MsgType::Msg(logmsg.clone()),
         };
 
         let p = TestScratch::new_dir("log-message-collector-path_for_log");
@@ -341,7 +356,7 @@ mod tests {
                        worker.consumer(&
                                        LogMessage {
                                            from: make_from("foo"),
-                                           message: Left(BuildLogStart {
+                                           message: MsgType::Start(BuildLogStart {
                                                attempt_id: String::from("my-attempt-id"),
                                                identity: String::from("my-identity"),
                                                system: String::from("foobar-x8664"),
@@ -356,14 +371,14 @@ mod tests {
 
             logmsg.line_number = 5;
             logmsg.output = String::from("line-5");
-            job.message = Right(logmsg.clone());
+            job.message = MsgType::Msg(logmsg.clone());
             assert_eq!(vec![worker::Action::Ack], worker.consumer(&job));
 
             job.from.attempt_id = String::from("my-other-attempt");
             logmsg.attempt_id = String::from("my-other-attempt");
             logmsg.line_number = 3;
             logmsg.output = String::from("line-3");
-            job.message = Right(logmsg.clone());
+            job.message = MsgType::Msg(logmsg.clone());
             assert_eq!(vec![worker::Action::Ack], worker.consumer(&job));
         }