Merge pull request #375 from CPSSD/release-0.4.0

Release 0.4.0

Author: DarraghGriffin

.travis.yml

@@ -4,7 +4,10 @@ sudo: true
 language: rust
 rust:
     - nightly
-cache: cargo
+
+cache:
+    - cargo: true
+    - timeout: 1000
 
 addons:
     apt:
@@ -19,5 +22,5 @@ install:
     - sudo mv protoc3/include/* /usr/local/include/
 
 script:
-    - cargo build --verbose --all
-    - cargo test --verbose --all
+    - make build
+    - make test

Cargo.lock

@@ -4,11 +4,11 @@ all: build
 
 # Bulds the debug version of cerberus
 build:
-	cargo build --all
+	cargo build --verbose --all
 
 # Create the release version
 release:
-	cargo build --all --release
+	cargo build --verbose --all --release
 
 clean:
 	cargo clean
@@ -19,10 +19,14 @@ clean:
 test: unit-test integration-test
 
 unit-test:
-	cargo test --all
+	cargo test --verbose --all
 
 integration-test:
-	# TODO: Add the integration test
+	./tests/integration.sh
+	./tests/state_saving.sh
+
+multi-machine:
+	./tests/multi_machine.sh
 
 #############################################################
 

Makefile

@@ -30,7 +30,11 @@ $ cargo build --all
 
 ## Running benchmarks
 
-The following is required to run the benchmarking script:
+The following dependancies are required to run the benchmarking script:
+- python3-tk
+- matplotlib
+
+Ubuntu dependancy installation:
 ```
 apt-get install python3-pip python3-tk
 pip3 install numpy matplotlib

README.md

@@ -1,6 +1,6 @@
 [package]
 name = "master"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Cerberus Authors <cerberus@cpssd.net>"]
 
 [dependencies]
@@ -13,6 +13,7 @@ futures = "0.1"
 futures-cpupool = "0.1"
 grpc = "0.2.1"
 log = "0.3"
+protobuf = "1.4.1"
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"

master/Cargo.toml

@@ -1,14 +1,15 @@
 use errors::*;
+use std::sync::Arc;
 use std::path::{PathBuf, Path};
 use std::process::{Command, Stdio};
 
 use chrono::prelude::*;
 use serde_json;
 use uuid::Uuid;
 
+use util::data_layer::AbstractionLayer;
 use state::StateHandling;
 use cerberus_proto::mapreduce as pb;
-use queued_work_store::QueuedWork;
 
 /// `JobOptions` stores arguments used to construct a `Job`.
 #[derive(Default)]
@@ -79,7 +80,7 @@ pub enum SerializableJobStatus {
 }
 
 impl Job {
-    pub fn new(options: JobOptions) -> Result<Self> {
+    pub fn new_no_validate(options: JobOptions) -> Result<Self> {
         let input_directory = options.input_directory;
 
         let output_directory = match options.output_directory {
@@ -95,7 +96,7 @@ impl Job {
             }
         };
 
-        let job = Job {
+        Ok(Job {
             client_id: options.client_id,
             id: Uuid::new_v4().to_string(),
             binary_path: options.binary_path,
@@ -116,36 +117,70 @@ impl Job {
             time_completed: None,
 
             cpu_time: 0,
-        };
-        if options.validate_paths {
-            job.validate_input().chain_err(|| "Error validating input")?;
+        })
+    }
+
+    pub fn new(
+        options: JobOptions,
+        data_abstraction_layer: &Arc<AbstractionLayer + Send + Sync>,
+    ) -> Result<Self> {
+        let validate_paths = options.validate_paths;
+        let job = Job::new_no_validate(options).chain_err(
+            || "Unable to create job",
+        )?;
+
+        if validate_paths {
+            job.validate_input(data_abstraction_layer).chain_err(
+                || "Error validating input",
+            )?;
         }
         Ok(job)
     }
 
-    fn validate_input(&self) -> Result<()> {
+    fn validate_input(
+        &self,
+        data_abstraction_layer: &Arc<AbstractionLayer + Send + Sync>,
+    ) -> Result<()> {
         // Validate the existence of the input directory and the binary file.
-        let input_path = Path::new(self.input_directory.as_str().clone());
-        if !(input_path.exists() && input_path.is_dir()) {
+        let input_path = Path::new(&self.input_directory);
+        let is_dir = data_abstraction_layer.is_dir(input_path).chain_err(
+            || "Error checking if path is a directory",
+        )?;
+        if !is_dir {
             return Err(
-                format!("Input directory does not exist: {}", self.input_directory).into(),
+                format!(
+                    "Input directory does not exist: {:?}",
+                    data_abstraction_layer.absolute_path(input_path)
+                ).into(),
             );
         }
 
-        let binary_path = Path::new(self.binary_path.as_str().clone());
-        if !(binary_path.exists() && binary_path.is_file()) {
+        let binary_path = Path::new(&self.binary_path);
+        let is_file = data_abstraction_layer.is_file(binary_path).chain_err(
+            || "Error checking if path is a file",
+        )?;
+        if !is_file {
             return Err(
-                format!("Binary does not exist: {}", self.binary_path).into(),
+                format!(
+                    "Binary does not exist: {:?}",
+                    data_abstraction_layer.absolute_path(binary_path)
+                ).into(),
             );
         }
 
         // Binary exists, so run sanity-check on it to verify that it's a libcerberus binary.
-        self.run_sanity_check()
+        self.run_sanity_check(data_abstraction_layer)
     }
 
-    fn run_sanity_check(&self) -> Result<()> {
-        let binary_path = Path::new(self.binary_path.as_str());
-        let child = Command::new(binary_path)
+    fn run_sanity_check(
+        &self,
+        data_abstraction_layer: &Arc<AbstractionLayer + Send + Sync>,
+    ) -> Result<()> {
+        let binary_path = Path::new(&self.binary_path);
+        let absolute_path = data_abstraction_layer
+            .absolute_path(binary_path)
+            .chain_err(|| "unable to get absolute path")?;
+        let child = Command::new(absolute_path)
             .arg("sanity-check")
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
@@ -198,10 +233,10 @@ impl StateHandling for Job {
                 .chain_err(|| "Unable to convert input_directory")?,
             output_directory: serde_json::from_value(data["output_directory"].clone())
                 .chain_err(|| "Unable to convert output dir")?,
-            validate_paths: true,
+            validate_paths: false,
         };
 
-        let mut job = Job::new(options).chain_err(
+        let mut job = Job::new_no_validate(options).chain_err(
             || "Unable to create map reduce job",
         )?;
 
@@ -294,18 +329,6 @@ impl StateHandling for Job {
     }
 }
 
-impl QueuedWork for Job {
-    type Key = String;
-
-    fn get_work_bucket(&self) -> String {
-        self.client_id.clone()
-    }
-
-    fn get_work_id(&self) -> String {
-        self.id.clone()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -321,7 +344,7 @@ mod tests {
 
     #[test]
     fn test_defaults() {
-        let job = Job::new(get_test_job_options()).unwrap();
+        let job = Job::new_no_validate(get_test_job_options()).unwrap();
         // Assert that the default status for a map reduce job is Queued.
         assert_eq!(pb::Status::IN_QUEUE, job.status);
         // Assert that completed tasks starts at 0.
@@ -332,18 +355,10 @@ mod tests {
         assert_eq!(0, job.reduce_tasks_total);
     }
 
-    #[test]
-    fn test_queued_work_impl() {
-        let job = Job::new(get_test_job_options()).unwrap();
-
-        assert_eq!(job.get_work_bucket(), "client-1");
-        assert_eq!(job.get_work_id(), job.id);
-    }
-
     #[test]
     fn test_output_directory() {
-        let job1 = Job::new(get_test_job_options()).unwrap();
-        let job2 = Job::new(JobOptions {
+        let job1 = Job::new_no_validate(get_test_job_options()).unwrap();
+        let job2 = Job::new_no_validate(JobOptions {
             client_id: "client-1".to_owned(),
             binary_path: "/tmp/binary".to_owned(),
             input_directory: "/tmp/input/".to_owned(),