ico/pingql
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fet: reduce LOC by reducing comments

Browse Source
This commit is contained in:
nate
2026-03-28 18:05:29 +04:00
parent 005f635fab
commit 8e554498f0
20 changed files with 2 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/monitor/src/main.rs
+1 -4
View File
@@ -13,10 +13,7 @@ use tracing::{error, info};
#[tokio::main]
async fn main() -> Result<()> {
// Install default rustls crypto provider (required for cert expiry checks)
rustls::crypto::ring::default_provider()
.install_default()
.ok(); // ok() — ignore error if already installed
rustls::crypto::ring::default_provider().install_default().ok();
tracing_subscriber::fmt()
.with_env_filter(env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
apps/monitor/src/query.rs
+1 -52
View File
@@ -1,38 +1,3 @@
/// PingQL query evaluation against a check response.
///
/// Query shape (MongoDB-inspired):
///
/// Simple equality:
/// { "status": 200 }
///
/// Operators:
/// { "status": { "$eq": 200 } }
/// { "status": { "$ne": 500 } }
/// { "status": { "$gte": 200, "$lt": 300 } }
/// { "body": { "$contains": "healthy" } }
/// { "body": { "$startsWith": "OK" } }
/// { "body": { "$endsWith": "done" } }
/// { "body": { "$regex": "ok|healthy" } }
/// { "body": { "$exists": true } }
/// { "status": { "$in": [200, 201, 204] } }
///
/// CSS selector (HTML parsing):
/// { "$select": "span.status", "$eq": "operational" }
///
/// JSONPath:
/// { "$json": "$.data.status", "$eq": "ok" }
///
/// Response time:
/// { "$responseTime": { "$lt": 500 } }
///
/// Certificate expiry:
/// { "$certExpiry": { "$gt": 30 } }
///
/// Logical:
/// { "$and": [ { "status": 200 }, { "body": { "$contains": "ok" } } ] }
/// { "$or": [ { "status": 200 }, { "status": 204 } ] }
/// { "$not": { "status": 500 } }
use anyhow::{bail, Result};
use regex::Regex;
use scraper::{Html, Selector};
@@ -46,11 +11,9 @@ pub struct Response {
pub cert_expiry_days: Option<i64>,
}
/// Returns true if `query` matches `response`. No query = always up.
pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
match query {
Value::Object(map) => {
// $consider — "up" (default) or "down": flips result if conditions match
if let Some(consider) = map.get("$consider") {
let is_down = consider.as_str() == Some("down");
let rest: serde_json::Map<String, Value> = map.iter()
@@ -61,7 +24,6 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(if is_down { !matches } else { matches });
}
// $and / $or / $not
if let Some(and) = map.get("$and") {
let Value::Array(clauses) = and else { bail!("$and expects array") };
return Ok(clauses.iter().all(|c| evaluate(c, response).unwrap_or(false)));
@@ -74,19 +36,14 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(!evaluate(not, response)?);
}
// $responseTime
if let Some(cond) = map.get("$responseTime") {
let val = Value::Number(serde_json::Number::from(response.latency_ms.unwrap_or(0)));
return eval_condition(cond, &val, response);
}
// $certExpiry
if let Some(cond) = map.get("$certExpiry") {
let val = Value::Number(serde_json::Number::from(response.cert_expiry_days.unwrap_or(0)));
return eval_condition(cond, &val, response);
}
// $json — { "$json": { "$.path": { "$op": val } } }
if let Some(json_path_map) = map.get("$json") {
let path_map = match json_path_map {
Value::Object(m) => m,
@@ -99,13 +56,11 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(true);
}
// $select — { "$select": { "css.selector": { "$op": val } } }
if let Some(sel_map) = map.get("$select") {
let sel_obj = match sel_map {
Value::Object(m) => m,
_ => bail!("$select expects an object {{ selector: condition }}"),
};
// Parse HTML once for all selectors
let doc = Html::parse_document(&response.body);
for (selector, condition) in sel_obj {
let sel = Selector::parse(selector)
@@ -118,7 +73,6 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(true);
}
// Field-level checks
for (field, condition) in map {
let field_val = resolve_field(field, response);
if !eval_condition(condition, &field_val, response)? {
@@ -154,7 +108,6 @@ fn resolve_json_path(body: &str, path: &str) -> Value {
if path.is_empty() { return obj; }
let mut current = &obj;
for part in path.split('.') {
// Handle array indexing like "items[0]"
if let Some(idx_start) = part.find('[') {
let key = &part[..idx_start];
if !key.is_empty() {
@@ -184,7 +137,6 @@ fn resolve_json_path(body: &str, path: &str) -> Value {
fn eval_condition(condition: &Value, field_val: &Value, response: &Response) -> Result<bool> {
match condition {
// Shorthand: { "status": 200 }
Value::Number(n) => Ok(field_val.as_f64() == n.as_f64()),
Value::String(s) => Ok(field_val.as_str() == Some(s.as_str())),
Value::Bool(b) => Ok(field_val.as_bool() == Some(*b)),
@@ -242,11 +194,8 @@ fn eval_op(op: &str, field_val: &Value, val: &Value, response: &Response) -> Res
}
}
"$select" => {
// Nested: { "body": { "$select": "css", "$eq": "val" } }
let sel_str = val.as_str().unwrap_or("");
let selected = css_select(&response.body, sel_str);
// If no comparison operator follows, just check existence
selected.is_some()
css_select(&response.body, sel_str).is_some()
}
_ => {
tracing::warn!("Unknown query operator: {op}");
apps/monitor/src/runner.rs
-39
View File
@@ -8,7 +8,6 @@ use std::time::Instant;
use tokio::sync::Mutex;
use tracing::{debug, warn};
// Cache native root certs per OS thread to avoid reloading from disk on every check.
thread_local! {
static ROOT_CERTS: Arc<Vec<ureq::tls::Certificate<'static>>> = Arc::new(
rustls_native_certs::load_native_certs()
@@ -19,7 +18,6 @@ thread_local! {
);
}
/// Fetch due monitors from coordinator, run them, post results back.
pub async fn fetch_and_run(
client: &reqwest::Client,
coordinator_url: &str,
@@ -27,8 +25,6 @@ pub async fn fetch_and_run(
region: &str,
in_flight: &Arc<Mutex<HashSet<String>>>,
) -> Result<usize> {
// Fetch monitors due within the next 2s — nodes receive exact scheduled_at_ms
// and sleep until that moment, so all regions fire in tight coordination.
let url = if region.is_empty() {
format!("{coordinator_url}/internal/due?lookahead_ms=2000")
} else {
@@ -45,12 +41,10 @@ pub async fn fetch_and_run(
let n = monitors.len();
if n == 0 { return Ok(0); }
// Shared read-only strings — clone the Arc instead of allocating per monitor
let coordinator_url: Arc<str> = Arc::from(coordinator_url);
let token: Arc<str> = Arc::from(token);
let region: Arc<str> = Arc::from(region);
// Spawn all checks — fire and forget, skip if already in-flight
let mut spawned = 0usize;
for monitor in monitors {
{
@@ -66,7 +60,6 @@ pub async fn fetch_and_run(
let coordinator_url = Arc::clone(&coordinator_url);
let token = Arc::clone(&token);
let region_owned = Arc::clone(&region);
// run_id: hash(monitor_id, interval_bucket) — same across all regions for this window
let run_id_owned = {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
@@ -77,7 +70,6 @@ pub async fn fetch_and_run(
bucket.hash(&mut h);
format!("{:016x}", h.finish())
};
// Convert scheduled_at_ms to an ISO string for storage in the ping
let scheduled_at_iso = monitor.scheduled_at_ms.map(|ms| {
chrono::DateTime::<chrono::Utc>::from_timestamp_millis(ms)
.map(|dt| dt.to_rfc3339())
@@ -85,7 +77,6 @@ pub async fn fetch_and_run(
});
let in_flight = in_flight.clone();
tokio::spawn(async move {
// Sleep until the exact scheduled moment — tight multi-region coordination
if let Some(ms) = monitor.scheduled_at_ms {
let now_ms = chrono::Utc::now().timestamp_millis();
if ms > now_ms {
@@ -94,7 +85,6 @@ pub async fn fetch_and_run(
}
}
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, scheduled_at_iso.clone(), &region_owned, &run_id_owned)).await {
Ok(r) => r,
@@ -113,8 +103,6 @@ pub async fn fetch_and_run(
run_id: Some(run_id_owned.clone()),
},
};
// Post result first, then clear in-flight — this prevents the next
// poll from picking up the monitor again before the ping is persisted.
if let Err(e) = post_result(&client, &coordinator_url, &token, result).await {
warn!("Failed to post result for {}: {e}", monitor.id);
}
@@ -126,10 +114,7 @@ pub async fn fetch_and_run(
}
async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Option<String>, region: &str, run_id: &str) -> PingResult {
// Record when the check actually started (used as checked_at in the ping)
let checked_at = chrono::Utc::now().to_rfc3339();
// 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()?;
let now = chrono::Utc::now();
@@ -138,15 +123,10 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
let start = Instant::now();
// Build request with method, headers, body, timeout
let method = monitor.method.as_deref().unwrap_or("GET").to_uppercase();
let timeout = std::time::Duration::from_millis(monitor.timeout_ms.unwrap_or(30000));
let is_https = monitor.url.starts_with("https://");
// Run the check in a real OS thread using ureq (blocking, synchronous HTTP).
// ureq sets SO_RCVTIMEO/SO_SNDTIMEO at the socket level, which reliably
// interrupts even a hanging TLS handshake — unlike async reqwest which
// cannot cancel syscall-level blocks via future cancellation.
let url = monitor.url.clone();
let req_headers = monitor.request_headers.clone();
let req_body = monitor.request_body.clone();
@@ -190,9 +170,6 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
},
Ok((status, headers, body)) => {
// Start cert expiry check in background — don't block result posting.
// We'll use None for cert_expiry_days in query evaluation since it
// shouldn't delay the main result by seconds of extra TLS handshake.
let cert_handle = if is_https {
let cert_url = monitor.url.clone();
Some(tokio::spawn(async move {
@@ -210,9 +187,6 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
let query = &monitor.query;
// Evaluate query if present (cert_expiry_days not yet available —
// $certExpiry queries will use None here; the actual value is
// attached to the result once the background check completes)
let (up, query_error) = if let Some(q) = query {
let response = Response {
status,
@@ -225,16 +199,13 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
Ok(result) => (result, None),
Err(e) => {
warn!("Query error for {}: {e}", monitor.id);
// Fall back to status-based up/down
(status < 400, Some(e.to_string()))
}
}
} else {
// Default: up if 2xx/3xx
(status < 400, None)
};
// Await the cert check now (it's been running concurrently during query eval)
let cert_expiry_days = match cert_handle {
Some(h) => h.await.unwrap_or(None),
None => None,
@@ -265,10 +236,6 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
}
}
/// Run an HTTP check synchronously using ureq.
/// ureq applies timeouts at the socket/IO level (not just future cancellation),
/// which reliably interrupts hanging TLS handshakes.
/// Must be called from a std::thread (not async context).
fn run_check_blocking(
url: &str,
method: &str,
@@ -276,7 +243,6 @@ fn run_check_blocking(
body: Option<&str>,
timeout: std::time::Duration,
) -> Result<(u16, HashMap<String, String>, String), String> {
// Reuse cached root certs (loaded once per OS thread via thread_local)
let root_certs = ROOT_CERTS.with(|c| Arc::clone(c));
let tls = ureq::tls::TlsConfig::builder()
@@ -352,8 +318,6 @@ fn run_check_blocking(
Ok((status, resp_headers, body_out))
}
/// Check SSL certificate expiry for a given HTTPS URL.
/// Returns the number of days until the certificate expires.
async fn check_cert_expiry(url: &str) -> Result<Option<i64>> {
use rustls::ClientConfig;
use rustls::pki_types::ServerName;
@@ -361,12 +325,10 @@ async fn check_cert_expiry(url: &str) -> Result<Option<i64>> {
use tokio_rustls::TlsConnector;
use x509_parser::prelude::*;
// Parse host and port from URL
let url_parsed = reqwest::Url::parse(url)?;
let host = url_parsed.host_str().unwrap_or("");
let port = url_parsed.port().unwrap_or(443);
// Build a rustls config that captures certificates
let mut root_store = rustls::RootCertStore::empty();
root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
@@ -380,7 +342,6 @@ async fn check_cert_expiry(url: &str) -> Result<Option<i64>> {
let stream = TcpStream::connect(format!("{host}:{port}")).await?;
let tls_stream = connector.connect(server_name, stream).await?;
// Get peer certificates
let (_, conn) = tls_stream.get_ref();
let certs = conn.peer_certificates().unwrap_or(&[]);