Menu
Public repository / 3a463af917761506796f6f58d832f707a1e1af3a
AkurAI-Monitor / src/routes.rs
Latest change a8ff0f6c293e8e373c1ef695ede698e856d940fe - AKURAI-MONITOR-6 share metric row mapping by Ólafur Búi Ólafsson
use axum::Json;
use axum::extract::Query;
use axum::http::{HeaderMap, StatusCode};
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use serde_json::json;
use crate::{config, db, framework_version, health, stream};
use mysql::{params, prelude::Queryable};
pub async fn health() -> impl IntoResponse {
Json(json!({
"app": "rust-monitor",
"version": env!("CARGO_PKG_VERSION"),
"status": "ok",
"framework": "AkurAI-Framework",
"framework_version": framework_version::FRAMEWORK_VERSION,
}))
}
/// GET /api/ingestion/health — reports ingestion pipeline status for each configured source,
/// including last activity timestamps, errors, and stale detection.
pub async fn ingestion_health() -> impl IntoResponse {
let sources = health::snapshot();
let now = chrono::Utc::now().timestamp();
let cfg = config::get();
let stale_secs = cfg.log_stale_secs.max(60);
let stale_count = sources
.iter()
.filter(|source| {
source
.last_activity_ts
.map(|ts| now - ts > stale_secs)
.unwrap_or(true)
})
.count();
let error_count = sources
.iter()
.filter(|source| source.last_error.is_some())
.count();
let status = if error_count > 0 {
"degraded"
} else if stale_count > 0 {
"stale"
} else {
"ok"
};
let sources_with_stale: Vec<_> = sources
.into_iter()
.map(|s| {
let stale = s
.last_activity_ts
.map(|ts| now - ts > stale_secs)
.unwrap_or(true);
json!({
"source": s.source,
"kind": s.kind,
"last_activity_ts": s.last_activity_ts,
"stale": stale,
"last_error": s.last_error,
"last_error_ts": s.last_error_ts,
"error_count": s.error_count,
})
})
.collect();
Json(json!({
"status": status,
"stale_count": stale_count,
"error_count": error_count,
"sources": sources_with_stale,
"configured": {
"journald_units": cfg.journal_units,
"log_files": cfg.log_files,
},
"stale_threshold_secs": stale_secs,
"now": now,
}))
}
#[derive(Serialize)]
struct MetricPoint {
host_id: Option<i64>,
application_id: Option<i64>,
name: String,
value: f64,
ts: i64,
}
fn metric_point(
(host_id, application_id, name, value, ts): (Option<i64>, Option<i64>, String, f64, i64),
) -> MetricPoint {
MetricPoint {
host_id,
application_id,
name,
value,
ts,
}
}
pub async fn status() -> impl IntoResponse {
// Latest value of each metric
let metrics = db::with_db(|conn| {
conn.query_map(
"SELECT host_id, NULLIF(application_id, 0), name, value, ts
FROM current_metrics
ORDER BY host_id, application_id, name",
metric_point,
)
.unwrap_or_default()
});
Json(json!({
"metrics": metrics,
"framework_latest": framework_version::FRAMEWORK_VERSION,
}))
}
#[derive(Deserialize)]
pub struct MetricsQuery {
pub name: Option<String>,
pub hours: Option<i64>,
pub host_id: Option<i64>,
pub application_id: Option<i64>,
pub names: Option<String>,
pub points: Option<i64>,
}
pub async fn metrics(Query(q): Query<MetricsQuery>) -> impl IntoResponse {
let hours = q.hours.unwrap_or(24).max(1);
let cutoff = chrono::Utc::now()
.timestamp()
.saturating_sub(hours.saturating_mul(3600));
let names = q
.names
.as_deref()
.map(|value| {
value
.split(',')
.map(str::trim)
.filter(|name| !name.is_empty())
.take(100)
.map(str::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_default();
let metrics = if names.is_empty() {
db::with_db(|conn| {
conn.exec_map(
"SELECT host_id, application_id, name, value, ts
FROM (
SELECT host_id, application_id, name, value, ts
FROM metrics
WHERE ts >= :cutoff
AND (:name IS NULL OR name = :name)
AND (:host_id IS NULL OR host_id = :host_id)
AND (:application_id IS NULL OR application_id = :application_id)
ORDER BY ts DESC
LIMIT 1000
) AS recent
ORDER BY ts",
params! {
"cutoff" => cutoff,
"name" => &q.name,
"host_id" => q.host_id,
"application_id" => q.application_id,
},
metric_point,
)
.unwrap_or_default()
})
} else {
let points = q.points.unwrap_or(300).clamp(50, 1000);
let seconds = hours.saturating_mul(3600);
let bucket_secs = seconds.saturating_add(points - 1) / points;
db::with_db(|conn| {
query_chart_metrics(
conn,
&names,
cutoff,
bucket_secs,
q.host_id,
q.application_id,
)
})
};
Json(json!({"metrics": metrics, "count": metrics.len()}))
}
fn query_chart_metrics(
conn: &mut mysql::PooledConn,
names: &[String],
cutoff: i64,
bucket_secs: i64,
host_id: Option<i64>,
application_id: Option<i64>,
) -> Vec<MetricPoint> {
let placeholders = vec!["?"; names.len()].join(",");
let mut params: Vec<mysql::Value> = names.iter().map(|name| name.as_str().into()).collect();
params.push(cutoff.into());
let mut filters = String::new();
if let Some(value) = host_id {
filters.push_str(" AND host_id = ?");
params.push(value.into());
}
if let Some(value) = application_id {
filters.push_str(" AND application_id = ?");
params.push(value.into());
}
params.push(cutoff.into());
params.push(bucket_secs.into());
let sql = format!(
"SELECT host_id, application_id, name, AVG(value), MAX(ts)
FROM metrics
WHERE name IN ({placeholders})
AND ts >= ?
{filters}
GROUP BY host_id, application_id, name, FLOOR((ts - ?) / ?)
ORDER BY MAX(ts)"
);
conn.exec_map(sql, params, metric_point).unwrap_or_default()
}
#[derive(Deserialize)]
pub struct LogsQuery {
pub source: Option<String>,
pub level: Option<String>,
pub hours: Option<i64>,
pub host_id: Option<i64>,
pub application_id: Option<i64>,
pub before_id: Option<i64>,
pub limit: Option<i64>,
}
#[derive(Serialize)]
struct LogEntry {
id: i64,
host_id: Option<i64>,
application_id: Option<i64>,
source: String,
line: String,
level: Option<String>,
target: Option<String>,
pid: Option<i64>,
origin: Option<String>,
fields: Option<serde_json::Value>,
ts: i64,
}
pub async fn logs(Query(q): Query<LogsQuery>) -> impl IntoResponse {
let hours = q.hours.unwrap_or(1).clamp(1, 24 * 30);
let limit = q.limit.unwrap_or(500).clamp(1, 1000);
let cutoff = chrono::Utc::now().timestamp() - (hours * 3600);
let entries = db::with_db(|conn| {
conn.exec_map(
"SELECT id, host_id, application_id, source, line, level, target, pid, origin, fields, ts
FROM logs
WHERE ts >= :cutoff
AND (:source IS NULL OR source = :source)
AND (:host_id IS NULL OR host_id = :host_id)
AND (:application_id IS NULL OR application_id = :application_id)
AND (:before_id IS NULL OR id < :before_id)
AND (:level IS NULL OR level = :level)
ORDER BY id DESC
LIMIT :limit",
params! {
"cutoff" => cutoff,
"source" => &q.source,
"host_id" => q.host_id,
"application_id" => q.application_id,
"before_id" => q.before_id,
"level" => &q.level,
"limit" => limit,
},
#[allow(clippy::type_complexity)]
|(id, host_id, application_id, source, line, level, target, pid, origin, fields, ts): (
i64,
Option<i64>,
Option<i64>,
String,
String,
Option<String>,
Option<String>,
Option<i64>,
Option<String>,
Option<String>,
i64,
)| {
LogEntry {
id,
host_id,
application_id,
source,
line,
level,
target,
pid,
origin,
fields: fields.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok()),
ts,
}
},
)
.unwrap_or_default()
});
let next_before_id = entries.last().map(|entry| entry.id);
Json(json!({"logs": entries, "count": entries.len(), "next_before_id": next_before_id}))
}
#[derive(Deserialize)]
pub struct AlertsQuery {
pub hours: Option<i64>,
pub host_id: Option<i64>,
pub application_id: Option<i64>,
}
#[derive(Serialize)]
struct AlertEvent {
id: i64,
host_id: Option<i64>,
application_id: Option<i64>,
rule_name: String,
metric_name: String,
threshold: f64,
value: f64,
triggered_at: i64,
resolved_at: Option<i64>,
}
pub async fn alerts(Query(q): Query<AlertsQuery>) -> impl IntoResponse {
let hours = q.hours.unwrap_or(24);
let cutoff = chrono::Utc::now().timestamp() - (hours * 3600);
let events = db::with_db(|conn| {
conn.exec_map(
"SELECT e.id, r.host_id, r.application_id, r.name, r.metric_name,
r.threshold, e.value, e.triggered_at, e.resolved_at
FROM alert_events e JOIN alert_rules r ON e.rule_id = r.id
WHERE (e.triggered_at >= :cutoff OR e.resolved_at IS NULL)
AND (:host_id IS NULL OR r.host_id = :host_id)
AND (:application_id IS NULL OR r.application_id = :application_id)
ORDER BY e.triggered_at DESC",
params! {
"cutoff" => cutoff,
"host_id" => q.host_id,
"application_id" => q.application_id,
},
|(
id,
host_id,
application_id,
rule_name,
metric_name,
threshold,
value,
triggered_at,
resolved_at,
)| AlertEvent {
id,
host_id,
application_id,
rule_name,
metric_name,
threshold,
value,
triggered_at,
resolved_at,
},
)
.unwrap_or_default()
});
let active = events.iter().filter(|e| e.resolved_at.is_none()).count();
Json(json!({"alerts": events, "active": active, "total": events.len()}))
}
#[derive(Deserialize)]
pub struct IngestLine {
pub source: String,
pub line: String,
pub ts: Option<i64>,
pub host_id: Option<i64>,
pub application_id: Option<i64>,
pub level: Option<String>,
pub target: Option<String>,
pub pid: Option<i64>,
pub origin: Option<String>,
pub fields: Option<serde_json::Value>,
}
#[derive(Deserialize)]
pub struct IngestBody {
pub logs: Vec<IngestLine>,
}
/// POST /api/ingest — accept log lines shipped from other applications.
/// Authenticated by a bearer token (MONITOR_INGEST_TOKEN), not OIDC.
pub async fn ingest(headers: HeaderMap, Json(body): Json<IngestBody>) -> impl IntoResponse {
let token = config::get().ingest_token.as_str();
if token.is_empty() {
return (StatusCode::SERVICE_UNAVAILABLE, "ingest disabled").into_response();
}
let provided = headers
.get(axum::http::header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.strip_prefix("Bearer "))
.or_else(|| headers.get("x-ingest-token").and_then(|v| v.to_str().ok()));
// Length-aware constant-ish comparison
if provided.map(|p| p.as_bytes().ct_eq(token.as_bytes())) != Some(true) {
return (StatusCode::UNAUTHORIZED, "invalid ingest token").into_response();
}
if body.logs.is_empty() {
return Json(json!({"inserted": 0})).into_response();
}
let invalid_ownership = db::with_db(|conn| {
body.logs.iter().any(|line| {
let Some(application_id) = line.application_id else {
return false;
};
let Some(host_id) = line.host_id else {
return true;
};
conn.exec_first::<i64, _, _>(
"SELECT COUNT(*) FROM applications WHERE id = ? AND host_id = ?",
(application_id, host_id),
)
.ok()
.flatten()
.unwrap_or(0)
== 0
})
});
if invalid_ownership {
return (
StatusCode::BAD_REQUEST,
"application does not belong to host",
)
.into_response();
}
let now = chrono::Utc::now().timestamp();
let inserted = db::with_db(|conn| {
let mut tx = conn.start_transaction(mysql::TxOpts::default())?;
let mut inserted = 0i64;
for line in &body.logs {
let ts = line.ts.unwrap_or(now);
let fields_json = line.fields.as_ref().map(|v| v.to_string());
tx.exec_drop(
"INSERT INTO logs
(host_id, application_id, source, line, level, target, pid, origin, fields, ts)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
line.host_id,
line.application_id,
&line.source,
&line.line,
&line.level,
&line.target,
line.pid,
&line.origin,
&fields_json,
ts,
),
)?;
inserted += 1;
crate::pibun_push::push_error(&line.source, &line.line, line.level.as_deref());
}
tx.commit()?;
Ok::<_, mysql::Error>(inserted)
});
let Ok(inserted) = inserted else {
return (StatusCode::INTERNAL_SERVER_ERROR, "failed to store logs").into_response();
};
// Broadcast to live log views so shipped logs appear in real time
let logs: Vec<_> = body
.logs
.iter()
.map(|l| {
json!({
"host_id": l.host_id,
"application_id": l.application_id,
"source": l.source,
"line": l.line,
"level": l.level,
"target": l.target,
"pid": l.pid,
"origin": l.origin,
"fields": l.fields,
"ts": l.ts.unwrap_or(now)
})
})
.collect();
stream::publish("log", json!({"logs": logs}).to_string());
Json(json!({"inserted": inserted})).into_response()
}
/// Tiny constant-time byte comparison to avoid token timing leaks.
trait CtEq {
fn ct_eq(&self, other: &[u8]) -> bool;
}
impl CtEq for [u8] {
fn ct_eq(&self, other: &[u8]) -> bool {
if self.len() != other.len() {
return false;
}
let mut diff = 0u8;
for (a, b) in self.iter().zip(other.iter()) {
diff |= a ^ b;
}
diff == 0
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::testing::TestDb;
use mysql::prelude::Queryable;
#[test]
fn chart_metrics_batches_requested_series_and_downsamples() {
let Some(db) = TestDb::new() else { return };
let mut conn = db.conn();
crate::schema::create_tables(&mut conn);
let host: i64 = conn
.query_first("SELECT id FROM hosts WHERE name = 'local'")
.unwrap()
.unwrap();
for ts in 0..120 {
for (name, offset) in [("cpu.usage", 0.0), ("mem.used_pct", 100.0)] {
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, ?, ?, ?)",
(host, name, ts as f64 + offset, ts),
)
.unwrap();
}
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'ignored', 1, ?)",
(host, ts),
)
.unwrap();
}
let names = vec!["cpu.usage".to_string(), "mem.used_pct".to_string()];
let metrics = query_chart_metrics(&mut conn, &names, 0, 10, None, None);
assert_eq!(metrics.len(), 24);
assert!(metrics.iter().all(|metric| metric.name != "ignored"));
let first_cpu = metrics
.iter()
.find(|metric| metric.name == "cpu.usage" && metric.ts == 9)
.unwrap();
assert_eq!(first_cpu.value, 4.5);
}
}