tawasuyu/arje

1//! Endpoint Prometheus en TCP. Formato text/plain (exposition format 0.0.4).
2//!
3//! Sin dependencias adicionales — la cardinalidad de nuestras métricas es
4//! pequeña y el formato es trivial. Si crece, sustituir por la crate
5//! `prometheus` con su Registry + encoders.
6
7use crate::introspect::BrainState;
8use crate::rules::EventKind;
9use std::net::SocketAddr;
10use tokio::io::{AsyncReadExt, AsyncWriteExt};
11use tokio::net::{TcpListener, TcpStream};
12use tracing::{info, trace, warn};
13
14/// Lanza el listener Prometheus. Devuelve cuando bind() falla; en caso
15/// contrario corre indefinidamente. Pensado para `tokio::spawn`.
16pub async fn serve_metrics(state: BrainState, addr: SocketAddr) -> anyhow::Result<()> {
17    let listener = TcpListener::bind(addr).await?;
18    info!(?addr, "prometheus /metrics escuchando");
19    loop {
20        match listener.accept().await {
21            Ok((stream, peer)) => {
22                trace!(?peer, "metrics scrape");
23                let s = state.clone();
24                tokio::spawn(async move {
25                    if let Err(e) = handle_scrape(stream, s).await {
26                        warn!(?e, "metrics conn ended");
27                    }
28                });
29            }
30            Err(e) => {
31                warn!(?e, "metrics accept failed");
32                return Ok(());
33            }
34        }
35    }
36}
37
38async fn handle_scrape(mut stream: TcpStream, state: BrainState) -> anyhow::Result<()> {
39    // Drenamos el request line + headers sin parsear (cualquier path
40    // responde igual — Prometheus envía GET /metrics típicamente).
41    let mut buf = [0u8; 1024];
42    let _ = stream.read(&mut buf).await;
43    let body = format_metrics(&state).await;
44    let resp = format!(
45        "HTTP/1.1 200 OK\r\n\
46         Content-Type: text/plain; version=0.0.4\r\n\
47         Content-Length: {}\r\n\
48         Connection: close\r\n\
49         \r\n\
50         {}",
51        body.len(), body
52    );
53    stream.write_all(resp.as_bytes()).await?;
54    stream.shutdown().await?;
55    Ok(())
56}
57
58async fn format_metrics(state: &BrainState) -> String {
59    let obs = state.observer.read().await;
60    let engine = state.engine.read().await;
61
62    let mut out = String::with_capacity(2048);
63
64    // ---- Entropía ----
65    out.push_str("# HELP ente_brain_entropy_bits Shannon entropy of marginal event distribution.\n");
66    out.push_str("# TYPE ente_brain_entropy_bits gauge\n");
67    out.push_str(&format!("ente_brain_entropy_bits {:.6}\n", obs.shannon_entropy()));
68
69    // ---- Tamaño de muestra ----
70    out.push_str("# HELP ente_brain_events_total Total events recorded by the observer.\n");
71    out.push_str("# TYPE ente_brain_events_total counter\n");
72    out.push_str(&format!("ente_brain_events_total {}\n", obs.total()));
73
74    // ---- Distinct kinds ----
75    out.push_str("# HELP ente_brain_distinct_kinds Number of distinct EventKind tags seen.\n");
76    out.push_str("# TYPE ente_brain_distinct_kinds gauge\n");
77    out.push_str(&format!("ente_brain_distinct_kinds {}\n", obs.marginals().len()));
78
79    // ---- Window ocupación ----
80    out.push_str("# HELP ente_brain_window_size Current sliding window length.\n");
81    out.push_str("# TYPE ente_brain_window_size gauge\n");
82    out.push_str(&format!("ente_brain_window_size {}\n", obs.current_window()));
83
84    // ---- Reglas vivas ----
85    out.push_str("# HELP ente_brain_rules_active Number of rules currently in the engine.\n");
86    out.push_str("# TYPE ente_brain_rules_active gauge\n");
87    out.push_str(&format!("ente_brain_rules_active {}\n", engine.len()));
88
89    // ---- Eventos por kind ----
90    out.push_str("# HELP ente_brain_events_by_kind Events by EventKind tag.\n");
91    out.push_str("# TYPE ente_brain_events_by_kind counter\n");
92    for (k, c) in obs.marginals() {
93        out.push_str(&format!(
94            "ente_brain_events_by_kind{{kind=\"{}\"}} {}\n",
95            kind_label(k), c
96        ));
97    }
98
99    // ---- Cristales detectados (con params actuales) ----
100    let crystals = crate::detect_crystals(&obs, &state.params);
101    out.push_str("# HELP ente_brain_crystals_total Number of crystals detected with current params.\n");
102    out.push_str("# TYPE ente_brain_crystals_total gauge\n");
103    out.push_str(&format!("ente_brain_crystals_total {}\n", crystals.len()));
104
105    out
106}
107
108fn kind_label(k: &EventKind) -> &'static str {
109    match k {
110        EventKind::EnteSpawned => "EnteSpawned",
111        EventKind::EnteDied => "EnteDied",
112        EventKind::BusAnnounce => "BusAnnounce",
113        EventKind::BusInvoke => "BusInvoke",
114        EventKind::BusInvokeOf(_) => "BusInvokeOf",
115        EventKind::DeviceAdded => "DeviceAdded",
116        EventKind::DeviceRemoved => "DeviceRemoved",
117        EventKind::Custom(_) => "Custom",
118    }
119}