feat: multi-region monitor support — region selector in UI, region flag on pings
This commit is contained in:
@@ -26,8 +26,9 @@ async fn main() -> Result<()> {
|
||||
.unwrap_or_else(|_| "http://localhost:3000".into());
|
||||
let monitor_token = env::var("MONITOR_TOKEN")
|
||||
.expect("MONITOR_TOKEN must be set");
|
||||
let region = env::var("REGION").unwrap_or_default();
|
||||
|
||||
info!("PingQL monitor starting, coordinator: {coordinator_url}");
|
||||
info!("PingQL monitor starting, coordinator: {coordinator_url}, region: {}", if region.is_empty() { "all" } else { ®ion });
|
||||
|
||||
let client = reqwest::Client::builder()
|
||||
.user_agent("PingQL-Monitor/0.1")
|
||||
@@ -37,7 +38,7 @@ async fn main() -> Result<()> {
|
||||
let in_flight: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
|
||||
|
||||
loop {
|
||||
match runner::fetch_and_run(&client, &coordinator_url, &monitor_token, &in_flight).await {
|
||||
match runner::fetch_and_run(&client, &coordinator_url, &monitor_token, ®ion, &in_flight).await {
|
||||
Ok(n) => { if n > 0 { info!("Spawned {n} checks"); } },
|
||||
Err(e) => error!("Check cycle failed: {e}"),
|
||||
}
|
||||
|
||||
@@ -13,11 +13,17 @@ pub async fn fetch_and_run(
|
||||
client: &reqwest::Client,
|
||||
coordinator_url: &str,
|
||||
token: &str,
|
||||
region: &str,
|
||||
in_flight: &Arc<Mutex<HashSet<String>>>,
|
||||
) -> Result<usize> {
|
||||
// Fetch due monitors
|
||||
// Fetch due monitors for this region
|
||||
let url = if region.is_empty() {
|
||||
format!("{coordinator_url}/internal/due")
|
||||
} else {
|
||||
format!("{coordinator_url}/internal/due?region={}", region)
|
||||
};
|
||||
let monitors: Vec<Monitor> = client
|
||||
.get(format!("{coordinator_url}/internal/due"))
|
||||
.get(&url)
|
||||
.header("x-monitor-token", token)
|
||||
.send()
|
||||
.await?
|
||||
@@ -42,12 +48,13 @@ pub async fn fetch_and_run(
|
||||
let client = client.clone();
|
||||
let coordinator_url = coordinator_url.to_string();
|
||||
let token = token.to_string();
|
||||
let region_owned = region.to_string();
|
||||
let in_flight = in_flight.clone();
|
||||
tokio::spawn(async move {
|
||||
let timeout_ms = monitor.timeout_ms.unwrap_or(30000);
|
||||
// Hard deadline: timeout + 5s buffer, so hung checks always resolve
|
||||
let deadline = std::time::Duration::from_millis(timeout_ms + 5000);
|
||||
let result = match tokio::time::timeout(deadline, run_check(&client, &monitor, monitor.scheduled_at.clone())).await {
|
||||
let result = match tokio::time::timeout(deadline, run_check(&client, &monitor, monitor.scheduled_at.clone(), ®ion_owned)).await {
|
||||
Ok(r) => r,
|
||||
Err(_) => PingResult {
|
||||
monitor_id: monitor.id.clone(),
|
||||
@@ -59,6 +66,7 @@ pub async fn fetch_and_run(
|
||||
error: Some(format!("timed out after {}ms", timeout_ms)),
|
||||
cert_expiry_days: None,
|
||||
meta: None,
|
||||
region: if region_owned.is_empty() { None } else { Some(region_owned.clone()) },
|
||||
},
|
||||
};
|
||||
// Post result first, then clear in-flight — this prevents the next
|
||||
@@ -73,7 +81,7 @@ pub async fn fetch_and_run(
|
||||
Ok(spawned)
|
||||
}
|
||||
|
||||
async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Option<String>) -> PingResult {
|
||||
async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Option<String>, region: &str) -> PingResult {
|
||||
// Compute jitter: how late we actually started vs when we were scheduled
|
||||
let jitter_ms: Option<i64> = scheduled_at.as_deref().and_then(|s| {
|
||||
let scheduled = chrono::DateTime::parse_from_rfc3339(s).ok()?;
|
||||
@@ -126,6 +134,7 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
|
||||
error: Some(e.clone()),
|
||||
cert_expiry_days: None,
|
||||
meta: None,
|
||||
region: if region.is_empty() { None } else { Some(region.to_string()) },
|
||||
}
|
||||
},
|
||||
Ok((status_code, headers, body)) => {
|
||||
@@ -185,6 +194,7 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
|
||||
error: query_error,
|
||||
cert_expiry_days,
|
||||
meta: Some(meta),
|
||||
region: if region.is_empty() { None } else { Some(region.to_string()) },
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ pub struct Monitor {
|
||||
pub interval_s: i64,
|
||||
pub query: Option<Value>,
|
||||
pub scheduled_at: Option<String>,
|
||||
pub regions: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
@@ -26,4 +27,5 @@ pub struct PingResult {
|
||||
pub error: Option<String>,
|
||||
pub cert_expiry_days: Option<i64>,
|
||||
pub meta: Option<Value>,
|
||||
pub region: Option<String>,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user