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::{Json, Router};
9use hydracache_actuator_axum::HydraCacheActuator;
10use hydracache_client_transport_axum::{
11    HYDRACACHE_ADMIN_HEADER, HYDRACACHE_CLIENT_ID_HEADER, HYDRACACHE_TENANT_HEADER,
12};
13use serde::Serialize;
14use thiserror::Error;
15
16use crate::bootstrap::{ServerAdminActionError, ServerRuntime};
17use crate::services::DrainOutcome;
18use hydracache_observability::PrometheusExporter;
19
20/// Liveness path used by Kubernetes probes.
21pub const ADMIN_HEALTHZ_PATH: &str = "/healthz";
22/// Readiness path used by Kubernetes probes.
23pub const ADMIN_READYZ_PATH: &str = "/readyz";
24/// Prometheus metrics path on the internal admin surface.
25pub const ADMIN_METRICS_PATH: &str = "/metrics";
26/// Read-only Management Center console path on the internal admin surface.
27pub const ADMIN_CONSOLE_PATH: &str = "/console";
28/// Read-only cluster overview path on the internal admin surface.
29pub const ADMIN_CLUSTER_OVERVIEW_PATH: &str = "/cluster/overview";
30/// Read-only per-cache actuator path on the internal admin surface.
31pub const ADMIN_ACTUATOR_PATH: &str = "/actuator/hydracache";
32/// Operator status path.
33pub const ADMIN_STATUS_PATH: &str = "/admin/status";
34/// Operator drain action path.
35pub const ADMIN_DRAIN_PATH: &str = "/admin/drain";
36/// Operator reshard action path.
37pub const ADMIN_RESHARD_PATH: &str = "/admin/reshard";
38/// Operator backup action path.
39pub const ADMIN_BACKUP_PATH: &str = "/admin/backup";
40
41/// Shared runtime state for the admin HTTP surface.
42pub type SharedServerRuntime = Arc<Mutex<ServerRuntime>>;
43
44/// Axum route owner for the internal admin/operator surface.
45#[derive(Debug, Clone)]
46pub struct AdminHttpSurface {
47    runtime: SharedServerRuntime,
48}
49
50impl AdminHttpSurface {
51    /// Create an admin surface from a server runtime.
52    pub fn new(runtime: ServerRuntime) -> Self {
53        Self {
54            runtime: Arc::new(Mutex::new(runtime)),
55        }
56    }
57
58    /// Create an admin surface from shared runtime state.
59    pub fn from_shared(runtime: SharedServerRuntime) -> Self {
60        Self { runtime }
61    }
62
63    /// Return shared runtime state for tests and embedding code.
64    pub fn runtime(&self) -> SharedServerRuntime {
65        Arc::clone(&self.runtime)
66    }
67
68    /// Return the axum router for `/healthz`, `/readyz`, and `/admin/*`.
69    pub fn routes(&self) -> Router {
70        let actuator_registry = self
71            .runtime
72            .lock()
73            .expect("server runtime mutex")
74            .metrics_registry();
75        Router::new()
76            .route(ADMIN_HEALTHZ_PATH, get(healthz))
77            .route(ADMIN_READYZ_PATH, get(readyz))
78            .route(ADMIN_METRICS_PATH, get(metrics))
79            .route(ADMIN_CONSOLE_PATH, get(console_index))
80            .route("/console/", get(console_index))
81            .route("/console/index.html", get(console_index))
82            .route("/console/app.js", get(console_app))
83            .route("/console/style.css", get(console_style))
84            .route(ADMIN_CLUSTER_OVERVIEW_PATH, get(cluster_overview))
85            .route(ADMIN_STATUS_PATH, get(admin_status))
86            .route(ADMIN_DRAIN_PATH, get(admin_drain).post(admin_drain))
87            .route(ADMIN_RESHARD_PATH, post(admin_reshard))
88            .route(ADMIN_BACKUP_PATH, post(admin_backup))
89            .with_state(Arc::clone(&self.runtime))
90            .nest(
91                ADMIN_ACTUATOR_PATH,
92                HydraCacheActuator::routes_for(actuator_registry),
93            )
94    }
95}
96
97async fn healthz(State(runtime): State<SharedServerRuntime>) -> Response {
98    let health = runtime.lock().expect("server runtime mutex").health();
99    (StatusCode::OK, Json(health)).into_response()
100}
101
102async fn readyz(State(runtime): State<SharedServerRuntime>) -> Response {
103    let ready = runtime.lock().expect("server runtime mutex").ready();
104    let status = if ready.ready {
105        StatusCode::OK
106    } else {
107        StatusCode::SERVICE_UNAVAILABLE
108    };
109    (status, Json(ready)).into_response()
110}
111
112async fn metrics(State(runtime): State<SharedServerRuntime>) -> Response {
113    let registry = runtime
114        .lock()
115        .expect("server runtime mutex")
116        .metrics_registry();
117    let text = PrometheusExporter::new(registry).render().await;
118    ([(CONTENT_TYPE, "text/plain; version=0.0.4")], text).into_response()
119}
120
121async fn console_index() -> Response {
122    (
123        [(CONTENT_TYPE, "text/html; charset=utf-8")],
124        include_str!("../console/index.html"),
125    )
126        .into_response()
127}
128
129async fn console_app() -> Response {
130    (
131        [(CONTENT_TYPE, "text/javascript; charset=utf-8")],
132        include_str!("../console/app.js"),
133    )
134        .into_response()
135}
136
137async fn console_style() -> Response {
138    (
139        [(CONTENT_TYPE, "text/css; charset=utf-8")],
140        include_str!("../console/style.css"),
141    )
142        .into_response()
143}
144
145async fn cluster_overview(State(runtime): State<SharedServerRuntime>) -> Response {
146    let overview = runtime
147        .lock()
148        .expect("server runtime mutex")
149        .cluster_overview();
150    (StatusCode::OK, Json(overview)).into_response()
151}
152
153async fn admin_status(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
154    if let Err(error) = require_admin(&headers) {
155        return error.into_response();
156    }
157    let status = runtime.lock().expect("server runtime mutex").admin_status();
158    (StatusCode::OK, Json(status)).into_response()
159}
160
161async fn admin_drain(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
162    if let Err(error) = require_admin(&headers) {
163        return error.into_response();
164    }
165    let drain = runtime
166        .lock()
167        .expect("server runtime mutex")
168        .request_admin_drain();
169    (
170        StatusCode::OK,
171        Json(AdminDrainReply {
172            action: "drain",
173            outcome: "accepted",
174            drain,
175        }),
176    )
177        .into_response()
178}
179
180async fn admin_reshard(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
181    if let Err(error) = require_admin(&headers) {
182        return error.into_response();
183    }
184    runtime
185        .lock()
186        .expect("server runtime mutex")
187        .request_reshard()
188        .map(|action| (StatusCode::OK, Json(action)).into_response())
189        .unwrap_or_else(|error| AdminHttpError::from(error).into_response())
190}
191
192async fn admin_backup(State(runtime): State<SharedServerRuntime>, headers: HeaderMap) -> Response {
193    if let Err(error) = require_admin(&headers) {
194        return error.into_response();
195    }
196    runtime
197        .lock()
198        .expect("server runtime mutex")
199        .request_backup()
200        .map(|action| (StatusCode::OK, Json(action)).into_response())
201        .unwrap_or_else(|error| AdminHttpError::from(error).into_response())
202}
203
204fn require_admin(headers: &HeaderMap) -> Result<(), AdminHttpError> {
205    let has_identity = header_value(headers, HYDRACACHE_CLIENT_ID_HEADER).is_some()
206        && header_value(headers, HYDRACACHE_TENANT_HEADER).is_some();
207    if !has_identity {
208        return Err(AdminHttpError::Unauthenticated);
209    }
210    let admin = headers
211        .get(HYDRACACHE_ADMIN_HEADER)
212        .and_then(|value| value.to_str().ok())
213        .is_some_and(|value| matches!(value, "true" | "1"));
214    if !admin {
215        return Err(AdminHttpError::Unauthorized);
216    }
217    Ok(())
218}
219
220fn header_value(headers: &HeaderMap, name: &'static str) -> Option<String> {
221    headers
222        .get(name)
223        .and_then(|value| value.to_str().ok())
224        .filter(|value| !value.trim().is_empty())
225        .map(ToOwned::to_owned)
226}
227
228/// Admin drain response.
229#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
230pub struct AdminDrainReply {
231    /// Stable action name.
232    pub action: &'static str,
233    /// Stable outcome string.
234    pub outcome: &'static str,
235    /// Drain result from the runtime.
236    pub drain: DrainOutcome,
237}
238
239/// JSON reply for rejected admin calls.
240#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
241pub struct AdminErrorReply {
242    /// Stable outcome string.
243    pub outcome: &'static str,
244    /// Redacted detail safe for operator Conditions.
245    pub detail: String,
246}
247
248impl AdminErrorReply {
249    fn rejected(detail: impl Into<String>) -> Self {
250        Self {
251            outcome: "rejected",
252            detail: detail.into(),
253        }
254    }
255}
256
257/// Admin HTTP boundary errors.
258#[derive(Debug, Clone, PartialEq, Eq, Error)]
259pub enum AdminHttpError {
260    /// Admin identity was absent or incomplete.
261    #[error("admin identity is required")]
262    Unauthenticated,
263    /// Caller identity is not privileged for admin actions.
264    #[error("admin privileges are required")]
265    Unauthorized,
266    /// Runtime refused the requested admin action.
267    #[error("{0}")]
268    Action(#[from] ServerAdminActionError),
269}
270
271impl IntoResponse for AdminHttpError {
272    fn into_response(self) -> Response {
273        let status = match self {
274            Self::Unauthenticated => StatusCode::UNAUTHORIZED,
275            Self::Unauthorized => StatusCode::FORBIDDEN,
276            Self::Action(ServerAdminActionError::NotReady(_)) => StatusCode::SERVICE_UNAVAILABLE,
277            Self::Action(
278                ServerAdminActionError::RequiresMember(_) | ServerAdminActionError::BackupDisabled,
279            ) => StatusCode::CONFLICT,
280        };
281        (status, Json(AdminErrorReply::rejected(self.to_string()))).into_response()
282    }
283}
hydracache_server/admin_http.rs

hydracache_server/
admin_http.rs
