Merge pull request #19 from runabove/filter

Filter
ovh · Feb 27, 2017 · 2990611 · 2990611
2 parents 27b2593 + c67484c
commit 2990611
Show file tree

Hide file tree

Showing 6 changed files with 76 additions and 45 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "beamium"
-version = "1.3.1"
+version = "1.4.0"
 authors = [ "d33d33 <[email protected]>" ]
 
 build = "build.rs"

diff --git a/src/config.rs b/src/config.rs
@@ -208,28 +208,30 @@ fn load_path<P: AsRef<Path>>(file_path: P, config: &mut Config) -> Result<(), Co
                 let period = try!(cast::u64(period)
                     .map_err(|_| format!("sources.{}.period is invalid", name)));
                 let format = if v["format"].is_badvalue() {
+                    SourceFormat::Prometheus
+                } else {
+                    let f = try!(v["format"]
+                        .as_str()
+                        .ok_or(format!("sinks.{}.format should be a string", name)));
+
+                    if f == "prometheus" {
                         SourceFormat::Prometheus
+                    } else if f == "sensision" {
+                        SourceFormat::Sensision
                     } else {
-                        let f = try!(v["format"]
-                            .as_str()
-                            .ok_or(format!("sinks.{}.format should be a string", name)));
-
-                        if f == "prometheus" {
-                            SourceFormat::Prometheus
-                        } else if f == "sensision" {
-                            SourceFormat::Sensision
-                        } else {
-                            return Err(format!("sinks.{}.format should be 'Prometheus' or 'sensision'", name).into())
-                        }
-                    };
+                        return Err(format!("sinks.{}.format should be 'Prometheus' or 'sensision'",
+                                           name)
+                            .into());
+                    }
+                };
                 let metrics = if v["metrics"].is_badvalue() {
                     None
                 } else {
                     let mut metrics = Vec::new();
                     let values = try!(v["metrics"].as_vec().ok_or("metrics should be an array"));
                     for v in values {
                         let value = try!(regex::Regex::new(try!(v.as_str()
-                                .ok_or(format!("metrics.{} is invalid", name)))));
+                            .ok_or(format!("metrics.{} is invalid", name)))));
                         metrics.push(String::from(r"^(\S*)\s") + value.as_str());
                     }
 
@@ -267,9 +269,13 @@ fn load_path<P: AsRef<Path>>(file_path: P, config: &mut Config) -> Result<(), Co
                 let selector = if v["selector"].is_badvalue() {
                     None
                 } else {
-                    Some(try!(regex::Regex::new(try!(v["selector"]
-                        .as_str()
-                        .ok_or(format!("sinks.{}.selector is invalid", name))))))
+                    Some(try!(regex::Regex::new(format!("^{}",
+                                                        try!(v["selector"]
+                                                            .as_str()
+                                                            .ok_or(format!("sinks.{}.selector \
+                                                                            is invalid",
+                                                                           name))))
+                        .as_str())))
                 };
 
                 let ttl = if v["ttl"].is_badvalue() {

diff --git a/src/main.rs b/src/main.rs
@@ -95,7 +95,6 @@ fn main() {
     }
 
     // Synchronisation stuff
-    // let signal = chan_signal::notify(&[Signal::INT, Signal::TERM]);
     let sigint = Arc::new(AtomicBool::new(false));
     let mut handles = Vec::with_capacity(config.sources.len() + 1 + config.sinks.len());
 

diff --git a/src/router.rs b/src/router.rs
@@ -35,9 +35,13 @@ pub fn router(sinks: &Vec<config::Sink>,
     loop {
         let start = time::now_utc();
 
-        match route(sinks, parameters, &labels) {
+        match route(sinks, parameters, &labels, sigint.clone()) {
             Err(err) => error!("route fail: {}", err),
-            Ok(_) => info!("route success"),
+            Ok(size) => {
+                if size > 0 {
+                    info!("route success - {}", size)
+                }
+            }
         }
 
         let elapsed = (time::now_utc() - start).num_milliseconds() as u64;
@@ -58,10 +62,15 @@ pub fn router(sinks: &Vec<config::Sink>,
 /// Route handle sources forwarding.
 fn route(sinks: &Vec<config::Sink>,
          parameters: &config::Parameters,
-         labels: &String)
-         -> Result<(), Box<Error>> {
-    debug!("route");
+         labels: &String,
+         sigint: Arc<AtomicBool>)
+         -> Result<usize, Box<Error>> {
+    let mut proc_size = 0;
+
     loop {
+        if sigint.load(Ordering::Relaxed) {
+            return Ok(proc_size);
+        }
         let entries = try!(fs::read_dir(&parameters.source_dir));
         let mut files = Vec::with_capacity(parameters.batch_count as usize);
         let mut metrics: Vec<String> = Vec::new();
@@ -127,6 +136,8 @@ fn route(sinks: &Vec<config::Sink>,
             batch_size += file.len();
         }
 
+        proc_size += metrics.len();
+
         // Nothing to do
         if files.len() == 0 {
             break;
@@ -185,7 +196,7 @@ fn route(sinks: &Vec<config::Sink>,
         }
     }
 
-    Ok(())
+    Ok(proc_size)
 }
 
 /// Read a file as String

diff --git a/src/sink.rs b/src/sink.rs
@@ -28,9 +28,13 @@ pub fn sink(sink: &config::Sink, parameters: &config::Parameters, sigint: Arc<At
     loop {
         let start = time::now_utc();
 
-        match send(sink, parameters) {
+        match send(sink, parameters, sigint.clone()) {
             Err(err) => error!("post fail: {}", err),
-            Ok(_) => info!("post success"),
+            Ok(size) => {
+                if size > 0 {
+                    info!("post success - {}", size)
+                }
+            }
         }
 
         let res = cappe(sink, parameters);
@@ -54,10 +58,17 @@ pub fn sink(sink: &config::Sink, parameters: &config::Parameters, sigint: Arc<At
 }
 
 /// Send sink metrics to Warp10.
-fn send(sink: &config::Sink, parameters: &config::Parameters) -> Result<(), Box<Error>> {
-    debug!("post {}", &sink.url);
+fn send(sink: &config::Sink,
+        parameters: &config::Parameters,
+        sigint: Arc<AtomicBool>)
+        -> Result<usize, Box<Error>> {
+    let mut proc_size = 0;
 
     loop {
+        if sigint.load(Ordering::Relaxed) {
+            return Ok(proc_size);
+        }
+
         let entries = try!(files(&parameters.sink_dir, &sink.name));
         let mut files = Vec::with_capacity(parameters.batch_count as usize);
         let mut metrics = String::new();
@@ -82,6 +93,8 @@ fn send(sink: &config::Sink, parameters: &config::Parameters) -> Result<(), Box<
             metrics.push_str("\n");
         }
 
+        proc_size += metrics.len();
+
         // Nothing to do
         if metrics.len() == 0 {
             break;
@@ -97,7 +110,7 @@ fn send(sink: &config::Sink, parameters: &config::Parameters) -> Result<(), Box<
         let mut headers = hyper::header::Headers::new();
         headers.set_raw(sink.token_header.clone(), vec![sink.token.clone().into()]);
 
-        debug!("post metrics");
+        debug!("post {}", &sink.url);
         let request = client.post(&sink.url).headers(headers).body(&metrics);
         let mut res = try!(request.send());
         if !res.status.is_success() {
@@ -115,7 +128,7 @@ fn send(sink: &config::Sink, parameters: &config::Parameters) -> Result<(), Box<
         }
     }
 
-    Ok(())
+    Ok(proc_size)
 }
 
 fn cappe(sink: &config::Sink, parameters: &config::Parameters) -> Result<(), Box<Error>> {
@@ -159,23 +172,25 @@ fn read(path: PathBuf) -> Result<String, Box<Error>> {
 }
 
 fn files(dir: &str, sink_name: &str) -> Result<Vec<fs::DirEntry>, Box<Error>> {
-    let mut entries: Vec<fs::DirEntry> = try!(fs::read_dir(dir)).filter_map(|entry| {
-        if entry.is_err() {
-            return None;
-        }
-        let entry = entry.unwrap();
-        if entry.path().extension() != Some(OsStr::new("metrics")) {
-            return None;
-        }
+    let mut entries: Vec<fs::DirEntry> = try!(fs::read_dir(dir))
+        .filter_map(|entry| {
+            if entry.is_err() {
+                return None;
+            }
+            let entry = entry.unwrap();
+            if entry.path().extension() != Some(OsStr::new("metrics")) {
+                return None;
+            }
 
-        let file_name = String::from(entry.file_name().to_str().unwrap_or(""));
+            let file_name = String::from(entry.file_name().to_str().unwrap_or(""));
 
-        if !file_name.starts_with(sink_name) {
-            return None;
-        }
+            if !file_name.starts_with(sink_name) {
+                return None;
+            }
 
-        Some(entry)
-    }).collect();
+            Some(entry)
+        })
+        .collect();
 
     entries.sort_by(|a, b| b.file_name().cmp(&a.file_name()));