Adding basic support for Kuma

data/config.yaml

@@ -12,6 +12,7 @@ regions:
   - name: region-north
     send_interval: 5s
     miss_threshold: 3
+    kuma_url: https://status.kuma.example/api/push/xxx
     groups:
       - name: default
         fail_threshold: 4

src/cli/init.rs

@@ -37,7 +37,8 @@ pub fn init_config() -> Result<(), Error> {
             }],
             name: region_name,
             send_interval: Some("5s".to_string()),
-            miss_threshold: Some(3)
+            miss_threshold: Some(3),
+            kuma_url: None
         })
     }
 

src/relay/api.rs

@@ -54,7 +54,7 @@ impl ServerApi {
         serde_json::from_str::<RegionConfig>(&body).map_err(|err| Error::new("Failed to decode JSON region config", err))
     }
 
-    pub async fn update_region_state(&self, group_results: Vec<GroupResultInput>, last_update: &str) -> Result<Option<String>, Error> {
+    pub async fn update_region_state(&self, group_results: &Vec<GroupResultInput>, last_update: &str) -> Result<Option<String>, Error> {
 
         let json_state = serde_json::to_string(&group_results)
             .map_err(|err| Error::new("Could not parse region state to JSON", err))?;
@@ -80,4 +80,33 @@ impl ServerApi {
         Ok(None)
     }
 
+    pub async fn trigger_kuma_update(&self, kuma_url: &str, total_groups: usize, unstable_groups: usize, last_ping: Option<f32>) -> Result<(), Error> {
+
+        let message = if total_groups == unstable_groups {
+            format!("OK {} healthy", total_groups)
+        } else {
+            format!("WARN {} unstable", unstable_groups)
+        };
+
+        let mut kuma_full_url = format!("{}?status=up&msg={}", kuma_url, message);
+
+        if let Some(ping) = last_ping {
+            let ping_url = format!("&ping={}", ping);
+            kuma_full_url.push_str(&ping_url);
+        }
+
+        let http_response = self.client.get(kuma_full_url)
+            .send()
+            .await
+            .map_err(|err| Error::new("Could not fetch configuration from server", err))?;
+
+        if http_response.status() != 200 {
+            return Err(
+                Error::basic(format!("Expected 200 for Kuma update, found {}", http_response.status()))
+            );
+        }
+
+        Ok(())
+    }
+
 }

src/relay/service.rs

@@ -39,6 +39,8 @@ pub async fn launch(base_url: String, token: String, region_name: String) -> Res
         loop {
 
             let mut group_results: Vec<GroupResultInput> = vec![];
+            let mut last_kuma_ping: Option<f32> = None;
+
             for group in &region_config.groups {
 
                 // Each monitoring group in a region has multiple tests (ping, http, ...) to ensure
@@ -69,6 +71,11 @@ pub async fn launch(base_url: String, token: String, region_name: String) -> Res
                             }
 
                             for (metric_key, metric_value) in test.metrics.unwrap_or_default() {
+
+                                if metric_key == "ping_rtt" {
+                                    last_kuma_ping = Some(metric_value);
+                                }
+
                                 group_metrics.push(MetricInput {
                                     name: metric_key,
                                     labels: HashMap::from([
@@ -98,7 +105,7 @@ pub async fn launch(base_url: String, token: String, region_name: String) -> Res
                 });
             }
 
-            let update_result = api.update_region_state(group_results, &last_update).await;
+            let update_result = api.update_region_state(&group_results, &last_update).await;
             match update_result {
                 Ok(Some(watchdog_update)) => {
 
@@ -116,6 +123,16 @@ pub async fn launch(base_url: String, token: String, region_name: String) -> Res
                 _ => {}
             }
 
+            if let Some(kuma_url) = &region_config.kuma_url {
+
+                let total_groups = group_results.len();
+                let unstable_groups = group_results.iter().filter(|x| x.has_warnings || !x.working).count();
+
+                api.trigger_kuma_update(kuma_url, total_groups, unstable_groups, last_kuma_ping).await.unwrap_or_else(|err| {
+                    eprintln!("Error while triggering Kuma update: {}", err);
+                });
+            }
+
             let mut cancel_loop = false;
 
             tokio::select! {

src/server/config.rs

@@ -42,6 +42,7 @@ pub struct RegionConfigInput {
     pub name: String,
     pub send_interval: Option<String>,
     pub miss_threshold: Option<u64>,
+    pub kuma_url: Option<String>,
     pub groups: Vec<GroupConfigInput>
 }
 
@@ -65,6 +66,7 @@ pub struct RegionConfig {
     pub name: String,
     pub interval_ms: u64,
     pub threshold_ms: u64,
+    pub kuma_url: Option<String>,
     pub groups: Vec<GroupConfig>
 }
 
@@ -135,6 +137,7 @@ impl TryFrom<ConfigInput> for Config{
                 // We add 1000 to let the network the network request be processed
                 // after the interval multiple
                 threshold_ms: region_interval_ms * region_miss_threshold + 1000,
+                kuma_url: region_input.kuma_url.clone(),
                 groups
             };
             regions.push(region);