From b60686eb6fb0776da5ba091647180863065490c9 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Tue, 7 Jul 2026 04:36:55 +0800 Subject: [PATCH] feat(admin): add replication diff/mrf and batch job admin endpoints (#4332) Add MinIO-compatible admin endpoints under the /v3 prefix, wired to real RustFS subsystems where they exist and documented as compatibility-only where they do not. Replication (backlog#611): - POST /v3/replication/diff: bounded on-demand scan of object versions, returning versions whose replication status is PENDING or FAILED. Wired to list_object_versions and the per-version replication status. Requires the bucket to exist and have a replication configuration. The scan is capped (REPLICATION_DIFF_MAX_SCAN) and reports IsTruncated when partial. - GET /v3/replication/mrf: reports the failed-replication backlog (MinIO's MRF concept) from the replication stats handle: aggregate failed count and size per target plus in-queue counters. RustFS records failed replications as aggregate counters and persists MRF entries without a runtime query API, so per-object MRF rows are not enumerable; PerObjectEntriesAvailable is always false to make that explicit. Batch jobs (backlog#613): - POST /v3/start-job, GET /v3/list-jobs, GET /v3/status-job, GET /v3/describe-job, DELETE /v3/cancel-job. - RustFS has no batch-job execution engine (no scheduler, queue, or worker), so these implement MinIO request/response/error semantics only and never fake execution or success: start-job parses and validates the job definition then returns NotImplemented for known job types and InvalidRequest for unknown ones; list-jobs returns an empty list; status/describe/cancel validate jobId and return a no-such-job error. All routes are registered with admin auth using existing AdminAction variants (ReplicationDiff, GetReplicationMetricsAction, StartBatchJobAction, ListBatchJobsAction, DescribeBatchJobAction, CancelBatchJobAction) and are covered by the admin route-registration matrix and new handler unit tests. Refs rustfs/backlog#611 #613 --- rustfs/src/admin/handlers/batch_job.rs | 336 ++++++++++++++++++++ rustfs/src/admin/handlers/mod.rs | 1 + rustfs/src/admin/handlers/replication.rs | 312 +++++++++++++++++- rustfs/src/admin/mod.rs | 5 +- rustfs/src/admin/route_policy.rs | 35 ++ rustfs/src/admin/route_registration_test.rs | 14 + 6 files changed, 699 insertions(+), 4 deletions(-) create mode 100644 rustfs/src/admin/handlers/batch_job.rs diff --git a/rustfs/src/admin/handlers/batch_job.rs b/rustfs/src/admin/handlers/batch_job.rs new file mode 100644 index 000000000..707f98d28 --- /dev/null +++ b/rustfs/src/admin/handlers/batch_job.rs @@ -0,0 +1,336 @@ +// 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 batch-job admin endpoints. +//! +//! RustFS does not (yet) ship a batch-job execution engine: there is no job +//! scheduler, queue, or background worker that can run `replicate`, `keyrotate`, +//! or `expire` batch jobs. These handlers therefore implement the MinIO request, +//! response, and error *semantics* only, so that MinIO-compatible admin clients +//! (mc admin batch, madmin) receive deliberate, stable answers instead of a +//! routing 404 or a silently faked success: +//! +//! * `POST /v3/start-job` parses the job definition, validates the declared job +//! type, and returns a deliberate `NotImplemented` compatibility error because +//! RustFS has no backend that can execute the job. Unknown job types are +//! rejected with `InvalidRequest`. No job is ever accepted or persisted, so no +//! execution or success is ever faked. +//! * `GET /v3/list-jobs` returns an empty job list — no batch job can be running +//! because none can be started. +//! * `GET /v3/status-job`, `GET /v3/describe-job`, and `DELETE /v3/cancel-job` +//! validate their `jobId` query parameter and return a `NoSuchJob`-style error, +//! since no job with that id can exist. +//! +//! When RustFS grows a real batch-job engine, these handlers should be rewired to +//! it; the request parsing and response shapes here are intended to stay stable. + +use crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::utils::read_compatible_admin_body; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue, Uri}; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_credentials::Credentials; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; +use std::collections::HashMap; +use tracing::warn; + +/// Job types recognised by the MinIO batch-job admin API. +/// +/// RustFS cannot execute any of these today; the enum exists so that +/// `start-job` can distinguish "known job type, unsupported backend" +/// (`NotImplemented`) from "unknown job type" (`InvalidRequest`). +const KNOWN_JOB_TYPES: &[&str] = &["replicate", "keyrotate", "expire"]; + +fn extract_query_params(uri: &Uri) -> HashMap { + let mut params = HashMap::new(); + if let Some(query) = uri.query() { + for (key, value) in url::form_urlencoded::parse(query.as_bytes()) { + params.insert(key.into_owned(), value.into_owned()); + } + } + params +} + +async fn validate_batch_job_admin_request(req: &S3Request, action: AdminAction) -> S3Result { + let Some(input_cred) = req.credentials.as_ref() else { + 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(action)], remote_addr).await?; + + Ok(cred) +} + +fn json_response(status: StatusCode, value: &T) -> S3Result> { + let data = serde_json::to_vec(value).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("{e}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + Ok(S3Response::with_headers((status, Body::from(data)), headers)) +} + +/// Best-effort detection of the declared job type from a MinIO batch-job +/// definition body. +/// +/// MinIO batch jobs are YAML documents whose top-level key names the job type, +/// e.g. `replicate:`, `keyrotate:`, or `expire:`. RustFS has no YAML dependency +/// and rejects every job anyway, so instead of pulling in a full parser we look +/// for the first recognised top-level job-type key. Returns `Some(job_type)` +/// when a known key is found, `None` otherwise. +fn detect_job_type(body: &[u8]) -> Option { + let text = std::str::from_utf8(body).ok()?; + for raw_line in text.lines() { + let line = raw_line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + // A top-level mapping key looks like `replicate:` with no leading indentation. + if raw_line.starts_with(char::is_whitespace) { + continue; + } + let Some((key, _)) = line.split_once(':') else { + continue; + }; + let key = key.trim().trim_matches('"').to_ascii_lowercase(); + if KNOWN_JOB_TYPES.contains(&key.as_str()) { + return Some(key); + } + } + None +} + +#[derive(Debug, Serialize)] +struct ListBatchJobsResult { + jobs: Vec, +} + +pub fn register_batch_job_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/start-job").as_str(), + AdminOperation(&StartBatchJobHandler {}), + )?; + + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/list-jobs").as_str(), + AdminOperation(&ListBatchJobsHandler {}), + )?; + + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/status-job").as_str(), + AdminOperation(&StatusBatchJobHandler {}), + )?; + + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/describe-job").as_str(), + AdminOperation(&DescribeBatchJobHandler {}), + )?; + + r.insert( + Method::DELETE, + format!("{}{}", ADMIN_PREFIX, "/v3/cancel-job").as_str(), + AdminOperation(&CancelBatchJobHandler {}), + )?; + + Ok(()) +} + +/// `POST /v3/start-job` +/// +/// Parses and validates the submitted job definition, then returns a deliberate +/// `NotImplemented` compatibility error: RustFS has no batch-job engine to run +/// the job. Unknown job types are rejected with `InvalidRequest`. No job is +/// accepted, persisted, or reported as started. +pub struct StartBatchJobHandler {} + +#[async_trait::async_trait] +impl Operation for StartBatchJobHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_batch_job_admin_request(&req, AdminAction::StartBatchJobAction).await?; + + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?; + + if body.trim_ascii().is_empty() { + return Err(s3_error!(InvalidRequest, "batch job definition is required")); + } + + match detect_job_type(&body) { + Some(job_type) => { + warn!( + job_type = %job_type, + "batch job start requested but RustFS has no batch-job execution backend" + ); + // Known job type, but no backend can run it: honest compatibility error. + Err(S3Error::with_message( + S3ErrorCode::NotImplemented, + format!("batch job type '{job_type}' is not supported: RustFS has no batch-job execution backend"), + )) + } + None => Err(s3_error!( + InvalidRequest, + "unable to determine batch job type from definition; expected one of: replicate, keyrotate, expire" + )), + } + } +} + +/// `GET /v3/list-jobs` +/// +/// No batch job can be running (none can be started), so this always returns an +/// empty job list. The optional `jobType` filter is validated for compatibility. +pub struct ListBatchJobsHandler {} + +#[async_trait::async_trait] +impl Operation for ListBatchJobsHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_batch_job_admin_request(&req, AdminAction::ListBatchJobsAction).await?; + + let queries = extract_query_params(&req.uri); + if let Some(job_type) = queries.get("jobType").filter(|v| !v.is_empty()) { + let job_type = job_type.to_ascii_lowercase(); + if !KNOWN_JOB_TYPES.contains(&job_type.as_str()) { + return Err(s3_error!(InvalidRequest, "invalid jobType filter: {job_type}")); + } + } + + json_response(StatusCode::OK, &ListBatchJobsResult { jobs: Vec::new() }) + } +} + +/// `GET /v3/status-job` +/// +/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style +/// error rather than a fabricated status. +pub struct StatusBatchJobHandler {} + +#[async_trait::async_trait] +impl Operation for StatusBatchJobHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_batch_job_admin_request(&req, AdminAction::DescribeBatchJobAction).await?; + let job_id = require_job_id(&req.uri)?; + Err(no_such_job(&job_id)) + } +} + +/// `GET /v3/describe-job` +/// +/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style +/// error rather than a fabricated definition. +pub struct DescribeBatchJobHandler {} + +#[async_trait::async_trait] +impl Operation for DescribeBatchJobHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_batch_job_admin_request(&req, AdminAction::DescribeBatchJobAction).await?; + let job_id = require_job_id(&req.uri)?; + Err(no_such_job(&job_id)) + } +} + +/// `DELETE /v3/cancel-job` +/// +/// No job can exist, so a well-formed `jobId` always yields a `NoSuchJob`-style +/// error rather than a fabricated cancellation. +pub struct CancelBatchJobHandler {} + +#[async_trait::async_trait] +impl Operation for CancelBatchJobHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_batch_job_admin_request(&req, AdminAction::CancelBatchJobAction).await?; + let job_id = require_job_id(&req.uri)?; + Err(no_such_job(&job_id)) + } +} + +fn require_job_id(uri: &Uri) -> S3Result { + let queries = extract_query_params(uri); + match queries.get("jobId").filter(|v| !v.is_empty()) { + Some(job_id) => Ok(job_id.clone()), + None => Err(s3_error!(InvalidRequest, "jobId is required")), + } +} + +fn no_such_job(job_id: &str) -> S3Error { + S3Error::with_message(S3ErrorCode::NoSuchKey, format!("no such batch job: {job_id}")) +} + +#[cfg(test)] +mod tests { + use super::{detect_job_type, extract_query_params, require_job_id}; + use http::Uri; + + #[test] + fn detect_job_type_recognises_top_level_replicate_key() { + let body = b"replicate:\n apiVersion: v1\n source:\n bucket: src\n"; + assert_eq!(detect_job_type(body), Some("replicate".to_string())); + } + + #[test] + fn detect_job_type_recognises_keyrotate_and_expire() { + assert_eq!(detect_job_type(b"keyrotate:\n foo: bar\n"), Some("keyrotate".to_string())); + assert_eq!(detect_job_type(b"expire:\n foo: bar\n"), Some("expire".to_string())); + } + + #[test] + fn detect_job_type_skips_comments_and_blank_lines() { + let body = b"# a comment\n\nreplicate:\n foo: bar\n"; + assert_eq!(detect_job_type(body), Some("replicate".to_string())); + } + + #[test] + fn detect_job_type_ignores_indented_keys() { + // An indented `replicate:` is a nested field, not the top-level job type. + let body = b"something:\n replicate: true\n"; + assert_eq!(detect_job_type(body), None); + } + + #[test] + fn detect_job_type_returns_none_for_unknown_type() { + assert_eq!(detect_job_type(b"transmogrify:\n foo: bar\n"), None); + } + + #[test] + fn extract_query_params_decodes_job_id() { + let uri: Uri = "/rustfs/admin/v3/status-job?jobId=abc%2F123" + .parse() + .expect("uri should parse"); + let params = extract_query_params(&uri); + assert_eq!(params.get("jobId"), Some(&"abc/123".to_string())); + } + + #[test] + fn require_job_id_rejects_missing_and_empty() { + let missing: Uri = "/rustfs/admin/v3/status-job".parse().expect("uri should parse"); + assert!(require_job_id(&missing).is_err()); + + let empty: Uri = "/rustfs/admin/v3/status-job?jobId=".parse().expect("uri should parse"); + assert!(require_job_id(&empty).is_err()); + + let present: Uri = "/rustfs/admin/v3/status-job?jobId=job-1".parse().expect("uri should parse"); + assert_eq!(require_job_id(&present).expect("job id"), "job-1"); + } +} diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index f3ad76c2d..2e92343e0 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -15,6 +15,7 @@ pub mod account_info; pub mod audit; mod audit_runtime_config; +pub mod batch_job; pub mod bucket_meta; pub mod cluster_snapshot; pub mod config_admin; diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index f6f0d314a..9dc7bcb1f 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -19,10 +19,11 @@ use crate::admin::runtime_sources::{current_object_store_handle, current_replica use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; -use crate::admin::storage_api::bucket::replication::BucketStats; +use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType}; use crate::admin::storage_api::bucket::target::{BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat}; use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys}; use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions}; +use crate::admin::storage_api::contract::list::ListOperations as _; use crate::admin::storage_api::error::StorageError; use crate::admin::utils::read_compatible_admin_body; use crate::auth::{check_key_valid, get_session_token}; @@ -36,7 +37,7 @@ use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::time::Duration; use time::OffsetDateTime; @@ -247,6 +248,18 @@ pub fn register_replication_route(r: &mut S3Router) -> std::io:: AdminOperation(&RemoveRemoteTargetHandler {}), )?; + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/replication/diff").as_str(), + AdminOperation(&ReplicationDiffHandler {}), + )?; + + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/replication/mrf").as_str(), + AdminOperation(&ReplicationMrfHandler {}), + )?; + Ok(()) } @@ -584,6 +597,301 @@ impl Operation for RemoveRemoteTargetHandler { } } +/// Upper bound on the number of object versions scanned per `POST +/// /v3/replication/diff` request. RustFS has no persisted per-object +/// replication-diff index, so the diff is computed by scanning object versions +/// on demand. Cap the work so a single admin call cannot walk an arbitrarily +/// large bucket. When the scan is truncated, `is_truncated` is set on the +/// response so clients know the diff is partial. +const REPLICATION_DIFF_MAX_SCAN: usize = 10_000; + +/// Number of object versions requested per `list_object_versions` page while +/// computing a replication diff. +const REPLICATION_DIFF_PAGE_SIZE: i32 = 1_000; + +/// A single object version whose replication is not yet complete, reported by +/// `POST /v3/replication/diff`. Field names mirror MinIO's `madmin.DiffInfo` +/// so MinIO-compatible admin clients can parse the response. +#[derive(Debug, Serialize)] +struct ReplicationDiffEntry { + #[serde(rename = "Object")] + object: String, + #[serde(rename = "VersionID", skip_serializing_if = "Option::is_none")] + version_id: Option, + #[serde(rename = "Size")] + size: i64, + #[serde(rename = "IsDeleteMarker")] + is_delete_marker: bool, + #[serde(rename = "ReplicationStatus")] + replication_status: String, + #[serde(rename = "LastModified", skip_serializing_if = "Option::is_none")] + last_modified: Option, +} + +/// Response body for `POST /v3/replication/diff`. +/// +/// `entries` lists object versions with a `PENDING` or `FAILED` replication +/// status. `is_truncated` indicates the on-demand scan hit +/// [`REPLICATION_DIFF_MAX_SCAN`] before reaching the end of the bucket, so the +/// diff is partial and should be re-run with a narrower prefix. +#[derive(Debug, Serialize)] +struct ReplicationDiffResponse { + #[serde(rename = "Entries")] + entries: Vec, + #[serde(rename = "IsTruncated")] + is_truncated: bool, + #[serde(rename = "ScannedVersions")] + scanned_versions: usize, +} + +#[derive(Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct ReplicationDiffRequest { + #[serde(default)] + prefix: String, +} + +/// `POST /v3/replication/diff` +/// +/// Computes, on demand, the set of object versions in a bucket whose replication +/// is still `PENDING` or has `FAILED`. RustFS stores the replication status on +/// each object version (`x-amz-replication-status`) but has no pre-built diff +/// index, so this handler scans object versions (bounded by +/// [`REPLICATION_DIFF_MAX_SCAN`]) and returns the not-yet-replicated versions. +/// +/// The bucket must exist and have a replication configuration, matching MinIO's +/// behavior of returning `ReplicationConfigurationNotFoundError` otherwise. +pub struct ReplicationDiffHandler {} + +#[async_trait::async_trait] +impl Operation for ReplicationDiffHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_replication_admin_request(&req, AdminAction::ReplicationDiff).await?; + + let queries = extract_query_params(&req.uri); + let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else { + return Err(s3_error!(InvalidRequest, "bucket is required")); + }; + + let Some(store) = current_object_store_handle() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + // A replication diff is only meaningful for a bucket that is configured + // for replication; mirror MinIO's not-found semantics otherwise. + if let Err(err) = get_replication_config(&bucket).await { + if err == StorageError::ConfigNotFound { + return Err(S3Error::with_message( + S3ErrorCode::ReplicationConfigurationNotFoundError, + "replication configuration not found".to_string(), + )); + } + return Err(ApiError::from(err).into()); + } + + // Optional prefix can be supplied either as a query parameter (MinIO + // clients) or, for RustFS clients, as a small JSON body. + let mut prefix = queries.get("prefix").cloned().unwrap_or_default(); + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key) + .await + .unwrap_or_default(); + if prefix.is_empty() && !body.trim_ascii().is_empty() { + match serde_json::from_slice::(&body) { + Ok(parsed) => prefix = parsed.prefix, + Err(e) => return Err(s3_error!(InvalidRequest, "invalid replication diff request body: {e}")), + } + } + + let mut entries: Vec = Vec::new(); + let mut scanned_versions: usize = 0; + let mut marker: Option = None; + let mut version_marker: Option = None; + let mut is_truncated = false; + + 'scan: loop { + let listing = store + .clone() + .list_object_versions(&bucket, &prefix, marker.clone(), version_marker.clone(), None, REPLICATION_DIFF_PAGE_SIZE) + .await + .map_err(ApiError::from)?; + + for object in &listing.objects { + scanned_versions += 1; + + if matches!(object.replication_status, ReplicationStatusType::Pending | ReplicationStatusType::Failed) { + entries.push(ReplicationDiffEntry { + object: object.name.clone(), + version_id: object.version_id.map(|v| v.to_string()), + size: object.size, + is_delete_marker: object.delete_marker, + replication_status: object.replication_status.as_str().to_string(), + last_modified: object + .mod_time + .and_then(|t| t.format(&time::format_description::well_known::Rfc3339).ok()), + }); + } + + if scanned_versions >= REPLICATION_DIFF_MAX_SCAN { + // We stopped early; the diff is partial. + is_truncated = + listing.is_truncated || listing.next_marker.is_some() || listing.next_version_idmarker.is_some(); + break 'scan; + } + } + + if !listing.is_truncated { + break; + } + marker = listing.next_marker; + version_marker = listing.next_version_idmarker; + if marker.is_none() && version_marker.is_none() { + break; + } + } + + debug!( + bucket = %bucket, + prefix = %prefix, + scanned = scanned_versions, + pending_or_failed = entries.len(), + truncated = is_truncated, + "computed replication diff" + ); + + let response = ReplicationDiffResponse { + entries, + is_truncated, + scanned_versions, + }; + let data = serde_json::to_vec(&response) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers)) + } +} + +/// Failed-replication backlog for one remote target (ARN), summarised from the +/// replication stats. RustFS tracks failed replications as aggregate counters, +/// not an enumerable per-object MRF queue, so this reports the aggregate. +#[derive(Debug, Serialize)] +struct MrfTargetBacklog { + #[serde(rename = "ARN")] + arn: String, + #[serde(rename = "FailedCount")] + failed_count: i64, + #[serde(rename = "FailedSize")] + failed_size: i64, +} + +/// Response body for `GET /v3/replication/mrf`. +/// +/// MinIO streams individual MRF (most-recently-failed) entries. RustFS persists +/// MRF entries to disk but does not expose a runtime query over that queue, so +/// this endpoint reports the aggregate failed-replication backlog and the +/// in-queue counters instead. See the handler docs and the PR limitations note. +#[derive(Debug, Serialize)] +struct MrfResponse { + #[serde(rename = "Bucket")] + bucket: String, + #[serde(rename = "Targets")] + targets: Vec, + #[serde(rename = "TotalFailedCount")] + total_failed_count: i64, + #[serde(rename = "TotalFailedSize")] + total_failed_size: i64, + #[serde(rename = "QueuedCount")] + queued_count: i64, + #[serde(rename = "QueuedSize")] + queued_size: i64, + #[serde(rename = "PerObjectEntriesAvailable")] + per_object_entries_available: bool, +} + +/// `GET /v3/replication/mrf` +/// +/// Reports the failed-replication backlog (MinIO's MRF concept) for a bucket. +/// +/// Compatibility note: MinIO returns a stream of individual MRF entries drained +/// from its in-memory MRF queue. RustFS records failed replications as aggregate +/// counters in the replication stats and persists MRF entries to disk without a +/// runtime query API, so this handler returns the aggregate failed and queued +/// counts per target instead of per-object rows. `PerObjectEntriesAvailable` is +/// always `false` to make that limitation explicit to clients. +pub struct ReplicationMrfHandler {} + +#[async_trait::async_trait] +impl Operation for ReplicationMrfHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_replication_admin_request(&req, AdminAction::GetReplicationMetricsAction).await?; + + let queries = extract_query_params(&req.uri); + let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else { + return Err(s3_error!(InvalidRequest, "bucket is required")); + }; + + let Some(store) = current_object_store_handle() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .map_err(ApiError::from)?; + + if let Err(err) = get_replication_config(&bucket).await { + if err == StorageError::ConfigNotFound { + return Err(S3Error::with_message( + S3ErrorCode::ReplicationConfigurationNotFoundError, + "replication configuration not found".to_string(), + )); + } + return Err(ApiError::from(err).into()); + } + + let bucket_stats = match current_replication_stats_handle() { + Some(s) => s.get_latest_replication_stats(&bucket).await, + None => BucketStats::default(), + }; + + let mut targets: Vec = Vec::new(); + let mut total_failed_count: i64 = 0; + let mut total_failed_size: i64 = 0; + for (arn, stat) in &bucket_stats.replication_stats.stats { + total_failed_count += stat.failed.count; + total_failed_size += stat.failed.size; + targets.push(MrfTargetBacklog { + arn: arn.clone(), + failed_count: stat.failed.count, + failed_size: stat.failed.size, + }); + } + targets.sort_by(|a, b| a.arn.cmp(&b.arn)); + + let queued = &bucket_stats.replication_stats.q_stat.curr; + let response = MrfResponse { + bucket, + targets, + total_failed_count, + total_failed_size, + queued_count: queued.count, + queued_size: queued.bytes, + per_object_entries_available: false, + }; + + let data = serde_json::to_vec(&response) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers)) + } +} + #[cfg(test)] mod tests { use super::{RemoteTargetRequest, extract_query_params, validate_remote_target_tls_settings}; diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index f8126222d..af6677a8e 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -33,8 +33,8 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, bucket_meta, cluster_snapshot, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc, - plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, + audit, batch_job, bucket_meta, cluster_snapshot, config_admin, 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, }; use router::{AdminOperation, S3Router}; @@ -79,6 +79,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() plugins_instances::register_plugin_instance_route(r)?; replication_handler::register_replication_route(r)?; + batch_job::register_batch_job_route(r)?; site_replication::register_site_replication_route(r)?; profile_admin::register_profiling_route(r)?; tls_debug::register_tls_debug_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index f9f27dc6a..b9aae7f27 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -82,6 +82,12 @@ const SITE_REPLICATION_OPERATION: AdminActionRef = AdminActionRef::new("SiteRepl const SITE_REPLICATION_REMOVE: AdminActionRef = AdminActionRef::new("SiteReplicationRemoveAction"); const SITE_REPLICATION_RESYNC: AdminActionRef = AdminActionRef::new("SiteReplicationResyncAction"); const STORAGE_INFO: AdminActionRef = AdminActionRef::new("StorageInfoAdminAction"); +// MinIO admin compat: batch jobs and replication diff. +const START_BATCH_JOB: AdminActionRef = AdminActionRef::new("StartBatchJobAction"); +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"); #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum DeferredRoutePolicyReason { @@ -1172,6 +1178,35 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ COMMIT_TABLE, RouteRiskLevel::High, ), + // MinIO admin compat: batch job lifecycle (backlog#613). + admin(HttpMethod::Post, "/rustfs/admin/v3/start-job", START_BATCH_JOB, RouteRiskLevel::High), + admin(HttpMethod::Get, "/rustfs/admin/v3/list-jobs", LIST_BATCH_JOBS, RouteRiskLevel::Sensitive), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/status-job", + DESCRIBE_BATCH_JOB, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/describe-job", + DESCRIBE_BATCH_JOB, + RouteRiskLevel::Sensitive, + ), + admin(HttpMethod::Delete, "/rustfs/admin/v3/cancel-job", CANCEL_BATCH_JOB, RouteRiskLevel::High), + // MinIO admin compat: replication diff / MRF inspection (backlog#611). + admin( + HttpMethod::Post, + "/rustfs/admin/v3/replication/diff", + REPLICATION_DIFF, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/replication/mrf", + GET_REPLICATION_METRICS, + RouteRiskLevel::Sensitive, + ), ]; pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[ diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 438a70e9a..3653b9ccc 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -244,6 +244,13 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::GET, "/v3/replicationmetrics"), admin_route(Method::PUT, "/v3/set-remote-target"), admin_route(Method::DELETE, "/v3/remove-remote-target"), + admin_route(Method::POST, "/v3/replication/diff"), + admin_route(Method::GET, "/v3/replication/mrf"), + admin_route(Method::POST, "/v3/start-job"), + admin_route(Method::GET, "/v3/list-jobs"), + admin_route(Method::GET, "/v3/status-job"), + admin_route(Method::GET, "/v3/describe-job"), + admin_route(Method::DELETE, "/v3/cancel-job"), admin_route(Method::PUT, "/v3/site-replication/add"), admin_route(Method::PUT, "/v3/site-replication/remove"), admin_route(Method::GET, "/v3/site-replication/info"), @@ -1125,6 +1132,13 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/import-bucket-metadata")); assert_route(&router, Method::GET, &admin_path("/v3/list-remote-targets")); assert_route(&router, Method::PUT, &admin_path("/v3/set-remote-target")); + assert_route(&router, Method::POST, &admin_path("/v3/replication/diff")); + assert_route(&router, Method::GET, &admin_path("/v3/replication/mrf")); + assert_route(&router, Method::POST, &admin_path("/v3/start-job")); + assert_route(&router, Method::GET, &admin_path("/v3/list-jobs")); + assert_route(&router, Method::GET, &admin_path("/v3/status-job")); + assert_route(&router, Method::GET, &admin_path("/v3/describe-job")); + assert_route(&router, Method::DELETE, &admin_path("/v3/cancel-job")); assert_route(&router, Method::PUT, &admin_path("/v3/site-replication/add")); assert_route(&router, Method::PUT, &admin_path("/v3/site-replication/remove")); assert_route(&router, Method::GET, &admin_path("/v3/site-replication/info"));