hydracache_server/
admin_http.rs

1use std::sync::{Arc, Mutex};
2
3use axum::extract::State;
4use axum::http::header::CONTENT_TYPE;
5use axum::http::{HeaderMap, StatusCode};
6use axum::response::{IntoResponse, Response};
7use axum::routing::{get, post};
8use axum::{Extension, Json, Router};
9use hydracache_actuator_axum::HydraCacheActuator;
10use hydracache_client_transport_axum::{
11    ClientSurfaceDiagnosticReset, HYDRACACHE_ADMIN_HEADER, HYDRACACHE_CLIENT_ID_HEADER,
12    HYDRACACHE_TENANT_HEADER,
13};
14use serde::Serialize;
15use thiserror::Error;
16
17use crate::bootstrap::{ServerAdminActionError, ServerRuntime};
18use crate::cluster_status::RaftCompactionError;
19use crate::hc2::Hc2ClientPlaneService;
20use crate::services::DrainOutcome;
21use hydracache_observability::PrometheusExporter;
22
23/// Liveness path used by Kubernetes probes.
24pub const ADMIN_HEALTHZ_PATH: &str = "/healthz";
25/// Readiness path used by Kubernetes probes.
26pub const ADMIN_READYZ_PATH: &str = "/readyz";
27/// Prometheus metrics path on the internal admin surface.
28pub const ADMIN_METRICS_PATH: &str = "/metrics";
29/// Read-only Management Center console path on the internal admin surface.
30pub const ADMIN_CONSOLE_PATH: &str = "/console";
31/// Read-only cluster overview path on the internal admin surface.
32pub const ADMIN_CLUSTER_OVERVIEW_PATH: &str = "/cluster/overview";
33/// Read-only per-cache actuator path on the internal admin surface.
34pub const ADMIN_ACTUATOR_PATH: &str = "/actuator/hydracache";
35/// Operator status path.
36pub const ADMIN_STATUS_PATH: &str = "/admin/status";
37/// Operator drain action path.
38pub const ADMIN_DRAIN_PATH: &str = "/admin/drain";
39/// Operator reshard action path.
40pub const ADMIN_RESHARD_PATH: &str = "/admin/reshard";
41/// Operator backup action path.
42pub const ADMIN_BACKUP_PATH: &str = "/admin/backup";
43/// Explicit, off-by-default disk-backed Raft compaction control and status path.
44pub const ADMIN_RAFT_COMPACTION_PATH: &str = "/admin/raft/compaction";
45/// Off-by-default, local-only destructive diagnostic reset path.
46pub const ADMIN_DIAGNOSTIC_RESET_PATH: &str = "/admin/diagnostics/reset";
47
48/// Shared runtime state for the admin HTTP surface.
49pub type SharedServerRuntime = Arc<Mutex<ServerRuntime>>;
50
51/// Axum route owner for the internal admin/operator surface.
52#[derive(Debug, Clone)]
53pub struct AdminHttpSurface {
54    runtime: SharedServerRuntime,
55    hc2_metrics: Option<Hc2ClientPlaneService>,
56}
57
58impl AdminHttpSurface {
59    /// Create an admin surface from a server runtime.
60    pub fn new(runtime: ServerRuntime) -> Self {
61        Self {
62            runtime: Arc::new(Mutex::new(runtime)),
63            hc2_metrics: None,
64        }
65    }
66
67    /// Create an admin surface from shared runtime state.
68    pub fn from_shared(runtime: SharedServerRuntime) -> Self {
69        Self {
70            runtime,
71            hc2_metrics: None,
72        }
73    }
74
75    /// Attach the selected production HC/2 listener to the internal metrics
76    /// surface. This does not expose metrics on either public client port.
77    pub fn with_hc2_metrics(mut self, service: Hc2ClientPlaneService) -> Self {
78        self.hc2_metrics = Some(service);
79        self
80    }
81
82    /// Return shared runtime state for tests and embedding code.
83    pub fn runtime(&self) -> SharedServerRuntime {
84        Arc::clone(&self.runtime)
85    }
86
87    /// Return the axum router for `/healthz`, `/readyz`, and `/admin/*`.
88    pub fn routes(&self) -> Router {
89        let actuator_registry = self
90            .runtime
91            .lock()
92            .expect("server runtime mutex")
93            .metrics_registry();
94        let routes = Router::new()
95            .route(ADMIN_HEALTHZ_PATH, get(healthz))
96            .route(ADMIN_READYZ_PATH, get(readyz))
97            .route(ADMIN_METRICS_PATH, get(metrics))
98            .route(ADMIN_CONSOLE_PATH, get(console_index))
99            .route("/console/", get(console_index))
100            .route("/console/index.html", get(console_index))
101            .route("/console/app.js", get(console_app))
102            .route("/console/style.css", get(console_style))
103            .route(ADMIN_CLUSTER_OVERVIEW_PATH, get(cluster_overview))
104            .route(ADMIN_STATUS_PATH, get(admin_status))
105            .route(ADMIN_DRAIN_PATH, get(admin_drain).post(admin_drain))
106            .route(ADMIN_RESHARD_PATH, post(admin_reshard))
107            .route(ADMIN_BACKUP_PATH, post(admin_backup))
108            .route(ADMIN_DIAGNOSTIC_RESET_PATH, post(admin_diagnostic_reset))
109            .route(
110                ADMIN_RAFT_COMPACTION_PATH,
111                get(admin_raft_compaction_status).post(admin_raft_compaction),
112            )
113            .with_state(Arc::clone(&self.runtime))
114            .nest(
115                ADMIN_ACTUATOR_PATH,
116                HydraCacheActuator::routes_for(actuator_registry),
117            );
118        if let Some(service) = self.hc2_metrics.clone() {
119            routes.layer(Extension(service))
120        } else {
121            routes
122        }
123    }
124}
125
126async fn healthz(State(runtime): State<SharedServerRuntime>) -> Response {
127    let health = runtime.lock().expect("server runtime mutex").health();
128    (StatusCode::OK, Json(health)).into_response()
129}
130
131async fn readyz(State(runtime): State<SharedServerRuntime>) -> Response {
132    let ready = runtime.lock().expect("server runtime mutex").ready();
133    let status = if ready.ready {
134        StatusCode::OK
135    } else {
136        StatusCode::SERVICE_UNAVAILABLE
137    };
138    (status, Json(ready)).into_response()
139}
140
141async fn metrics(
142    State(runtime): State<SharedServerRuntime>,
143    hc2: Option<Extension<Hc2ClientPlaneService>>,
144) -> Response {
145    let registry = runtime
146        .lock()
147        .expect("server runtime mutex")
148        .metrics_registry();
149    let mut text = PrometheusExporter::new(registry).render().await;
150    if let Some(Extension(service)) = hc2 {
151        text.push_str(&service.prometheus_metrics());
152    }
153    ([(CONTENT_TYPE, "text/plain; version=0.0.4")], text).into_response()
154}
155
156async fn console_index() -> Response {
157    (
158        [(CONTENT_TYPE, "text/html; charset=utf-8")],
159        include_str!("../console/index.html"),
160    )
161        .into_response()
162}
163
164async fn console_app() -> Response {
165    (
166        [(CONTENT_TYPE, "text/javascript; charset=utf-8")],
167        include_str!("../console/app.js"),
168    )
169        .into_response()
170}
171
172async fn console_style() -> Response {
173    (
174        [(CONTENT_TYPE, "text/css; charset=utf-8")],
175        include_str!("../console/style.css"),
176    )
177        .into_response()
178}
179
180async fn cluster_overview(State(runtime): State<SharedServerRuntime>) -> Response {
181    let overview = runtime
182        .lock()
183        .expect("server runtime mutex")
184        .cluster_overview();
185    (StatusCode::OK, Json(overview)).into_response()
186}
187
188async fn admin_status(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
189    if let Err(error) = require_admin(&headers) {
190        return error.into_response();
191    }
192    let status = runtime.lock().expect("server runtime mutex").admin_status();
193    (StatusCode::OK, Json(status)).into_response()
194}
195
196async fn admin_drain(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
197    if let Err(error) = require_admin(&headers) {
198        return error.into_response();
199    }
200    let drain = runtime
201        .lock()
202        .expect("server runtime mutex")
203        .request_admin_drain();
204    (
205        StatusCode::OK,
206        Json(AdminDrainReply {
207            action: "drain",
208            outcome: "accepted",
209            drain,
210        }),
211    )
212        .into_response()
213}
214
215async fn admin_reshard(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
216    if let Err(error) = require_admin(&headers) {
217        return error.into_response();
218    }
219    runtime
220        .lock()
221        .expect("server runtime mutex")
222        .request_reshard()
223        .map(|action| (StatusCode::OK, Json(action)).into_response())
224        .unwrap_or_else(|error| AdminHttpError::from(error).into_response())
225}
226
227async fn admin_backup(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
228    if let Err(error) = require_admin(&headers) {
229        return error.into_response();
230    }
231    runtime
232        .lock()
233        .expect("server runtime mutex")
234        .request_backup()
235        .map(|action| {
236            (
237                StatusCode::ACCEPTED,
238                Json(AdminBackupRequestAcceptance {
239                    action: action.action,
240                    outcome: action.outcome,
241                    detail: action.detail,
242                    authority: "request_only",
243                    durable_artifact_created: false,
244                    restore_point_available: false,
245                }),
246            )
247                .into_response()
248        })
249        .unwrap_or_else(|error| AdminHttpError::from(error).into_response())
250}
251
252async fn admin_diagnostic_reset(
253    State(runtime): State<SharedServerRuntime>,
254    hc2: Option<Extension<Hc2ClientPlaneService>>,
255    headers: HeaderMap,
256) -> Response {
257    if let Err(error) = require_admin(&headers) {
258        return error.into_response();
259    }
260    let (cache, client_state) = {
261        let runtime = runtime.lock().expect("server runtime mutex");
262        if !runtime.diagnostic_reset_enabled() {
263            return AdminHttpError::DiagnosticResetDisabled.into_response();
264        }
265        runtime.diagnostic_reset_targets()
266    };
267    if client_state
268        .as_ref()
269        .is_some_and(|state| state.active_subscriptions() != 0)
270        || hc2.as_ref().is_some_and(|Extension(service)| {
271            let accounting = service.accounting();
272            accounting.active_connections != 0
273                || accounting.pending_invocations != 0
274                || accounting.active_subscriptions != 0
275                || accounting.active_sessions != 0
276        })
277    {
278        return AdminHttpError::DiagnosticResetBusy.into_response();
279    }
280
281    let embedded_before = cache.diagnostics().await.estimated_entries;
282    if let Err(error) = cache.flush().await {
283        return AdminHttpError::DiagnosticResetFailed(error.to_string()).into_response();
284    }
285    let client = match client_state
286        .map(|state| state.reset_retained_state_for_diagnostics())
287        .transpose()
288    {
289        Ok(reset) => reset,
290        Err(error) => {
291            return AdminHttpError::DiagnosticResetFailed(error.to_string()).into_response();
292        }
293    };
294    let embedded_after = cache.diagnostics().await.estimated_entries;
295    let client_is_zero = client.as_ref().is_none_or(|reset| {
296        reset.after.store_entries == 0
297            && reset.after.idempotency_outcomes == 0
298            && reset.after.conditional.records == 0
299            && reset.after.conditional.locks == 0
300            && reset.after.conditional.session_heartbeats == 0
301    });
302    if embedded_after != 0 || !client_is_zero {
303        return AdminHttpError::DiagnosticResetFailed(
304            "owner counts remained non-zero after reset".to_owned(),
305        )
306        .into_response();
307    }
308
309    (
310        StatusCode::OK,
311        Json(AdminDiagnosticResetReply {
312            action: "diagnostic_reset",
313            outcome: "completed",
314            embedded_before,
315            embedded_after,
316            client,
317        }),
318    )
319        .into_response()
320}
321
322async fn admin_raft_compaction_status(
323    State(runtime): State<SharedServerRuntime>,
324    headers: HeaderMap,
325) -> Response {
326    if let Err(error) = require_admin(&headers) {
327        return error.into_response();
328    }
329    runtime
330        .lock()
331        .expect("server runtime mutex")
332        .raft_compaction_status()
333        .map(|status| (StatusCode::OK, Json(status)).into_response())
334        .unwrap_or_else(|error| {
335            AdminHttpError::from(ServerAdminActionError::from(error)).into_response()
336        })
337}
338
339async fn admin_raft_compaction(
340    State(runtime): State<SharedServerRuntime>,
341    headers: HeaderMap,
342) -> Response {
343    if let Err(error) = require_admin(&headers) {
344        return error.into_response();
345    }
346    runtime
347        .lock()
348        .expect("server runtime mutex")
349        .request_raft_compaction()
350        .map(|status| (StatusCode::OK, Json(status)).into_response())
351        .unwrap_or_else(|error| AdminHttpError::from(error).into_response())
352}
353
354/// Honest response boundary for the currently request-only backup admin seam.
355///
356/// A successful HTTP response confirms only that configuration and runtime
357/// preconditions accepted the request. The daemon does not yet own a live
358/// value-plane backup source, a durable object-store writer, or restore-point
359/// authority, so neither boolean may be inferred from `outcome = "accepted"`.
360#[derive(Debug, Serialize)]
361struct AdminBackupRequestAcceptance {
362    action: &'static str,
363    outcome: &'static str,
364    detail: String,
365    authority: &'static str,
366    durable_artifact_created: bool,
367    restore_point_available: bool,
368}
369
370fn require_admin(headers: &HeaderMap) -> Result<(), AdminHttpError> {
371    let has_identity = header_value(headers, HYDRACACHE_CLIENT_ID_HEADER).is_some()
372        && header_value(headers, HYDRACACHE_TENANT_HEADER).is_some();
373    if !has_identity {
374        return Err(AdminHttpError::Unauthenticated);
375    }
376    let admin = headers
377        .get(HYDRACACHE_ADMIN_HEADER)
378        .and_then(|value| value.to_str().ok())
379        .is_some_and(|value| matches!(value, "true" | "1"));
380    if !admin {
381        return Err(AdminHttpError::Unauthorized);
382    }
383    Ok(())
384}
385
386fn header_value(headers: &HeaderMap, name: &'static str) -> Option<String> {
387    headers
388        .get(name)
389        .and_then(|value| value.to_str().ok())
390        .filter(|value| !value.trim().is_empty())
391        .map(ToOwned::to_owned)
392}
393
394/// Admin drain response.
395#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
396pub struct AdminDrainReply {
397    /// Stable action name.
398    pub action: &'static str,
399    /// Stable outcome string.
400    pub outcome: &'static str,
401    /// Drain result from the runtime.
402    pub drain: DrainOutcome,
403}
404
405/// Verified owner counts returned by the local diagnostic reset.
406#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
407pub struct AdminDiagnosticResetReply {
408    /// Stable destructive action name.
409    pub action: &'static str,
410    /// Stable successful outcome.
411    pub outcome: &'static str,
412    /// Embedded cache entries observed before cleanup.
413    pub embedded_before: u64,
414    /// Embedded cache entries observed after cleanup.
415    pub embedded_after: u64,
416    /// Shared HC/1, HC/2 and RESP dispatch owner counts, when configured.
417    pub client: Option<ClientSurfaceDiagnosticReset>,
418}
419
420/// JSON reply for rejected admin calls.
421#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
422pub struct AdminErrorReply {
423    /// Stable outcome string.
424    pub outcome: &'static str,
425    /// Redacted detail safe for operator Conditions.
426    pub detail: String,
427}
428
429impl AdminErrorReply {
430    fn rejected(detail: impl Into<String>) -> Self {
431        Self {
432            outcome: "rejected",
433            detail: detail.into(),
434        }
435    }
436}
437
438/// Admin HTTP boundary errors.
439#[derive(Debug, Clone, PartialEq, Eq, Error)]
440pub enum AdminHttpError {
441    /// Admin identity was absent or incomplete.
442    #[error("admin identity is required")]
443    Unauthenticated,
444    /// Caller identity is not privileged for admin actions.
445    #[error("admin privileges are required")]
446    Unauthorized,
447    /// Diagnostic reset is not explicitly enabled.
448    #[error("diagnostic reset is disabled")]
449    DiagnosticResetDisabled,
450    /// Active client resources make destructive reset unsafe.
451    #[error("diagnostic reset requires a quiescent client surface")]
452    DiagnosticResetBusy,
453    /// A reset owner failed cleanup or its zero assertion.
454    #[error("diagnostic reset failed: {0}")]
455    DiagnosticResetFailed(String),
456    /// Runtime refused the requested admin action.
457    #[error("{0}")]
458    Action(#[from] ServerAdminActionError),
459}
460
461impl IntoResponse for AdminHttpError {
462    fn into_response(self) -> Response {
463        let status = match self {
464            Self::Unauthenticated => StatusCode::UNAUTHORIZED,
465            Self::Unauthorized => StatusCode::FORBIDDEN,
466            Self::DiagnosticResetDisabled => StatusCode::NOT_FOUND,
467            Self::DiagnosticResetBusy => StatusCode::CONFLICT,
468            Self::DiagnosticResetFailed(_) => StatusCode::INTERNAL_SERVER_ERROR,
469            Self::Action(ServerAdminActionError::NotReady(_)) => StatusCode::SERVICE_UNAVAILABLE,
470            Self::Action(
471                ServerAdminActionError::RequiresMember(_) | ServerAdminActionError::BackupDisabled,
472            ) => StatusCode::CONFLICT,
473            Self::Action(ServerAdminActionError::RaftCompaction(
474                RaftCompactionError::Disabled | RaftCompactionError::Unavailable,
475            )) => StatusCode::CONFLICT,
476            Self::Action(ServerAdminActionError::RaftCompaction(RaftCompactionError::Runtime(
477                _,
478            ))) => StatusCode::INTERNAL_SERVER_ERROR,
479        };
480        (status, Json(AdminErrorReply::rejected(self.to_string()))).into_response()
481    }
482}
hydracache_server/admin_http.rs

hydracache_server/
admin_http.rs
