tawasuyu/brahman

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    let audit = state.audit.read().await;
62
63    let mut out = String::with_capacity(2048);
64
65    // ---- Entropía ----
66    out.push_str("# HELP ente_brain_entropy_bits Shannon entropy of marginal event distribution.\n");
67    out.push_str("# TYPE ente_brain_entropy_bits gauge\n");
68    out.push_str(&format!("ente_brain_entropy_bits {:.6}\n", obs.shannon_entropy()));
69
70    // ---- Tamaño de muestra ----
71    out.push_str("# HELP ente_brain_events_total Total events recorded by the observer.\n");
72    out.push_str("# TYPE ente_brain_events_total counter\n");
73    out.push_str(&format!("ente_brain_events_total {}\n", obs.total()));
74
75    // ---- Distinct kinds ----
76    out.push_str("# HELP ente_brain_distinct_kinds Number of distinct EventKind tags seen.\n");
77    out.push_str("# TYPE ente_brain_distinct_kinds gauge\n");
78    out.push_str(&format!("ente_brain_distinct_kinds {}\n", obs.marginals().len()));
79
80    // ---- Window ocupación ----
81    out.push_str("# HELP ente_brain_window_size Current sliding window length.\n");
82    out.push_str("# TYPE ente_brain_window_size gauge\n");
83    out.push_str(&format!("ente_brain_window_size {}\n", obs.current_window()));
84
85    // ---- Reglas vivas ----
86    out.push_str("# HELP ente_brain_rules_active Number of rules currently in the engine.\n");
87    out.push_str("# TYPE ente_brain_rules_active gauge\n");
88    out.push_str(&format!("ente_brain_rules_active {}\n", engine.len()));
89
90    // ---- Eventos por kind ----
91    out.push_str("# HELP ente_brain_events_by_kind Events by EventKind tag.\n");
92    out.push_str("# TYPE ente_brain_events_by_kind counter\n");
93    for (k, c) in obs.marginals() {
94        out.push_str(&format!(
95            "ente_brain_events_by_kind{{kind=\"{}\"}} {}\n",
96            kind_label(k), c
97        ));
98    }
99
100    // ---- Cristales detectados (con params actuales) ----
101    let crystals = crate::detect_crystals(&obs, &state.params);
102    out.push_str("# HELP ente_brain_crystals_total Number of crystals detected with current params.\n");
103    out.push_str("# TYPE ente_brain_crystals_total gauge\n");
104    out.push_str(&format!("ente_brain_crystals_total {}\n", crystals.len()));
105
106    // ---- Audit log ----
107    out.push_str("# HELP ente_brain_audit_chain_length Total entries persisted to CAS.\n");
108    out.push_str("# TYPE ente_brain_audit_chain_length counter\n");
109    out.push_str(&format!("ente_brain_audit_chain_length {}\n", audit.flushed_count()));
110
111    out.push_str("# HELP ente_brain_audit_in_memory Entries currently in the in-memory ring.\n");
112    out.push_str("# TYPE ente_brain_audit_in_memory gauge\n");
113    out.push_str(&format!("ente_brain_audit_in_memory {}\n", audit.len()));
114
115    out.push_str("# HELP ente_brain_audit_subscribers Active stream-audit subscribers.\n");
116    out.push_str("# TYPE ente_brain_audit_subscribers gauge\n");
117    out.push_str(&format!("ente_brain_audit_subscribers {}\n", audit.subscriber_count()));
118
119    if let Some(age) = audit.last_flush_age_secs() {
120        out.push_str("# HELP ente_brain_audit_last_flush_age_seconds Time since last flush to CAS.\n");
121        out.push_str("# TYPE ente_brain_audit_last_flush_age_seconds gauge\n");
122        out.push_str(&format!("ente_brain_audit_last_flush_age_seconds {:.3}\n", age));
123    }
124    if let Some(sha) = audit.last_flushed_sha() {
125        // Info-style metric con head sha como label. Útil para dashboards
126        // que quieran mostrar "current head".
127        out.push_str("# HELP ente_brain_audit_head_info Current head SHA of the audit chain.\n");
128        out.push_str("# TYPE ente_brain_audit_head_info gauge\n");
129        out.push_str(&format!(
130            "ente_brain_audit_head_info{{sha=\"{}\"}} 1\n",
131            ente_cas::hex(&sha)
132        ));
133    }
134
135    // ---- Histogramas de gaps temporales (top-32 pares más frecuentes) ----
136    out.push_str("# HELP ente_brain_pair_gap_seconds Time gap between correlated events.\n");
137    out.push_str("# TYPE ente_brain_pair_gap_seconds histogram\n");
138    let limits = crate::observer::GapHistogram::bucket_limits();
139    for ((a, b), hist) in obs.top_gap_pairs(32) {
140        let labels = format!(r#"a="{}",b="{}""#, kind_label(a), kind_label(b));
141        for (i, &limit) in limits.iter().enumerate() {
142            out.push_str(&format!(
143                "ente_brain_pair_gap_seconds_bucket{{{},le=\"{}\"}} {}\n",
144                labels, limit, hist.buckets[i]
145            ));
146        }
147        out.push_str(&format!(
148            "ente_brain_pair_gap_seconds_bucket{{{},le=\"+Inf\"}} {}\n",
149            labels, hist.count
150        ));
151        out.push_str(&format!(
152            "ente_brain_pair_gap_seconds_sum{{{}}} {:.6}\n",
153            labels, hist.sum_secs
154        ));
155        out.push_str(&format!(
156            "ente_brain_pair_gap_seconds_count{{{}}} {}\n",
157            labels, hist.count
158        ));
159    }
160
161    out
162}
163
164fn kind_label(k: &EventKind) -> &'static str {
165    match k {
166        EventKind::EnteSpawned => "EnteSpawned",
167        EventKind::EnteDied => "EnteDied",
168        EventKind::BusAnnounce => "BusAnnounce",
169        EventKind::BusInvoke => "BusInvoke",
170        EventKind::BusInvokeOf(_) => "BusInvokeOf",
171        EventKind::DeviceAdded => "DeviceAdded",
172        EventKind::DeviceRemoved => "DeviceRemoved",
173        EventKind::Custom(_) => "Custom",
174    }
175}