feat(sherlock): Added the logic to normalize data frame

sherlock/src/inference.rs

@@ -14,6 +14,7 @@ use ::slice_of_array::prelude::*;
 lazy_static! {
     static ref TENSORFLOW_END_POINT: String = var("TENSORFLOW_END_POINT").unwrap_or("http://localhost:8501/v1/models".to_string());
     static ref POOL_DURATION: String = var("POOL_DURATION").unwrap_or("60".to_string());
+    static ref DATA_COLLECT: String = var("DATA_COLLECT_ONLY").unwrap_or("off".to_string());
     static ref ANOMLAY_GAUGE: GaugeVec = register_gauge_vec!(
         opts!(
             "anomaly_score",
@@ -44,7 +45,7 @@ fn parse_config() -> Result<HashMap<String, InferenceData>, Box<dyn std::error::
     Ok(services)
 }
 
-async fn query_model(service: &str, input: [[[f64; 1]; 9]; 10]) -> Result<f64, Box<dyn std::error::Error>> {
+async fn query_model(service: &str, input: [[[f64; 3]; 9]; 10]) -> Result<f64, Box<dyn std::error::Error>> {
     let endpoint = format!("{}/{}:predict", TENSORFLOW_END_POINT.as_str(), service);
 
     let query = json!({
@@ -82,9 +83,12 @@ async fn save(service: &str, input: [[[f64; 3]; 9]; 10]) -> Result<(), Box<dyn s
 async fn calculate_anomaly_score(service: &str, args: &InferenceData) -> Result<(), Box<dyn std::error::Error>> {
     println!("Calculate anomaly score for {} using {}", service, &args.model_name);
     let input = build_telemetry_matrix(&service).await?;
-    save(&service, input).await?;
-    // let score = query_model(&args.model_name, input).await?;
-    let score = 0.4;
+    let mut score = 0.4;
+    if *DATA_COLLECT == "on"{
+        save(&service, input).await?;
+    }else{
+        score = query_model(&args.model_name, input).await?;
+    }
     ANOMLAY_GAUGE.with_label_values(&[service, &args.namespace]).set(score);
     println!("Anomaly score for {}: {}", service, score);
     Ok(())

sherlock/src/query.rs

@@ -1,4 +1,4 @@
-use std::env::var;
+use std::{env::var, f64::{INFINITY, NEG_INFINITY}, cmp};
 use prometheus_http_query::{Client, RangeVector};
 use chrono::{Duration, Local, DateTime};
 use lazy_static::lazy_static;
@@ -37,6 +37,13 @@ async fn query_prometheus(query: &str, time: DateTime<Local>) -> Result<f64,  Bo
     Ok(value)
 }
 
+fn after_pad(value: f64, add: bool) -> f64 {
+    if add {
+        return value + (value * 0.25)
+    }
+    return value - (value * 0.25)
+}
+
 pub async fn build_telemetry_matrix(service: &str) -> Result<[[[f64; 3]; 9]; 10],  Box<dyn std::error::Error>> {
     let mut data: [[[f64; 3]; 9]; 10] = [[[0.0; 3]; 9]; 10];
 
@@ -53,17 +60,47 @@ pub async fn build_telemetry_matrix(service: &str) -> Result<[[[f64; 3]; 9]; 10]
         Local::now() - Duration::minutes(256),
     ];
 
+    let mut low: [[f64; 3]; 9] = [[INFINITY; 3]; 9];
+    let mut high: [[f64; 3]; 9] = [[NEG_INFINITY; 3]; 9];
+
     for (x, time_step) in time_steps.iter().enumerate() {
         for (y, metric) in METRICS.iter().enumerate() {
             for (z, sample) in SAMPLES.iter().enumerate(){
                 let query = &metric.replace("SERVICE_NAME", service).replace("SAMPLE", sample);
                 match query_prometheus(query, *time_step).await {
-                    Ok(value) => data[x][y][z] = value,
+                    Ok(value) => {
+                        data[x][y][z] = value;
+                        if after_pad(value, true) > high[y][z] {
+                            high[y][z] = after_pad(value, true)
+                        }if after_pad(value, false) < low[y][z] {
+                            low[y][z] = after_pad(value, false)
+                        }
+                    },
                     Err(e) => return Err(e),
                 }
             }
         }
     }
 
+    for (_, timestep) in data.iter_mut().enumerate() {
+        for (y, metric) in timestep.iter_mut().enumerate() {
+            for (z, peroid) in metric.iter_mut().enumerate(){
+
+                let max = high[y][z];
+                let min = low[y][z];
+
+                let mut lower_bound = max - min;
+
+                if lower_bound <= 0.0{
+                    lower_bound = 1.0;
+                }
+
+                let value =  (*peroid - min) / lower_bound;
+
+                *peroid = value
+            }
+        }
+    }
+
     Ok(data)
 }

sherlock/training/.gitignore

@@ -1 +1 @@
-data/*.json
+*.json