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

refactor query language

Browse Source
This commit is contained in:
nate
2026-04-09 23:30:57 +04:00
parent 783427ab34
commit 9b2f1de4e7
10 changed files with 111 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/monitor/src/query.rs
+21 -21
View File
@@ -9,6 +9,7 @@ pub struct Response {
pub headers: std::collections::HashMap<String, String>,
pub latency_ms: Option<u64>,
pub cert_expiry_days: Option<i64>,
pub cert_issuer: Option<String>,
}
pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
@@ -36,7 +37,11 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(!evaluate(not, response)?);
}
if let Some(cond) = map.get("$responseTime") {
if let Some(cond) = map.get("size") {
let val = Value::Number(serde_json::Number::from(response.body.len()));
return eval_condition(cond, &val, response);
}
if let Some(cond) = map.get("$time") {
let val = Value::Number(serde_json::Number::from(response.latency_ms.unwrap_or(0)));
return eval_condition(cond, &val, response);
}
@@ -44,6 +49,12 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
let val = Value::Number(serde_json::Number::from(response.cert_expiry_days.unwrap_or(0)));
return eval_condition(cond, &val, response);
}
if let Some(cond) = map.get("$certIssuer") {
let val = response.cert_issuer.as_ref()
.map(|s| Value::String(s.clone()))
.unwrap_or(Value::Null);
return eval_condition(cond, &val, response);
}
if let Some(json_path_map) = map.get("$json") {
let path_map = match json_path_map {
Value::Object(m) => m,
@@ -56,10 +67,10 @@ pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
return Ok(true);
}
if let Some(sel_map) = map.get("$select") {
if let Some(sel_map) = map.get("$html") {
let sel_obj = match sel_map {
Value::Object(m) => m,
_ => bail!("$select expects an object {{ selector: condition }}"),
_ => bail!("$html expects an object {{ selector: condition }}"),
};
let doc = Html::parse_document(&response.body);
for (selector, condition) in sel_obj {
@@ -152,27 +163,27 @@ fn eval_condition(condition: &Value, field_val: &Value, response: &Response) ->
}
}
fn eval_op(op: &str, field_val: &Value, val: &Value, response: &Response) -> Result<bool> {
fn eval_op(op: &str, field_val: &Value, val: &Value, _response: &Response) -> Result<bool> {
let ok = match op {
"$eq" => field_val == val,
"$ne" => field_val != val,
"$gt" => cmp_num(field_val, val, |a,b| a > b),
"$gte" => cmp_num(field_val, val, |a,b| a >= b),
"$ge" => cmp_num(field_val, val, |a,b| a >= b),
"$lt" => cmp_num(field_val, val, |a,b| a < b),
"$lte" => cmp_num(field_val, val, |a,b| a <= b),
"$contains" => {
"$le" => cmp_num(field_val, val, |a,b| a <= b),
"$co" => {
let needle = val.as_str().unwrap_or("");
field_val.as_str().map(|s| s.contains(needle)).unwrap_or(false)
}
"$startsWith" => {
"$sw" => {
let needle = val.as_str().unwrap_or("");
field_val.as_str().map(|s| s.starts_with(needle)).unwrap_or(false)
}
"$endsWith" => {
"$ew" => {
let needle = val.as_str().unwrap_or("");
field_val.as_str().map(|s| s.ends_with(needle)).unwrap_or(false)
}
"$regex" => {
"$re" => {
let pattern = val.as_str().unwrap_or("");
if pattern.len() > 200 { return Ok(false); }
let re = match Regex::new(pattern) {
@@ -186,17 +197,6 @@ fn eval_op(op: &str, field_val: &Value, val: &Value, response: &Response) -> Res
let exists = !field_val.is_null();
exists == should_exist
}
"$in" => {
if let Value::Array(arr) = val {
arr.contains(field_val)
} else {
false
}
}
"$select" => {
let sel_str = val.as_str().unwrap_or("");
css_select(&response.body, sel_str).is_some()
}
_ => {
tracing::warn!("Unknown query operator: {op}");
false
apps/monitor/src/runner.rs
+21 -6
View File
@@ -105,6 +105,7 @@ pub async fn fetch_and_run(
up: false,
error: Some(format!("timed out after {}ms", timeout_ms)),
cert_expiry_days: None,
cert_issuer: None,
meta: None,
region: Some(region_owned.to_string()),
run_id: Some(run_id_owned.clone()),
@@ -166,11 +167,12 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
let url = monitor.url.clone();
let req_headers = monitor.request_headers.clone();
let req_body = monitor.request_body.clone();
let max_redirects = monitor.max_redirects;
let (tx, rx) = tokio::sync::oneshot::channel::<Result<(u16, HashMap<String, String>, String), String>>();
std::thread::spawn(move || {
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
run_check_blocking(&url, &method, req_headers.as_ref(), req_body.as_deref(), timeout)
run_check_blocking(&url, &method, req_headers.as_ref(), req_body.as_deref(), timeout, max_redirects)
}));
let _ = tx.send(match result {
Ok(r) => r,
@@ -199,6 +201,7 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
up: false,
error: Some(e.clone()),
cert_expiry_days: None,
cert_issuer: None,
meta: None,
region: Some(region.to_string()),
run_id: Some(run_id.to_string()),
@@ -211,9 +214,9 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
Some(tokio::spawn(async move {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
check_cert_expiry(&cert_url),
check_cert(&cert_url),
).await {
Ok(Ok(days)) => days,
Ok(Ok(info)) => info,
_ => None,
}
}))
@@ -230,6 +233,7 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
headers: headers.clone(),
latency_ms: Some(latency_ms),
cert_expiry_days: None,
cert_issuer: None,
};
match query::evaluate(q, &response) {
Ok(result) => (result, None),
@@ -242,10 +246,12 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
((200..300).contains(&status), None)
};
let cert_expiry_days = match cert_handle {
let cert_info = match cert_handle {
Some(h) => h.await.unwrap_or(None),
None => None,
};
let cert_expiry_days = cert_info.as_ref().map(|c| c.expiry_days);
let cert_issuer = cert_info.map(|c| c.issuer);
let meta = json!({
"headers": headers,
@@ -264,6 +270,7 @@ async fn run_check(client: &reqwest::Client, monitor: &Monitor, scheduled_at: Op
up,
error: query_error,
cert_expiry_days,
cert_issuer,
meta: Some(meta),
region: Some(region.to_string()),
run_id: Some(run_id.to_string()),
@@ -278,6 +285,7 @@ fn run_check_blocking(
headers: Option<&HashMap<String, String>>,
body: Option<&str>,
timeout: std::time::Duration,
max_redirects: u32,
) -> Result<(u16, HashMap<String, String>, String), String> {
let root_certs = ROOT_CERTS.with(|c| Arc::clone(c));
@@ -289,6 +297,7 @@ fn run_check_blocking(
.timeout_global(Some(timeout))
.timeout_connect(Some(timeout))
.http_status_as_error(false)
.max_redirects(max_redirects)
.user_agent("Mozilla/5.0 (compatible; PingQL/1.0; +https://pingql.com)")
.tls_config(tls)
.build()
@@ -359,7 +368,12 @@ fn run_check_blocking(
Ok((status, resp_headers, body_out))
}
async fn check_cert_expiry(url: &str) -> Result<Option<i64>> {
struct CertInfo {
expiry_days: i64,
issuer: String,
}
async fn check_cert(url: &str) -> Result<Option<CertInfo>> {
use rustls::ClientConfig;
use rustls::pki_types::ServerName;
use tokio::net::TcpStream;
@@ -394,7 +408,8 @@ async fn check_cert_expiry(url: &str) -> Result<Option<i64>> {
.unwrap()
.as_secs() as i64;
let days = (not_after - now) / 86400;
return Ok(Some(days));
let issuer = cert.issuer().to_string();
return Ok(Some(CertInfo { expiry_days: days, issuer }));
}
Ok(None)
apps/monitor/src/types.rs
+5
View File
@@ -1,4 +1,6 @@
use serde::{Deserialize, Deserializer, Serialize};
fn default_max_redirects() -> u32 { 1 }
use serde_json::Value;
use std::collections::HashMap;
@@ -27,6 +29,8 @@ pub struct Monitor {
pub max_retries: u32,
#[serde(default)]
pub retry_interval_s: u64,
#[serde(default = "default_max_redirects")]
pub max_redirects: u32,
pub query: Option<Value>,
pub scheduled_at: Option<String>, // ISO string for backward compat in PingResult
#[serde(deserialize_with = "deserialize_ms")]
@@ -45,6 +49,7 @@ pub struct PingResult {
pub up: bool,
pub error: Option<String>,
pub cert_expiry_days: Option<i64>,
pub cert_issuer: Option<String>,
pub meta: Option<Value>,
pub region: Option<String>,
pub run_id: Option<String>,