Merge pull request #13 from runabove/metrics

Metrics

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "beamium"
-version = "1.0.3"
+version = "1.1.0"
 authors = [ "d33d33 <kevin.georges@corp.ovh.com>" ]
 
 build = "build.rs"

README.md

@@ -31,10 +31,12 @@ Config is composed of four parts:
 Beamium can have none to many Prometheus endpoints. A *source* is defined as follow:
 ``` yaml
 sources: # Sources definitions (Optional)
-  source1:                             # Source name         (Required)
-    url: http://127.0.0.1:9100/metrics # Prometheus endpoint (Required)
-    period: 10000                      # Polling interval    (Required)
-    format: prometheus                 # Polling format      (Optional, default: prometheus, value: [prometheus, sensision])
+  source1:                             # Source name                  (Required)
+    url: http://127.0.0.1:9100/metrics # Prometheus endpoint          (Required)
+    period: 10000                      # Polling interval             (Required)
+    format: prometheus                 # Polling format               (Optional, default: prometheus, value: [prometheus, sensision])
+    metrics:                           # Filter fetched metrics       (Optional)
+      - node.*                         # Regex used to select metrics (Required)
 ```
 
 #### Sinks

src/config.rs

@@ -34,6 +34,7 @@ pub struct Source {
     pub url: String,
     pub period: u64,
     pub format: SourceFormat,
+    pub metrics: Option<regex::RegexSet>,
 }
 
 #[derive(Debug)]
@@ -217,12 +218,26 @@ fn load_path<P: AsRef<Path>>(file_path: P, config: &mut Config) -> Result<(), Co
                             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)))));
+                        metrics.push(String::from(r"^(\S*)\s") + value.as_str());
+                    }
+
+                    Some(try!(regex::RegexSet::new(&metrics)))
+                };
 
                 config.sources.push(Source {
                     name: String::from(name),
                     url: String::from(url),
                     period: period,
                     format: format,
+                    metrics: metrics,
                 })
             }
         }
@@ -250,7 +265,7 @@ fn load_path<P: AsRef<Path>>(file_path: P, config: &mut Config) -> Result<(), Co
                 } else {
                     Some(try!(regex::Regex::new(try!(v["selector"]
                         .as_str()
-                        .ok_or(format!("sinks.{}.selector should be a string", name))))))
+                        .ok_or(format!("sinks.{}.selector is invalid", name))))))
                 };
 
                 config.sinks.push(Sink {

src/source.rs

@@ -87,9 +87,17 @@ fn fetch(source: &config::Source,
 
         for line in body.lines() {
             let line = match source.format {
-                config::SourceFormat::Sensision => line.trim().into(),
+                config::SourceFormat::Sensision => {
+                    match format_sensision(line.trim(), labels) {
+                        Err(_) => {
+                            warn!("bad row {}", &line);
+                            continue;
+                        }
+                        Ok(v) => v,
+                    }
+                },
                 config::SourceFormat::Prometheus => {
-                    match format(line.trim(), labels, now) {
+                    match format_prometheus(line.trim(), labels, now) {
                         Err(_) => {
                             warn!("bad row {}", &line);
                             continue;
@@ -103,6 +111,12 @@ fn fetch(source: &config::Source,
                 continue;
             }
 
+            if source.metrics.is_some() {
+                if !source.metrics.as_ref().unwrap().is_match(&line) {
+                    continue;
+                }
+            }
+
             try!(file.write(line.as_bytes()));
             try!(file.write(b"\n"));
         }
@@ -119,7 +133,7 @@ fn fetch(source: &config::Source,
 }
 
 /// Format Warp10 metrics from Prometheus one.
-fn format(line: &str, labels: &String, now: i64) -> Result<String, Box<Error>> {
+fn format_prometheus(line: &str, labels: &String, now: i64) -> Result<String, Box<Error>> {
     // Skip comments
     if line.starts_with("#") {
         return Ok(String::new());
@@ -143,24 +157,50 @@ fn format(line: &str, labels: &String, now: i64) -> Result<String, Box<Error>> {
         })
         .unwrap_or(now);
 
-    // Manage labels
-    let class = if labels.is_empty() {
-        String::from(class)
+    // Format class
+    let mut parts = class.splitn(2, "{");
+    let class = String::from(try!(parts.next().ok_or("no_class")));
+    let plabels = parts.next();
+    let mut slabels = if plabels.is_some() {
+        let mut labels = plabels.unwrap().split("\",")
+            .map(|v| v.replace("=", "%3D")) // escape
+            .map(|v| v.replace("%3D\"", "=")) // remove left double quote
+            .map(|v| v.replace("\"}", "")) // remove right double quote
+            .map(|v| v.replace(",", "%2C")) // escape
+            .map(|v| v.replace("}", "%7D")) // escape
+            .map(|v| v.replace(r"\\", r"\")) // unescape
+            .map(|v| v.replace("\\\"", "\"")) // unescape
+            .map(|v| v.replace(r"\n", "%0A")) // unescape
+            .fold(String::new(), |acc, x| acc + &x + ",");
+        labels.pop();
+        labels
     } else {
-        if class.ends_with("}") {
-            // Has Prometheus labels
-            let mut cl = String::from(class);
-            cl.pop();
-            if !class.ends_with("{") {
-                // Non empty labels, append ours
-                cl.push(',')
-            }
-            format!("{}{}}}", cl, labels)
-        } else {
-            // No labels
-            format!("{}{{{}}}", class, labels)
-        }
+        String::new()
     };
 
+    if !labels.is_empty() {
+        if !slabels.is_empty() {
+            slabels += ",";
+        }
+        slabels += labels;
+    }
+
+    let class = format!("{}{{{}}}", class, slabels);
+
     Ok(format!("{}// {} {}", timestamp, class, value))
 }
+
+/// Format Warp10 metrics from Sensision one.
+fn format_sensision(line: &str, labels: &String) -> Result<String, Box<Error>> {
+    if labels.is_empty() {
+        return Ok(String::from(line));
+    }
+    let mut parts = line.splitn(2, "{");
+
+    let class = String::from(try!(parts.next().ok_or("no_class")));
+    let plabels = String::from(try!(parts.next().ok_or("no_labels")));
+
+    let slabels = labels.clone() + if plabels.trim().starts_with("}") {""} else {","} + &plabels;
+
+    Ok(format!("{}{{{}", class, slabels))
+}