diff --git a/crates/consensus/primary-metrics/src/consensus.rs b/crates/consensus/primary-metrics/src/consensus.rs index 94f35687..edaecb3d 100644 --- a/crates/consensus/primary-metrics/src/consensus.rs +++ b/crates/consensus/primary-metrics/src/consensus.rs @@ -53,6 +53,20 @@ pub struct ConsensusMetrics { pub leader_commits: IntCounterVec, /// number of bad nodes in the committee pub num_of_bad_nodes: IntGauge, + /// This node's header vote requests rejected by a peer, labeled by the rejecting validator + /// (`authority`) and `reason` (`too_old` | `too_old_skipped` | `epoch_mismatch` | + /// `epoch_mismatch_stale`). Node-local: high totals mean this node is falling behind and being + /// rejected by the committee. The `*_skipped` / `*_stale` variants are rejections that were + /// *not* counted toward peer demotion (a too-old rejection the cert store already covered + /// while the DAG progressed, or an epoch rejection from a peer that is itself stale) — tracked + /// under distinct reasons so the raw rejection rate stays visible. + pub vote_request_rejections: IntCounterVec, + /// Size of the committee for the current epoch. + pub committee_size: IntGauge, + /// Committed certificates per validator (by `authority`) — a direct per-validator liveness + /// signal: which validators' certificates actually land in committed sub-dags. Consistent + /// across nodes (every node commits the same sub-dags), unlike node-local rejection counts. + pub validator_participation: IntCounterVec, } impl ConsensusMetrics { @@ -126,6 +140,23 @@ impl ConsensusMetrics { "The number of bad nodes in the new leader schedule", registry )?, + vote_request_rejections: register_int_counter_vec_with_registry!( + "vote_request_rejections", + "This node's header vote requests rejected by a peer, by rejecting validator (authority) and reason", + &["authority", "reason"], + registry + )?, + committee_size: register_int_gauge_with_registry!( + "committee_size", + "Size of the committee for the current epoch", + registry + )?, + validator_participation: register_int_counter_vec_with_registry!( + "validator_participation", + "Committed certificates per validator (by authority) - a per-validator liveness signal", + &["authority"], + registry + )?, }) } } diff --git a/crates/consensus/primary/src/certifier.rs b/crates/consensus/primary/src/certifier.rs index 9aee1c42..4995583c 100644 --- a/crates/consensus/primary/src/certifier.rs +++ b/crates/consensus/primary/src/certifier.rs @@ -403,6 +403,14 @@ impl Certifier { header_round, limit_round, cert_store_round, committed_round, "ignoring too-old rejection: cert store covers limit round and DAG still progressing (transient proposer lag)" ); + // Count the rejection under a distinct reason even though it isn't held + // against the peer for demotion — otherwise the metric would show zero + // while a node continuously trips the skip path, masking real trouble. + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "too_old_skipped"]) + .inc(); return VoteErrorAction::Continue; } warn!( @@ -414,6 +422,11 @@ impl Certifier { ); } let outcome = self.vote_failures.record_too_old(peer_id.clone()); + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "too_old"]) + .inc(); warn!( target: "primary::certifier", auth=?self.authority_id, @@ -448,10 +461,22 @@ impl Certifier { peer_epoch, our_epoch, "ignoring epoch rejection from stale peer" ); + // Count it under a distinct reason even though a stale peer can't demote us — + // mirrors `too_old_skipped`, keeping the raw rejection rate visible. + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "epoch_mismatch_stale"]) + .inc(); return VoteErrorAction::Continue; } let outcome = self.vote_failures.record_epoch_mismatch(peer_id.clone()); + self.consensus_bus + .consensus_metrics() + .vote_request_rejections + .with_label_values(&[&peer_id.to_string(), "epoch_mismatch"]) + .inc(); warn!( target: "primary::certifier", auth=?self.authority_id, diff --git a/crates/consensus/primary/src/consensus/bullshark.rs b/crates/consensus/primary/src/consensus/bullshark.rs index f3a1d4ff..6417cb87 100644 --- a/crates/consensus/primary/src/consensus/bullshark.rs +++ b/crates/consensus/primary/src/consensus/bullshark.rs @@ -188,6 +188,19 @@ impl Bullshark { self.metrics.committed_certificates.report(total_committed_certificates); + // Per-validator participation: count each committed certificate by its author. Distinct + // from the leader-only series (`leader_election` / `leader_commits`), this shows which + // validators are actually contributing certificates each commit — a direct liveness + // signal, and consistent across nodes since every node commits the same sub-dags. + for sub_dag in &committed_sub_dags { + for certificate in &sub_dag.certificates { + self.metrics + .validator_participation + .with_label_values(&[&certificate.origin().to_string()]) + .inc(); + } + } + Ok((Outcome::Commit, committed_sub_dags)) } diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs index cd1e7c09..ab908db3 100644 --- a/crates/middleware/orchestrator/src/epoch_manager/core.rs +++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs @@ -97,14 +97,22 @@ where let mut node_task_manager = TaskManager::new(NODE_TASK_MANAGER); let node_task_spawner = node_task_manager.get_spawner(); - // Bind the liveness healthcheck before any boot-time cold work (crash reconcile + backlog + // Bind the healthcheck before any boot-time cold work (crash reconcile + backlog // migration). Both are synchronous and can be multi-minute on a large DB; binding the probe // first keeps it answering throughout (the migration is `spawn_blocking`'d so the runtime // stays free), so a short-deadline probe cannot blackout and restart-loop the node during - // recovery. NOTE: the probe is liveness-only by design (it does not report readiness), so - // it is green while these run. + // recovery. LIVENESS (any path but `/readyz`) stays green throughout — the process is up. + // READINESS (`/readyz`) reads the live node mode and reports `503` until the node is voting + // or observing, so it is correctly not-ready during this boot recovery. A bind failure + // (e.g. the port is already in use) is propagated rather than silently swallowed — the + // operator asked for this endpoint. if let Some(port) = self.builder.healthcheck { - let _ = HealthcheckServer::spawn(node_task_manager.get_spawner(), port).await; + HealthcheckServer::spawn( + node_task_manager.get_spawner(), + port, + self.consensus_bus.node_mode().subscribe(), + ) + .await?; } // Heal any crash-interrupted archive before serving, while consensus and execution have not @@ -495,9 +503,12 @@ where // This needs to be created early so required machinery for other tasks exists when needed. let mut worker = worker_node.new_worker().await?; worker.set_batch_tracker(self.consensus_bus.batch_tracker().clone()); - let current_epoch = primary.current_committee().await.epoch(); + let current_committee = primary.current_committee().await; + let current_epoch = current_committee.epoch(); - self.consensus_bus.consensus_metrics().current_epoch.set(current_epoch as i64); + let consensus_metrics = self.consensus_bus.consensus_metrics(); + consensus_metrics.current_epoch.set(current_epoch as i64); + consensus_metrics.committee_size.set(current_committee.size() as i64); // Produce a "dummy" epoch 0 EpochRecord if missing. // This will let us use simple code to find any epoch including 0 at startup. diff --git a/crates/middleware/orchestrator/src/types/health.rs b/crates/middleware/orchestrator/src/types/health.rs index 54c517c2..554b9336 100644 --- a/crates/middleware/orchestrator/src/types/health.rs +++ b/crates/middleware/orchestrator/src/types/health.rs @@ -1,52 +1,79 @@ -//! Simple TCP healthcheck endpoint for monitoring service availability. +//! TCP health + readiness endpoint for monitoring service availability. //! -//! Implements a minimal HTTP/1.1 server that responds with status 200 to all requests. -//! This is designed for integration with GCP load balancers and similar health monitoring systems. +//! A minimal HTTP/1.1 server exposing two probes: +//! - **liveness** (any path, e.g. `GET /`): always `200 OK` while the process is accepting +//! connections — the pre-existing behaviour. +//! - **readiness** (`GET /readyz`, alias `GET /ready`): `200 OK` only when the node is actually +//! participating — voting (`CvvActive`) or an operational observer — and `503 Service +//! Unavailable` while a validator is still catching up (`CvvInactive`). This lets a load balancer +//! / on-call distinguish "process up" from "actually voting in the current epoch". +//! +//! Designed for integration with GCP load balancers and similar health monitoring systems. use std::{io::ErrorKind, net::SocketAddr, time::Duration}; +use rayls_consensus_primary::NodeMode; use rayls_infrastructure_types::TaskSpawner; -use tokio::{io::AsyncWriteExt, net::TcpListener, time::sleep}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::watch, + time::sleep, +}; use tracing::info; -/// Minimal HTTP health check responder for service monitoring. +/// Liveness / readiness HTTP responder for service monitoring. /// -/// Binds to a TCP port and responds with HTTP 200 to any connection. -/// Uses raw TCP sockets for minimal overhead and dependencies. +/// Binds a TCP port and answers: +/// - liveness (any path): `200 OK` — the process is up. +/// - readiness (`/readyz`): `200` when voting/observing, `503` while catching up. /// /// # Security Considerations /// -/// This endpoint accepts connections from any source and responds unconditionally. -/// -/// Node operators must ensure the endpoint is protected by a firewall. -/// This service is off by default, but can be enabled through the CLI node command. -/// Each connection is handled synchronously in the main accept loop. -/// No connection limits or rate limiting are implemented. -/// Connections are immediately closed after sending response. +/// This endpoint accepts connections from any source. Node operators must protect it with a +/// firewall. It is off by default and enabled via the CLI node command. Each connection is +/// handled in the accept loop and closed immediately after the response. /// -/// To enable on node startup, use `rayls-network node --enable-healthcheck`. +/// To enable on node startup, use `rayls-network node --healthcheck `. /// See `rayls-network-cli::node` for more info. #[derive(Debug)] pub(crate) struct HealthcheckServer; +const LIVE_200: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; +const READY_200: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nREADY"; +const NOT_READY_503: &[u8] = + b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 9\r\n\r\nNOT_READY"; + +/// True if the HTTP request line targets the readiness path (`/readyz` or `/ready`). +fn request_targets_readiness(req: &[u8]) -> bool { + // Request line looks like `GET /readyz HTTP/1.1`; match the path token, ignoring the + // method and version. + let line = req.split(|&b| b == b'\r' || b == b'\n').next().unwrap_or(req); + let mut tokens = line.split(|&b| b == b' ').filter(|t| !t.is_empty()); + let _method = tokens.next(); + matches!(tokens.next(), Some(path) if path == b"/readyz" || path == b"/ready") +} + +/// Readiness response for a node mode: ready (`200`) when voting (`CvvActive`) or an operational +/// observer; not ready (`503`) while a validator is still catching up (`CvvInactive`). +fn readiness_response(mode: NodeMode) -> &'static [u8] { + if mode.is_active_cvv() || mode.is_observer() { + READY_200 + } else { + NOT_READY_503 + } +} + impl HealthcheckServer { - /// Spawns the health check server task and returns the bound address. - /// - /// Binds to port specified by `HEALTHCHECK_PORT` environment variable, - /// or lets the OS assign a port if unset or set to 0. - /// - /// # Network Binding - /// - /// Binds to 0.0.0.0 (all interfaces) to allow external health checkers. - /// This makes the service accessible on all network interfaces including public IPs. + /// Spawn the health/readiness server, returning the bound address. /// - /// # Protocol - /// - /// Implements minimal HTTP/1.1 with a fixed response: - /// - Status: 200 OK - /// - Body: "OK" (2 bytes) - /// - No request parsing or validation - /// - No custom headers to avoid information disclosure - pub(crate) async fn spawn(task_spawner: TaskSpawner, port: u16) -> eyre::Result { + /// Binds `0.0.0.0:port` (all interfaces, for external health checkers). `node_mode` is a + /// watch receiver of the node's consensus mode (from the consensus bus); the readiness probe + /// reads its current value per request. + pub(crate) async fn spawn( + task_spawner: TaskSpawner, + port: u16, + node_mode: watch::Receiver, + ) -> eyre::Result { // IMPORTANT: use firewall to protect this endpoint let addr: SocketAddr = ([0, 0, 0, 0], port).into(); let listener = TcpListener::bind(addr).await?; @@ -54,20 +81,46 @@ impl HealthcheckServer { info!(target: "epoch-manager", ?listen_on, "healthcheck listening"); task_spawner.spawn_task("healthcheck", async move { - // minimal valid HTTP - let response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; - - let mut backoff = Duration::from_millis(100); + let initial_backoff = Duration::from_millis(100); + let mut backoff = initial_backoff; let max_backoff = Duration::from_secs(5); loop { match listener.accept().await { Ok((mut socket, _)) => { - // write response, ignore errors (client disconnect, etc.) - // then drop connection - if let Err(e) = socket.write_all(response).await { - tracing::error!(target: "healthcheck", ?e, "error writing healthcheck response"); - } + // Reset backoff after a healthy accept so a past transient error doesn't + // keep the delay elevated. + backoff = initial_backoff; + + // Handle each connection in its own task: the request read is bounded by a + // short timeout, but a client that connects and sends nothing would still + // hold the accept loop for the full timeout, delaying every other probe. + // A per-connection task keeps the accept loop free. `node_mode` is a cheap + // watch receiver, cloned per connection. + let node_mode = node_mode.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 256]; + let wants_readiness = matches!( + tokio::time::timeout( + Duration::from_millis(200), + socket.read(&mut buf), + ) + .await, + Ok(Ok(n)) if n > 0 && request_targets_readiness(&buf[..n]) + ); + + let response = if wants_readiness { + readiness_response(*node_mode.borrow()) + } else { + // Any other path (incl. `/`) is a liveness probe: process is up. + LIVE_200 + }; + + // write response, ignore errors (client disconnect, etc.), then drop. + if let Err(e) = socket.write_all(response).await { + tracing::error!(target: "healthcheck", ?e, "error writing healthcheck response"); + } + }); } Err(ref e) if matches!( e.kind(), @@ -76,15 +129,14 @@ impl HealthcheckServer { | ErrorKind::ConnectionAborted | ErrorKind::ConnectionReset | ErrorKind::Other - )=> { + ) => { // transient errors that can be ignored tracing::warn!(target: "healthcheck", ?e, "transient error accepting healthcheck connection"); sleep(backoff).await; backoff = (backoff * 2).min(max_backoff); - } Err(e) => { - // unexpected errors should be logged and break the loop to avoid spinning on fatal errors + // unexpected errors: log and break to avoid spinning on fatal errors tracing::error!(target: "healthcheck", ?e, "error accepting healthcheck connection"); break; } @@ -98,60 +150,52 @@ impl HealthcheckServer { #[cfg(test)] mod tests { + use super::*; use rayls_infrastructure_types::{get_available_tcp_port, TaskManager}; - use std::time::Duration; - use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, - net::TcpStream, - }; + use tokio::net::TcpStream; + + async fn request(addr: SocketAddr, path: &str) -> String { + let mut stream = TcpStream::connect(addr).await.unwrap(); + stream.write_all(format!("GET {path} HTTP/1.1\r\n\r\n").as_bytes()).await.unwrap(); + let mut response = vec![0u8; 1024]; + let n = stream.read(&mut response).await.unwrap(); + String::from_utf8_lossy(&response[..n]).into_owned() + } + + #[test] + fn readiness_path_is_matched() { + assert!(request_targets_readiness(b"GET /readyz HTTP/1.1\r\n\r\n")); + assert!(request_targets_readiness(b"GET /ready HTTP/1.1\r\n\r\n")); + assert!(!request_targets_readiness(b"GET / HTTP/1.1\r\n\r\n")); + assert!(!request_targets_readiness(b"GET /healthz HTTP/1.1\r\n\r\n")); + } - use crate::types::HealthcheckServer; + #[test] + fn readiness_response_tracks_mode() { + assert_eq!(readiness_response(NodeMode::CvvActive), READY_200); + assert_eq!(readiness_response(NodeMode::Observer), READY_200); + assert_eq!(readiness_response(NodeMode::CvvInactive), NOT_READY_503); + } #[tokio::test] - async fn test_tcp_healthcheck() -> eyre::Result<()> { + async fn liveness_always_ok_and_readiness_follows_mode() -> eyre::Result<()> { let task_manager = TaskManager::default(); let task_spawner = task_manager.get_spawner(); + let (tx, rx) = watch::channel(NodeMode::CvvInactive); let port = get_available_tcp_port("127.0.0.1").expect("tcp port assigned by host"); - // spawn server and get the bound address - let addr = HealthcheckServer::spawn(task_spawner.clone(), port).await?; - - // give server time to start listening + let addr = HealthcheckServer::spawn(task_spawner, port, rx).await?; tokio::time::sleep(Duration::from_millis(10)).await; - tokio::time::timeout(Duration::from_millis(500), async move { - // request healthcheck - let mut stream = TcpStream::connect(addr).await?; - - // send minimal HTTP request - stream.write_all(b"GET / HTTP/1.1\r\n\r\n").await?; - - // read response - let mut response = vec![0u8; 1024]; - let n = stream.read(&mut response).await?; - response.truncate(n); - let response_str = String::from_utf8_lossy(&response); - - // verify http status line - assert!( - response_str.starts_with("HTTP/1.1 200 OK"), - "Expected 200 OK, got: {}", - response_str - ); - - // verify body - assert!(response_str.ends_with("OK"), "Expected body 'OK', got: {}", response_str); - - // verify content-length header - assert!( - response_str.contains("Content-Length: 2"), - "Missing or incorrect Content-Length header" - ); - - Ok::<(), eyre::Error>(()) - }) - .await - .expect("response received")?; + // Liveness: always 200, even while catching up. + assert!(request(addr, "/").await.starts_with("HTTP/1.1 200 OK")); + + // Readiness: 503 while CvvInactive (catching up)... + assert!(request(addr, "/readyz").await.starts_with("HTTP/1.1 503")); + + // ...and 200 once the node is voting. + tx.send(NodeMode::CvvActive).unwrap(); + assert!(request(addr, "/readyz").await.starts_with("HTTP/1.1 200 OK")); Ok(()) } diff --git a/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml b/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml new file mode 100644 index 00000000..8c95269a --- /dev/null +++ b/etc/monitoring/grafana/provisioning/dashboards/dashboards.yml @@ -0,0 +1,13 @@ +# Grafana dashboard provisioning — auto-loads the JSON dashboards in this directory on start. +apiVersion: 1 + +providers: + - name: rayls + orgId: 1 + type: file + disableDeletion: false + updateIntervalSeconds: 30 + allowUiUpdates: true + options: + path: /etc/grafana/provisioning/dashboards + foldersFromFilesStructure: false diff --git a/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json b/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json new file mode 100644 index 00000000..73e20538 --- /dev/null +++ b/etc/monitoring/grafana/provisioning/dashboards/rayls-consensus.json @@ -0,0 +1,149 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "title": "Rayls — Consensus & Validator Health", + "uid": "rayls-consensus", + "tags": ["rayls", "consensus"], + "timezone": "browser", + "schemaVersion": 39, + "version": 1, + "time": { "from": "now-1h", "to": "now" }, + "refresh": "30s", + "templating": { + "list": [ + { + "name": "DS", + "label": "Prometheus", + "type": "datasource", + "query": "prometheus", + "current": {}, + "hide": 0, + "refresh": 1 + } + ] + }, + "panels": [ + { + "id": 1, + "title": "Current epoch", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 0, "y": 0 }, + "targets": [{ "expr": "current_epoch", "refId": "A" }] + }, + { + "id": 2, + "title": "Committee size", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 6, "y": 0 }, + "targets": [{ "expr": "committee_size", "refId": "A" }] + }, + { + "id": 3, + "title": "Bad nodes (leader schedule)", + "type": "stat", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 4, "w": 6, "x": 12, "y": 0 }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "red", "value": 1 } + ] + } + }, + "overrides": [] + }, + "targets": [{ "expr": "num_of_bad_nodes", "refId": "A" }] + }, + { + "id": 4, + "title": "Consensus rounds (proposer vs. committed)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 4 }, + "description": "Gap between current_round and last_committed_round is the commit lag.", + "targets": [ + { "expr": "current_round", "legendFormat": "proposer round", "refId": "A" }, + { "expr": "last_committed_round", "legendFormat": "committed round", "refId": "B" } + ] + }, + { + "id": 5, + "title": "Commit round latency (p95)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 4 }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "targets": [ + { + "expr": "histogram_quantile(0.95, sum(rate(consensus_commit_rounds_latency_bucket[5m])) by (le))", + "legendFormat": "p95", + "refId": "A" + } + ] + }, + { + "id": 6, + "title": "Leader election outcomes (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 12 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(leader_election[5m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + }, + { + "id": 7, + "title": "Vote-request rejections by validator (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 12 }, + "description": "This node's header vote requests rejected by a peer. Rising totals mean this node is falling behind.", + "targets": [ + { + "expr": "sum by (authority, reason) (rate(vote_request_rejections[5m]))", + "legendFormat": "{{authority}} — {{reason}}", + "refId": "A" + } + ] + }, + { + "id": 8, + "title": "Connected peers", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 20 }, + "targets": [ + { + "expr": "connected_peers_count", + "legendFormat": "{{kad_type}}", + "refId": "A" + } + ] + }, + { + "id": 9, + "title": "Validator participation (committed certs/s)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "${DS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 20 }, + "description": "Rate of committed certificates per validator — a direct per-validator liveness signal, consistent across nodes. A validator whose line drops to zero has stopped contributing certificates.", + "targets": [ + { + "expr": "sum by (authority) (rate(validator_participation[5m]))", + "legendFormat": "{{authority}}", + "refId": "A" + } + ] + } + ] +}