Menu
Public repository / 811ef9b7c8ba017fa242f008efa90faab07a0454
AkurAI-Monitor / src/routes.rs
Latest change 811ef9b7c8ba017fa242f008efa90faab07a0454 - Adopt AkurAI-Framework UI layer and add framework-version tracking 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};
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,
}
pub async fn status() -> impl IntoResponse {
// Latest value of each metric
let metrics = db::with_db(|conn| {
let mut stmt = conn
.prepare(
"SELECT host_id, NULLIF(application_id, 0), name, value, ts
FROM current_metrics
ORDER BY host_id, application_id, name",
)
.unwrap();
stmt.query_map([], |row| {
Ok(MetricPoint {
host_id: row.get(0)?,
application_id: row.get(1)?,
name: row.get(2)?,
value: row.get(3)?,
ts: row.get(4)?,
})
})
.unwrap()
.filter_map(Result::ok)
.collect::<Vec<_>>()
});
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| {
let mut stmt = conn
.prepare(
"SELECT host_id, application_id, name, value, ts
FROM (
SELECT host_id, application_id, name, value, ts
FROM metrics
WHERE ts >= ?1
AND (?2 IS NULL OR name = ?2)
AND (?3 IS NULL OR host_id = ?3)
AND (?4 IS NULL OR application_id = ?4)
ORDER BY ts DESC
LIMIT 1000
)
ORDER BY ts",
)
.unwrap();
stmt.query_map(
rusqlite::params![cutoff, q.name, q.host_id, q.application_id],
|row| {
Ok(MetricPoint {
host_id: row.get(0)?,
application_id: row.get(1)?,
name: row.get(2)?,
value: row.get(3)?,
ts: row.get(4)?,
})
},
)
.unwrap()
.filter_map(Result::ok)
.collect::<Vec<_>>()
})
} 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: &rusqlite::Connection,
names: &[String],
cutoff: i64,
bucket_secs: i64,
host_id: Option<i64>,
application_id: Option<i64>,
) -> Vec<MetricPoint> {
let placeholders = (1..=names.len())
.map(|index| format!("?{index}"))
.collect::<Vec<_>>()
.join(",");
let cutoff_index = names.len() + 1;
let bucket_index = names.len() + 2;
let mut params = names
.iter()
.cloned()
.map(rusqlite::types::Value::Text)
.collect::<Vec<_>>();
params.push(rusqlite::types::Value::Integer(cutoff));
params.push(rusqlite::types::Value::Integer(bucket_secs));
let mut filters = String::new();
if let Some(value) = host_id {
params.push(rusqlite::types::Value::Integer(value));
filters.push_str(&format!(" AND host_id = ?{}", params.len()));
}
if let Some(value) = application_id {
params.push(rusqlite::types::Value::Integer(value));
filters.push_str(&format!(" AND application_id = ?{}", params.len()));
}
let sql = format!(
"SELECT host_id, application_id, name, AVG(value), MAX(ts)
FROM metrics
WHERE name IN ({placeholders})
AND ts >= ?{cutoff_index}
{filters}
GROUP BY host_id, application_id, name,
CAST((ts - ?{cutoff_index}) / ?{bucket_index} AS INTEGER)
ORDER BY ts"
);
let mut stmt = conn.prepare(&sql).unwrap();
stmt.query_map(rusqlite::params_from_iter(params), |row| {
Ok(MetricPoint {
host_id: row.get(0)?,
application_id: row.get(1)?,
name: row.get(2)?,
value: row.get(3)?,
ts: row.get(4)?,
})
})
.unwrap()
.filter_map(Result::ok)
.collect()
}
#[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| {
let mut stmt = conn
.prepare(
"SELECT id, host_id, application_id, source, line, level, target, pid, origin, fields, ts
FROM logs
WHERE ts >= ?1
AND (?2 IS NULL OR source = ?2)
AND (?3 IS NULL OR host_id = ?3)
AND (?4 IS NULL OR application_id = ?4)
AND (?5 IS NULL OR id < ?5)
AND (?6 IS NULL OR level = ?6)
ORDER BY id DESC
LIMIT ?7",
)
.unwrap();
stmt.query_map(
rusqlite::params![
cutoff,
q.source,
q.host_id,
q.application_id,
q.before_id,
q.level,
limit
],
|row| {
let fields: Option<String> = row.get(9)?;
Ok(LogEntry {
id: row.get(0)?,
host_id: row.get(1)?,
application_id: row.get(2)?,
source: row.get(3)?,
line: row.get(4)?,
level: row.get(5)?,
target: row.get(6)?,
pid: row.get(7)?,
origin: row.get(8)?,
fields: fields.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok()),
ts: row.get(10)?,
})
},
)
.unwrap()
.filter_map(Result::ok)
.collect::<Vec<_>>()
});
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| {
let mut stmt = conn
.prepare(
"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 >= ?1 OR e.resolved_at IS NULL)
AND (?2 IS NULL OR r.host_id = ?2)
AND (?3 IS NULL OR r.application_id = ?3)
ORDER BY e.triggered_at DESC",
)
.unwrap();
stmt.query_map(
rusqlite::params![cutoff, q.host_id, q.application_id],
|row| {
Ok(AlertEvent {
id: row.get(0)?,
host_id: row.get(1)?,
application_id: row.get(2)?,
rule_name: row.get(3)?,
metric_name: row.get(4)?,
threshold: row.get(5)?,
value: row.get(6)?,
triggered_at: row.get(7)?,
resolved_at: row.get(8)?,
})
},
)
.unwrap()
.filter_map(|r| r.ok())
.collect::<Vec<_>>()
});
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.query_row(
"SELECT COUNT(*) FROM applications WHERE id = ?1 AND host_id = ?2",
rusqlite::params![application_id, host_id],
|row| row.get::<_, i64>(0),
)
.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 tx = conn.unchecked_transaction()?;
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.execute(
"INSERT INTO logs
(host_id, application_id, source, line, level, target, pid, origin, fields, ts)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
rusqlite::params![
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::<_, rusqlite::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::*;
#[test]
fn chart_metrics_batches_requested_series_and_downsamples() {
let conn = rusqlite::Connection::open_in_memory().unwrap();
conn.execute_batch(
"CREATE TABLE metrics (
id INTEGER PRIMARY KEY,
host_id INTEGER,
application_id INTEGER,
name TEXT NOT NULL,
value REAL NOT NULL,
ts INTEGER NOT NULL
);",
)
.unwrap();
for ts in 0..120 {
for (name, offset) in [("cpu.usage", 0.0), ("mem.used_pct", 100.0)] {
conn.execute(
"INSERT INTO metrics (host_id, name, value, ts)
VALUES (1, ?1, ?2, ?3)",
rusqlite::params![name, ts as f64 + offset, ts],
)
.unwrap();
}
conn.execute(
"INSERT INTO metrics (host_id, name, value, ts)
VALUES (1, 'ignored', 1, ?1)",
[ts],
)
.unwrap();
}
let names = vec!["cpu.usage".to_string(), "mem.used_pct".to_string()];
let metrics = query_chart_metrics(&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);
}
}