From 5c72d76acf392b582c59ae0863229156eb13d90c Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:46:02 +0300 Subject: [PATCH] fix(provisioning): skip CreateBucket for buckets that already exist Closes #286 --- CHANGELOG.md | 2 + crates/rustfs-admin/src/s3_ops.rs | 27 ++++++++++++ src/reconcile/provisioning.rs | 70 ++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3331562..740d50a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,8 @@ All notable changes to RustFS Operator are documented in this file. The format i ### Fixed +- Check an existing Tenant bucket with HeadBucket before CreateBucket, so a bucket under + continuous writes no longer stays Pending on RustFS lock-timeout 503 responses. - Use `docker.io/rustfs/operator` as the default Helm image repository for the Operator, Console, and COSI driver, while preserving repository overrides through Helm values. - Reject multi-pool Tenants containing a single-node single-disk pool before applying pool workloads, matching RustFS startup constraints. diff --git a/crates/rustfs-admin/src/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs index ff43c3e..20b0221 100644 --- a/crates/rustfs-admin/src/s3_ops.rs +++ b/crates/rustfs-admin/src/s3_ops.rs @@ -748,6 +748,12 @@ impl RustfsAdminClient { return Err(RustfsClientError::RequestBuildFailed); } + // RustFS takes the bucket's exclusive locks before it finds an existing + // bucket, so a repeated PUT times out with 503 while objects are written. + if self.bucket_exists(bucket).await? { + return Ok(CreateBucketResult::AlreadyOwnedByYou); + } + let path = format!("/{bucket}"); let body = create_bucket_body(region); let content_type = (!body.is_empty()).then_some("application/xml"); @@ -807,6 +813,27 @@ impl RustfsAdminClient { )) } + /// Report whether HeadBucket succeeds with these credentials. Any other + /// response is left for CreateBucket to classify. + async fn bucket_exists(&self, bucket: &str) -> Result { + let path = format!("/{bucket}"); + let signed = self.sign_request("HEAD", &path, "", "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + + let response = self + .http_client + .head(format!("{}{}", self.base_url.trim_end_matches('/'), path)) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + Ok(response.status().is_success()) + } + /// Delete a bucket. Missing buckets are treated as success (idempotent). pub async fn delete_bucket(&self, bucket: &str) -> Result<(), RustfsClientError> { if bucket.trim().is_empty() { diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 6380c48..d16f42e 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -3761,7 +3761,7 @@ mod tests { extract::State, http::{Request, StatusCode}, response::{IntoResponse, Response}, - routing::{any, get, put}, + routing::{any, get, head, put}, }; use http_body_util::BodyExt; use k8s_openapi::ByteString; @@ -6530,6 +6530,73 @@ mod tests { server.abort(); } + #[tokio::test] + async fn existing_bucket_skips_create_bucket() { + let router = Router::new().route( + "/app-data", + head(|| async { StatusCode::OK }).put(|| async { StatusCode::SERVICE_UNAVAILABLE }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + let bucket = ProvisioningBucket { + name: "app-data".to_string(), + ..Default::default() + }; + + let item = ensure_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + item.message.as_deref(), + Some("RustFS bucket already exists") + ); + server.abort(); + } + + #[tokio::test] + async fn missing_bucket_is_created() { + let router = Router::new().route( + "/app-data", + head(|| async { StatusCode::NOT_FOUND }).put(|| async { StatusCode::OK }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("test server should bind"); + let addr = listener.local_addr().expect("listener should have address"); + let server = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("test server should serve") + }); + let client = + RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + let bucket = ProvisioningBucket { + name: "app-data".to_string(), + ..Default::default() + }; + + let item = ensure_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(item.message.as_deref(), Some("RustFS bucket was created")); + server.abort(); + } + #[tokio::test] async fn create_bucket_400_fails_without_retry() { let router = Router::new().route("/app-data", put(|| async { StatusCode::BAD_REQUEST })); @@ -7076,6 +7143,7 @@ mod tests { let router = Router::new().route( "/app-data", put(|| async { StatusCode::OK }) + .head(|| async { StatusCode::NOT_FOUND }) .get(move |req: Request| { let live_policy = route_live_policy.clone(); async move {