Fix heartbeat metrics read from CH

Author: the2pizza

src/bin/api/core/clickhouse/mod.rs

@@ -1,24 +1,20 @@
 use clickhouse::Client as ChClient;
-use std::sync::Arc;
-use tokio::sync::Mutex;
 
 pub mod query;
 
 #[derive(Clone)]
 pub struct ChContext {
-    inner: Arc<Mutex<ChClient>>,
+    url: String,
 }
 
 impl ChContext {
     pub fn new(url: &str) -> Self {
-        let client = ChClient::default().with_url(url);
-
         Self {
-            inner: Arc::new(Mutex::new(client)),
+            url: url.to_string(),
         }
     }
 
-    pub fn client(&self) -> Arc<Mutex<ChClient>> {
-        self.inner.clone()
+    pub fn client(&self) -> ChClient {
+        ChClient::default().with_url(&self.url)
     }
 }

src/bin/api/core/clickhouse/query.rs

@@ -40,22 +40,30 @@ impl Queries for ChContext {
         T: DeserializeOwned + Debug + Send + Clone + Sync + 'static,
     {
         let query = format!(
-            "SELECT 
-                  toInt64(toUnixTimestamp(toDateTime(argMax(Timestamp, Timestamp)))) AS timestamp,
-                  Path AS metric,
-                  toFloat64(argMax(Value, Timestamp)) AS value
-            FROM default.graphite_data
-            WHERE Path LIKE '{}.{}.{}.heartbeat'
-            GROUP BY Path",
+            r#"SELECT Path AS metric,
+          toFloat64(argMax(Value, Timestamp)) AS value,
+          toInt64(toUnixTimestamp(toDateTime(argMax(Timestamp, Timestamp)))) AS timestamp     
+FROM default.graphite_data
+WHERE Path = '{}.{}.{}.heartbeat'
+GROUP BY Path"#,
             env, hostname, uuid
         );
 
-        log::debug!("CH Query {}", query);
+        println!("CH Query\n{}", query);
 
         let client = self.client();
-        let client = client.lock().await;
 
         let result = client.query(&query).fetch_all::<Metric<T>>().await;
+
+        match result {
+            Ok(ref rows) => {
+                log::debug!("Fetched {} rows from CH", rows.len());
+            }
+            Err(ref e) => {
+                log::error!("Failed to fetch heartbeat from CH: {:?}", e);
+            }
+        }
+
         result.ok().and_then(|mut rows| rows.pop())
     }
 
@@ -69,30 +77,28 @@ impl Queries for ChContext {
     {
         let start_str = start.format("%Y-%m-%d %H:%M:%S").to_string();
         let query = format!(
-            "SELECT 
-                   anyLast(timestamp) as timestamp,
-                   metric, 
-                   toInt64(sum(metric_value)) AS value
-               FROM (
-                 SELECT
-                   toInt64(toUnixTimestamp(toDateTime(anyLast(Timestamp)))) AS timestamp,
-                   extract(Path, '[^.]+$') AS metric,
-                   anyLast(Value) AS metric_value
-                 FROM default.graphite_data
-                 WHERE Path LIKE '%.%.{conn_id}.conn_stat.%'
-                    AND Timestamp >= toDateTime('{start}')
-                    AND Timestamp < toDateTime('{start}') + INTERVAL 1 DAY
-                 GROUP BY Path
-                )
-            GROUP BY metric",
+            r#"SELECT  metric, 
+           toInt64(sum(metric_value)) AS value,
+           anyLast(timestamp) AS timestamp 
+FROM (
+  SELECT
+    toInt64(toUnixTimestamp(toDateTime(anyLast(Timestamp)))) AS timestamp,
+    extract(Path, '[^.]+$') AS metric,
+    anyLast(Value) AS metric_value
+  FROM default.graphite_data
+  WHERE Path LIKE '%.%.{conn_id}.conn_stat.%'
+     AND Timestamp >= toDateTime('{start}')
+     AND Timestamp < toDateTime('{start}') + INTERVAL 1 DAY
+  GROUP BY Path
+ )
+GROUP BY metric"#,
             conn_id = conn_id,
             start = start_str,
         );
 
-        log::debug!("CH Query {}", query);
+        println!("CH Query\n{}", query);
 
         let client = self.client();
-        let client = client.lock().await;
 
         let result = client.query(&query).fetch_all::<Metric<T>>().await;
 

src/bin/api/main.rs

@@ -38,7 +38,7 @@ async fn main() -> Result<()> {
         .expect("required config path as an argument");
     println!("Config file {:?}", config_path);
 
-    let mut settings = ApiSettings::new(&config_path);
+    let settings = ApiSettings::new(&config_path);
 
     settings.validate().expect("Wrong settings file");
     println!(">>> Settings: {:?}", settings.clone());
@@ -61,7 +61,13 @@ async fn main() -> Result<()> {
 
     let debug = settings.debug.enabled;
 
-    let db = PgContext::init(&settings.pg).await.expect("DB error");
+    let db = match PgContext::init(&settings.pg).await {
+        Ok(db) => db,
+        Err(err) => {
+            log::error!("Failed to init DB: {}", err);
+            return Err(err.into());
+        }
+    };
 
     let ch = ChContext::new(&settings.clickhouse.address);
     let publisher = ZmqPublisher::new(&settings.zmq.endpoint).await;

src/metrics/metrics.rs

@@ -50,9 +50,9 @@ impl<T: fmt::Display> fmt::Display for Metric<T> {
 
 impl<T: std::fmt::Display + std::fmt::Debug> Metric<T> {
     pub async fn send(&self, server: &str) -> Result<(), io::Error> {
-        let metric_string = format!("{}\n", self.to_string());
+        let metric_string = format!("{self}\n");
 
-        log::debug!("Sending carbon metric string: {}", metric_string);
+        log::debug!("Sending carbon metric string: {:?}", metric_string);
 
         match TcpStream::connect(server).await {
             Ok(mut stream) => {