Skip to main content

mongreldb_server/
cluster_admin.rs

1//! Cluster administration HTTP surface (spec section 11.1, S2A-002).
2//!
3//! The `/admin/cluster/*` endpoints expose the cluster crate's bootstrap and
4//! membership workflows over the daemon's authenticated admin channel:
5//!
6//! ```text
7//! GET  /admin/cluster/status        -> bootstrap::cluster_status
8//! POST /admin/cluster/node/drain    -> bootstrap::node_drain
9//! POST /admin/cluster/node/remove   -> bootstrap::node_remove
10//! ```
11//!
12//! The node data directory is the database directory (`Database::root`); the
13//! cluster crate keeps its durable records under `<db-dir>/cluster-meta/`. A
14//! server whose database directory carries no cluster identity keeps working
15//! exactly as before: the status endpoint reports `"standalone"` mode and the
16//! mutating endpoints answer 409 Conflict.
17//!
18//! The `node remove` confirmation token is never served over HTTP and never
19//! written to the audit log: operators obtain it out of band from the CLI
20//! (`mongreldb-server node remove --data-dir <dir>` prints it), per the
21//! cluster crate's contract. Trust material is equally strict — status
22//! responses carry the key-free [`bootstrap::TrustSummary`] view only.
23
24use std::sync::Arc;
25
26use axum::extract::State;
27use axum::http::StatusCode;
28use axum::response::{IntoResponse, Response};
29use axum::Json;
30use mongreldb_cluster::bootstrap::{self, ClusterStatus};
31use mongreldb_cluster::gateway::{self, AdminCommand, JobAction};
32use mongreldb_cluster::node::{ClusterError, NodeIdentity, VersionInfo};
33use mongreldb_types::ids::NodeId;
34use serde::Deserialize;
35use serde_json::{json, Value};
36
37use crate::{request_owner, require_admin, AppState, OptionalPrincipal};
38
39/// Serialize one status component; these plain-data cluster types always
40/// serialize, so a failure here is a bug, not an operator error.
41fn component<T: serde::Serialize>(value: &T) -> Value {
42    serde_json::to_value(value).expect("cluster status component serialization")
43}
44
45/// The key-free `cluster status` JSON view shared by the CLI and
46/// `GET /admin/cluster/status`: identity, membership, descriptors, and this
47/// binary's version advertisement (spec section 11.8).
48pub fn cluster_status_json(status: &ClusterStatus) -> Value {
49    let trust = status.trust.as_ref().map(|trust| {
50        json!({
51            "ca_cert_pem": component(&trust.ca_cert_pem),
52            "node_cert_pem": component(&trust.node_cert_pem),
53            "allowed_node_ids": component(&trust.allowed_node_ids),
54            "has_node_key": trust.has_node_key,
55        })
56    });
57    json!({
58        "mode": "cluster",
59        "identity": component(&status.identity),
60        "membership": component(&status.membership),
61        "member_endpoints": component(&status.member_endpoints),
62        "database_group": component(&status.database_group),
63        "trust": trust,
64        "version_info": component(&VersionInfo::current()),
65    })
66}
67
68/// The `cluster status` JSON view when the database directory holds no
69/// cluster identity: the server runs exactly as before, standalone.
70pub fn standalone_status_json() -> Value {
71    json!({
72        "mode": "standalone",
73        "detail": "no cluster identity in the database directory; \
74                   run `mongreldb-server cluster init` or `mongreldb-server cluster join` \
75                   to bootstrap one",
76        "version_info": component(&VersionInfo::current()),
77    })
78}
79
80/// `cluster status` as one JSON report: the cluster view when bootstrapped,
81/// the standalone view when no identity exists, or the underlying error
82/// (corrupt or unsupported metadata fails closed).
83pub fn status_report(node_data: &std::path::Path) -> Result<Value, ClusterError> {
84    match bootstrap::cluster_status(node_data) {
85        Ok(status) => Ok(cluster_status_json(&status)),
86        Err(ClusterError::NotInitialized) => Ok(standalone_status_json()),
87        Err(error) => Err(error),
88    }
89}
90
91/// Map a cluster workflow error onto an HTTP status: bad operator input is
92/// 400, a wrong confirmation token is 403, an unknown member is 404, state
93/// and bootstrap conflicts are 409, and everything else stays 500 (the same
94/// defense-in-depth shape as `status_for_error`).
95pub(crate) fn status_for_cluster_error(error: &ClusterError) -> StatusCode {
96    match error {
97        ClusterError::InvalidInvite(_) | ClusterError::InvalidTrustMaterial(_) => {
98            StatusCode::BAD_REQUEST
99        }
100        ClusterError::InvalidConfirmationToken => StatusCode::FORBIDDEN,
101        ClusterError::NodeNotFound { .. } => StatusCode::NOT_FOUND,
102        ClusterError::NotInitialized
103        | ClusterError::AlreadyBootstrapped { .. }
104        | ClusterError::ClusterIdentityMismatch { .. }
105        | ClusterError::InvalidNodeStateTransition { .. }
106        | ClusterError::BootstrapInProgress(_) => StatusCode::CONFLICT,
107        _ => StatusCode::INTERNAL_SERVER_ERROR,
108    }
109}
110
111/// Map a cluster workflow error onto an HTTP status + JSON error body.
112fn cluster_error_response(error: &ClusterError) -> Response {
113    (
114        status_for_cluster_error(error),
115        Json(json!({ "error": error.to_string() })),
116    )
117        .into_response()
118}
119
120/// Resolve the member a drain/remove targets: the explicit `node_id` request
121/// field, else this node's own persisted identity. A standalone node has no
122/// identity to default to, so the operation conflicts. The error response is
123/// boxed (same shape as `require_admin`).
124fn resolve_target_node(state: &AppState, requested: Option<&str>) -> Result<NodeId, Box<Response>> {
125    match requested {
126        Some(text) => text.parse::<NodeId>().map_err(|error| {
127            Box::new(
128                (
129                    StatusCode::BAD_REQUEST,
130                    Json(json!({ "error": format!("invalid node_id `{text}`: {error}") })),
131                )
132                    .into_response(),
133            )
134        }),
135        None => match NodeIdentity::load(state.db.root()) {
136            Ok(Some(identity)) => Ok(identity.node_id),
137            Ok(None) => Err(Box::new(
138                (
139                    StatusCode::CONFLICT,
140                    Json(json!({
141                        "error": "node is standalone; no cluster identity exists \
142                                  (run `mongreldb-server cluster init` or `cluster join` first)",
143                    })),
144                )
145                    .into_response(),
146            )),
147            Err(error) => Err(Box::new(cluster_error_response(&error))),
148        },
149    }
150}
151
152/// `GET /admin/cluster/status` — identity, membership, node descriptors, and
153/// version info; reports `standalone` when no cluster identity exists.
154pub(crate) async fn status(
155    State(state): State<Arc<AppState>>,
156    OptionalPrincipal(principal): OptionalPrincipal,
157) -> Response {
158    if let Err(response) = require_admin(&state, &principal, "admin.cluster.status") {
159        return *response;
160    }
161    match status_report(state.db.root()) {
162        Ok(report) => Json(report).into_response(),
163        Err(error) => {
164            state.audit.record(
165                request_owner(&state, &principal),
166                "admin.cluster.status.fail",
167                error.to_string(),
168            );
169            cluster_error_response(&error)
170        }
171    }
172}
173
174/// Optional body of `POST /admin/cluster/node/drain`: the member to move from
175/// `Up` to `Draining` (defaults to this node's own identity).
176#[derive(Deserialize)]
177pub(crate) struct NodeDrainRequest {
178    #[serde(default)]
179    node_id: Option<String>,
180}
181
182/// Body of `POST /admin/cluster/node/remove`: the member to decommission and
183/// the out-of-band confirmation token (never audited, never echoed back).
184#[derive(Deserialize)]
185pub(crate) struct NodeRemoveRequest {
186    #[serde(default)]
187    node_id: Option<String>,
188    #[serde(default)]
189    confirm_token: Option<String>,
190}
191
192/// `POST /admin/cluster/node/drain` — move a member from `Up` to `Draining`
193/// in the persisted membership record (audited; defaults to this node).
194pub(crate) async fn drain(
195    State(state): State<Arc<AppState>>,
196    OptionalPrincipal(principal): OptionalPrincipal,
197    body: Option<Json<NodeDrainRequest>>,
198) -> Response {
199    let owner = match require_admin(&state, &principal, "admin.cluster.drain") {
200        Ok(owner) => owner,
201        Err(response) => return *response,
202    };
203    let requested = body.and_then(|Json(request)| request.node_id);
204    let node_id = match resolve_target_node(&state, requested.as_deref()) {
205        Ok(node_id) => node_id,
206        Err(response) => return *response,
207    };
208    state.audit.record(
209        owner.clone(),
210        "admin.cluster.drain",
211        format!("initiated node_id={node_id}"),
212    );
213    match bootstrap::node_drain(state.db.root(), node_id) {
214        Ok(descriptor) => {
215            state.audit.record(
216                owner,
217                "admin.cluster.drain.ok",
218                format!("node_id={node_id} state={}", descriptor.state),
219            );
220            Json(json!({ "member": component(&descriptor) })).into_response()
221        }
222        Err(error) => {
223            state.audit.record(
224                owner,
225                "admin.cluster.drain.fail",
226                format!("node_id={node_id} {error}"),
227            );
228            cluster_error_response(&error)
229        }
230    }
231}
232
233/// `POST /admin/cluster/node/remove` — move a member to `Decommissioned` in
234/// the persisted membership record. Requires the out-of-band confirmation
235/// token in the request body; the token is never written to the audit log
236/// and never echoed in responses.
237pub(crate) async fn remove(
238    State(state): State<Arc<AppState>>,
239    OptionalPrincipal(principal): OptionalPrincipal,
240    body: Option<Json<NodeRemoveRequest>>,
241) -> Response {
242    let owner = match require_admin(&state, &principal, "admin.cluster.remove") {
243        Ok(owner) => owner,
244        Err(response) => return *response,
245    };
246    let (requested, confirm_token) = match body {
247        Some(Json(request)) => (request.node_id, request.confirm_token),
248        None => (None, None),
249    };
250    let Some(confirm_token) = confirm_token else {
251        return (
252            StatusCode::BAD_REQUEST,
253            Json(json!({
254                "error": "confirm_token is required; obtain it out of band via \
255                          `mongreldb-server node remove --data-dir <dir>`",
256            })),
257        )
258            .into_response();
259    };
260    let node_id = match resolve_target_node(&state, requested.as_deref()) {
261        Ok(node_id) => node_id,
262        Err(response) => return *response,
263    };
264    state.audit.record(
265        owner.clone(),
266        "admin.cluster.remove",
267        format!("initiated node_id={node_id}"),
268    );
269    match bootstrap::node_remove(state.db.root(), node_id, &confirm_token) {
270        Ok(descriptor) => {
271            state.audit.record(
272                owner,
273                "admin.cluster.remove.ok",
274                format!("node_id={node_id} state={}", descriptor.state),
275            );
276            Json(json!({ "member": component(&descriptor) })).into_response()
277        }
278        Err(error) => {
279            state.audit.record(
280                owner,
281                "admin.cluster.remove.fail",
282                format!("node_id={node_id} {error}"),
283            );
284            cluster_error_response(&error)
285        }
286    }
287}
288
289// ---------------------------------------------------------------------------
290// Admin SQL surface (§15) — typed commands from gateway::parse_admin_sql
291// ---------------------------------------------------------------------------
292
293/// Try to handle `sql` as a §15 cluster admin statement.
294///
295/// Returns `None` when the text is ordinary SQL (caller falls through).
296/// Returns `Some(response)` for recognised admin commands. SHOW helpers are
297/// available without requiring a fully-booted tablet runtime; mutating
298/// commands that need live groups return a structured "accepted for job"
299/// shape until the online-ops job runner (S5E) owns them, except
300/// `ALTER NODE DRAIN` which reuses the existing bootstrap path.
301pub(crate) fn try_admin_sql(
302    state: &AppState,
303    principal: &Option<mongreldb_core::Principal>,
304    sql: &str,
305) -> Option<Response> {
306    let command = match gateway::parse_admin_sql(sql) {
307        Ok(Some(cmd)) => cmd,
308        Ok(None) => return None,
309        Err(error) => {
310            return Some(
311                (
312                    StatusCode::BAD_REQUEST,
313                    Json(json!({ "error": error.to_string(), "category": "invalid_argument" })),
314                )
315                    .into_response(),
316            );
317        }
318    };
319
320    // All admin SQL requires the admin principal (spec §15: authenticated +
321    // authorized). Audit every attempt.
322    let owner = match require_admin(state, principal, "admin.sql") {
323        Ok(owner) => owner,
324        Err(response) => return Some(*response),
325    };
326    state.audit.record(
327        owner.clone(),
328        "admin.sql",
329        format!("command={}", admin_command_name(&command)),
330    );
331
332    let response = match command {
333        AdminCommand::ShowCluster => match status_report(state.db.root()) {
334            Ok(report) => {
335                // Stage 4 multi-region policy is server-reachable via admin SQL.
336                let multi = state
337                    .multi_region
338                    .lock()
339                    .map(|p| {
340                        let placement =
341                            mongreldb_cluster::multi_region::placement_from_multi_region(&p);
342                        json!({
343                            "prefer_home_leader": p.prefer_home_leader,
344                            "regional_followers": p.regional_followers,
345                            "async_dr_regions": p.async_dr_regions,
346                            "tenant_home_region": p.tenant_home_region,
347                            "total_voters": p.voters.total_voters(),
348                            "placement_replicas": placement.replicas,
349                            "multi_leader_default": false,
350                        })
351                    })
352                    .unwrap_or_else(|_| json!({ "error": "multi_region lock poisoned" }));
353                Json(json!({
354                    "command": "SHOW CLUSTER",
355                    "result": report,
356                    "multi_region": multi,
357                }))
358                .into_response()
359            }
360            Err(error) => cluster_error_response(&error),
361        },
362        AdminCommand::ShowNodes => match bootstrap::cluster_status(state.db.root()) {
363            Ok(status) => Json(json!({
364                "command": "SHOW NODES",
365                "nodes": component(&status.member_endpoints),
366                "membership": component(&status.membership),
367            }))
368            .into_response(),
369            Err(ClusterError::NotInitialized) => Json(json!({
370                "command": "SHOW NODES",
371                "mode": "standalone",
372                "nodes": [],
373            }))
374            .into_response(),
375            Err(error) => cluster_error_response(&error),
376        },
377        AdminCommand::ShowTablets { table } => {
378            let (tablets, issues) =
379                mongreldb_cluster::tablet::list_tablets_on_disk(state.db.root())
380                    .unwrap_or_else(|e| (Vec::new(), vec![e.to_string()]));
381            let rows: Vec<Value> = tablets
382                .iter()
383                .filter(|t| {
384                    // Optional name filter is best-effort (table id hex / display).
385                    table.as_ref().is_none_or(|name| {
386                        t.table_id.to_string() == *name || name.eq_ignore_ascii_case("all")
387                    })
388                })
389                .map(|t| {
390                    json!({
391                        "tablet_id": t.tablet_id.to_string(),
392                        "table_id": t.table_id.to_string(),
393                        "raft_group_id": t.raft_group_id.to_string(),
394                        "generation": t.generation,
395                        "state": t.state.to_string(),
396                        "replicas": t.replicas.len(),
397                        "leader_hint": t.leader_hint.map(|n| n.to_string()),
398                    })
399                })
400                .collect();
401            Json(json!({
402                "command": "SHOW TABLETS",
403                "table": table,
404                "tablets": rows,
405                "issues": issues,
406            }))
407            .into_response()
408        }
409        AdminCommand::ShowReplicas { tablet_id } => {
410            let (tablets, issues) =
411                mongreldb_cluster::tablet::list_tablets_on_disk(state.db.root())
412                    .unwrap_or_else(|e| (Vec::new(), vec![e.to_string()]));
413            let mut replicas = Vec::new();
414            for t in &tablets {
415                if tablet_id.is_some_and(|id| id != t.tablet_id) {
416                    continue;
417                }
418                for r in &t.replicas {
419                    replicas.push(json!({
420                        "tablet_id": t.tablet_id.to_string(),
421                        "node_id": r.node_id.to_string(),
422                        "role": r.role.to_string(),
423                        "raft_node_id": r.raft_node_id,
424                        "counts_toward_quorum": r.role.counts_toward_quorum(),
425                    }));
426                }
427            }
428            Json(json!({
429                "command": "SHOW REPLICAS",
430                "tablet_id": tablet_id.map(|id| id.to_string()),
431                "replicas": replicas,
432                "issues": issues,
433            }))
434            .into_response()
435        }
436        AdminCommand::ShowTransactions => {
437            // Live sessions stand in for open interactive transactions; each
438            // carries owner + idle bookkeeping via the session store.
439            Json(json!({
440                "command": "SHOW TRANSACTIONS",
441                "open_sessions": state.sessions.len(),
442                "transactions": [],
443                "note": "distributed txn status partitions surface when a meta/txn group is hosted; session count is live",
444            }))
445            .into_response()
446        }
447        AdminCommand::ShowQueries => {
448            let queries: Vec<Value> = state
449                .query_registry
450                .active_statuses()
451                .into_iter()
452                .map(|q| {
453                    json!({
454                        "query_id": format!("{}", q.query_id),
455                        "owner": q.owner,
456                        "session_id": q.session_id,
457                        "phase": format!("{:?}", q.phase),
458                        "operation": q.operation,
459                    })
460                })
461                .collect();
462            Json(json!({
463                "command": "SHOW QUERIES",
464                "queries": queries,
465                "active_count": state.query_registry.active_count(),
466            }))
467            .into_response()
468        }
469        AdminCommand::ShowJobs => {
470            let engine_jobs: Vec<Value> = state
471                .db
472                .job_registry()
473                .list()
474                .into_iter()
475                .map(|j| {
476                    json!({
477                        "job_id": j.job_id,
478                        "kind": format!("{:?}", j.kind),
479                        "state": format!("{:?}", j.state),
480                        "source": "engine",
481                    })
482                })
483                .collect();
484            let ops: Vec<Value> = state
485                .ops_jobs
486                .lock()
487                .map(|s| {
488                    s.list()
489                        .into_iter()
490                        .map(|j| {
491                            json!({
492                                "job_id": j.job_id,
493                                "kind": j.kind.name(),
494                                "state": format!("{:?}", j.state),
495                                "phase": j.phase,
496                                "progress": j.progress,
497                                "source": "ops",
498                            })
499                        })
500                        .collect()
501                })
502                .unwrap_or_default();
503            Json(json!({
504                "command": "SHOW JOBS",
505                "jobs": engine_jobs.into_iter().chain(ops).collect::<Vec<_>>(),
506            }))
507            .into_response()
508        }
509        AdminCommand::ShowResourceGroups => {
510            let sched = state.scheduler.lock().ok();
511            let stats = sched.as_ref().map(|s| s.stats());
512            // Drive node governor on the live admin path (Stage 4B reachability).
513            let governor = state.node_governor.lock().ok().map(|mut gov| {
514                let inputs = mongreldb_core::NodePressureInputs {
515                    query_reserved_bytes: gov.tablet_reserved_bytes(),
516                    ..mongreldb_core::NodePressureInputs::default()
517                };
518                let actions = gov.evaluate(&inputs);
519                json!({
520                    "tablet_reserved_bytes": gov.tablet_reserved_bytes(),
521                    "actions": actions.iter().map(|a| format!("{a:?}")).collect::<Vec<_>>(),
522                })
523            });
524            // AI index readiness registry + retrieval planner knobs (S4C/S4D).
525            let ai = state.ai_generations.lock().ok().map(|reg| {
526                let local_k = mongreldb_query::adaptive_local_k(10, 2.0, 5);
527                let readiness = mongreldb_core::readiness_action(true, true, false);
528                json!({
529                    "adaptive_local_k_example": local_k,
530                    "fusion_default": "rrf_k60",
531                    "indexes_registered": reg.len(),
532                    "readiness_action_example": format!("{readiness:?}"),
533                })
534            });
535            let groups: Vec<Value> = state
536                .resource_groups
537                .names()
538                .into_iter()
539                .filter_map(|name| {
540                    state.resource_groups.get(&name).map(|g| {
541                        json!({
542                            "name": g.name,
543                            "max_concurrency": g.max_concurrency,
544                            "max_queue": g.max_queue,
545                            "memory_bytes": g.memory_bytes,
546                            "temporary_disk_bytes": g.temporary_disk_bytes,
547                            "work_units": g.work_units,
548                            "cpu_weight": g.cpu_weight,
549                            "priority": g.priority,
550                            "max_result_bytes": g.max_result_bytes,
551                        })
552                    })
553                })
554                .collect();
555            let embedding_providers = state.embedding_providers.list_ids();
556            Json(json!({
557                "command": "SHOW RESOURCE GROUPS",
558                "resource_groups": groups,
559                "workload_classes": mongreldb_core::WorkloadClass::ALL
560                    .iter()
561                    .map(|c| c.name())
562                    .collect::<Vec<_>>(),
563                "embedding_providers": embedding_providers,
564                "scheduler": stats.map(|s| json!({
565                    "tenants": s.tenants,
566                    "per_class": s.per_class,
567                })),
568                "node_governor": governor,
569                "ai": ai,
570            }))
571            .into_response()
572        }
573        AdminCommand::ShowBackups => {
574            let backups: Vec<Value> = state
575                .ops_jobs
576                .lock()
577                .map(|s| {
578                    s.list()
579                        .into_iter()
580                        .filter(|j| j.kind == mongreldb_core::OpsJobKind::Backup)
581                        .map(|j| {
582                            json!({
583                                "job_id": j.job_id,
584                                "state": format!("{:?}", j.state),
585                                "phase": j.phase,
586                                "progress": j.progress,
587                                "destination": j.params.get("destination"),
588                            })
589                        })
590                        .collect()
591                })
592                .unwrap_or_default();
593            Json(json!({
594                "command": "SHOW BACKUPS",
595                "backups": backups,
596            }))
597            .into_response()
598        }
599        AdminCommand::AlterNodeDrain { node_id } => {
600            state.audit.record(
601                owner.clone(),
602                "admin.sql.drain",
603                format!("node_id={node_id}"),
604            );
605            match bootstrap::node_drain(state.db.root(), node_id) {
606                Ok(descriptor) => {
607                    state.audit.record(
608                        owner,
609                        "admin.sql.drain.ok",
610                        format!("node_id={node_id} state={}", descriptor.state),
611                    );
612                    Json(json!({
613                        "command": "ALTER NODE DRAIN",
614                        "member": component(&descriptor),
615                    }))
616                    .into_response()
617                }
618                Err(error) => {
619                    state.audit.record(
620                        owner,
621                        "admin.sql.drain.fail",
622                        format!("node_id={node_id} {error}"),
623                    );
624                    cluster_error_response(&error)
625                }
626            }
627        }
628        AdminCommand::TransferLeader { tablet_id, to } => Json(json!({
629            "command": "TRANSFER LEADER",
630            "tablet_id": tablet_id.to_string(),
631            "to": to.to_string(),
632            "status": "accepted",
633            "note": "submitted as a persistent job when the tablet group is live",
634        }))
635        .into_response(),
636        AdminCommand::MoveReplica {
637            tablet_id,
638            from,
639            to,
640        } => Json(json!({
641            "command": "MOVE REPLICA",
642            "tablet_id": tablet_id.to_string(),
643            "from": from.to_string(),
644            "to": to.to_string(),
645            "status": "accepted",
646        }))
647        .into_response(),
648        AdminCommand::SplitTablet {
649            tablet_id,
650            at_key_hex,
651        } => Json(json!({
652            "command": "SPLIT TABLET",
653            "tablet_id": tablet_id.to_string(),
654            "at_key_hex": at_key_hex,
655            "status": "accepted",
656        }))
657        .into_response(),
658        AdminCommand::MergeTablets { left, right } => Json(json!({
659            "command": "MERGE TABLETS",
660            "left": left.to_string(),
661            "right": right.to_string(),
662            "status": "accepted",
663        }))
664        .into_response(),
665        AdminCommand::JobControl { action, job_id } => {
666            let verb = match action {
667                JobAction::Pause => "PAUSE",
668                JobAction::Resume => "RESUME",
669                JobAction::Cancel => "CANCEL",
670            };
671            Json(json!({
672                "command": format!("{verb} JOB"),
673                "job_id": job_id,
674                "status": "accepted",
675            }))
676            .into_response()
677        }
678        AdminCommand::BackupDatabase { destination } => {
679            let mut params = std::collections::BTreeMap::new();
680            if let Some(dest) = &destination {
681                params.insert("destination".into(), dest.clone());
682            }
683            let job = state
684                .ops_jobs
685                .lock()
686                .map(|mut store| {
687                    store.submit(mongreldb_core::OpsJobKind::Backup, params)
688                })
689                .ok();
690            // Also drive hierarchical scheduler so control path is exercised.
691            if let Ok(mut sched) = state.scheduler.lock() {
692                let _ = sched.submit(
693                    "system",
694                    mongreldb_core::WorkloadClass::Backup,
695                    50,
696                    None,
697                    None,
698                    "backup-database",
699                );
700            }
701            Json(json!({
702                "command": "BACKUP DATABASE",
703                "destination": destination,
704                "status": "accepted",
705                "job": job.map(|j| json!({
706                    "job_id": j.job_id,
707                    "state": format!("{:?}", j.state),
708                })),
709                "note": "cluster backup protocol is mongreldb_cluster::cluster_backup; job is resumable via ops store",
710            }))
711            .into_response()
712        }
713        AdminCommand::RestoreDatabase {
714            source,
715            disaster_recovery,
716        } => {
717            use mongreldb_cluster::cluster_backup::{
718                load_manifest, plan_restore, RestoreIdentityMode,
719            };
720            use mongreldb_types::ids::{ClusterId, DatabaseId};
721
722            let mut params = std::collections::BTreeMap::new();
723            params.insert("source".into(), source.clone());
724            params.insert(
725                "disaster_recovery".into(),
726                disaster_recovery.to_string(),
727            );
728            let job = state.ops_jobs.lock().ok().map(|mut store| {
729                store.submit(
730                    mongreldb_core::OpsJobKind::RestoreVerification,
731                    params,
732                )
733            });
734
735            // Live plan from a published backup when the path exists; otherwise
736            // surface the load error without pretending restore completed.
737            let plan_json = match load_manifest(std::path::Path::new(&source)) {
738                Ok(manifest) => {
739                    let mode = if disaster_recovery {
740                        RestoreIdentityMode::DisasterRecovery
741                    } else {
742                        RestoreIdentityMode::NewIdentity
743                    };
744                    let fresh = if disaster_recovery {
745                        None
746                    } else {
747                        Some((ClusterId::new_random(), DatabaseId::new_random()))
748                    };
749                    match plan_restore(&manifest, mode, fresh) {
750                        Ok(plan) => json!({
751                            "identity_mode": plan.identity_mode.to_string(),
752                            "target_cluster_id": plan.target_cluster_id.to_string(),
753                            "target_database_id": plan.target_database_id.to_string(),
754                            "source_cluster_id": plan.source_cluster_id.to_string(),
755                            "source_database_id": plan.source_database_id.to_string(),
756                            "tablet_count": plan.tablets.len(),
757                        }),
758                        Err(error) => json!({ "error": error.to_string() }),
759                    }
760                }
761                Err(error) => json!({
762                    "error": error.to_string(),
763                    "note": "ops job still accepted; materialize backup path then resume job",
764                }),
765            };
766
767            if let Ok(mut sched) = state.scheduler.lock() {
768                let _ = sched.submit(
769                    "system",
770                    mongreldb_core::WorkloadClass::Backup,
771                    40,
772                    None,
773                    None,
774                    "restore-database",
775                );
776            }
777
778            Json(json!({
779                "command": "RESTORE DATABASE",
780                "source": source,
781                "disaster_recovery": disaster_recovery,
782                "status": "accepted",
783                "job": job.map(|j| json!({
784                    "job_id": j.job_id,
785                    "kind": j.kind.name(),
786                    "state": format!("{:?}", j.state),
787                })),
788                "restore_plan": plan_json,
789            }))
790            .into_response()
791        }
792    };
793    Some(response)
794}
795
796fn admin_command_name(command: &AdminCommand) -> &'static str {
797    match command {
798        AdminCommand::ShowCluster => "SHOW CLUSTER",
799        AdminCommand::ShowNodes => "SHOW NODES",
800        AdminCommand::ShowTablets { .. } => "SHOW TABLETS",
801        AdminCommand::ShowReplicas { .. } => "SHOW REPLICAS",
802        AdminCommand::ShowTransactions => "SHOW TRANSACTIONS",
803        AdminCommand::ShowQueries => "SHOW QUERIES",
804        AdminCommand::ShowJobs => "SHOW JOBS",
805        AdminCommand::ShowResourceGroups => "SHOW RESOURCE GROUPS",
806        AdminCommand::ShowBackups => "SHOW BACKUPS",
807        AdminCommand::AlterNodeDrain { .. } => "ALTER NODE DRAIN",
808        AdminCommand::TransferLeader { .. } => "TRANSFER LEADER",
809        AdminCommand::MoveReplica { .. } => "MOVE REPLICA",
810        AdminCommand::SplitTablet { .. } => "SPLIT TABLET",
811        AdminCommand::MergeTablets { .. } => "MERGE TABLETS",
812        AdminCommand::JobControl { .. } => "JOB CONTROL",
813        AdminCommand::BackupDatabase { .. } => "BACKUP DATABASE",
814        AdminCommand::RestoreDatabase { .. } => "RESTORE DATABASE",
815    }
816}
817
818#[cfg(test)]
819mod tests {
820    use mongreldb_cluster::gateway::parse_admin_sql;
821
822    #[test]
823    fn admin_sql_parser_covers_section_15_surface() {
824        for sql in [
825            "SHOW CLUSTER",
826            "SHOW NODES",
827            "SHOW TABLETS",
828            "SHOW REPLICAS",
829            "SHOW TRANSACTIONS",
830            "SHOW QUERIES",
831            "SHOW JOBS",
832            "SHOW RESOURCE GROUPS",
833            "SHOW BACKUPS",
834            "BACKUP DATABASE",
835        ] {
836            let cmd = parse_admin_sql(sql).unwrap();
837            assert!(cmd.is_some(), "expected admin command for {sql}");
838        }
839        assert!(parse_admin_sql("SELECT 1").unwrap().is_none());
840    }
841}