From 6d06b574f25988ada9cba518e1092a7a0e45e949 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Fri, 5 Jun 2026 21:52:13 +0800 Subject: [PATCH] feat(table-catalog): add REST catalog route surface (#3211) * feat(table-catalog): add REST catalog route surface * fix(table-catalog): avoid test-only router import * fix(table-catalog): align unsupported REST responses --------- Co-authored-by: Henry Guo --- rustfs/src/admin/handlers/mod.rs | 2 + rustfs/src/admin/handlers/table_catalog.rs | 260 ++++++++++++++++++++ rustfs/src/admin/mod.rs | 3 +- rustfs/src/admin/route_registration_test.rs | 25 +- rustfs/src/admin/router.rs | 1 + rustfs/src/server/layer.rs | 7 +- rustfs/src/server/mod.rs | 2 +- rustfs/src/server/prefix.rs | 7 +- rustfs/src/server/readiness.rs | 2 + 9 files changed, 302 insertions(+), 7 deletions(-) create mode 100644 rustfs/src/admin/handlers/table_catalog.rs diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 0b246e7fc..ca64cc06f 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -45,6 +45,7 @@ pub mod service_account; pub mod site_replication; pub mod sts; pub mod system; +pub mod table_catalog; mod target_descriptor; pub mod tier; pub mod tls_debug; @@ -95,6 +96,7 @@ mod tests { let _site_replication_add_handler = site_replication::SiteReplicationAddHandler {}; let _site_replication_info_handler = site_replication::SiteReplicationInfoHandler {}; let _site_replication_status_handler = site_replication::SiteReplicationStatusHandler {}; + let _table_catalog_config_handler = table_catalog::GetCatalogConfigHandler {}; // Just verify they can be created without panicking // Test passes if we reach this point without panicking diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs new file mode 100644 index 000000000..d5c676d65 --- /dev/null +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -0,0 +1,260 @@ +// 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, + router::{AdminOperation, Operation, S3Router}, +}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{RemoteAddr, TABLE_CATALOG_PREFIX}; +use crate::table_catalog::DEFAULT_WAREHOUSE_ID; +use http::{HeaderMap, HeaderValue, StatusCode}; +use hyper::Method; +use matchit::Params; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; +use serde::Serialize; +use std::collections::BTreeMap; + +const JSON_CONTENT_TYPE: &str = "application/json"; +const WAREHOUSE_PROPERTY: &str = "warehouse"; +const UNSUPPORTED_OPERATION_TYPE: &str = "UnsupportedOperationException"; +const UNSUPPORTED_OPERATION_STATUS: StatusCode = StatusCode::NOT_ACCEPTABLE; + +static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {}; +static LIST_NAMESPACES_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::GetTableNamespaceAction, + operation: "list namespaces", +}; +static CREATE_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::SetTableNamespaceAction, + operation: "create namespace", +}; +static GET_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::GetTableNamespaceAction, + operation: "get namespace", +}; +static DROP_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::DeleteTableNamespaceAction, + operation: "drop namespace", +}; +static LIST_TABLES_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::GetTableAction, + operation: "list tables", +}; +static CREATE_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::CreateTableAction, + operation: "create table", +}; +static REGISTER_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::RegisterTableAction, + operation: "register table", +}; +static LOAD_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::GetTableAction, + operation: "load table", +}; +static COMMIT_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::CommitTableAction, + operation: "commit table", +}; +static DROP_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler { + action: AdminAction::DeleteTableAction, + operation: "drop table", +}; + +#[derive(Debug, Serialize)] +struct CatalogConfigResponse { + defaults: BTreeMap<&'static str, &'static str>, + overrides: BTreeMap<&'static str, &'static str>, + endpoints: Vec<&'static str>, +} + +#[derive(Debug, Serialize)] +struct ErrorResponse { + error: RestCatalogError, +} + +#[derive(Debug, Serialize)] +struct RestCatalogError { + message: String, + #[serde(rename = "type")] + error_type: &'static str, + code: u16, +} + +pub fn register_table_catalog_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{TABLE_CATALOG_PREFIX}/config").as_str(), + AdminOperation(&GET_CONFIG_HANDLER), + )?; + r.insert( + Method::GET, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces").as_str(), + AdminOperation(&LIST_NAMESPACES_HANDLER), + )?; + r.insert( + Method::POST, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces").as_str(), + AdminOperation(&CREATE_NAMESPACE_HANDLER), + )?; + r.insert( + Method::GET, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(), + AdminOperation(&GET_NAMESPACE_HANDLER), + )?; + r.insert( + Method::DELETE, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(), + AdminOperation(&DROP_NAMESPACE_HANDLER), + )?; + r.insert( + Method::GET, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(), + AdminOperation(&LIST_TABLES_HANDLER), + )?; + r.insert( + Method::POST, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(), + AdminOperation(&CREATE_TABLE_HANDLER), + )?; + r.insert( + Method::POST, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/register").as_str(), + AdminOperation(®ISTER_TABLE_HANDLER), + )?; + r.insert( + Method::GET, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(), + AdminOperation(&LOAD_TABLE_HANDLER), + )?; + r.insert( + Method::POST, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(), + AdminOperation(&COMMIT_TABLE_HANDLER), + )?; + r.insert( + Method::DELETE, + format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(), + AdminOperation(&DROP_TABLE_HANDLER), + )?; + + Ok(()) +} + +fn catalog_config_response() -> CatalogConfigResponse { + CatalogConfigResponse { + defaults: BTreeMap::from([(WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID)]), + overrides: BTreeMap::new(), + endpoints: Vec::new(), + } +} + +fn build_json_response(status: StatusCode, body: &T) -> S3Result> { + let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE)); + Ok(S3Response::with_headers((status, Body::from(data)), headers)) +} + +async fn authorize_table_catalog_request(req: &S3Request, action: AdminAction) -> S3Result<()> { + let Some(input_cred) = &req.credentials 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?; + + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(action)], + req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), + ) + .await +} + +fn unsupported_response(operation: &str) -> S3Result> { + build_json_response( + UNSUPPORTED_OPERATION_STATUS, + &ErrorResponse { + error: RestCatalogError { + message: format!("Iceberg REST Catalog {operation} is not implemented yet"), + error_type: UNSUPPORTED_OPERATION_TYPE, + code: UNSUPPORTED_OPERATION_STATUS.as_u16(), + }, + }, + ) +} + +pub struct GetCatalogConfigHandler {} + +#[async_trait::async_trait] +impl Operation for GetCatalogConfigHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_table_catalog_request(&req, AdminAction::GetTableCatalogAction).await?; + build_json_response(StatusCode::OK, &catalog_config_response()) + } +} + +pub struct UnsupportedCatalogOperationHandler { + action: AdminAction, + operation: &'static str, +} + +#[async_trait::async_trait] +impl Operation for UnsupportedCatalogOperationHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_table_catalog_request(&req, self.action).await?; + unsupported_response(self.operation) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn catalog_config_response_uses_default_warehouse_without_unsupported_endpoints() { + let response = catalog_config_response(); + + assert_eq!(response.defaults.get(WAREHOUSE_PROPERTY), Some(&DEFAULT_WAREHOUSE_ID)); + assert!(response.overrides.is_empty()); + assert!(response.endpoints.is_empty()); + } + + #[test] + fn unsupported_response_uses_rest_catalog_unsupported_status() { + let response = unsupported_response("create table").expect("unsupported response should build"); + + assert_eq!(response.output.0, UNSUPPORTED_OPERATION_STATUS); + } + + #[test] + fn table_catalog_handlers_require_table_admin_actions() { + let src = include_str!("table_catalog.rs"); + + assert!(src.contains("AdminAction::GetTableCatalogAction")); + assert!(src.contains("AdminAction::GetTableNamespaceAction")); + assert!(src.contains("AdminAction::SetTableNamespaceAction")); + assert!(src.contains("AdminAction::DeleteTableNamespaceAction")); + assert!(src.contains("AdminAction::CreateTableAction")); + assert!(src.contains("AdminAction::RegisterTableAction")); + assert!(src.contains("AdminAction::CommitTableAction")); + assert!(src.contains("AdminAction::DeleteTableAction")); + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 81e54d93e..9384356e6 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -28,7 +28,7 @@ mod route_registration_test; use handlers::{ audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, - profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user, + profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -70,6 +70,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result tls_debug::register_tls_debug_route(&mut r)?; kms::register_kms_route(&mut r)?; oidc::register_oidc_route(&mut r)?; + table_catalog::register_table_catalog_route(&mut r)?; Ok(r) } diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 46c5d4364..87b3e1e56 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -15,11 +15,15 @@ use crate::admin::{ handlers::{ audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, - profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user, + profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, + user, }, router::{AdminOperation, S3Router}, }; -use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH}; +use crate::server::{ + ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, + TABLE_CATALOG_PREFIX, +}; use hyper::Method; use serial_test::serial; use temp_env::with_var; @@ -32,6 +36,10 @@ fn compat_admin_alias_path(path: &str) -> String { format!("{}{}", MINIO_ADMIN_PREFIX, path) } +fn table_catalog_path(path: &str) -> String { + format!("{}{}", TABLE_CATALOG_PREFIX, path) +} + fn assert_route(router: &S3Router, method: Method, path: &str) { assert!( router.contains_route(method.clone(), path), @@ -64,6 +72,7 @@ fn register_admin_routes(router: &mut S3Router) { tls_debug::register_tls_debug_route(router).expect("register tls debug route"); kms::register_kms_route(router).expect("register kms route"); oidc::register_oidc_route(router).expect("register oidc route"); + table_catalog::register_table_catalog_route(router).expect("register table catalog route"); } // register_admin_routes reads ENV_HEALTH_ENDPOINT_ENABLE to decide whether @@ -126,6 +135,18 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/config")); assert_route(&router, Method::GET, &admin_path("/v3/scanner/status")); + assert_route(&router, Method::GET, &table_catalog_path("/config")); + assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces")); + assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces")); + assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales")); + assert_route(&router, Method::DELETE, &table_catalog_path("/analytics/namespaces/sales")); + assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables")); + assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables")); + assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/register")); + assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); + assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); + assert_route(&router, Method::DELETE, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); + assert_route(&router, Method::POST, &admin_path("/v3/service")); assert_route(&router, Method::GET, &admin_path("/v3/info")); assert_route(&router, Method::GET, &admin_path("/v3/storageinfo")); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index c5fbc7111..2f1c2fd6d 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -2458,6 +2458,7 @@ mod tests { fn is_admin_path_accepts_rustfs_and_compat_prefixes() { assert!(is_admin_path("/rustfs/admin/v3/info")); assert!(is_admin_path("/minio/admin/v3/info")); + assert!(is_admin_path(&format!("{}/config", crate::server::TABLE_CATALOG_PREFIX))); assert!(!is_admin_path("/bucket/object")); assert!(!is_admin_path("/rustfs/administrator/object")); assert!(!is_admin_path("/minio/administrator/object")); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index d70809deb..7e962c11b 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -19,8 +19,8 @@ use crate::server::cors; use crate::server::hybrid::HybridBody; use crate::server::{ ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, - MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, - has_path_prefix, is_admin_path, + MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TABLE_CATALOG_PREFIX, active_http_requests, + collect_dependency_readiness_report, has_path_prefix, is_admin_path, }; use crate::storage::apply_cors_headers; use crate::storage::request_context::{RequestContext, extract_request_id_from_headers}; @@ -864,6 +864,7 @@ fn is_object_attributes_request(req: &HttpRequest) -> bool { || has_path_prefix(path, MINIO_ADMIN_PREFIX) || has_path_prefix(path, RUSTFS_ADMIN_PREFIX) || has_path_prefix(path, MINIO_ADMIN_V3_PREFIX) + || has_path_prefix(path, TABLE_CATALOG_PREFIX) || has_path_prefix(path, CONSOLE_PREFIX) || has_path_prefix(path, RPC_PREFIX) { @@ -908,6 +909,7 @@ impl ConditionalCorsLayer { // Exclude Admin, Console, RPC, and configured special paths !has_path_prefix(path, ADMIN_PREFIX) && !has_path_prefix(path, MINIO_ADMIN_PREFIX) + && !has_path_prefix(path, TABLE_CATALOG_PREFIX) && !has_path_prefix(path, RPC_PREFIX) && !is_console_path(path) && !Self::EXCLUDED_EXACT_PATHS.contains(&path) @@ -1958,6 +1960,7 @@ mod tests { assert!(ConditionalCorsLayer::is_s3_path("/")); assert!(!ConditionalCorsLayer::is_s3_path("/rustfs/admin/v3/info")); assert!(!ConditionalCorsLayer::is_s3_path("/minio/admin/v3/info")); + assert!(!ConditionalCorsLayer::is_s3_path(&format!("{TABLE_CATALOG_PREFIX}/config"))); assert!(ConditionalCorsLayer::is_s3_path("/minio/adminx/object")); assert!(!ConditionalCorsLayer::is_s3_path("/health")); assert!(!ConditionalCorsLayer::is_s3_path("/health/ready")); diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index f569a2225..fa7e147d1 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -49,7 +49,7 @@ pub(crate) use module_switch::{ pub(crate) use prefix::{ ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, - TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path, + TABLE_CATALOG_PREFIX, TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path, }; pub(crate) use readiness::DependencyReadiness; pub(crate) use readiness::DependencyReadinessReport; diff --git a/rustfs/src/server/prefix.rs b/rustfs/src/server/prefix.rs index 1d875d15b..4426ae096 100644 --- a/rustfs/src/server/prefix.rs +++ b/rustfs/src/server/prefix.rs @@ -43,9 +43,14 @@ pub(crate) const ADMIN_PREFIX: &str = "/rustfs/admin"; /// This alias allows stock MinIO admin tooling to reach RustFS handlers. pub(crate) const MINIO_ADMIN_PREFIX: &str = "/minio/admin"; +/// Iceberg REST Catalog prefix for RustFS S3 Tables control-plane routes. +pub(crate) const TABLE_CATALOG_PREFIX: &str = "/iceberg/v1"; + /// Returns true for the admin prefix itself or slash-delimited children. pub(crate) fn is_admin_path(path: &str) -> bool { - has_path_prefix(path, ADMIN_PREFIX) || has_path_prefix(path, MINIO_ADMIN_PREFIX) + has_path_prefix(path, ADMIN_PREFIX) + || has_path_prefix(path, MINIO_ADMIN_PREFIX) + || has_path_prefix(path, TABLE_CATALOG_PREFIX) } pub(crate) fn has_path_prefix(path: &str, prefix: &str) -> bool { diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 48813e925..9f9eabd45 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -140,6 +140,7 @@ fn is_probe_path(path: &str) -> bool { let is_prefix_probe = has_path_prefix(path, crate::server::RUSTFS_ADMIN_PREFIX) || has_path_prefix(path, crate::server::MINIO_ADMIN_V3_PREFIX) + || has_path_prefix(path, crate::server::TABLE_CATALOG_PREFIX) || has_path_prefix(path, crate::server::CONSOLE_PREFIX) || has_path_prefix(path, crate::server::RPC_PREFIX) || has_path_prefix(path, crate::server::ADMIN_PREFIX) @@ -719,6 +720,7 @@ mod tests { fn probe_path_checks_admin_boundaries() { assert!(is_probe_path("/minio/admin/v3/info")); assert!(is_probe_path("/rustfs/admin/v3/info")); + assert!(is_probe_path(&format!("{}/config", crate::server::TABLE_CATALOG_PREFIX))); assert!(is_probe_path("/rustfs/console/")); assert!(!is_probe_path("/minio/adminx/object")); assert!(!is_probe_path("/rustfs/adminx/object"));