From 55880fb39801aaaa788c55245efa67d71b9f21e2 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Jul 2026 12:13:45 +0800 Subject: [PATCH] feat(admin): add manual transition run endpoint Add a bounded POST /rustfs/admin/v3/ilm/transition/run API for backlog #1477 and cover the route, policy, query parsing, and partial status contract needed by #1481. Console operations from #1480 are intentionally left for a later client integration. Co-Authored-By: heihutu --- rustfs/src/admin/handlers/ilm_transition.rs | 207 ++++++++++++++++++++ rustfs/src/admin/handlers/mod.rs | 2 + rustfs/src/admin/mod.rs | 3 +- rustfs/src/admin/route_policy.rs | 7 + rustfs/src/admin/route_registration_test.rs | 2 + rustfs/src/admin/storage_api.rs | 15 ++ 6 files changed, 235 insertions(+), 1 deletion(-) create mode 100644 rustfs/src/admin/handlers/ilm_transition.rs diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs new file mode 100644 index 000000000..9603d2248 --- /dev/null +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -0,0 +1,207 @@ +// 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. + +use crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::object_store_from_extensions; +use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; +use crate::admin::storage_api::lifecycle::{ + ManualTransitionRunOptions, ManualTransitionRunReport, enqueue_transition_for_existing_objects_scoped, +}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue}; +use hyper::{Method, StatusCode}; +use matchit::Params; +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, Serialize}; + +const JSON_CONTENT_TYPE: &str = "application/json"; +const DEFAULT_MANUAL_TRANSITION_MAX_OBJECTS: u64 = 10_000; +const MAX_MANUAL_TRANSITION_OBJECTS: u64 = 100_000; + +#[derive(Debug, Deserialize, Default)] +#[serde(deny_unknown_fields)] +struct ManualTransitionRunQuery { + bucket: Option, + prefix: Option, + tier: Option, + #[serde(rename = "dryRun")] + dry_run: Option, + #[serde(rename = "maxObjects")] + max_objects: Option, +} + +#[derive(Debug, Serialize)] +struct ManualTransitionRunResponse { + state: &'static str, + mode: &'static str, + job_id: Option, + status_endpoint: Option, + report: ManualTransitionRunReport, +} + +pub fn register_ilm_transition_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/ilm/transition/run").as_str(), + AdminOperation(&ManualTransitionRunHandler {}), + )?; + + Ok(()) +} + +fn parse_manual_transition_query(query: Option<&str>) -> S3Result<(String, ManualTransitionRunOptions)> { + let query: ManualTransitionRunQuery = match query { + Some(query) => serde_urlencoded::from_bytes(query.as_bytes()) + .map_err(|_| s3_error!(InvalidArgument, "invalid manual transition query"))?, + None => ManualTransitionRunQuery::default(), + }; + + let bucket = query + .bucket + .as_deref() + .map(str::trim) + .filter(|bucket| !bucket.is_empty()) + .ok_or_else(|| s3_error!(InvalidRequest, "bucket is required"))?; + if is_reserved_or_invalid_bucket(bucket, false) { + return Err(s3_error!(InvalidBucketName, "invalid bucket name")); + } + + let max_objects = query.max_objects.unwrap_or(DEFAULT_MANUAL_TRANSITION_MAX_OBJECTS); + if max_objects == 0 || max_objects > MAX_MANUAL_TRANSITION_OBJECTS { + return Err(s3_error!(InvalidArgument, "maxObjects is outside the allowed range")); + } + + Ok(( + bucket.to_string(), + ManualTransitionRunOptions { + prefix: query.prefix.unwrap_or_default(), + tier: query.tier.map(|tier| tier.trim().to_string()).filter(|tier| !tier.is_empty()), + dry_run: query.dry_run.unwrap_or(false), + max_objects: Some(max_objects), + }, + )) +} + +async fn authorize_manual_transition_request(req: &S3Request) -> S3Result<()> { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(InvalidRequest, "authentication 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(|addr| addr.0)); + + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::SetTierAction)], + remote_addr, + ) + .await +} + +fn response_state(report: &ManualTransitionRunReport) -> &'static str { + if report.truncated_by_limit || report.has_partial_enqueue() { + "partial" + } else { + "completed" + } +} + +fn json_response(response: &ManualTransitionRunResponse) -> S3Result> { + let body = serde_json::to_vec(response).map_err(|err| { + S3Error::with_message(S3ErrorCode::InternalError, format!("failed to encode manual transition response: {err}")) + })?; + let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE) + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid content type: {err}")))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, content_type); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), headers)) +} + +pub struct ManualTransitionRunHandler {} + +#[async_trait::async_trait] +impl Operation for ManualTransitionRunHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_manual_transition_request(&req).await?; + let (bucket, options) = parse_manual_transition_query(req.uri.query())?; + let Some(store) = object_store_from_extensions(&req.extensions) else { + return Err(s3_error!(InternalError, "object store is not initialized")); + }; + + let report = enqueue_transition_for_existing_objects_scoped(store, &bucket, options) + .await + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("manual transition run failed: {err}")))?; + let response = ManualTransitionRunResponse { + state: response_state(&report), + mode: "enqueue_only", + job_id: None, + status_endpoint: None, + report, + }; + + json_response(&response) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn manual_transition_query_defaults_to_bounded_run() { + let (bucket, options) = + parse_manual_transition_query(Some("bucket=data&prefix=logs/&tier=warm")).expect("valid query should parse"); + + assert_eq!(bucket, "data"); + assert_eq!(options.prefix, "logs/"); + assert_eq!(options.tier.as_deref(), Some("warm")); + assert!(!options.dry_run); + assert_eq!(options.max_objects, Some(DEFAULT_MANUAL_TRANSITION_MAX_OBJECTS)); + } + + #[test] + fn manual_transition_query_rejects_server_info_style_unscoped_request() { + let err = parse_manual_transition_query(Some("dryRun=true")).expect_err("bucket must be required"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn manual_transition_query_rejects_unbounded_budget() { + let err = parse_manual_transition_query(Some("bucket=data&maxObjects=0")).expect_err("zero budget must fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + } + + #[test] + fn manual_transition_response_reports_partial_for_queue_pressure() { + let report = ManualTransitionRunReport { + skipped_queue_full: 1, + ..Default::default() + }; + + assert_eq!(response_state(&report), "partial"); + } +} diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 0e8415e78..0eb22f22b 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -28,6 +28,7 @@ pub mod heal; pub mod health; pub(crate) mod iam_error; pub mod idp_compat; +pub mod ilm_transition; pub mod is_admin; pub mod kms; pub mod kms_dynamic; @@ -118,6 +119,7 @@ mod tests { let _list_remote_target_handler = replication::ListRemoteTargetHandler {}; let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {}; let _scanner_status_handler = scanner::ScannerStatusHandler {}; + let _manual_transition_handler = ilm_transition::ManualTransitionRunHandler {}; let _site_replication_add_handler = site_replication::SiteReplicationAddHandler {}; let _site_replication_info_handler = site_replication::SiteReplicationInfoHandler {}; let _site_replication_status_handler = site_replication::SiteReplicationStatusHandler {}; diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 222ff169b..6d265906e 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -34,7 +34,7 @@ mod route_registration_test; use handlers::{ audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler, extensions, - heal, health, idp_compat, kms, module_switch, object_data_cache, object_zip_download, oidc, plugins_catalog, + heal, health, idp_compat, ilm_transition, kms, module_switch, object_data_cache, 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, }; @@ -76,6 +76,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() bucket_meta::register_bucket_meta_route(r)?; config_admin::register_config_route(r)?; scanner::register_scanner_route(r)?; + ilm_transition::register_ilm_transition_route(r)?; object_data_cache::register_object_data_cache_route(r)?; audit::register_audit_target_route(r)?; module_switch::register_module_switch_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index af765cc70..11b2b6242 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -412,6 +412,7 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High), admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High), admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive), + admin(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER, RouteRiskLevel::High), admin( HttpMethod::Get, "/rustfs/admin/v3/audit/target/list", @@ -1791,6 +1792,12 @@ mod tests { } } + #[test] + fn route_policy_requires_set_tier_for_manual_transition_run() { + assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER); + assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SERVER_INFO); + } + #[test] fn route_policy_keeps_contextual_auth_deferred() { assert_deferred( diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index e633e8fc9..024bf6042 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -194,6 +194,7 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::PUT, "/v3/tier"), admin_route_sample(Method::POST, "/v3/tier/{tiername}", "/v3/tier/HOT"), admin_route(Method::POST, "/v3/tier/clear"), + admin_route(Method::POST, "/v3/ilm/transition/run"), admin_route(Method::PUT, "/v3/set-bucket-quota"), admin_route(Method::GET, "/v3/get-bucket-quota"), admin_route_sample(Method::PUT, "/v3/quota/{bucket}", "/v3/quota/test-bucket"), @@ -830,6 +831,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &admin_path("/v3/config")); assert_route(&router, Method::PUT, &admin_path("/v3/config")); assert_route(&router, Method::GET, &admin_path("/v3/scanner/status")); + assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run")); assert_route(&router, Method::GET, &table_catalog_path("/config")); assert_route(&router, Method::PUT, &table_catalog_path("/buckets/analytics")); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 864fd0bce..eaa916932 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -193,6 +193,21 @@ pub(crate) mod bucket_target_sys { } pub(crate) mod lifecycle { + pub(crate) type ManualTransitionRunOptions = + super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunOptions; + pub(crate) type ManualTransitionRunReport = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunReport; + + pub(crate) async fn enqueue_transition_for_existing_objects_scoped( + api: std::sync::Arc, + bucket: &str, + options: ManualTransitionRunOptions, + ) -> super::Result { + super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects_scoped( + api, bucket, options, + ) + .await + } + pub(crate) mod tier_last_day_stats { #[cfg(test)] pub(crate) type LastDayTierStats = super::super::ecstore_bucket::lifecycle::tier_last_day_stats::LastDayTierStats;