Send metrics from dataplane to control-plane's queue

conductor/Cargo.toml

@@ -27,6 +27,9 @@ actix-web-opentelemetry = { version = "0.13.0", features = ["metrics-prometheus"
 opentelemetry = { version = "0.18", features = ["metrics", "rt-tokio-current-thread"] }
 opentelemetry-prometheus = "0.11"
 sqlx = "0.6.3"
+anyhow =  "1.0.82"
+serde_yaml = "0.9.34"
+reqwest = { version = "0.12.3", features = ["json"] }
 
 [dependencies.kube]
 features = ["runtime", "client", "derive"]

conductor/Dockerfile

@@ -14,6 +14,7 @@ COPY Cargo.toml .
 COPY Cargo.lock .
 COPY ./src ./src
 COPY ./migrations ./migrations
+COPY metrics.yml .
 
 RUN cargo install --path .
 

conductor/justfile

@@ -10,7 +10,9 @@ default:
 
 install-traefik:
 	kubectl create namespace traefik || true
-	helm upgrade --install --namespace=traefik --values=./testdata/traefik-values.yaml traefik traefik/traefik
+	helm upgrade --install --namespace=traefik --version=20.8.0 --values=./testdata/traefik-values.yaml traefik traefik/traefik
+	# https://github.com/traefik/traefik-helm-chart/issues/757#issuecomment-1753995542
+	kubectl apply -f https://raw.githubusercontent.com/traefik/traefik/v3.0.0-beta2/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml
 
 install-operator:
   just install-cert-manager
@@ -74,6 +76,7 @@ watch:
 	RUST_LOG={{RUST_LOG}} \
 	CONTROL_PLANE_EVENTS_QUEUE=myqueue_control_plane \
 	DATA_PLANE_EVENTS_QUEUE=myqueue_data_plane \
+	METRICS_EVENTS_QUEUE=metrics_events \
 	DATA_PLANE_BASEDOMAIN=coredb-development.com \
 	CF_TEMPLATE_BUCKET=cdb-plat-use1-dev-eks-data-1-conductor-cf-templates \
 	BACKUP_ARCHIVE_BUCKET=cdb-plat-use1-dev-instance-backups \

conductor/metrics.yml

@@ -0,0 +1,10 @@
+metrics:
+- name: is_standby
+  query: |
+    sum by (instance_id, pod) (
+      label_replace(
+        cnpg_pg_replication_in_recovery{} * on(namespace) group_left(label_tembo_io_instance_id)
+        kube_namespace_labels{label_tembo_io_instance_id!=""},
+        "instance_id", "$1", "label_tembo_io_instance_id", "(.*)"
+      )
+    )

conductor/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod aws;
 pub mod errors;
 pub mod extensions;
+pub mod metrics;
 pub mod monitoring;
 pub mod routes;
 pub mod types;

conductor/src/main.rs

@@ -25,10 +25,12 @@ use std::env;
 use std::sync::{Arc, Mutex};
 use std::{thread, time};
 
+use crate::metrics_reporter::run_metrics_reporter;
 use crate::status_reporter::run_status_reporter;
 use conductor::routes::health::background_threads_running;
 use types::{CRUDevent, Event};
 
+mod metrics_reporter;
 mod status_reporter;
 
 // Amount of time to wait after requeueing a message for an expected failure,
@@ -45,6 +47,8 @@ async fn run(metrics: CustomMetrics) -> Result<(), ConductorError> {
         env::var("POSTGRES_QUEUE_CONNECTION").expect("POSTGRES_QUEUE_CONNECTION must be set");
     let control_plane_events_queue =
         env::var("CONTROL_PLANE_EVENTS_QUEUE").expect("CONTROL_PLANE_EVENTS_QUEUE must be set");
+    let metrics_events_queue =
+        env::var("METRICS_EVENTS_QUEUE").expect("METRICS_EVENTS_QUEUE must be set");
     let data_plane_events_queue =
         env::var("DATA_PLANE_EVENTS_QUEUE").expect("DATA_PLANE_EVENTS_QUEUE must be set");
     let data_plane_basedomain =
@@ -69,6 +73,7 @@ async fn run(metrics: CustomMetrics) -> Result<(), ConductorError> {
     // Create queues if they do not exist
     queue.create(&control_plane_events_queue).await?;
     queue.create(&data_plane_events_queue).await?;
+    queue.create(&metrics_events_queue).await?;
 
     // Infer the runtime environment and try to create a Kubernetes Client
     let client = Client::try_default().await?;
@@ -579,7 +584,8 @@ async fn main() -> std::io::Result<()> {
         .expect("Failed to remember our background threads");
 
     let conductor_enabled = from_env_default("CONDUCTOR_ENABLED", "true");
-    let watcher_enabled = from_env_default("WATCHER_ENABLED", "true");
+    let status_reporter_enabled = from_env_default("WATCHER_ENABLED", "true");
+    let metrics_reported_enabled = from_env_default("METRICS_REPORTER_ENABLED", "true");
 
     if conductor_enabled != "false" {
         info!("Starting conductor");
@@ -616,7 +622,7 @@ async fn main() -> std::io::Result<()> {
         }));
     }
 
-    if watcher_enabled != "false" {
+    if status_reporter_enabled != "false" {
         info!("Starting status reporter");
         background_threads_locked.push(tokio::spawn({
             let custom_metrics_copy = custom_metrics.clone();
@@ -633,13 +639,31 @@ async fn main() -> std::io::Result<()> {
                             error!("error in conductor: {:?}", err);
                         }
                     }
-                    warn!("conductor exited, sleeping for 1 second");
+                    warn!("status_reporter exited, sleeping for 1 second");
                     thread::sleep(time::Duration::from_secs(1));
                 }
             }
         }));
     }
 
+    if metrics_reported_enabled != "false" {
+        info!("Starting status reporter");
+        let custom_metrics_copy = custom_metrics.clone();
+        background_threads_locked.push(tokio::spawn(async move {
+            let custom_metrics = &custom_metrics_copy;
+            if let Err(err) = run_metrics_reporter().await {
+                custom_metrics
+                    .conductor_errors
+                    .add(&opentelemetry::Context::current(), 1, &[]);
+
+                error!("error in metrics_reporter: {err}")
+            }
+
+            warn!("metrics_reporter exited, sleeping for 1 second");
+            thread::sleep(time::Duration::from_secs(1));
+        }));
+    }
+
     std::mem::drop(background_threads_locked);
 
     let server_port = env::var("PORT")

conductor/src/metrics.rs

@@ -0,0 +1,168 @@
+/// Types to deserialize Prometheus query responses
+pub mod prometheus {
+    use std::fmt;
+
+    use serde::de;
+    use serde::de::SeqAccess;
+    use serde::de::Visitor;
+    use serde::Deserialize;
+    use serde::Deserializer;
+    use serde::Serialize;
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    #[serde(rename_all = "camelCase")]
+    pub struct Metrics {
+        pub status: String,
+        pub data: MetricsData,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    #[serde(rename_all = "camelCase")]
+    pub struct MetricsData {
+        pub result_type: String,
+        pub result: Vec<MetricsResult>,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    #[serde(rename_all = "camelCase")]
+    pub struct MetricsResult {
+        // This value does not come in the Prometheus response,
+        // we add it in later.
+        pub metric: MetricLabels,
+        #[serde(deserialize_with = "custom_deserialize_tuple")]
+        pub value: (i64, i64),
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    pub struct MetricLabels {
+        pub instance_id: String,
+        pub pod: String,
+    }
+
+    fn custom_deserialize_tuple<'de, D>(deserializer: D) -> Result<(i64, i64), D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        struct TupleVisitor;
+
+        impl<'de> Visitor<'de> for TupleVisitor {
+            type Value = (i64, i64);
+
+            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+                formatter.write_str("a tuple of (f64, String)")
+            }
+
+            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: SeqAccess<'de>,
+            {
+                let f64_val: f64 = seq
+                    .next_element()?
+                    .ok_or_else(|| de::Error::invalid_length(0, &self))?;
+                let str_val: &str = seq
+                    .next_element()?
+                    .ok_or_else(|| de::Error::invalid_length(1, &self))?;
+
+                let timestamp = f64_val.trunc() as i64;
+                let parsed_int = str_val
+                    .parse::<i64>()
+                    .map_err(|_| de::Error::custom("Failed to parse string into integer"))?;
+
+                Ok((timestamp, parsed_int))
+            }
+        }
+
+        deserializer.deserialize_seq(TupleVisitor)
+    }
+}
+
+/// Data Plane metrics as packaged to be sent to Control Plane
+pub mod dataplane_metrics {
+    use serde::{Deserialize, Serialize};
+
+    use super::prometheus::MetricsResult;
+
+    #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+    #[serde(rename_all = "camelCase")]
+    pub struct DataPlaneMetrics {
+        /// Name of the corresponding metric
+        pub name: String,
+        /// Results of this metric for all instances
+        pub result: Vec<MetricsResult>,
+    }
+
+    impl DataPlaneMetrics {
+        pub fn from_query_result(name: &str, query_result: Vec<MetricsResult>) -> Self {
+            Self {
+                name: name.into(),
+                result: query_result,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::metrics::prometheus::{MetricLabels, Metrics, MetricsData, MetricsResult};
+
+    const QUERY_RESPONSE: &str = r#"
+    {
+        "status":"success",
+        "data":{
+           "resultType":"vector",
+           "result":[
+              {
+                 "metric":{
+                    "instance_id":"inst_0000000000000_AAAA0_1",
+                    "pod":"org-dummt-inst-dummy1"
+                 },
+                 "value":[
+                    1713365010.028,
+                    "0"
+                 ]
+              },
+              {
+                 "metric":{
+                    "instance_id":"inst_0000000000001_AAAB0_1",
+                    "pod":"org-dummy-2-inst-dummy-1"
+                 },
+                 "value":[
+                    1713365023.028,
+                    "1005"
+                 ]
+              }
+           ]
+        }
+     }
+    "#;
+
+    #[test]
+    fn deserializes_prometheus_responses_correctly() {
+        let response: Metrics = serde_json::from_str(QUERY_RESPONSE).unwrap();
+
+        let expected = Metrics {
+            status: "success".into(),
+            data: MetricsData {
+                result_type: "vector".into(),
+                result: vec![
+                    MetricsResult {
+                        metric: MetricLabels {
+                            instance_id: "inst_0000000000000_AAAA0_1".into(),
+                            pod: "org-dummt-inst-dummy1".into(),
+                        },
+                        value: (1713365010, 0),
+                    },
+                    MetricsResult {
+                        metric: MetricLabels {
+                            instance_id: "inst_0000000000001_AAAB0_1".into(),
+                            pod: "org-dummy-2-inst-dummy-1".into(),
+                        },
+                        value: (1713365023, 1005),
+                    },
+                ],
+            },
+        };
+
+        assert_eq!(response, expected);
+    }
+}