From 3420f28c17ab76e4636a039ff6cc673e2f0455f8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 7 Jul 2026 04:46:42 +0800 Subject: [PATCH] feat(admin): add MinIO-compatible diagnostics endpoints (#4333) feat(admin): add MinIO-compatible diagnostics and service-control endpoints Register and implement MinIO admin API compatibility routes: - POST /v3/service: restart/stop trigger the existing graceful-shutdown path; freeze/unfreeze record advisory state (request admission is not yet gated, surfaced honestly via effective=false) - profiling/trace and healthinfo/obdinfo/log diagnostics endpoints - GET /v3/top/locks and POST /v3/force-unlock, backed by new FastObjectLockManager::list_locks()/force_unlock() introspection - speedtest routes where feasible Refs rustfs/backlog#604 #606 #607 #615 --- crates/lock/src/fast_lock/manager.rs | 76 ++ crates/lock/src/fast_lock/shard.rs | 81 ++ rustfs/src/admin/handlers/diagnostics.rs | 993 ++++++++++++++++++++ rustfs/src/admin/handlers/mod.rs | 1 + rustfs/src/admin/handlers/profile_admin.rs | 220 ++++- rustfs/src/admin/handlers/system.rs | 339 ++++++- rustfs/src/admin/mod.rs | 3 +- rustfs/src/admin/route_policy.rs | 36 + rustfs/src/admin/route_registration_test.rs | 16 + 9 files changed, 1755 insertions(+), 10 deletions(-) create mode 100644 rustfs/src/admin/handlers/diagnostics.rs diff --git a/crates/lock/src/fast_lock/manager.rs b/crates/lock/src/fast_lock/manager.rs index 7b7801d49..d1fc2e2df 100644 --- a/crates/lock/src/fast_lock/manager.rs +++ b/crates/lock/src/fast_lock/manager.rs @@ -290,6 +290,28 @@ impl FastObjectLockManager { shard.get_lock_info(key) } + /// Enumerate every currently held lock across all shards. + /// + /// Powers the admin "top locks" view. Order is shard-then-insertion and is + /// not otherwise stable across calls. + pub fn list_locks(&self) -> Vec { + let mut infos = Vec::new(); + for shard in &self.shards { + infos.extend(shard.list_locks()); + } + infos + } + + /// Force-release every holder of the lock on `key`. + /// + /// Returns the number of owners released (0 if the resource was not locked). + /// This bypasses guard tracking and is intended only for administrative + /// recovery of a stuck resource. + pub fn force_unlock(&self, key: &crate::fast_lock::types::ObjectKey) -> usize { + let shard = self.get_shard(key); + shard.force_release_all(key) + } + /// Get aggregated metrics pub fn get_metrics(&self) -> crate::fast_lock::metrics::AggregatedMetrics { let shard_metrics: Vec<_> = self.shards.iter().map(|shard| shard.metrics().snapshot()).collect(); @@ -467,6 +489,7 @@ impl LockManager for FastObjectLockManager { #[cfg(test)] mod tests { use super::*; + use crate::fast_lock::types::LockMode; fn make_request(manager: &FastObjectLockManager, shard_id: usize, suffix: usize) -> ObjectLockRequest { let mut candidate = 0usize; @@ -507,4 +530,57 @@ mod tests { manager.shutdown().await; } + + #[tokio::test] + async fn test_list_locks_reports_held_locks() { + let manager = FastObjectLockManager::new(); + let write_key = ObjectKey::new("bucket", "write-object"); + let read_key = ObjectKey::new("bucket", "read-object"); + + let _write_guard = manager + .acquire_write_lock(write_key.clone(), "writer") + .await + .expect("write lock should acquire"); + let _read_guard = manager + .acquire_read_lock(read_key.clone(), "reader") + .await + .expect("read lock should acquire"); + + let mut locks = manager.list_locks(); + locks.sort_by(|a, b| a.key.object.cmp(&b.key.object)); + assert_eq!(locks.len(), 2); + + let read = locks.iter().find(|l| l.key == read_key).expect("read lock listed"); + assert_eq!(read.mode, LockMode::Shared); + assert_eq!(read.owner.as_ref(), "reader"); + + let write = locks.iter().find(|l| l.key == write_key).expect("write lock listed"); + assert_eq!(write.mode, LockMode::Exclusive); + assert_eq!(write.owner.as_ref(), "writer"); + + manager.shutdown().await; + } + + #[tokio::test] + async fn test_force_unlock_releases_stuck_lock() { + let manager = FastObjectLockManager::new(); + let key = ObjectKey::new("bucket", "stuck-object"); + + let guard = manager + .acquire_write_lock(key.clone(), "owner") + .await + .expect("write lock should acquire"); + // Leak the guard so it cannot release on drop, mimicking a stuck lock. + std::mem::forget(guard); + assert_eq!(manager.total_lock_count(), 1); + + let released = manager.force_unlock(&key); + assert_eq!(released, 1); + assert!(manager.list_locks().is_empty()); + + // Force-unlocking an already-clear resource is a no-op. + assert_eq!(manager.force_unlock(&key), 0); + + manager.shutdown().await; + } } diff --git a/crates/lock/src/fast_lock/shard.rs b/crates/lock/src/fast_lock/shard.rs index 258106d36..fb2e56c71 100644 --- a/crates/lock/src/fast_lock/shard.rs +++ b/crates/lock/src/fast_lock/shard.rs @@ -508,6 +508,87 @@ impl LockShard { None } + /// Enumerate every currently held lock in this shard. + /// + /// Exclusive locks yield a single entry; shared locks yield one entry per + /// distinct owner so administrative "top locks" views can attribute every + /// holder. Entries for objects that are tracked but not currently locked + /// (e.g. pooled-but-idle state) are skipped. + pub fn list_locks(&self) -> Vec { + let objects = self.objects.read(); + let mut infos = Vec::new(); + for (key, state) in objects.iter() { + let Some(mode) = state.current_mode() else { + continue; + }; + let priority = *state.priority.read(); + match mode { + LockMode::Exclusive => { + if let Some(info) = state.current_owner.read().clone() { + let expires_at = info + .acquired_at + .checked_add(info.lock_timeout) + .unwrap_or_else(|| info.acquired_at + crate::fast_lock::DEFAULT_LOCK_TIMEOUT); + infos.push(crate::fast_lock::types::ObjectLockInfo { + key: key.clone(), + mode, + owner: info.owner, + acquired_at: info.acquired_at, + expires_at, + priority, + }); + } + } + LockMode::Shared => { + for entry in state.shared_owners.read().iter() { + let expires_at = entry + .acquired_at + .checked_add(entry.lock_timeout) + .unwrap_or_else(|| entry.acquired_at + crate::fast_lock::DEFAULT_LOCK_TIMEOUT); + infos.push(crate::fast_lock::types::ObjectLockInfo { + key: key.clone(), + mode, + owner: entry.owner.clone(), + acquired_at: entry.acquired_at, + expires_at, + priority, + }); + } + } + } + } + infos + } + + /// Force-release every holder of a lock on `key`, regardless of owner. + /// + /// Returns the number of owners that were released. Used by the admin + /// force-unlock path to clear a stuck resource. + pub fn force_release_all(&self, key: &ObjectKey) -> usize { + let owners_modes: Vec<(Arc, LockMode)> = { + let objects = self.objects.read(); + let Some(state) = objects.get(key) else { + return 0; + }; + let mut pairs = Vec::new(); + if let Some(info) = state.current_owner.read().clone() { + pairs.push((info.owner, LockMode::Exclusive)); + } + for entry in state.shared_owners.read().iter() { + pairs.push((entry.owner.clone(), LockMode::Shared)); + } + pairs + }; + + let mut released = 0; + for (owner, mode) in owners_modes { + if self.release_lock(key, &owner, mode) { + released += 1; + } + } + released + } + /// Get current load factor of the shard pub fn current_load_factor(&self) -> f64 { let objects = self.objects.read(); diff --git a/rustfs/src/admin/handlers/diagnostics.rs b/rustfs/src/admin/handlers/diagnostics.rs new file mode 100644 index 000000000..0564d21b0 --- /dev/null +++ b/rustfs/src/admin/handlers/diagnostics.rs @@ -0,0 +1,993 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! MinIO-compatible admin diagnostics endpoints. +//! +//! Implements the lock inspection / force-unlock, health-info, log, and +//! speedtest families of the admin `/v3` API surface. Each handler wires to a +//! real RustFS subsystem where one exists (namespace lock manager, `sysinfo` +//! host telemetry, `StorageInfo` per-drive throughput) and returns +//! MinIO-compatible request/response semantics — with an explicit, +//! honestly-labeled capability note — where RustFS does not yet carry the +//! backing infrastructure (in-process log ring buffer, cross-node object +//! speedtest harness). + +use crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::storage_api::access::spawn_traced; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use bytes::Bytes; +use futures::{Stream, StreamExt}; +use http::{HeaderMap, HeaderValue, Uri}; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_lock::{LockMode, ObjectKey, get_global_lock_manager}; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::header::CONTENT_TYPE; +use s3s::stream::{ByteStream, DynByteStream}; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, StdError, s3_error}; +use serde::{Deserialize, Serialize}; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::{Duration, SystemTime}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tracing::warn; + +const CONTENT_TYPE_JSON: &str = "application/json"; +const CONTENT_TYPE_NDJSON: &str = "application/x-ndjson"; + +/// Cap on how many locks a single `top/locks` response enumerates, matching the +/// MinIO default page size and bounding response size on busy clusters. +const TOP_LOCKS_DEFAULT_LIMIT: usize = 1000; +const TOP_LOCKS_MAX_LIMIT: usize = 10000; + +// --------------------------------------------------------------------------- +// Route registration +// --------------------------------------------------------------------------- + +pub fn register_diagnostics_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/top/locks").as_str(), + AdminOperation(&TopLocksHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/force-unlock").as_str(), + AdminOperation(&ForceUnlockHandler {}), + )?; + + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/healthinfo").as_str(), + AdminOperation(&HealthInfoHandler {}), + )?; + // MinIO exposes the same collector under the legacy `obdinfo` alias. + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/obdinfo").as_str(), + AdminOperation(&HealthInfoHandler {}), + )?; + + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/log").as_str(), + AdminOperation(&ConsoleLogHandler {}), + )?; + + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest").as_str(), + AdminOperation(&SpeedtestHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest/object").as_str(), + AdminOperation(&SpeedtestHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest/drive").as_str(), + AdminOperation(&SpeedtestHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest/net").as_str(), + AdminOperation(&SpeedtestHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest/site").as_str(), + AdminOperation(&SpeedtestHandler {}), + )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/speedtest/client/devnull").as_str(), + AdminOperation(&SpeedtestClientDevnullHandler {}), + )?; + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Shared auth helper +// --------------------------------------------------------------------------- + +async fn authorize(req: &S3Request, action: AdminAction) -> S3Result<()> { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(AccessDenied, "Signature is required")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + + validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await +} + +fn json_response(status: StatusCode, value: &T) -> S3Result> { + let data = serde_json::to_vec(value) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize response: {e}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE_JSON)); + Ok(S3Response::with_headers((status, Body::from(data)), headers)) +} + +async fn read_body(input: Body) -> S3Result> { + let mut input = input; + let body = input + .store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?; + Ok(body.to_vec()) +} + +// --------------------------------------------------------------------------- +// #615 GET /v3/top/locks — real namespace-lock enumeration +// --------------------------------------------------------------------------- + +/// A single held namespace lock, shaped for the admin "top locks" view. +#[derive(Debug, Clone, Serialize)] +pub struct LockEntry { + /// `bucket/object` resource the lock is held on. + pub resource: String, + pub bucket: String, + pub object: String, + /// `None` means the latest version. + #[serde(skip_serializing_if = "Option::is_none")] + pub version: Option, + /// `"WRITE"` for exclusive, `"READ"` for shared. + #[serde(rename = "type")] + pub lock_type: &'static str, + pub owner: String, + pub priority: &'static str, + /// Wall-clock RFC3339 timestamp when the lock was acquired, if representable. + #[serde(skip_serializing_if = "Option::is_none")] + pub since: Option, + /// How long the lock has been held, in seconds. + pub elapsed_secs: u64, + /// Seconds until the lock's timeout expires (0 if already past). + pub ttl_secs: u64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct TopLocksResponse { + pub total: usize, + pub truncated: bool, + pub locks: Vec, + /// Present only when the lock subsystem is disabled by configuration. + #[serde(skip_serializing_if = "Option::is_none")] + pub capability_note: Option, +} + +fn lock_priority_label(priority: rustfs_lock::fast_lock::LockPriority) -> &'static str { + use rustfs_lock::fast_lock::LockPriority; + match priority { + LockPriority::Low => "LOW", + LockPriority::Normal => "NORMAL", + LockPriority::High => "HIGH", + LockPriority::Critical => "CRITICAL", + } +} + +fn system_time_to_rfc3339(t: SystemTime) -> Option { + let dt: time::OffsetDateTime = t.into(); + dt.format(&time::format_description::well_known::Rfc3339).ok() +} + +fn collect_top_locks(limit: usize) -> TopLocksResponse { + let manager = get_global_lock_manager(); + let Some(fast) = manager.as_fast_lock_manager() else { + return TopLocksResponse { + total: 0, + truncated: false, + locks: Vec::new(), + capability_note: Some( + "namespace lock subsystem is disabled (RUSTFS_LOCK_ENABLED=false); no locks are tracked".to_string(), + ), + }; + }; + + let now = SystemTime::now(); + let mut infos = fast.list_locks(); + // Longest-held first, matching MinIO's `top locks` ordering intent. + infos.sort_by_key(|i| i.acquired_at); + let total = infos.len(); + let truncated = total > limit; + + let locks = infos + .into_iter() + .take(limit) + .map(|info| { + let elapsed_secs = now.duration_since(info.acquired_at).unwrap_or(Duration::ZERO).as_secs(); + let ttl_secs = info.expires_at.duration_since(now).unwrap_or(Duration::ZERO).as_secs(); + LockEntry { + resource: format!("{}/{}", info.key.bucket, info.key.object), + bucket: info.key.bucket.to_string(), + object: info.key.object.to_string(), + version: info.key.version.as_ref().map(|v| v.to_string()), + lock_type: match info.mode { + LockMode::Exclusive => "WRITE", + LockMode::Shared => "READ", + }, + owner: info.owner.to_string(), + priority: lock_priority_label(info.priority), + since: system_time_to_rfc3339(info.acquired_at), + elapsed_secs, + ttl_secs, + } + }) + .collect(); + + TopLocksResponse { + total, + truncated, + locks, + capability_note: None, + } +} + +fn parse_top_locks_limit(uri: &Uri) -> usize { + query_value(uri, "count") + .and_then(|v| v.parse::().ok()) + .filter(|n| *n > 0) + .unwrap_or(TOP_LOCKS_DEFAULT_LIMIT) + .min(TOP_LOCKS_MAX_LIMIT) +} + +pub struct TopLocksHandler {} + +#[async_trait::async_trait] +impl Operation for TopLocksHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::TopLocksAdminAction).await?; + let limit = parse_top_locks_limit(&req.uri); + let response = collect_top_locks(limit); + json_response(StatusCode::OK, &response) + } +} + +// --------------------------------------------------------------------------- +// #615 POST /v3/force-unlock — real force-release +// --------------------------------------------------------------------------- + +#[derive(Debug, Deserialize)] +struct ForceUnlockRequest { + /// Resources to force-unlock. Each entry is a `bucket/object` path or a + /// bucket-only name. Accepts either an explicit list or the MinIO-style + /// single `resource` field. + #[serde(default)] + resources: Vec, + #[serde(default)] + resource: Option, +} + +#[derive(Debug, Serialize)] +struct ForceUnlockResult { + resource: String, + released_owners: usize, +} + +#[derive(Debug, Serialize)] +struct ForceUnlockResponse { + results: Vec, + total_released: usize, + #[serde(skip_serializing_if = "Option::is_none")] + capability_note: Option, +} + +fn parse_resource(raw: &str) -> Option { + let trimmed = raw.trim().trim_start_matches('/'); + if trimmed.is_empty() { + return None; + } + match trimmed.split_once('/') { + Some((bucket, object)) if !bucket.is_empty() && !object.is_empty() => Some(ObjectKey::new(bucket, object)), + // Bucket-only resource: MinIO treats the bucket name itself as the object key. + _ => Some(ObjectKey::new(trimmed, trimmed)), + } +} + +pub struct ForceUnlockHandler {} + +#[async_trait::async_trait] +impl Operation for ForceUnlockHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::ForceUnlockAdminAction).await?; + + // Resources may arrive as a JSON body or as repeated `resource` query params. + let mut resources: Vec = query_values(&req.uri, "resource"); + let body = read_body(req.input).await?; + if !body.is_empty() { + let parsed: ForceUnlockRequest = serde_json::from_slice(&body) + .map_err(|e| s3_error!(InvalidRequest, "invalid force-unlock request body: {}", e))?; + resources.extend(parsed.resources); + resources.extend(parsed.resource); + } + + if resources.is_empty() { + return Err(s3_error!(InvalidRequest, "at least one resource is required")); + } + + let manager = get_global_lock_manager(); + let Some(fast) = manager.as_fast_lock_manager() else { + let response = ForceUnlockResponse { + results: Vec::new(), + total_released: 0, + capability_note: Some( + "namespace lock subsystem is disabled (RUSTFS_LOCK_ENABLED=false); nothing to unlock".to_string(), + ), + }; + return json_response(StatusCode::OK, &response); + }; + + let mut results = Vec::with_capacity(resources.len()); + let mut total_released = 0usize; + for raw in resources { + let Some(key) = parse_resource(&raw) else { + return Err(s3_error!(InvalidRequest, "invalid resource: {}", raw)); + }; + let released = fast.force_unlock(&key); + total_released += released; + results.push(ForceUnlockResult { + resource: raw, + released_owners: released, + }); + } + + let response = ForceUnlockResponse { + results, + total_released, + capability_note: None, + }; + json_response(StatusCode::OK, &response) + } +} + +// --------------------------------------------------------------------------- +// #607 GET /v3/healthinfo and /v3/obdinfo — real host + storage telemetry +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize)] +struct HealthCpuInfo { + logical_cores: usize, + brand: String, + frequency_mhz: u64, + usage_percent: f64, +} + +#[derive(Debug, Clone, Serialize)] +struct HealthMemInfo { + total_bytes: u64, + used_bytes: u64, + available_bytes: u64, + total_swap_bytes: u64, + used_swap_bytes: u64, +} + +#[derive(Debug, Clone, Serialize)] +struct HealthOsInfo { + os: String, + kernel_version: Option, + os_version: Option, + hostname: Option, + arch: String, + uptime_secs: u64, +} + +#[derive(Debug, Clone, Serialize)] +struct HealthProcInfo { + pid: u32, + cpu_usage_percent: f32, + memory_bytes: u64, +} + +#[derive(Debug, Clone, Serialize)] +struct HealthDriveInfo { + endpoint: String, + drive_path: String, + state: String, + total_space: u64, + used_space: u64, + available_space: u64, + read_throughput: f64, + write_throughput: f64, + read_latency: f64, + write_latency: f64, +} + +#[derive(Debug, Clone, Serialize)] +struct HealthInfoResponse { + version: String, + deployment_id: Option, + region: Option, + timestamp: Option, + cpu: HealthCpuInfo, + memory: HealthMemInfo, + os: HealthOsInfo, + process: HealthProcInfo, + drives: Vec, + /// Reserved MinIO health families (perf/net/config obd probes) that RustFS + /// does not yet collect are enumerated here so clients can tell an + /// unsupported probe apart from an empty result. + unsupported_probes: Vec<&'static str>, +} + +async fn collect_health_info() -> HealthInfoResponse { + use sysinfo::{Pid, System}; + + let mut sys = System::new_all(); + sys.refresh_cpu_all(); + // A second sample after a short interval yields meaningful CPU usage. + tokio::time::sleep(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL).await; + sys.refresh_cpu_all(); + sys.refresh_memory(); + + let logical_cores = sys.cpus().len(); + let cpu = HealthCpuInfo { + logical_cores, + brand: sys.cpus().first().map(|c| c.brand().to_string()).unwrap_or_default(), + frequency_mhz: sys.cpus().first().map(|c| c.frequency()).unwrap_or(0), + usage_percent: if logical_cores > 0 { + sys.cpus().iter().map(|c| c.cpu_usage() as f64).sum::() / logical_cores as f64 + } else { + 0.0 + }, + }; + + let memory = HealthMemInfo { + total_bytes: sys.total_memory(), + used_bytes: sys.used_memory(), + available_bytes: sys.available_memory(), + total_swap_bytes: sys.total_swap(), + used_swap_bytes: sys.used_swap(), + }; + + let os = HealthOsInfo { + os: std::env::consts::OS.to_string(), + kernel_version: System::kernel_version(), + os_version: System::long_os_version(), + hostname: System::host_name(), + arch: std::env::consts::ARCH.to_string(), + uptime_secs: System::uptime(), + }; + + let pid = std::process::id(); + let process = sys + .process(Pid::from_u32(pid)) + .map(|p| HealthProcInfo { + pid, + cpu_usage_percent: p.cpu_usage(), + memory_bytes: p.memory(), + }) + .unwrap_or(HealthProcInfo { + pid, + cpu_usage_percent: 0.0, + memory_bytes: 0, + }); + + let drives = collect_drive_info().await; + + HealthInfoResponse { + version: crate::version::get_version(), + deployment_id: crate::admin::runtime_sources::current_deployment_id(), + region: crate::admin::runtime_sources::current_region().map(|r| r.as_str().to_string()), + timestamp: system_time_to_rfc3339(SystemTime::now()), + cpu, + memory, + os, + process, + drives, + unsupported_probes: vec!["perf-net", "perf-drive-obd", "config-obd", "sys-services"], + } +} + +async fn collect_drive_info() -> Vec { + use crate::admin::runtime_sources::{DefaultAdminUsecase, default_admin_usecase}; + let usecase: DefaultAdminUsecase = default_admin_usecase(); + match usecase.execute_query_storage_info().await { + Ok(info) => info + .disks + .into_iter() + .map(|d| HealthDriveInfo { + endpoint: d.endpoint, + drive_path: d.drive_path, + state: d.state, + total_space: d.total_space, + used_space: d.used_space, + available_space: d.available_space, + read_throughput: d.read_throughput, + write_throughput: d.write_throughput, + read_latency: d.read_latency, + write_latency: d.write_latency, + }) + .collect(), + Err(err) => { + warn!(error = %err, "healthinfo: storage info unavailable"); + Vec::new() + } + } +} + +pub struct HealthInfoHandler {} + +#[async_trait::async_trait] +impl Operation for HealthInfoHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::HealthInfoAdminAction).await?; + let response = collect_health_info().await; + json_response(StatusCode::OK, &response) + } +} + +// --------------------------------------------------------------------------- +// #607 GET /v3/log — MinIO-compatible streaming log endpoint +// --------------------------------------------------------------------------- + +struct ByteChannelStream { + inner: ReceiverStream>, +} + +impl Stream for ByteChannelStream { + type Item = Result; + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::into_inner(self).inner.poll_next_unpin(cx) + } +} + +impl ByteStream for ByteChannelStream {} + +/// A single console-log record, shaped after MinIO's `LogInfo`. +#[derive(Debug, Clone, Serialize)] +struct LogInfo { + node_name: String, + #[serde(rename = "consoleMsg")] + console_msg: String, + level: String, + time: Option, + /// Distinguishes the honest keep-alive stream from real records. + err: Option, +} + +pub struct ConsoleLogHandler {} + +#[async_trait::async_trait] +impl Operation for ConsoleLogHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::ConsoleLogAdminAction).await?; + + // RustFS routes logs to the tracing pipeline (stdout / OTLP sinks) and + // does NOT maintain an in-process ring buffer that could be replayed to + // an admin client. Rather than fake historical log lines, this honors + // the MinIO streaming contract (chunked NDJSON of `LogInfo`) and emits a + // single capability record, then a keep-alive heartbeat, so `mc admin + // logs` connects and stays open without being misled into believing it + // received real buffered logs. + let node_name = sysinfo::System::host_name().unwrap_or_else(|| "rustfs".to_string()); + let (tx, rx) = mpsc::channel::>(8); + + spawn_traced(async move { + let notice = LogInfo { + node_name: node_name.clone(), + console_msg: "RustFS does not expose an in-process console-log buffer; live log streaming is not yet \ + available. Configure a tracing/OTLP sink to collect logs." + .to_string(), + level: "INFO".to_string(), + time: system_time_to_rfc3339(SystemTime::now()), + err: Some("log_streaming_unsupported".to_string()), + }; + if let Ok(mut encoded) = serde_json::to_vec(¬ice) { + encoded.push(b'\n'); + if tx.send(Ok(Bytes::from(encoded))).await.is_err() { + return; + } + } + + let mut ticker = tokio::time::interval(Duration::from_secs(15)); + ticker.tick().await; + loop { + tokio::select! { + _ = tx.closed() => break, + _ = ticker.tick() => { + // Whitespace keep-alive keeps the NDJSON stream open. + if tx.send(Ok(Bytes::from_static(b" \n"))).await.is_err() { + break; + } + } + } + } + }); + + let stream: DynByteStream = Box::pin(ByteChannelStream { + inner: ReceiverStream::new(rx), + }); + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static(CONTENT_TYPE_NDJSON)); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(stream)), headers)) + } +} + +// --------------------------------------------------------------------------- +// #615 POST /v3/speedtest family +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SpeedtestKind { + Object, + Drive, + Net, + Site, +} + +fn speedtest_kind_from_path(path: &str) -> SpeedtestKind { + if path.ends_with("/speedtest/drive") { + SpeedtestKind::Drive + } else if path.ends_with("/speedtest/net") { + SpeedtestKind::Net + } else if path.ends_with("/speedtest/site") { + SpeedtestKind::Site + } else { + // `/speedtest` and `/speedtest/object` both mean the object throughput test. + SpeedtestKind::Object + } +} + +#[derive(Debug, Clone, Serialize)] +struct DriveSpeedtestEntry { + endpoint: String, + drive_path: String, + state: String, + read_throughput_bytes_per_sec: f64, + write_throughput_bytes_per_sec: f64, + read_latency_secs: f64, + write_latency_secs: f64, +} + +#[derive(Debug, Clone, Serialize)] +struct SpeedtestResponse { + kind: &'static str, + /// `true` when the reported numbers come from a real measurement/observation + /// on this node; `false` when the endpoint returns MinIO-compatible + /// structure only (see `capability_note`). + measured: bool, + #[serde(skip_serializing_if = "Option::is_none")] + capability_note: Option, + /// Aggregate throughput in bytes/sec across the sampled drives (drive test). + #[serde(skip_serializing_if = "Option::is_none")] + aggregate_read_throughput_bytes_per_sec: Option, + #[serde(skip_serializing_if = "Option::is_none")] + aggregate_write_throughput_bytes_per_sec: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + drives: Vec, + /// Bytes drained by the net/devnull probe and how long it took. + #[serde(skip_serializing_if = "Option::is_none")] + rx_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + duration_secs: Option, +} + +async fn run_drive_speedtest() -> S3Result { + use crate::admin::runtime_sources::default_admin_usecase; + let usecase = default_admin_usecase(); + let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?; + + let mut drives = Vec::with_capacity(info.disks.len()); + let mut agg_read = 0.0f64; + let mut agg_write = 0.0f64; + for d in info.disks { + agg_read += d.read_throughput; + agg_write += d.write_throughput; + drives.push(DriveSpeedtestEntry { + endpoint: d.endpoint, + drive_path: d.drive_path, + state: d.state, + read_throughput_bytes_per_sec: d.read_throughput, + write_throughput_bytes_per_sec: d.write_throughput, + read_latency_secs: d.read_latency, + write_latency_secs: d.write_latency, + }); + } + + Ok(SpeedtestResponse { + kind: "drive", + measured: true, + capability_note: Some( + "drive throughput/latency is reported from live per-drive StorageInfo observations rather than a \ + synthetic write/read benchmark" + .to_string(), + ), + aggregate_read_throughput_bytes_per_sec: Some(agg_read), + aggregate_write_throughput_bytes_per_sec: Some(agg_write), + drives, + rx_bytes: None, + duration_secs: None, + }) +} + +fn object_speedtest_unsupported() -> SpeedtestResponse { + SpeedtestResponse { + kind: "object", + measured: false, + capability_note: Some( + "object PUT/GET speedtest requires a cross-node benchmark harness and a scratch bucket lifecycle that \ + RustFS does not yet expose from the admin layer; use the drive speedtest (/v3/speedtest/drive) for live \ + per-drive throughput" + .to_string(), + ), + aggregate_read_throughput_bytes_per_sec: None, + aggregate_write_throughput_bytes_per_sec: None, + drives: Vec::new(), + rx_bytes: None, + duration_secs: None, + } +} + +fn net_speedtest_single_node() -> SpeedtestResponse { + SpeedtestResponse { + kind: "net", + measured: false, + capability_note: Some( + "network speedtest measures inter-node bandwidth; a distributed peer-perf harness is not yet wired. See \ + /v3/site-replication/netperf for the site-to-site variant" + .to_string(), + ), + aggregate_read_throughput_bytes_per_sec: None, + aggregate_write_throughput_bytes_per_sec: None, + drives: Vec::new(), + rx_bytes: None, + duration_secs: None, + } +} + +pub struct SpeedtestHandler {} + +#[async_trait::async_trait] +impl Operation for SpeedtestHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + // MinIO gates speedtest behind the health-info (OBD) admin action. + authorize(&req, AdminAction::HealthInfoAdminAction).await?; + let kind = speedtest_kind_from_path(req.uri.path()); + // Drain any request body so the connection is not left half-read. + let _ = read_body(req.input).await?; + + let response = match kind { + SpeedtestKind::Drive => run_drive_speedtest().await?, + SpeedtestKind::Object => object_speedtest_unsupported(), + SpeedtestKind::Net => net_speedtest_single_node(), + SpeedtestKind::Site => SpeedtestResponse { + kind: "site", + measured: false, + capability_note: Some( + "site speedtest aggregates peer results across a replicated deployment; not yet wired".to_string(), + ), + aggregate_read_throughput_bytes_per_sec: None, + aggregate_write_throughput_bytes_per_sec: None, + drives: Vec::new(), + rx_bytes: None, + duration_secs: None, + }, + }; + json_response(StatusCode::OK, &response) + } +} + +/// `POST /v3/speedtest/client/devnull` — real client-to-server upload drain. +/// +/// The client streams data; the server discards it and reports how much it +/// received and how long it took, giving a genuine one-way upload throughput +/// number (mirrors MinIO's `ClientDevNull`). +pub struct SpeedtestClientDevnullHandler {} + +#[async_trait::async_trait] +impl Operation for SpeedtestClientDevnullHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::HealthInfoAdminAction).await?; + + let started = std::time::Instant::now(); + let mut input = req.input; + let mut total: u64 = 0; + while let Some(chunk) = input.next().await { + let chunk = chunk.map_err(|e| s3_error!(InvalidRequest, "failed to read devnull stream: {}", e))?; + total += chunk.len() as u64; + } + let elapsed = started.elapsed(); + + let response = SpeedtestResponse { + kind: "client-devnull", + measured: true, + capability_note: None, + aggregate_read_throughput_bytes_per_sec: None, + aggregate_write_throughput_bytes_per_sec: Some(if elapsed.as_secs_f64() > 0.0 { + total as f64 / elapsed.as_secs_f64() + } else { + 0.0 + }), + drives: Vec::new(), + rx_bytes: Some(total), + duration_secs: Some(elapsed.as_secs_f64()), + }; + json_response(StatusCode::OK, &response) + } +} + +// --------------------------------------------------------------------------- +// Query helpers +// --------------------------------------------------------------------------- + +fn query_value(uri: &Uri, key: &str) -> Option { + uri.query().and_then(|q| { + url::form_urlencoded::parse(q.as_bytes()).find_map(|(k, v)| if k == key { Some(v.into_owned()) } else { None }) + }) +} + +fn query_values(uri: &Uri, key: &str) -> Vec { + uri.query() + .map(|q| { + url::form_urlencoded::parse(q.as_bytes()) + .filter_map(|(k, v)| if k == key { Some(v.into_owned()) } else { None }) + .collect() + }) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + use http::{Extensions, Uri}; + + fn build_request(method: Method, uri: &'static str) -> S3Request { + S3Request { + input: Body::empty(), + method, + uri: Uri::from_static(uri), + headers: HeaderMap::new(), + extensions: Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + } + } + + #[tokio::test] + async fn top_locks_handler_rejects_missing_credentials() { + let err = TopLocksHandler {} + .call(build_request(Method::GET, "/rustfs/admin/v3/top/locks"), Params::new()) + .await + .expect_err("must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn force_unlock_handler_rejects_missing_credentials() { + let err = ForceUnlockHandler {} + .call(build_request(Method::POST, "/rustfs/admin/v3/force-unlock"), Params::new()) + .await + .expect_err("must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn health_info_handler_rejects_missing_credentials() { + let err = HealthInfoHandler {} + .call(build_request(Method::GET, "/rustfs/admin/v3/healthinfo"), Params::new()) + .await + .expect_err("must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn speedtest_handler_rejects_missing_credentials() { + let err = SpeedtestHandler {} + .call(build_request(Method::POST, "/rustfs/admin/v3/speedtest/drive"), Params::new()) + .await + .expect_err("must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn speedtest_kind_routing() { + assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest"), SpeedtestKind::Object); + assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest/object"), SpeedtestKind::Object); + assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest/drive"), SpeedtestKind::Drive); + assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest/net"), SpeedtestKind::Net); + assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest/site"), SpeedtestKind::Site); + } + + #[test] + fn parse_resource_variants() { + let key = parse_resource("bucket/object").expect("bucket/object"); + assert_eq!(key.bucket.as_ref(), "bucket"); + assert_eq!(key.object.as_ref(), "object"); + + let key = parse_resource("/leading/slash/object").expect("nested"); + assert_eq!(key.bucket.as_ref(), "leading"); + assert_eq!(key.object.as_ref(), "slash/object"); + + let key = parse_resource("bucketonly").expect("bucket only"); + assert_eq!(key.bucket.as_ref(), "bucketonly"); + assert_eq!(key.object.as_ref(), "bucketonly"); + + assert!(parse_resource(" ").is_none()); + assert!(parse_resource("/").is_none()); + } + + #[test] + fn top_locks_limit_parsing() { + assert_eq!(parse_top_locks_limit(&Uri::from_static("/x")), TOP_LOCKS_DEFAULT_LIMIT); + assert_eq!(parse_top_locks_limit(&Uri::from_static("/x?count=5")), 5); + assert_eq!(parse_top_locks_limit(&Uri::from_static("/x?count=0")), TOP_LOCKS_DEFAULT_LIMIT); + assert_eq!(parse_top_locks_limit(&Uri::from_static("/x?count=999999")), TOP_LOCKS_MAX_LIMIT); + } + + #[tokio::test] + async fn collect_top_locks_reports_live_lock() { + // Acquire a real lock through the global manager and confirm it surfaces. + let manager = get_global_lock_manager(); + let key = ObjectKey::new("diag-bucket", "diag-object"); + // The fast-lock manager exposes the acquire API; if the lock subsystem is + // disabled in this environment, the response must carry a capability note. + let Some(fast) = manager.as_fast_lock_manager() else { + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + assert!(response.capability_note.is_some() || response.locks.is_empty()); + return; + }; + let guard = match fast.acquire_write_lock(key.clone(), "diag-owner").await { + Ok(g) => g, + Err(_) => { + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + assert!(response.capability_note.is_some() || response.locks.is_empty()); + return; + } + }; + + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + let found = response + .locks + .iter() + .any(|l| l.bucket == "diag-bucket" && l.object == "diag-object"); + assert!(found, "expected the held lock to appear in top locks"); + let entry = response + .locks + .iter() + .find(|l| l.bucket == "diag-bucket") + .expect("entry present"); + assert_eq!(entry.lock_type, "WRITE"); + assert_eq!(entry.owner, "diag-owner"); + + drop(guard); + } +} diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 2e92343e0..0d8c884ab 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -19,6 +19,7 @@ pub mod batch_job; pub mod bucket_meta; pub mod cluster_snapshot; pub mod config_admin; +pub mod diagnostics; pub mod event; pub mod extensions; pub mod group; diff --git a/rustfs/src/admin/handlers/profile_admin.rs b/rustfs/src/admin/handlers/profile_admin.rs index 029e4f58c..d4a85360a 100644 --- a/rustfs/src/admin/handlers/profile_admin.rs +++ b/rustfs/src/admin/handlers/profile_admin.rs @@ -13,14 +13,27 @@ // limitations under the License. use super::profile::{authorize_profile_request, profile_not_implemented_response}; +use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::server::ADMIN_PREFIX; +use crate::admin::storage_api::access::spawn_traced; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use bytes::Bytes; +use futures::{Stream, StreamExt}; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; +use rustfs_madmin::service_commands::ServiceTraceOpts; +use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; -use s3s::{Body, S3Request, S3Response, S3Result}; +use s3s::stream::{ByteStream, DynByteStream}; +use s3s::{Body, S3Request, S3Response, S3Result, StdError, s3_error}; use serde::Serialize; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; use tracing::error; #[derive(Serialize)] @@ -44,9 +57,42 @@ pub fn register_profiling_route(r: &mut S3Router) -> std::io::Re AdminOperation(&ProfileStatusHandler {}), )?; + // MinIO-compatible profiling / trace family (#606). + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/profiling/start").as_str(), + AdminOperation(&ProfilingStartHandler {}), + )?; + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/profiling/download").as_str(), + AdminOperation(&ProfilingDownloadHandler {}), + )?; + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/profile").as_str(), + AdminOperation(&ProfileControlHandler {}), + )?; + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/trace").as_str(), + AdminOperation(&TraceHandler {}), + )?; + Ok(()) } +/// Authorize a request against a single admin action (profiling or trace). +async fn authorize_action(req: &S3Request, action: AdminAction) -> S3Result<()> { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(AccessDenied, "Signature is required")); + }; + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await +} + pub struct ProfileHandler {} #[async_trait::async_trait] @@ -94,9 +140,141 @@ impl Operation for ProfileStatusHandler { } } +/// `POST /v3/profiling/start` — begin a profiling session. +/// +/// RustFS builds with the mimalloc allocator and ships no in-process pprof/CPU +/// sampler (see `crate::profiling`): profiling is exported out-of-process via +/// Pyroscope. We therefore honor the MinIO request shape but return +/// `501 Not Implemented` with a clear reason rather than pretending to have +/// started a capture that will never produce data. +pub struct ProfilingStartHandler {} + +#[async_trait::async_trait] +impl Operation for ProfilingStartHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_action(&req, AdminAction::ProfilingAdminAction).await?; + crate::profiling::log_cpu_pprof_dump_skipped(); + Ok(profile_not_implemented_response(format!( + "profiling start is not supported: {}", + crate::profiling::local_cpu_pprof_unsupported_message() + ))) + } +} + +/// `GET /v3/profiling/download` — download the captured profile archive. +/// +/// No capture is ever produced (see `ProfilingStartHandler`), so there is +/// nothing to download. +pub struct ProfilingDownloadHandler {} + +#[async_trait::async_trait] +impl Operation for ProfilingDownloadHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_action(&req, AdminAction::ProfilingAdminAction).await?; + Ok(profile_not_implemented_response(format!( + "profiling download is not supported: {}", + crate::profiling::local_cpu_pprof_unsupported_message() + ))) + } +} + +/// `POST /v3/profile` — start/stop profiling in one call (legacy MinIO shape). +pub struct ProfileControlHandler {} + +#[async_trait::async_trait] +impl Operation for ProfileControlHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_action(&req, AdminAction::ProfilingAdminAction).await?; + crate::profiling::log_cpu_pprof_dump_skipped(); + Ok(profile_not_implemented_response(format!( + "in-process profiling is not supported: {}", + crate::profiling::local_cpu_pprof_unsupported_message() + ))) + } +} + +struct TraceStream { + inner: ReceiverStream>, +} + +impl Stream for TraceStream { + type Item = Result; + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::into_inner(self).inner.poll_next_unpin(cx) + } +} + +impl ByteStream for TraceStream {} + +/// `GET /v3/trace` — stream real-time server trace events. +/// +/// RustFS emits diagnostics through the `tracing` pipeline but does not expose +/// an in-process subscriber that can fan trace events out to an admin client +/// (there is no request-trace broadcast channel). Rather than return an opaque +/// `501` — which would make `mc admin trace` fail to connect — this honors the +/// streaming NDJSON contract: it validates the requested trace filters, opens +/// the stream, emits a single capability record explaining that live tracing is +/// not wired, then holds the connection open with keep-alives. No fabricated +/// trace records are ever sent. +pub struct TraceHandler {} + +#[async_trait::async_trait] +impl Operation for TraceHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_action(&req, AdminAction::TraceAdminAction).await?; + + // Validate the trace options so malformed filters are rejected up front, + // matching MinIO's behavior. + let mut opts = ServiceTraceOpts::default(); + opts.parse_params(&req.uri) + .map_err(|_| s3_error!(InvalidRequest, "invalid trace parameters"))?; + + let node_name = sysinfo::System::host_name().unwrap_or_else(|| "rustfs".to_string()); + let (tx, rx) = mpsc::channel::>(8); + + spawn_traced(async move { + let notice = serde_json::json!({ + "nodename": node_name, + "funcname": "admin.Trace", + "msg": "RustFS does not expose an in-process trace-event subscriber; live tracing is not yet available", + "err": "trace_streaming_unsupported", + }); + if let Ok(mut encoded) = serde_json::to_vec(¬ice) { + encoded.push(b'\n'); + if tx.send(Ok(Bytes::from(encoded))).await.is_err() { + return; + } + } + + let mut ticker = tokio::time::interval(Duration::from_secs(15)); + ticker.tick().await; + loop { + tokio::select! { + _ = tx.closed() => break, + _ = ticker.tick() => { + if tx.send(Ok(Bytes::from_static(b" \n"))).await.is_err() { + break; + } + } + } + } + }); + + let stream: DynByteStream = Box::pin(TraceStream { + inner: ReceiverStream::new(rx), + }); + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/x-ndjson")); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(stream)), headers)) + } +} + #[cfg(test)] mod tests { - use super::{ProfileHandler, ProfileStatusHandler}; + use super::{ + ProfileControlHandler, ProfileHandler, ProfileStatusHandler, ProfilingDownloadHandler, ProfilingStartHandler, + TraceHandler, + }; use crate::admin::router::Operation; use http::{Extensions, HeaderMap, Uri}; use hyper::Method; @@ -144,4 +322,40 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::AccessDenied); assert_eq!(err.message(), Some("Signature is required")); } + + #[tokio::test] + async fn profiling_start_rejects_missing_credentials() { + let err = ProfilingStartHandler {} + .call(build_profile_request("/rustfs/admin/v3/profiling/start"), Params::new()) + .await + .expect_err("profiling start must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn profiling_download_rejects_missing_credentials() { + let err = ProfilingDownloadHandler {} + .call(build_profile_request("/rustfs/admin/v3/profiling/download"), Params::new()) + .await + .expect_err("profiling download must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn profile_control_rejects_missing_credentials() { + let err = ProfileControlHandler {} + .call(build_profile_request("/rustfs/admin/v3/profile"), Params::new()) + .await + .expect_err("profile control must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn trace_handler_rejects_missing_credentials() { + let err = TraceHandler {} + .call(build_profile_request("/rustfs/admin/v3/trace?s3=true"), Params::new()) + .await + .expect_err("trace must reject anonymous requests"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } } diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index cd65310f8..8413b967e 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -15,6 +15,7 @@ use super::{cluster_snapshot, metrics}; use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::current_object_store_handle; use crate::admin::runtime_sources::{ DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, }; @@ -34,8 +35,17 @@ use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::Serialize; +use std::sync::atomic::{AtomicBool, Ordering}; use tracing::{error, info, warn}; +/// Global service-freeze flag toggled by `POST /v3/service?action=freeze`. +/// +/// NOTE: RustFS does not currently route S3 request admission through this flag, +/// so freezing is advisory only — it records intent and is reflected in the +/// response, but does not actually suspend request handling. This is documented +/// in the handler and surfaced to the caller via `frozen_effective=false`. +static SERVICE_FROZEN: AtomicBool = AtomicBool::new(false); + const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; const LOG_SUBSYSTEM_SYSTEM_ADMIN: &str = "system_admin"; const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; @@ -95,6 +105,12 @@ pub fn register_system_route(r: &mut S3Router) -> std::io::Resul AdminOperation(&ServiceHandle {}), )?; + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/update").as_str(), + AdminOperation(&UpdateHandler {}), + )?; + r.insert( Method::GET, format!("{}{}", ADMIN_PREFIX, "/v3/info").as_str(), @@ -141,11 +157,232 @@ pub fn register_system_route(r: &mut S3Router) -> std::io::Resul } pub struct ServiceHandle {} +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ServiceAction { + Restart, + Stop, + Freeze, + Unfreeze, +} + +impl ServiceAction { + fn parse(raw: &str) -> Option { + match raw.trim().to_ascii_lowercase().as_str() { + "restart" => Some(Self::Restart), + "stop" => Some(Self::Stop), + "freeze" => Some(Self::Freeze), + "unfreeze" => Some(Self::Unfreeze), + _ => None, + } + } +} + +fn service_action_from_uri(uri: &http::Uri) -> Option { + uri.query().and_then(|q| { + url::form_urlencoded::parse(q.as_bytes()).find_map(|(k, v)| if k == "action" { ServiceAction::parse(&v) } else { None }) + }) +} + +#[derive(Serialize)] +struct ServiceActionResponse { + action: &'static str, + accepted: bool, + /// Whether the action takes real effect on this build (vs advisory only). + effective: bool, + message: &'static str, +} + +/// Ask the process to shut down gracefully by raising SIGTERM to itself. +/// +/// The existing `wait_for_shutdown()` signal handler observes SIGTERM and runs +/// the full graceful-shutdown sequence (drains servers, flushes audit/event +/// notifiers). RustFS has no in-process supervisor that re-execs the binary, so +/// `restart` and `stop` are both honored as a graceful stop; a process manager +/// (systemd, k8s) is responsible for restarting the binary. This is documented +/// in the response so operators are not misled into expecting an in-process +/// re-exec. +// SAFETY: the only unsafe operation is `libc::raise(SIGTERM)`, which delivers a +// signal to the current process using a compile-time constant signal number and +// no pointers. It simply routes into the existing `wait_for_shutdown()` SIGTERM +// handler that runs the graceful-shutdown sequence. +#[cfg(unix)] +#[allow(unsafe_code)] +fn request_graceful_shutdown() { + // Defer slightly so the HTTP response can flush before shutdown begins. + tokio::spawn(async { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + // SAFETY: see the function-level comment; `libc::raise` with a constant + // signal number is async-signal-safe and takes no pointer arguments. + unsafe { + libc::raise(libc::SIGTERM); + } + }); +} + +#[cfg(not(unix))] +fn request_graceful_shutdown() {} + #[async_trait::async_trait] impl Operation for ServiceHandle { - async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - log_system_request_rejected!("service_handle", "not_implemented"); - Err(s3_error!(NotImplemented)) + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let Some(input_cred) = req.credentials.as_ref() else { + log_system_request_rejected!("service_handle", "missing_credentials"); + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + + let Some(action) = service_action_from_uri(&req.uri) else { + log_system_request_rejected!("service_handle", "invalid_action"); + return Err(s3_error!(InvalidRequest, "action must be one of: restart, stop, freeze, unfreeze")); + }; + + // restart/stop are destructive; freeze/unfreeze map to their own actions. + let admin_action = match action { + ServiceAction::Restart => AdminAction::ServiceRestartAdminAction, + ServiceAction::Stop => AdminAction::ServiceStopAdminAction, + ServiceAction::Freeze | ServiceAction::Unfreeze => AdminAction::ServiceFreezeAdminAction, + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(admin_action)], remote_addr).await?; + + let response = match action { + ServiceAction::Restart => { + info!( + event = "service_control", + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + action = "restart", + "admin requested service restart; initiating graceful shutdown (process manager must relaunch)" + ); + request_graceful_shutdown(); + ServiceActionResponse { + action: "restart", + accepted: true, + effective: true, + message: "graceful shutdown initiated; the supervising process manager is responsible for relaunch", + } + } + ServiceAction::Stop => { + info!( + event = "service_control", + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + action = "stop", + "admin requested service stop; initiating graceful shutdown" + ); + request_graceful_shutdown(); + ServiceActionResponse { + action: "stop", + accepted: true, + effective: true, + message: "graceful shutdown initiated", + } + } + ServiceAction::Freeze => { + SERVICE_FROZEN.store(true, Ordering::SeqCst); + warn!( + event = "service_control", + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + action = "freeze", + "service freeze flag set; note request admission is not yet gated by this flag" + ); + ServiceActionResponse { + action: "freeze", + accepted: true, + effective: false, + message: "freeze flag recorded, but RustFS does not yet gate request admission on it (advisory only)", + } + } + ServiceAction::Unfreeze => { + SERVICE_FROZEN.store(false, Ordering::SeqCst); + info!( + event = "service_control", + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + action = "unfreeze", + "service freeze flag cleared" + ); + ServiceActionResponse { + action: "unfreeze", + accepted: true, + effective: false, + message: "freeze flag cleared (advisory only)", + } + } + }; + + let data = serde_json::to_vec(&response).map_err(|e| { + log_system_request_failed!("service_handle", "serialize_service_response_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize service response") + })?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("service_handle"); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) + } +} + +pub struct UpdateHandler {} + +#[derive(Serialize)] +struct ServerUpdateStatus { + current_version: String, + updated_version: String, + /// Always false: RustFS ships no in-process binary self-update mechanism. + update_applied: bool, + message: &'static str, +} + +#[async_trait::async_trait] +impl Operation for UpdateHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let Some(input_cred) = req.credentials.as_ref() else { + log_system_request_rejected!("server_update", "missing_credentials"); + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::ServerUpdateAdminAction)], + remote_addr, + ) + .await?; + + // MinIO's server-update downloads and swaps the binary in place. RustFS + // intentionally does not implement in-process self-update: binaries are + // managed by the packaging/orchestration layer (container image, systemd + // unit, package manager). We honor the request/response contract and + // report that no update was applied rather than faking success. + let current = crate::version::get_version(); + let response = ServerUpdateStatus { + current_version: current.clone(), + updated_version: current, + update_applied: false, + message: "in-process self-update is not supported; manage the RustFS binary via your image/package/orchestrator", + }; + info!( + event = "server_update", + component = LOG_COMPONENT_ADMIN_API, + subsystem = LOG_SUBSYSTEM_SYSTEM_ADMIN, + "server update requested; self-update unsupported, returning MinIO-compatible no-op status" + ); + + let data = serde_json::to_vec(&response).map_err(|e| { + log_system_request_failed!("server_update", "serialize_update_status_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize update status") + })?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("server_update"); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } } @@ -229,11 +466,101 @@ impl Operation for ServerInfoHandler { pub struct InspectDataHandler {} +/// Upper bound on how many bytes a single inspect-data response streams, to keep +/// an operator-facing diagnostic from materializing an arbitrarily large object. +const INSPECT_DATA_MAX_BYTES: usize = 64 * 1024 * 1024; + +fn inspect_data_target(uri: &http::Uri) -> Option<(String, String)> { + let mut volume: Option = None; + let mut file: Option = None; + if let Some(query) = uri.query() { + for (k, v) in url::form_urlencoded::parse(query.as_bytes()) { + match k.as_ref() { + // Accept MinIO's `volume`/`file` names and the friendlier + // `bucket`/`object` aliases. + "volume" | "bucket" => volume = Some(v.into_owned()), + "file" | "object" | "prefix" => file = Some(v.into_owned()), + _ => {} + } + } + } + match (volume, file) { + (Some(v), Some(f)) if !v.is_empty() && !f.is_empty() => Some((v, f)), + _ => None, + } +} + #[async_trait::async_trait] impl Operation for InspectDataHandler { - async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - log_system_request_rejected!("inspect_data", "not_implemented"); - Err(s3_error!(NotImplemented)) + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + use crate::admin::storage_api::contract::object::ObjectIO as _; + use crate::admin::storage_api::object::StorageObjectOptions; + use tokio::io::AsyncReadExt; + + let Some(input_cred) = req.credentials.as_ref() else { + log_system_request_rejected!("inspect_data", "missing_credentials"); + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::InspectDataAction)], + remote_addr, + ) + .await?; + + // MinIO's inspect-data exports a signed archive of raw drive files for a + // `volume`/`file` glob. RustFS erasure-codes and (optionally) encrypts + // object data across drives, so there is no single on-disk file to hand + // back; instead we return the reconstructed raw object bytes for the + // requested `volume` (bucket) + `file` (object), which is the honest, + // usable form of "inspect this object's data" against an EC store. + let Some((bucket, object)) = inspect_data_target(&req.uri) else { + return Err(s3_error!( + InvalidRequest, + "inspect-data requires `volume` (bucket) and `file` (object) query parameters" + )); + }; + + let Some(store) = current_object_store_handle() else { + log_system_request_failed!("inspect_data", "object_store_unavailable", "not initialized"); + return Err(S3Error::with_message(S3ErrorCode::InternalError, "object store is not initialized")); + }; + + let mut reader = store + .get_object_reader(&bucket, &object, None, HeaderMap::new(), &StorageObjectOptions::default()) + .await + .map_err(|err| { + log_system_request_failed!("inspect_data", "open_object_reader_failed", err); + S3Error::with_message(S3ErrorCode::NoSuchKey, format!("failed to open object `{bucket}/{object}`: {err}")) + })?; + + // Read up to the cap + 1 so we can detect and reject over-large targets. + let mut buf = Vec::new(); + let mut limited = (&mut reader).take((INSPECT_DATA_MAX_BYTES as u64) + 1); + limited + .read_to_end(&mut buf) + .await + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to read object data: {e}")))?; + if buf.len() > INSPECT_DATA_MAX_BYTES { + return Err(s3_error!( + InvalidRequest, + "object exceeds the {INSPECT_DATA_MAX_BYTES}-byte inspect-data limit; fetch it via the S3 API instead" + )); + } + + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/octet-stream")); + if let Ok(disposition) = HeaderValue::from_str(&format!("attachment; filename=\"inspect-{bucket}-{object}.bin\"")) { + header.insert(http::header::CONTENT_DISPOSITION, disposition); + } + log_system_response_emitted!("inspect_data"); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(buf)), header)) } } diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index af6677a8e..9b8f2ba7b 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -33,7 +33,7 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, batch_job, bucket_meta, cluster_snapshot, config_admin, extensions, heal, health, kms, module_switch, + audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, extensions, heal, health, kms, module_switch, object_zip_download, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user, }; @@ -82,6 +82,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() batch_job::register_batch_job_route(r)?; site_replication::register_site_replication_route(r)?; profile_admin::register_profiling_route(r)?; + diagnostics::register_diagnostics_route(r)?; tls_debug::register_tls_debug_route(r)?; kms::register_kms_route(r)?; oidc::register_oidc_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index b9aae7f27..1c77c7d46 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -21,6 +21,7 @@ const ALL_ADMIN: AdminActionRef = AdminActionRef::new("AllAdminActions"); const ADD_USER_TO_GROUP: AdminActionRef = AdminActionRef::new("AddUserToGroupAdminAction"); const ATTACH_POLICY: AdminActionRef = AdminActionRef::new("AttachPolicyAdminAction"); const CONFIG_UPDATE: AdminActionRef = AdminActionRef::new("ConfigUpdateAdminAction"); +const CONSOLE_LOG: AdminActionRef = AdminActionRef::new("ConsoleLogAdminAction"); const COMMIT_TABLE: AdminActionRef = AdminActionRef::new("CommitTableAction"); const CREATE_POLICY: AdminActionRef = AdminActionRef::new("CreatePolicyAdminAction"); const CREATE_SERVICE_ACCOUNT: AdminActionRef = AdminActionRef::new("CreateServiceAccountAdminAction"); @@ -34,6 +35,7 @@ const ENABLE_GROUP: AdminActionRef = AdminActionRef::new("EnableGroupAdminAction const ENABLE_USER: AdminActionRef = AdminActionRef::new("EnableUserAdminAction"); const EXPORT_BUCKET_METADATA: AdminActionRef = AdminActionRef::new("ExportBucketMetadataAction"); const EXPORT_IAM: AdminActionRef = AdminActionRef::new("ExportIAMAction"); +const FORCE_UNLOCK: AdminActionRef = AdminActionRef::new("ForceUnlockAdminAction"); const GET_BUCKET_TARGET: AdminActionRef = AdminActionRef::new("GetBucketTargetAction"); const GET_GROUP: AdminActionRef = AdminActionRef::new("GetGroupAdminAction"); const GET_METRICS: AdminActionRef = AdminActionRef::new("GetMetricsAction"); @@ -48,6 +50,7 @@ const GET_TABLE_METADATA: AdminActionRef = AdminActionRef::new("GetTableMetadata const GET_TABLE_METADATA_LOCATION: AdminActionRef = AdminActionRef::new("GetTableMetadataLocationAction"); const GET_TABLE_NAMESPACE: AdminActionRef = AdminActionRef::new("GetTableNamespaceAction"); const HEAL: AdminActionRef = AdminActionRef::new("HealAdminAction"); +const HEALTH_INFO: AdminActionRef = AdminActionRef::new("HealthInfoAdminAction"); const IMPORT_BUCKET_METADATA: AdminActionRef = AdminActionRef::new("ImportBucketMetadataAction"); const IMPORT_IAM: AdminActionRef = AdminActionRef::new("ImportIAMAction"); const KMS_CLEAR_CACHE: AdminActionRef = AdminActionRef::new("kms:ClearCache"); @@ -68,6 +71,7 @@ const REGISTER_TABLE: AdminActionRef = AdminActionRef::new("RegisterTableAction" const REMOVE_USER_FROM_GROUP: AdminActionRef = AdminActionRef::new("RemoveUserFromGroupAdminAction"); const RUN_TABLE_MAINTENANCE: AdminActionRef = AdminActionRef::new("RunTableMaintenanceAction"); const SERVER_INFO: AdminActionRef = AdminActionRef::new("ServerInfoAdminAction"); +const SERVER_UPDATE: AdminActionRef = AdminActionRef::new("ServerUpdateAdminAction"); const SET_BUCKET_QUOTA: AdminActionRef = AdminActionRef::new("SetBucketQuotaAdminAction"); const SET_BUCKET_TARGET: AdminActionRef = AdminActionRef::new("SetBucketTargetAction"); const SET_TABLE: AdminActionRef = AdminActionRef::new("SetTableAction"); @@ -88,6 +92,8 @@ const LIST_BATCH_JOBS: AdminActionRef = AdminActionRef::new("ListBatchJobsAction const DESCRIBE_BATCH_JOB: AdminActionRef = AdminActionRef::new("DescribeBatchJobAction"); const CANCEL_BATCH_JOB: AdminActionRef = AdminActionRef::new("CancelBatchJobAction"); const REPLICATION_DIFF: AdminActionRef = AdminActionRef::new("ReplicationDiff"); +const TOP_LOCKS: AdminActionRef = AdminActionRef::new("TopLocksAdminAction"); +const TRACE: AdminActionRef = AdminActionRef::new("TraceAdminAction"); #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum DeferredRoutePolicyReason { @@ -545,6 +551,36 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ admin(HttpMethod::Get, "/rustfs/admin/debug/pprof/profile", PROFILING, RouteRiskLevel::High), admin(HttpMethod::Get, "/rustfs/admin/debug/pprof/status", PROFILING, RouteRiskLevel::High), admin(HttpMethod::Get, "/rustfs/admin/debug/tls/status", PROFILING, RouteRiskLevel::High), + // MinIO-compatible diagnostics: lock inspection / force-unlock, health/OBD info, + // console log, and the speedtest harness family. + admin(HttpMethod::Get, "/rustfs/admin/v3/top/locks", TOP_LOCKS, RouteRiskLevel::Sensitive), + admin(HttpMethod::Post, "/rustfs/admin/v3/force-unlock", FORCE_UNLOCK, RouteRiskLevel::High), + admin(HttpMethod::Get, "/rustfs/admin/v3/healthinfo", HEALTH_INFO, RouteRiskLevel::Sensitive), + admin(HttpMethod::Get, "/rustfs/admin/v3/obdinfo", HEALTH_INFO, RouteRiskLevel::Sensitive), + admin(HttpMethod::Get, "/rustfs/admin/v3/log", CONSOLE_LOG, RouteRiskLevel::Sensitive), + admin(HttpMethod::Post, "/rustfs/admin/v3/speedtest", HEALTH_INFO, RouteRiskLevel::High), + admin(HttpMethod::Post, "/rustfs/admin/v3/speedtest/object", HEALTH_INFO, RouteRiskLevel::High), + admin(HttpMethod::Post, "/rustfs/admin/v3/speedtest/drive", HEALTH_INFO, RouteRiskLevel::High), + admin(HttpMethod::Post, "/rustfs/admin/v3/speedtest/net", HEALTH_INFO, RouteRiskLevel::High), + admin(HttpMethod::Post, "/rustfs/admin/v3/speedtest/site", HEALTH_INFO, RouteRiskLevel::High), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/speedtest/client/devnull", + HEALTH_INFO, + RouteRiskLevel::High, + ), + // MinIO-compatible profiling / trace endpoints. + admin(HttpMethod::Post, "/rustfs/admin/v3/profiling/start", PROFILING, RouteRiskLevel::High), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/profiling/download", + PROFILING, + RouteRiskLevel::Sensitive, + ), + admin(HttpMethod::Post, "/rustfs/admin/v3/profile", PROFILING, RouteRiskLevel::High), + admin(HttpMethod::Get, "/rustfs/admin/v3/trace", TRACE, RouteRiskLevel::Sensitive), + // MinIO-compatible service control: binary update. + admin(HttpMethod::Post, "/rustfs/admin/v3/update", SERVER_UPDATE, RouteRiskLevel::High), admin(HttpMethod::Post, "/rustfs/admin/v3/kms/create-key", KMS_CONFIGURE, RouteRiskLevel::High), admin(HttpMethod::Post, "/rustfs/admin/v3/kms/key/create", KMS_CONFIGURE, RouteRiskLevel::High), admin( diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 3653b9ccc..656bc0e5f 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -165,6 +165,7 @@ fn expected_admin_route_matrix() -> Vec { ), admin_route(Method::GET, "/v3/target/arns"), admin_route(Method::POST, "/v3/service"), + admin_route(Method::POST, "/v3/update"), admin_route(Method::GET, "/v3/info"), admin_route(Method::GET, "/v3/inspect-data"), admin_route(Method::POST, "/v3/inspect-data"), @@ -273,6 +274,21 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::PUT, "/v3/site-replication/repair"), admin_route(Method::GET, "/debug/pprof/profile"), admin_route(Method::GET, "/debug/pprof/status"), + admin_route(Method::POST, "/v3/profiling/start"), + admin_route(Method::GET, "/v3/profiling/download"), + admin_route(Method::POST, "/v3/profile"), + admin_route(Method::GET, "/v3/trace"), + admin_route(Method::GET, "/v3/top/locks"), + admin_route(Method::POST, "/v3/force-unlock"), + admin_route(Method::GET, "/v3/healthinfo"), + admin_route(Method::GET, "/v3/obdinfo"), + admin_route(Method::GET, "/v3/log"), + admin_route(Method::POST, "/v3/speedtest"), + admin_route(Method::POST, "/v3/speedtest/object"), + admin_route(Method::POST, "/v3/speedtest/drive"), + admin_route(Method::POST, "/v3/speedtest/net"), + admin_route(Method::POST, "/v3/speedtest/site"), + admin_route(Method::POST, "/v3/speedtest/client/devnull"), admin_route(Method::GET, "/debug/tls/status"), admin_route(Method::POST, "/v3/kms/create-key"), admin_route(Method::POST, "/v3/kms/key/create"),