lfg2025/archy
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 33 Wiki Activity

feat: add real-time metrics collection with ring buffer storage (MON-01)

Browse Source 
Implements monitoring/collector.rs that collects per-container CPU/RAM/network/disk,
system-wide metrics, RPC latency, and WebSocket connection count every 60 seconds.
Data stored in dual ring buffers: 1-min resolution (24h) and 15-min resolution (7d).
Three new RPC endpoints: monitoring.current, monitoring.history, monitoring.containers.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Dorian
2026-03-11 11:11:02 +00:00
parent 45032d937b
commit 592548066e
8 changed files with 1285 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
core/archipelago/src/api/handler.rs
View File

@@ -1,6 +1,8 @@
use crate::api::rpc::RpcHandler;
use crate::content_server;
use crate::electrs_status;
use crate::monitoring::MetricsStore;
use crate::network::dwn_store::DwnStore;
use crate::node_message as node_msg;
use crate::config::Config;
use crate::session::{self, SessionStore};
@@ -12,26 +14,39 @@ use hyper_ws_listener::WsStream;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_tungstenite::tungstenite::Message;
use std::time::Instant;
use tracing::{debug, info};
pub struct ApiHandler {
config: Config,
rpc_handler: Arc<RpcHandler>,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
session_store: SessionStore,
}
impl ApiHandler {
pub async fn new(config: Config, state_manager: Arc<StateManager>) -> Result<Self> {
pub async fn new(
config: Config,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
) -> Result<Self> {
let session_store = SessionStore::new();
let rpc_handler = Arc::new(
RpcHandler::new(config.clone(), state_manager.clone(), session_store.clone()).await?,
RpcHandler::new(
config.clone(),
state_manager.clone(),
metrics_store.clone(),
session_store.clone(),
)
.await?,
);
Ok(Self {
config,
rpc_handler,
state_manager,
metrics_store,
session_store,
})
}
@@ -105,7 +120,7 @@ impl ApiHandler {
if !self.is_authenticated(req.headers()).await {
return Ok(Self::unauthorized());
}
return Self::handle_websocket(req, self.state_manager.clone()).await;
return Self::handle_websocket(req, self.state_manager.clone(), self.metrics_store.clone()).await;
}
// Convert body to bytes for non-WS routes
@@ -163,6 +178,19 @@ impl ApiHandler {
Self::handle_lnd_proxy(path, &origin).await
}
// DWN health — unauthenticated
(Method::GET, "/dwn/health") => {
Self::handle_dwn_health(&self.config).await
}
// DWN message processing — authenticated
(Method::POST, "/dwn") => {
if !self.is_authenticated(&headers).await {
return Ok(Self::unauthorized());
}
Self::handle_dwn_message(body_bytes, &self.config).await
}
_ => Ok(Response::builder()
.status(StatusCode::NOT_FOUND)
.body(hyper::Body::from("Not Found"))
@@ -439,6 +467,7 @@ impl ApiHandler {
async fn handle_websocket(
req: Request<hyper::Body>,
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
) -> Result<Response<hyper::Body>> {
let (response, ws_fut_opt) = hyper_ws_listener::create_ws(req)
.map_err(|e| anyhow::anyhow!("WebSocket upgrade failed: {}", e))?;
@@ -456,6 +485,7 @@ impl ApiHandler {
return;
}
};
metrics_store.increment_ws();
info!("WebSocket /ws/db connected");
let (mut tx, mut rx) = ws_stream.split();
@@ -472,10 +502,18 @@ impl ApiHandler {
let mut state_rx = state_manager.subscribe();
let ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
tokio::pin!(ping_interval);
let mut last_client_activity = Instant::now();
const INACTIVITY_TIMEOUT_SECS: u64 = 300; // 5 minutes
loop {
tokio::select! {
_ = ping_interval.tick() => {
// Check inactivity timeout
if last_client_activity.elapsed().as_secs() >= INACTIVITY_TIMEOUT_SECS {
info!("WebSocket client inactive for {}s, closing", INACTIVITY_TIMEOUT_SECS);
let _ = tx.send(Message::Close(None)).await;
break;
}
if tx.send(Message::Ping(vec![])).await.is_err() {
debug!("Failed to send ping, connection likely closed");
break;
@@ -505,12 +543,23 @@ impl ApiHandler {
match msg {
Some(Ok(Message::Close(_))) => break,
Some(Ok(Message::Pong(_))) => {
last_client_activity = Instant::now();
debug!("Received pong");
}
Some(Ok(Message::Ping(data))) => {
last_client_activity = Instant::now();
let _ = tx.send(Message::Pong(data)).await;
}
Some(Ok(_)) => {}
Some(Ok(Message::Text(text))) => {
last_client_activity = Instant::now();
// Handle JSON ping from frontend
if text.contains("\"type\":\"ping\"") || text.contains("\"type\": \"ping\"") {
let _ = tx.send(Message::Text(r#"{"type":"pong"}"#.to_string())).await;
}
}
Some(Ok(_)) => {
last_client_activity = Instant::now();
}
Some(Err(e)) => {
debug!("WebSocket stream error: {}", e);
break;
@@ -520,6 +569,7 @@ impl ApiHandler {
}
}
}
metrics_store.decrement_ws();
info!("WebSocket /ws/db disconnected");
});
}
@@ -556,3 +606,139 @@ fn sanitize_html(s: &str) -> String {
.replace('"', "&quot;")
.replace('\'', "&#x27;")
}
impl ApiHandler {
/// DWN health endpoint — returns store stats.
async fn handle_dwn_health(config: &Config) -> Result<Response<hyper::Body>> {
match DwnStore::new(&config.data_dir).await {
Ok(store) => {
let stats = store.stats().await.unwrap_or(crate::network::dwn_store::StoreStats {
message_count: 0,
protocol_count: 0,
total_bytes: 0,
});
let body = serde_json::json!({
"status": "ok",
"message_count": stats.message_count,
"protocol_count": stats.protocol_count,
"total_bytes": stats.total_bytes,
});
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(hyper::Body::from(body.to_string()))
.unwrap())
}
Err(_) => Ok(Response::builder()
.status(StatusCode::SERVICE_UNAVAILABLE)
.header("Content-Type", "application/json")
.body(hyper::Body::from(r#"{"status":"unavailable"}"#))
.unwrap()),
}
}
/// DWN message processing endpoint — handles RecordsWrite, RecordsQuery, RecordsRead, RecordsDelete.
async fn handle_dwn_message(
body: hyper::body::Bytes,
config: &Config,
) -> Result<Response<hyper::Body>> {
let request: serde_json::Value = match serde_json::from_slice(&body) {
Ok(v) => v,
Err(e) => {
let err = serde_json::json!({"error": format!("Invalid JSON: {}", e)});
return Ok(Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("Content-Type", "application/json")
.body(hyper::Body::from(err.to_string()))
.unwrap());
}
};
let interface = request["message"]["descriptor"]["interface"]
.as_str()
.unwrap_or("");
let method = request["message"]["descriptor"]["method"]
.as_str()
.unwrap_or("");
let store = DwnStore::new(&config.data_dir).await?;
let result = match (interface, method) {
("Records", "Write") => {
let author = request["message"]["author"].as_str().unwrap_or("unknown");
let protocol = request["message"]["descriptor"]["protocol"].as_str();
let schema = request["message"]["descriptor"]["schema"].as_str();
let data_format = request["message"]["descriptor"]["dataFormat"].as_str();
let data = request["message"].get("data").cloned();
match store.write_message(author, protocol, schema, data_format, data).await {
Ok(msg) => serde_json::json!({"status": {"code": 202}, "entry": msg}),
Err(e) => serde_json::json!({"status": {"code": 500, "detail": e.to_string()}}),
}
}
("Records", "Query") => {
let query = crate::network::dwn_store::MessageQuery {
protocol: request["message"]["descriptor"]["filter"]["protocol"]
.as_str()
.map(|s| s.to_string()),
schema: request["message"]["descriptor"]["filter"]["schema"]
.as_str()
.map(|s| s.to_string()),
author: request["message"]["descriptor"]["filter"]["author"]
.as_str()
.map(|s| s.to_string()),
date_from: request["message"]["descriptor"]["filter"]["dateFrom"]
.as_str()
.map(|s| s.to_string()),
date_to: request["message"]["descriptor"]["filter"]["dateTo"]
.as_str()
.map(|s| s.to_string()),
limit: request["message"]["descriptor"]["filter"]["limit"]
.as_u64()
.map(|n| n as usize),
};
match store.query_messages(&query).await {
Ok(messages) => serde_json::json!({"status": {"code": 200}, "entries": messages}),
Err(e) => serde_json::json!({"status": {"code": 500, "detail": e.to_string()}}),
}
}
("Records", "Read") => {
let record_id = request["message"]["descriptor"]["recordId"]
.as_str()
.unwrap_or("");
match store.read_message(record_id).await {
Ok(Some(msg)) => serde_json::json!({"status": {"code": 200}, "entry": msg}),
Ok(None) => serde_json::json!({"status": {"code": 404, "detail": "Record not found"}}),
Err(e) => serde_json::json!({"status": {"code": 500, "detail": e.to_string()}}),
}
}
("Records", "Delete") => {
let record_id = request["message"]["descriptor"]["recordId"]
.as_str()
.unwrap_or("");
match store.delete_message(record_id).await {
Ok(true) => serde_json::json!({"status": {"code": 200}}),
Ok(false) => serde_json::json!({"status": {"code": 404, "detail": "Record not found"}}),
Err(e) => serde_json::json!({"status": {"code": 500, "detail": e.to_string()}}),
}
}
_ => {
serde_json::json!({"status": {"code": 400, "detail": format!("Unknown method: {}.{}", interface, method)}})
}
};
let status_code = result["status"]["code"].as_u64().unwrap_or(200);
let http_status = match status_code {
202 => StatusCode::ACCEPTED,
400 => StatusCode::BAD_REQUEST,
404 => StatusCode::NOT_FOUND,
500 => StatusCode::INTERNAL_SERVER_ERROR,
_ => StatusCode::OK,
};
Ok(Response::builder()
.status(http_status)
.header("Content-Type", "application/json")
.body(hyper::Body::from(result.to_string()))
.unwrap())
}
}