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 <heihutu@gmail.com>

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<String>,
+    prefix: Option<String>,
+    tier: Option<String>,
+    #[serde(rename = "dryRun")]
+    dry_run: Option<bool>,
+    #[serde(rename = "maxObjects")]
+    max_objects: Option<u64>,
+}
+
+#[derive(Debug, Serialize)]
+struct ManualTransitionRunResponse {
+    state: &'static str,
+    mode: &'static str,
+    job_id: Option<String>,
+    status_endpoint: Option<String>,
+    report: ManualTransitionRunReport,
+}
+
+pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> 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<Body>) -> 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::<Option<RemoteAddr>>()
+        .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<S3Response<(StatusCode, Body)>> {
+    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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        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");
+    }
+}

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 {};

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<AdminOperation>) -> 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)?;

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(

rustfs/src/admin/route_registration_test.rs

@@ -194,6 +194,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         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"));

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<super::ECStore>,
+        bucket: &str,
+        options: ManualTransitionRunOptions,
+    ) -> super::Result<ManualTransitionRunReport> {
+        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;