Merge pull request #3 from samply/bridgehead-monitoring

Bridgehead monitoring

Cargo.toml

@@ -12,6 +12,9 @@ clap = { version = "4.3", features = ["derive", "env"] }
 reqwest = { version = "0.11", features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] }
+bridgehead-monitoring-lib = { git = "https://github.com/samply/bridgehead-monitoring" }
+beam-lib = { git = "https://github.com/samply/beam", branch = "develop" }
+serde_json = "1.0.104"
 
 [profile.release]
 #opt-level = "z"     # Optimize for size.

src/bridgehead_health.rs

@@ -0,0 +1,77 @@
+use beam_lib::{TaskRequest, FailureStrategy, MsgId, AppId, TaskResult};
+use clap::Args;
+use bridgehead_monitoring_lib::Check;
+use anyhow::{Result, bail};
+use reqwest::{Url, header, StatusCode};
+use serde_json::Value;
+
+use crate::icinga::IcingaCode;
+
+#[derive(Debug, Args)]
+pub struct BridgeheadCheck {
+    #[arg(long, env, value_parser)]
+    beam_proxy_url: Url,
+
+    /// Beam app id of this application
+    #[arg(long, env, value_parser)]
+    beam_proxy_name: String,
+
+    // Beam app secret of this application
+    #[arg(long, env, value_parser)]
+    beam_proxy_secret: String,
+
+    #[arg(value_parser = parse_check)]
+    checks: Vec<Check>,
+
+    /// Receiving bridgehead-monitoring beam id
+    #[arg(long, value_parser)]
+    to: String
+}
+
+pub fn parse_check(input: &str) -> Result<Check> {
+    Ok(serde_json::from_value(Value::String(input.to_owned()))?)
+}
+
+pub async fn check_bridgehead(
+    BridgeheadCheck {
+        beam_proxy_url,
+        beam_proxy_name,
+        beam_proxy_secret,
+        checks,
+        to
+    }: BridgeheadCheck
+) -> Result<IcingaCode> {
+    let client = reqwest::Client::new();
+    let task = TaskRequest {
+        id: MsgId::new(),
+        from: AppId::new_unchecked(&beam_proxy_name),
+        to: vec![AppId::new_unchecked(to)],
+        body: checks.clone(),
+        ttl: "60s".to_string(),
+        failure_strategy: FailureStrategy::Discard,
+        metadata: Value::Null,
+    };
+    let res = client
+        .post(beam_proxy_url.join("/v1/tasks")?)
+        .header(header::AUTHORIZATION, format!("ApiKey {} {}", beam_proxy_name, beam_proxy_secret))
+        .json(&task)
+        .send()
+        .await?;
+    if res.status() != StatusCode::CREATED {
+        bail!("Failed to create task: Got status {}", res.status());
+    }
+    let res = client
+        .get(beam_proxy_url.join(&format!("/v1/tasks/{}/results?wait_count=1", task.id))?)
+        .header(header::AUTHORIZATION, format!("ApiKey {} {}", beam_proxy_name, beam_proxy_secret))
+        .send()
+        .await?;
+    if res.status() != StatusCode::OK {
+        bail!("Failed to retrive task: Got status {}", res.status());
+    }
+    let results = res.json::<Vec<TaskResult<Vec<String>>>>().await?.pop();
+    match results {
+        Some(task) => checks.into_iter().zip(task.body).for_each(|(check, result)| println!("{check}: {result}")),
+        None => bail!("Got no results from task"),
+    };
+    Ok(IcingaCode::Ok)
+}

src/main.rs

@@ -3,11 +3,13 @@ use std::process::ExitCode;
 use clap::{Parser, Subcommand};
 use icinga::IcingaCode;
 use proxy_health::query_proxy_health;
+use bridgehead_health::check_bridgehead;
 use anyhow::Context;
 use reqwest::Url;
 
 mod icinga;
 mod proxy_health;
+mod bridgehead_health;
 
 #[derive(Debug, Parser)]
 #[clap(
@@ -16,21 +18,23 @@ mod proxy_health;
     arg_required_else_help(true),
 )]
 struct CliArgs {
-    #[clap(long, env, value_parser)]
-    monitoring_api_key: String,
-
-    #[clap(long, env, value_parser)]
-    broker_url: Url,
-
+
     #[command(subcommand)]
     command: SubCommands,
 }
 
 #[derive(Debug, Subcommand)]
 enum SubCommands {
     Health {
+        #[arg(long, env, value_parser)]
+        monitoring_api_key: String,
+
+        #[arg(long, env, value_parser)]
+        broker_url: Url,
         name: String,
-    }
+    },
+    Bridgehead(bridgehead_health::BridgeheadCheck)
+
 }
 
 
@@ -39,11 +43,12 @@ async fn main() -> ExitCode {
     let args = CliArgs::parse();
 
     let result = match args.command {
-        SubCommands::Health { name } => query_proxy_health(&name, &args.monitoring_api_key, &args.broker_url).await.context("Failed to query proxy health"),
+        SubCommands::Health { name, monitoring_api_key, broker_url } => query_proxy_health(&name, &monitoring_api_key, &broker_url).await.context("Failed to query proxy health"),
+        SubCommands::Bridgehead(bridgehead_check) => check_bridgehead(bridgehead_check).await,
     };
     match result {
         Err(e) => {
-            print!("{e}");
+            println!("{e}");
             IcingaCode::Unknown
         },
         Ok(code) => code,