From 3e0accf7ee90121a42ef938c3c02ae28035876e8 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:36:11 +0200 Subject: [PATCH 1/2] feat(reconcile): manage per-pool PodDisruptionBudgets --- CHANGELOG.md | 2 + crates/rustfs-admin/src/lib.rs | 23 + deploy/k8s-dev/operator-rbac.yaml | 3 + deploy/rustfs-operator/crds/tenant-crd.yaml | 10 + .../templates/clusterrole.yaml | 5 + docs/operator-user-guide.md | 11 + docs/operator-user-guide.zh-CN.md | 11 + src/lib.rs | 5 + src/reconcile.rs | 11 +- src/reconcile/phases.rs | 27 +- src/reconcile/pod_disruption_budget.rs | 444 ++++++++++++++++++ src/reconcile/pool_lifecycle.rs | 6 +- src/sts/rustfs_client.rs | 8 +- src/tenant_monitor.rs | 1 + src/types/v1alpha1/tenant.rs | 9 + 15 files changed, 566 insertions(+), 10 deletions(-) create mode 100644 src/reconcile/pod_disruption_budget.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index a2b823cb..cc81ee75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,8 @@ All notable changes to RustFS Operator are documented in this file. The format i driver alongside the upstream provisioner sidecar, with dedicated ServiceAccount/RBAC. - Seven built-in RustFS Grafana dashboards, published by default as sidecar-discoverable ConfigMaps with configurable labels and namespace. +- Per-pool PodDisruptionBudgets with `minAvailable` derived from the erasure layout RustFS + reports, so node drains keep write quorum; opt out with `spec.disablePodDisruptionBudget`. ### Fixed diff --git a/crates/rustfs-admin/src/lib.rs b/crates/rustfs-admin/src/lib.rs index 352a6fad..539e5466 100644 --- a/crates/rustfs-admin/src/lib.rs +++ b/crates/rustfs-admin/src/lib.rs @@ -169,6 +169,29 @@ pub struct RustfsServerInfo { pub backend: Option, #[serde(default)] pub pools: Option>>, + #[serde(default)] + pub servers: Vec, +} + +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsServerProperties { + #[serde(default)] + pub endpoint: String, + #[serde(default)] + pub drives: Vec, +} + +/// Erasure placement of one drive as reported by `/rustfs/admin/v3/info`. +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsDriveInfo { + #[serde(default)] + pub endpoint: String, + #[serde(default)] + pub state: String, + #[serde(default)] + pub pool_index: Option, + #[serde(default)] + pub set_index: Option, } #[derive(Debug, Clone, serde::Deserialize, PartialEq)] diff --git a/deploy/k8s-dev/operator-rbac.yaml b/deploy/k8s-dev/operator-rbac.yaml index b4444972..fcec9645 100755 --- a/deploy/k8s-dev/operator-rbac.yaml +++ b/deploy/k8s-dev/operator-rbac.yaml @@ -54,6 +54,9 @@ rules: - apiGroups: ["apps"] resources: ["statefulsets", "replicasets", "deployments"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + - apiGroups: ["policy"] + resources: ["poddisruptionbudgets"] + verbs: ["get", "list", "watch", "create", "patch", "delete"] - apiGroups: ["cert-manager.io"] resources: ["certificates"] verbs: ["get", "list", "watch", "create", "patch", "update", "delete"] diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index 74a2934f..9d30a7fd 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -1561,6 +1561,16 @@ spec: required: - name type: object + disablePodDisruptionBudget: + description: |- + Disables the PodDisruptionBudgets the operator creates for each pool. + + By default every pool that can lose a server without losing write quorum gets a + PodDisruptionBudget named after its StatefulSet, with `minAvailable` derived from the + erasure layout reported by RustFS. Setting this to `true` deletes those budgets; + PodDisruptionBudgets created outside the operator are left untouched. + nullable: true + type: boolean encryption: description: |- Encryption / KMS configuration for server-side encryption. diff --git a/deploy/rustfs-operator/templates/clusterrole.yaml b/deploy/rustfs-operator/templates/clusterrole.yaml index 4698dc90..a7dced25 100755 --- a/deploy/rustfs-operator/templates/clusterrole.yaml +++ b/deploy/rustfs-operator/templates/clusterrole.yaml @@ -56,6 +56,11 @@ rules: resources: ["statefulsets", "replicasets", "deployments"] verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + # PodDisruptionBudgets for tenant pools + - apiGroups: ["policy"] + resources: ["poddisruptionbudgets"] + verbs: ["get", "list", "watch", "create", "patch", "delete"] + # cert-manager Certificate orchestration and readiness watches - apiGroups: ["cert-manager.io"] resources: ["certificates"] diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index cdeab964..13bb6543 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -643,6 +643,7 @@ Useful Tenant-level fields: | `lifecycle` | Kubernetes container lifecycle hooks. | | `podManagementPolicy` | StatefulSet pod management policy. | | `podDeletionPolicyWhenNodeIsDown` | Node-down pod deletion behavior. | +| `disablePodDisruptionBudget` | Set to `true` to stop creating per-pool PodDisruptionBudgets and delete the ones the operator created. | | `securityContext` | Pod SecurityContext overrides for all RustFS Pools. | | `containerSecurityContext` | RustFS container SecurityContext overrides for all Pools. | | `hostUsers` | Optional Pod `hostUsers`. `false` isolates the user namespace (`restricted-v3`). An OpenShift empty security-context pair also defaults this to `false`. | @@ -767,6 +768,16 @@ Multi-pool Tenants cannot contain a single-node single-disk pool. The Operator r Force deletion can have data consistency implications. Use it only when the storage backend and operational procedure are designed for that failure mode. Force deletion requires the Node object to be deleted or marked with an effective `node.kubernetes.io/out-of-service` taint that the target Pod does not tolerate, so volume detach fencing is explicit. Before using force policies, confirm the node is powered off or otherwise isolated; deleting the Node object is treated as that operational assertion. +#### Pod disruption budgets + +The Operator creates one PodDisruptionBudget per pool, named after the pool StatefulSet and selecting the pool Pods. `minAvailable` is the pool's server count minus the number of servers that can be unavailable at once while every erasure set keeps write quorum. It is computed from the drive placement RustFS reports through the admin API: + +- an erasure set of `d` drives with standard parity `p` keeps write quorum while `d - p` drives are online, or `d - p + 1` when data and parity drives are equal; +- a server that holds `k` drives of one set takes `k` drives down with it, so the set tolerates `floor(spare drives / k)` unavailable servers; +- the smallest value across the pool's sets is used. RustFS reports a single standard parity, so the Operator uses the lower of that value and the automatic parity of each pool's set size. + +Pools that cannot lose a server without losing write quorum, such as single-server pools or two-server pools, get no PodDisruptionBudget. Budgets are created and updated only while every pool is rolled out, ready, and not decommissioning, and only when RustFS reports every drive of the pool online; otherwise the existing budgets are kept and the Operator retries. A node drain stops when evicting the next Pod would go below `minAvailable`. Set `spec.disablePodDisruptionBudget: true` to delete the operator-created budgets; PodDisruptionBudgets created outside the Operator are not touched. + ### 7.5 TLS Tenant TLS is configured under `spec.tls`. diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index 7336b4fd..76bc9f31 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -603,6 +603,7 @@ ConfigMap 更新投射到 Pod。需要让更新进入运行中的 Pod 时，请 | `lifecycle` | Kubernetes 容器 lifecycle hook。 | | `podManagementPolicy` | StatefulSet pod management policy。 | | `podDeletionPolicyWhenNodeIsDown` | 节点 NotReady/Unknown 时的 Pod 删除策略。 | +| `disablePodDisruptionBudget` | 设为 `true` 时不再为各 Pool 创建 PodDisruptionBudget，并删除 Operator 已创建的 PodDisruptionBudget。 | | `securityContext` | 所有 RustFS Pool 的 Pod SecurityContext 覆盖。 | | `containerSecurityContext` | 所有 Pool 的 RustFS 容器 SecurityContext 覆盖。 | | `hostUsers` | 可选的 Pod `hostUsers`。`false` 会隔离用户命名空间（`restricted-v3`）。OpenShift 空 securityContext 对也会默认渲染为 `false`。 | @@ -711,6 +712,16 @@ Operator 会自动管理以下环境变量： 强制删除可能影响数据一致性。只有当存储后端和运维流程明确支持该故障处理方式时才应启用。强制删除要求 Node 对象已删除，或 Node 带有对目标 Pod 生效且未被该 Pod tolerate 的 `node.kubernetes.io/out-of-service` taint，确保 volume detach fencing 是显式的。启用 force 类策略前，必须先确认节点已经关机或被隔离；删除 Node 对象会被视为这一运维断言。 +#### Pod disruption budget + +Operator 为每个 Pool 创建一个 PodDisruptionBudget，名称与该 Pool 的 StatefulSet 相同，选择该 Pool 的 Pod。`minAvailable` 等于 Pool 的服务器数减去在所有纠删集保持写仲裁的前提下可以同时不可用的服务器数，依据 RustFS admin API 报告的磁盘分布计算： + +- 一个有 `d` 块盘、标准校验盘数为 `p` 的纠删集，在线盘数不少于 `d - p` 时保持写仲裁；数据盘与校验盘数量相等时需要 `d - p + 1`； +- 一台服务器持有该纠删集的 `k` 块盘时，它下线会带走 `k` 块盘，因此该纠删集可容忍 `floor(冗余盘数 / k)` 台服务器不可用； +- 取 Pool 内所有纠删集中的最小值。RustFS 只报告一个标准校验盘数，因此 Operator 取该值与每个 Pool 按纠删集大小自动计算的校验盘数中的较小值。 + +无法在保持写仲裁的情况下失去一台服务器的 Pool（例如单服务器或双服务器 Pool）不会创建 PodDisruptionBudget。只有当所有 Pool 都完成滚动更新、全部就绪且没有进行退役，并且 RustFS 报告该 Pool 的所有磁盘在线时，才会创建或更新 PodDisruptionBudget；否则保留现有 PodDisruptionBudget 并稍后重试。节点 drain 在驱逐下一个 Pod 会低于 `minAvailable` 时停止。设置 `spec.disablePodDisruptionBudget: true` 会删除 Operator 创建的 PodDisruptionBudget；Operator 之外创建的 PodDisruptionBudget 不受影响。 + ### 7.5 TLS Tenant TLS 通过 `spec.tls` 配置。 diff --git a/src/lib.rs b/src/lib.rs index 9c608bfd..b6076264 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,6 +29,7 @@ use hyper_util::server::conn::auto::Builder as HyperBuilder; use hyper_util::service::TowerToHyperService; use k8s_openapi::api::apps::v1 as appsv1; use k8s_openapi::api::core::v1 as corev1; +use k8s_openapi::api::policy::v1 as policyv1; use k8s_openapi::api::rbac::v1 as rbacv1; use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition; use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1; @@ -290,6 +291,10 @@ async fn run_controller( .owns( Api::::all(client.clone()), watcher::Config::default(), + ) + .owns( + Api::::all(client.clone()), + watcher::Config::default(), ); let certificate_gvk = cert_manager_certificate_gvk(); diff --git a/src/reconcile.rs b/src/reconcile.rs index 8220513b..bc504cf1 100755 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -31,6 +31,7 @@ use std::time::Duration; use tracing::{debug, info, warn}; mod phases; +mod pod_disruption_budget; mod pool_lifecycle; mod provisioning; mod tls; @@ -113,7 +114,7 @@ pub async fn reconcile_rustfs(tenant: Arc, ctx: Arc) -> Result< let lifecycle_decisions = reconcile_pool_lifecycle(&ctx, &latest_tenant, &ns).await?; - let summary = reconcile_pool_statefulsets( + let mut summary = reconcile_pool_statefulsets( &ctx, &latest_tenant, &ns, @@ -122,6 +123,14 @@ pub async fn reconcile_rustfs(tenant: Arc, ctx: Arc) -> Result< &removed_pool_cleanup, ) .await?; + let budget_requeue = pod_disruption_budget::reconcile_pod_disruption_budgets( + &ctx, + &latest_tenant, + &ns, + summary.workloads_settled(), + ) + .await?; + summary.set_pod_disruption_budget_requeue_after(budget_requeue); finalize_tenant_status(&ctx, &latest_tenant, summary, tls_plan, pod_cleanup_outcome).await } diff --git a/src/reconcile/phases.rs b/src/reconcile/phases.rs index b79fa744..bc4086d8 100644 --- a/src/reconcile/phases.rs +++ b/src/reconcile/phases.rs @@ -53,10 +53,30 @@ pub(super) struct PoolReconcileSummary { any_lifecycle_failed: bool, any_lifecycle_canceled: bool, lifecycle_requeue_after: Option, + pod_disruption_budget_requeue_after: Option, total_replicas: i32, ready_replicas: i32, } +impl PoolReconcileSummary { + /// Every pool is rolled out and ready, and no pool lifecycle work is pending. + pub(super) fn workloads_settled(&self) -> bool { + !self.any_updating + && !self.any_degraded + && !self.any_lifecycle_reconciling + && !self.any_removed_pool_cleanup_reconciling + && !self.any_lifecycle_decommissioned + && !self.any_lifecycle_failed + && !self.any_lifecycle_canceled + && self.total_replicas > 0 + && self.ready_replicas == self.total_replicas + } + + pub(super) fn set_pod_disruption_budget_requeue_after(&mut self, after: Option) { + self.pod_disruption_budget_requeue_after = after; + } +} + const REMOVED_POOL_CLEANUP_REQUEUE_INTERVAL: Duration = Duration::from_secs(10); const POD_DELETION_POLICY_REQUEUE_INTERVAL: Duration = Duration::from_secs(30); const SERVICE_RECREATE_REQUEUE_INTERVAL: Duration = Duration::from_secs(5); @@ -964,8 +984,11 @@ fn reconcile_requeue_after( }; earliest_requeue_after( - earliest_requeue_after(lifecycle_requeue, updating_requeue), - pod_cleanup_requeue, + earliest_requeue_after( + earliest_requeue_after(lifecycle_requeue, updating_requeue), + pod_cleanup_requeue, + ), + summary.pod_disruption_budget_requeue_after, ) } diff --git a/src/reconcile/pod_disruption_budget.rs b/src/reconcile/pod_disruption_budget.rs new file mode 100644 index 00000000..c8051b90 --- /dev/null +++ b/src/reconcile/pod_disruption_budget.rs @@ -0,0 +1,444 @@ +// Copyright 2025 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. + +//! Per-pool PodDisruptionBudgets derived from the erasure layout RustFS reports. + +use super::pool_lifecycle::{expected_pool_cmd_line, rustfs_admin_client, same_cmd_line}; +use super::{Error, context_result, object_owned_by_tenant}; +use crate::context::Context; +use crate::sts::rustfs_client::{RustfsPoolListItem, RustfsServerInfo}; +use crate::types::v1alpha1::pool::Pool; +use crate::types::v1alpha1::tenant::{RUSTFS_TENANT_LABEL, Tenant}; +use k8s_openapi::api::policy::v1::{PodDisruptionBudget, PodDisruptionBudgetSpec}; +use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1; +use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString; +use kube::ResourceExt; +use kube::api::{DeleteParams, ListParams, Preconditions}; +use std::collections::{BTreeMap, HashSet}; +use std::time::Duration; +use tracing::{debug, warn}; + +/// How long to wait before retrying when RustFS cannot report a usable layout yet. +pub(super) const POD_DISRUPTION_BUDGET_RETRY_INTERVAL: Duration = Duration::from_secs(60); + +const DRIVE_STATE_OK: &str = "ok"; + +/// Reconciles one PodDisruptionBudget per pool. +/// +/// Budgets of pools removed from the spec, and all budgets when +/// `spec.disablePodDisruptionBudget` is set, are deleted right away. Budgets are created or +/// updated only while every pool is settled (no rollout, decommission or unready Pods), because +/// the erasure layout is read from the running RustFS cluster. Returns a requeue interval when +/// the layout could not be read and the budgets may be stale. +pub(super) async fn reconcile_pod_disruption_budgets( + ctx: &Context, + tenant: &Tenant, + namespace: &str, + workloads_settled: bool, +) -> Result, Error> { + let existing = list_owned_budgets(ctx, tenant, namespace).await?; + let disabled = tenant.spec.disable_pod_disruption_budget == Some(true); + let desired_names = tenant + .spec + .pools + .iter() + .map(|pool| budget_name(tenant, pool)) + .collect::>(); + + for budget in &existing { + if disabled || !desired_names.contains(&budget.name_any()) { + delete_budget(ctx, tenant, namespace, budget).await?; + } + } + if disabled || !workloads_settled { + return Ok(None); + } + + let layouts = match read_pool_layouts(ctx, tenant).await { + Ok(layouts) => layouts, + Err(message) => { + warn!( + tenant = %tenant.name(), + namespace = %namespace, + message = %message, + "cannot read RustFS erasure layout; keeping existing PodDisruptionBudgets" + ); + return Ok(Some(POD_DISRUPTION_BUDGET_RETRY_INTERVAL)); + } + }; + + let mut requeue = None; + for pool in &tenant.spec.pools { + let name = budget_name(tenant, pool); + let min_available = match pool_min_available(ctx, tenant, namespace, pool, &layouts) { + Ok(min_available) => min_available, + Err(message) => { + warn!( + tenant = %tenant.name(), + namespace = %namespace, + pool = %pool.name, + message = %message, + "cannot derive PodDisruptionBudget from RustFS erasure layout; keeping the existing one" + ); + requeue = Some(POD_DISRUPTION_BUDGET_RETRY_INTERVAL); + continue; + } + }; + + match min_available { + Some(min_available) => { + let budget = new_pool_budget(tenant, namespace, pool, min_available); + context_result(ctx.apply(&budget, namespace).await, ctx, tenant).await?; + } + None => { + debug!( + tenant = %tenant.name(), + namespace = %namespace, + pool = %pool.name, + "pool cannot lose a server without losing write quorum; no PodDisruptionBudget" + ); + if let Some(budget) = existing.iter().find(|budget| budget.name_any() == name) { + delete_budget(ctx, tenant, namespace, budget).await?; + } + } + } + } + + Ok(requeue) +} + +struct PoolLayouts { + info: RustfsServerInfo, + pools: Vec, + parity: usize, +} + +async fn read_pool_layouts(ctx: &Context, tenant: &Tenant) -> Result { + let client = rustfs_admin_client(ctx, tenant) + .await + .map_err(|error| format!("create RustFS admin client: {error}"))?; + let info = client + .server_info() + .await + .map_err(|error| format!("read RustFS server info: {error}"))?; + let pools = client + .list_pools() + .await + .map_err(|error| format!("list RustFS pools: {error}"))?; + let parity = info + .backend + .as_ref() + .and_then(|backend| backend.standard_sc_parity) + .ok_or_else(|| "RustFS server info has no standard storage class parity".to_string())?; + let parity = usize::try_from(parity) + .map_err(|_| format!("RustFS standard storage class parity {parity} is out of range"))?; + + Ok(PoolLayouts { + info, + pools, + parity, + }) +} + +/// `Ok(None)` means the pool cannot tolerate losing any server. +fn pool_min_available( + ctx: &Context, + tenant: &Tenant, + namespace: &str, + pool: &Pool, + layouts: &PoolLayouts, +) -> Result, String> { + let cmd_line = expected_pool_cmd_line(tenant, namespace, pool, ctx.cluster_domain())?; + let pool_id = layouts + .pools + .iter() + .find(|item| same_cmd_line(&item.cmd_line, &cmd_line)) + .map(|item| item.id) + .ok_or_else(|| format!("RustFS pool list has no pool with cmdline '{cmd_line}'"))?; + let expected_drives = usize::try_from(pool.servers) + .ok() + .zip(usize::try_from(pool.persistence.volumes_per_server).ok()) + .map(|(servers, volumes)| servers.saturating_mul(volumes)) + .ok_or_else(|| "pool has invalid servers or volumesPerServer".to_string())?; + let sets = pool_erasure_sets(&layouts.info, pool_id, expected_drives)?; + + let losses = tolerated_server_losses(&sets, layouts.parity); + if losses == 0 { + return Ok(None); + } + let losses = i32::try_from(losses).unwrap_or(i32::MAX); + Ok(Some(pool.servers.saturating_sub(losses).max(1))) +} + +/// Drive count per server for every erasure set of a pool, keyed by set index. +type ErasureSets = BTreeMap>; + +fn pool_erasure_sets( + info: &RustfsServerInfo, + pool_id: usize, + expected_drives: usize, +) -> Result { + let mut sets = ErasureSets::new(); + let mut drives = 0usize; + + for drive in info.servers.iter().flat_map(|server| server.drives.iter()) { + if drive + .pool_index + .and_then(|index| usize::try_from(index).ok()) + != Some(pool_id) + { + continue; + } + if drive.state != DRIVE_STATE_OK { + return Err(format!( + "drive '{}' is in state '{}'", + drive.endpoint, drive.state + )); + } + let set_index = drive + .set_index + .filter(|index| *index >= 0) + .ok_or_else(|| format!("drive '{}' has no erasure set index", drive.endpoint))?; + let host = endpoint_host(&drive.endpoint) + .ok_or_else(|| format!("drive endpoint '{}' has no host", drive.endpoint))?; + + *sets + .entry(set_index) + .or_default() + .entry(host.to_string()) + .or_default() += 1; + drives += 1; + } + + if drives != expected_drives { + return Err(format!( + "RustFS reports {drives} drives for pool {pool_id}, the spec declares {expected_drives}" + )); + } + Ok(sets) +} + +/// Number of servers of a pool that may be down at once while every erasure set keeps write +/// quorum. +/// +/// RustFS reports one standard parity for the whole cluster, while automatic parity is +/// resolved per pool from its set size, so the smaller of the two is used. +fn tolerated_server_losses(sets: &ErasureSets, reported_parity: usize) -> usize { + sets.values() + .map(|servers| { + let drives = servers.values().sum::(); + let parity = reported_parity.min(default_parity_count(drives)); + let data = drives.saturating_sub(parity); + let write_quorum = data + usize::from(data == parity); + let spare_drives = drives.saturating_sub(write_quorum); + let drives_per_server = servers.values().copied().max().unwrap_or(1).max(1); + spare_drives / drives_per_server + }) + .min() + .unwrap_or(0) +} + +/// Mirrors RustFS `default_parity_count` for automatic standard parity. +fn default_parity_count(drives: usize) -> usize { + match drives { + 0 | 1 => 0, + 2 | 3 => 1, + 4 | 5 => 2, + 6 | 7 => 3, + _ => 4, + } +} + +fn endpoint_host(endpoint: &str) -> Option<&str> { + let rest = endpoint + .split_once("://") + .map_or(endpoint, |(_, rest)| rest); + let host = rest.split('/').next().unwrap_or_default(); + (!host.is_empty()).then_some(host) +} + +fn budget_name(tenant: &Tenant, pool: &Pool) -> String { + format!("{}-{}", tenant.name(), pool.name) +} + +fn new_pool_budget( + tenant: &Tenant, + namespace: &str, + pool: &Pool, + min_available: i32, +) -> PodDisruptionBudget { + PodDisruptionBudget { + metadata: metav1::ObjectMeta { + name: Some(budget_name(tenant, pool)), + namespace: Some(namespace.to_string()), + owner_references: Some(vec![tenant.new_owner_ref()]), + labels: Some(tenant.pool_labels(pool)), + ..Default::default() + }, + spec: Some(PodDisruptionBudgetSpec { + min_available: Some(IntOrString::Int(min_available)), + selector: Some(metav1::LabelSelector { + match_labels: Some(tenant.pool_selector_labels(pool)), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + } +} + +async fn list_owned_budgets( + ctx: &Context, + tenant: &Tenant, + namespace: &str, +) -> Result, Error> { + let params = ListParams::default().labels(&format!("{RUSTFS_TENANT_LABEL}={}", tenant.name())); + let budgets = context_result( + ctx.list_with_params::(namespace, ¶ms) + .await, + ctx, + tenant, + ) + .await?; + + Ok(budgets + .items + .into_iter() + .filter(|budget| object_owned_by_tenant(&budget.metadata, tenant)) + .collect()) +} + +async fn delete_budget( + ctx: &Context, + tenant: &Tenant, + namespace: &str, + budget: &PodDisruptionBudget, +) -> Result<(), Error> { + let params = DeleteParams { + preconditions: Some(Preconditions { + uid: budget.metadata.uid.clone(), + resource_version: None, + }), + ..Default::default() + }; + match ctx + .delete_with_params::(&budget.name_any(), namespace, ¶ms) + .await + { + Err(error) if crate::context::is_kube_not_found(&error) => Ok(()), + result => context_result(result, ctx, tenant).await, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sts::rustfs_client::{RustfsDriveInfo, RustfsServerProperties}; + + fn info(servers: usize, volumes: usize, set_size: usize) -> RustfsServerInfo { + // RustFS expands `pod-{0...N}/rustfs{0...V}` with the server varying fastest. + let endpoints = (0..volumes) + .flat_map(|volume| { + (0..servers).map(move |server| { + ( + server, + format!("http://t-p-{server}.t-hl.ns.svc:9000/data/rustfs{volume}"), + ) + }) + }) + .collect::>(); + let mut properties = (0..servers) + .map(|server| RustfsServerProperties { + endpoint: format!("t-p-{server}.t-hl.ns.svc:9000"), + drives: Vec::new(), + }) + .collect::>(); + for (index, (server, endpoint)) in endpoints.into_iter().enumerate() { + properties[server].drives.push(RustfsDriveInfo { + endpoint, + state: DRIVE_STATE_OK.to_string(), + pool_index: Some(0), + set_index: Some(i32::try_from(index / set_size).unwrap()), + }); + } + RustfsServerInfo { + servers: properties, + ..Default::default() + } + } + + fn losses(servers: usize, volumes: usize, set_size: usize, parity: usize) -> usize { + let sets = pool_erasure_sets(&info(servers, volumes, set_size), 0, servers * volumes) + .expect("layout"); + tolerated_server_losses(&sets, parity) + } + + #[test] + fn three_servers_with_one_drive_tolerate_one_server() { + assert_eq!(losses(3, 1, 3, 1), 1); + } + + #[test] + fn equal_data_and_parity_needs_an_extra_drive() { + // 4 drives with EC:2 keep write quorum only with 3 drives online. + assert_eq!(losses(4, 1, 4, 2), 1); + assert_eq!(losses(2, 1, 2, 1), 0); + } + + #[test] + fn drives_per_server_in_a_set_limit_server_losses() { + // 8 servers x 2 drives, one set of 16, EC:4: each server holds two drives of the set. + assert_eq!(losses(8, 2, 16, 4), 2); + // 3 servers x 4 drives, one set of 12, EC:4: each server holds four drives. + assert_eq!(losses(3, 4, 12, 4), 1); + assert_eq!(losses(3, 4, 12, 2), 0); + } + + #[test] + fn single_server_pools_tolerate_nothing() { + assert_eq!(losses(1, 4, 4, 2), 0); + assert_eq!(losses(1, 1, 1, 0), 0); + } + + #[test] + fn reported_parity_is_capped_by_automatic_parity_of_the_pool() { + // Pool 0 reports EC:4; a pool of 3 drives resolves automatic parity to 1. + assert_eq!(losses(3, 1, 3, 4), 1); + } + + #[test] + fn layout_rejects_missing_or_unhealthy_drives() { + let mut layout = info(3, 1, 3); + assert!(pool_erasure_sets(&layout, 0, 4).is_err()); + layout.servers[1].drives[0].state = "offline".to_string(); + assert!(pool_erasure_sets(&layout, 0, 3).is_err()); + } + + #[test] + fn layout_ignores_other_pools() { + let mut layout = info(3, 1, 3); + layout.servers[0].drives[0].pool_index = Some(1); + let error = pool_erasure_sets(&layout, 0, 3).expect_err("one drive belongs to pool 1"); + assert!(error.contains("reports 2 drives")); + } + + #[test] + fn endpoint_host_keeps_port_and_drops_path() { + assert_eq!( + endpoint_host("https://t-p-0.t-hl.ns.svc:9000/data/rustfs0"), + Some("t-p-0.t-hl.ns.svc:9000") + ); + assert_eq!(endpoint_host("/data/rustfs0"), None); + } +} diff --git a/src/reconcile/pool_lifecycle.rs b/src/reconcile/pool_lifecycle.rs index 816c0bc6..959433a2 100644 --- a/src/reconcile/pool_lifecycle.rs +++ b/src/reconcile/pool_lifecycle.rs @@ -449,7 +449,7 @@ async fn reconcile_single_pool_lifecycle( } } -async fn rustfs_admin_client( +pub(super) async fn rustfs_admin_client( ctx: &Context, tenant: &Tenant, ) -> Result { @@ -510,7 +510,7 @@ async fn find_rustfs_pool( }) } -fn expected_pool_cmd_line( +pub(super) fn expected_pool_cmd_line( tenant: &Tenant, namespace: &str, pool: &Pool, @@ -536,7 +536,7 @@ fn expected_pool_cmd_line( Ok(tenant.rustfs_pool_volume_spec(pool, scheme, namespace, cluster_domain)) } -fn same_cmd_line(left: &str, right: &str) -> bool { +pub(super) fn same_cmd_line(left: &str, right: &str) -> bool { left.trim() == right.trim() } diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index f07f5cca..dc7e208b 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -26,10 +26,10 @@ pub use rustfs_admin::{ BucketLifecycleRuleStatus, BucketLifecycleTransition, BucketObjectLockConfiguration, BucketObjectLockDefaultRetention, BucketObjectLockRetentionMode, BucketObjectLockRetentionPeriod, BucketVersioningState, BucketVersioningUpdate, - CreateBucketResult, RustfsClientError, RustfsCredentials, RustfsErasureBackend, - RustfsErasureSetInfo, RustfsPoolDecommissionInfo, RustfsPoolListItem, RustfsPoolStatus, - RustfsServerInfo, RustfsServerUsage, RustfsTier, RustfsTierConfig, RustfsUserInfo, - canonicalize_bucket_lifecycle_xml, + CreateBucketResult, RustfsClientError, RustfsCredentials, RustfsDriveInfo, + RustfsErasureBackend, RustfsErasureSetInfo, RustfsPoolDecommissionInfo, RustfsPoolListItem, + RustfsPoolStatus, RustfsServerInfo, RustfsServerProperties, RustfsServerUsage, RustfsTier, + RustfsTierConfig, RustfsUserInfo, canonicalize_bucket_lifecycle_xml, }; /// Tenant-aware wrapper around the kube-agnostic RustFS admin client. diff --git a/src/tenant_monitor.rs b/src/tenant_monitor.rs index 6b428e8f..105e3add 100644 --- a/src/tenant_monitor.rs +++ b/src/tenant_monitor.rs @@ -287,6 +287,7 @@ mod tests { drives_per_set: vec![4], }), pools: Some(pools), + servers: Vec::new(), }; let metrics = storage_metrics_from_info(&info); diff --git a/src/types/v1alpha1/tenant.rs b/src/types/v1alpha1/tenant.rs index 5a387e76..8e2a961e 100755 --- a/src/types/v1alpha1/tenant.rs +++ b/src/types/v1alpha1/tenant.rs @@ -144,6 +144,15 @@ pub struct TenantSpec { #[serde(default, skip_serializing_if = "Option::is_none")] pub pod_deletion_policy_when_node_is_down: Option, + /// Disables the PodDisruptionBudgets the operator creates for each pool. + /// + /// By default every pool that can lose a server without losing write quorum gets a + /// PodDisruptionBudget named after its StatefulSet, with `minAvailable` derived from the + /// erasure layout reported by RustFS. Setting this to `true` deletes those budgets; + /// PodDisruptionBudgets created outside the operator are left untouched. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub disable_pod_disruption_budget: Option, + /// Additional RustFS container environment variables. /// /// `RUSTFS_KMS_*` is reserved; configure KMS through `spec.encryption`. From e44d1242d68b764641ab20e8a480ee15ed625be8 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Fri, 30 Oct 2026 13:16:00 +0200 Subject: [PATCH 2/2] test(reconcile): cover the PodDisruptionBudget reconcile step Read the erasure layout through an injected future so the reconcile can run against a mocked API server, and document removing user-created budgets before upgrading. --- docs/operator-user-guide.md | 2 +- docs/operator-user-guide.zh-CN.md | 2 +- src/reconcile/pod_disruption_budget.rs | 299 ++++++++++++++++++++++++- 3 files changed, 300 insertions(+), 3 deletions(-) diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index 13bb6543..7a9c4890 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -776,7 +776,7 @@ The Operator creates one PodDisruptionBudget per pool, named after the pool Stat - a server that holds `k` drives of one set takes `k` drives down with it, so the set tolerates `floor(spare drives / k)` unavailable servers; - the smallest value across the pool's sets is used. RustFS reports a single standard parity, so the Operator uses the lower of that value and the automatic parity of each pool's set size. -Pools that cannot lose a server without losing write quorum, such as single-server pools or two-server pools, get no PodDisruptionBudget. Budgets are created and updated only while every pool is rolled out, ready, and not decommissioning, and only when RustFS reports every drive of the pool online; otherwise the existing budgets are kept and the Operator retries. A node drain stops when evicting the next Pod would go below `minAvailable`. Set `spec.disablePodDisruptionBudget: true` to delete the operator-created budgets; PodDisruptionBudgets created outside the Operator are not touched. +Pools that cannot lose a server without losing write quorum, such as single-server pools or two-server pools, get no PodDisruptionBudget. Budgets are created and updated only while every pool is rolled out, ready, and not decommissioning, and only when RustFS reports every drive of the pool online; otherwise the existing budgets are kept and the Operator retries. A node drain stops when evicting the next Pod would go below `minAvailable`. Set `spec.disablePodDisruptionBudget: true` to delete the operator-created budgets; PodDisruptionBudgets created outside the Operator are not touched. Remove any PodDisruptionBudget of your own that selects pool Pods before upgrading: Kubernetes rejects evictions of a Pod selected by more than one budget, and a budget with the same name blocks the reconcile. ### 7.5 TLS diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index 76bc9f31..8f01d9fb 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -720,7 +720,7 @@ Operator 为每个 Pool 创建一个 PodDisruptionBudget，名称与该 Pool 的 - 一台服务器持有该纠删集的 `k` 块盘时，它下线会带走 `k` 块盘，因此该纠删集可容忍 `floor(冗余盘数 / k)` 台服务器不可用； - 取 Pool 内所有纠删集中的最小值。RustFS 只报告一个标准校验盘数，因此 Operator 取该值与每个 Pool 按纠删集大小自动计算的校验盘数中的较小值。 -无法在保持写仲裁的情况下失去一台服务器的 Pool（例如单服务器或双服务器 Pool）不会创建 PodDisruptionBudget。只有当所有 Pool 都完成滚动更新、全部就绪且没有进行退役，并且 RustFS 报告该 Pool 的所有磁盘在线时，才会创建或更新 PodDisruptionBudget；否则保留现有 PodDisruptionBudget 并稍后重试。节点 drain 在驱逐下一个 Pod 会低于 `minAvailable` 时停止。设置 `spec.disablePodDisruptionBudget: true` 会删除 Operator 创建的 PodDisruptionBudget；Operator 之外创建的 PodDisruptionBudget 不受影响。 +无法在保持写仲裁的情况下失去一台服务器的 Pool（例如单服务器或双服务器 Pool）不会创建 PodDisruptionBudget。只有当所有 Pool 都完成滚动更新、全部就绪且没有进行退役，并且 RustFS 报告该 Pool 的所有磁盘在线时，才会创建或更新 PodDisruptionBudget；否则保留现有 PodDisruptionBudget 并稍后重试。节点 drain 在驱逐下一个 Pod 会低于 `minAvailable` 时停止。设置 `spec.disablePodDisruptionBudget: true` 会删除 Operator 创建的 PodDisruptionBudget；Operator 之外创建的 PodDisruptionBudget 不受影响。升级前请删除自行创建的、选择 Pool Pod 的 PodDisruptionBudget：Kubernetes 会拒绝驱逐被多个 PodDisruptionBudget 选中的 Pod，同名的 PodDisruptionBudget 会阻塞 reconcile。 ### 7.5 TLS diff --git a/src/reconcile/pod_disruption_budget.rs b/src/reconcile/pod_disruption_budget.rs index c8051b90..88eac459 100644 --- a/src/reconcile/pod_disruption_budget.rs +++ b/src/reconcile/pod_disruption_budget.rs @@ -46,6 +46,24 @@ pub(super) async fn reconcile_pod_disruption_budgets( tenant: &Tenant, namespace: &str, workloads_settled: bool, +) -> Result, Error> { + reconcile_budgets_with_layouts( + ctx, + tenant, + namespace, + workloads_settled, + read_pool_layouts(ctx, tenant), + ) + .await +} + +/// `read_layouts` is awaited only when budgets are enabled and every pool is settled. +async fn reconcile_budgets_with_layouts( + ctx: &Context, + tenant: &Tenant, + namespace: &str, + workloads_settled: bool, + read_layouts: impl Future>, ) -> Result, Error> { let existing = list_owned_budgets(ctx, tenant, namespace).await?; let disabled = tenant.spec.disable_pod_disruption_budget == Some(true); @@ -65,7 +83,7 @@ pub(super) async fn reconcile_pod_disruption_budgets( return Ok(None); } - let layouts = match read_pool_layouts(ctx, tenant).await { + let layouts = match read_layouts.await { Ok(layouts) => layouts, Err(message) => { warn!( @@ -345,6 +363,13 @@ async fn delete_budget( mod tests { use super::*; use crate::sts::rustfs_client::{RustfsDriveInfo, RustfsServerProperties}; + use http::{Method, Request, Response, StatusCode}; + use http_body_util::BodyExt; + use kube::{Client, client::Body}; + use serde_json::{Value, json}; + use std::convert::Infallible; + use std::sync::{Arc, Mutex}; + use tower::service_fn; fn info(servers: usize, volumes: usize, set_size: usize) -> RustfsServerInfo { // RustFS expands `pod-{0...N}/rustfs{0...V}` with the server varying fastest. @@ -441,4 +466,276 @@ mod tests { ); assert_eq!(endpoint_host("/data/rustfs0"), None); } + + const BUDGETS_PATH: &str = "/apis/policy/v1/namespaces/default/poddisruptionbudgets"; + + #[derive(Debug)] + struct KubeCall { + method: Method, + path: String, + body: Option, + } + + type KubeCalls = Arc>>; + + fn tenant_with_pools(servers: i32, pools: &[&str]) -> Tenant { + let mut tenant = crate::tests::create_test_tenant(None, None); + let mut template = tenant.spec.pools[0].clone(); + template.servers = servers; + template.persistence.volumes_per_server = 1; + tenant.spec.pools = pools + .iter() + .map(|name| Pool { + name: (*name).to_string(), + ..template.clone() + }) + .collect(); + tenant + } + + fn existing_budget(tenant: &Tenant, pool: &str, owned: bool) -> PodDisruptionBudget { + PodDisruptionBudget { + metadata: metav1::ObjectMeta { + name: Some(format!("{}-{pool}", tenant.name())), + namespace: Some("default".to_string()), + uid: Some(format!("{pool}-budget-uid")), + labels: Some(BTreeMap::from([( + RUSTFS_TENANT_LABEL.to_string(), + tenant.name(), + )])), + owner_references: owned.then(|| vec![tenant.new_owner_ref()]), + ..Default::default() + }, + ..Default::default() + } + } + + fn mock_kube(existing: Vec) -> (Context, KubeCalls) { + let calls = KubeCalls::default(); + let service = service_fn({ + let calls = Arc::clone(&calls); + move |request: Request| { + let calls = Arc::clone(&calls); + let existing = existing.clone(); + async move { + let method = request.method().clone(); + let path = request.uri().path().to_string(); + let bytes = request + .into_body() + .collect() + .await + .expect("request body should be readable") + .to_bytes(); + let body = (!bytes.is_empty()).then(|| { + serde_json::from_slice::(&bytes).expect("request body is JSON") + }); + let response = match method { + Method::GET if path == BUDGETS_PATH => json!({ + "apiVersion": "policy/v1", + "kind": "PodDisruptionBudgetList", + "metadata": {}, + "items": existing + }), + Method::PATCH if path.starts_with(BUDGETS_PATH) => { + body.clone().expect("apply carries the budget") + } + Method::DELETE if path.starts_with(BUDGETS_PATH) => json!({ + "apiVersion": "v1", + "kind": "Status", + "status": "Success" + }), + _ => panic!("unexpected Kubernetes request: {method} {path}"), + }; + calls + .lock() + .expect("calls lock") + .push(KubeCall { method, path, body }); + Ok::<_, Infallible>( + Response::builder() + .status(StatusCode::OK) + .body(Body::from( + serde_json::to_vec(&response).expect("response serializes"), + )) + .expect("response builds"), + ) + } + } + }); + (Context::new(Client::new(service, "default")), calls) + } + + fn layouts_for(ctx: &Context, tenant: &Tenant, parity: usize) -> PoolLayouts { + let pool = &tenant.spec.pools[0]; + let servers = usize::try_from(pool.servers).expect("servers"); + let cmd_line = expected_pool_cmd_line(tenant, "default", pool, ctx.cluster_domain()) + .expect("pool cmdline"); + PoolLayouts { + info: info(servers, 1, servers), + pools: vec![ + serde_json::from_value(json!({ + "id": 0, + "cmdline": cmd_line, + "lastUpdate": "", + "status": "active" + })) + .expect("pool list item"), + ], + parity, + } + } + + async fn layouts_never_read() -> Result { + panic!("the erasure layout must not be read") + } + + fn requests(calls: &KubeCalls) -> Vec<(Method, String)> { + calls + .lock() + .expect("calls lock") + .iter() + .map(|call| (call.method.clone(), call.path.clone())) + .collect() + } + + fn budget_path(name: &str) -> String { + format!("{BUDGETS_PATH}/{name}") + } + + #[tokio::test] + async fn settled_pool_gets_a_budget_from_its_layout() { + let tenant = tenant_with_pools(4, &["pool-0"]); + let (ctx, calls) = mock_kube(Vec::new()); + let layouts = layouts_for(&ctx, &tenant, 2); + + let requeue = + reconcile_budgets_with_layouts(&ctx, &tenant, "default", true, async { Ok(layouts) }) + .await + .expect("reconcile"); + + assert_eq!(requeue, None); + let calls = calls.lock().expect("calls lock"); + assert_eq!(calls.len(), 2); + assert_eq!(calls[1].method, Method::PATCH); + assert_eq!(calls[1].path, budget_path("test-tenant-pool-0")); + let budget: PodDisruptionBudget = + serde_json::from_value(calls[1].body.clone().expect("budget")).expect("budget"); + let spec = budget.spec.expect("spec"); + assert_eq!(spec.min_available, Some(IntOrString::Int(3))); + assert_eq!( + spec.selector.and_then(|selector| selector.match_labels), + Some(tenant.pool_selector_labels(&tenant.spec.pools[0])) + ); + assert!(object_owned_by_tenant(&budget.metadata, &tenant)); + } + + #[tokio::test] + async fn pool_without_spare_servers_loses_its_budget() { + let tenant = tenant_with_pools(2, &["pool-0"]); + let (ctx, calls) = mock_kube(vec![existing_budget(&tenant, "pool-0", true)]); + let layouts = layouts_for(&ctx, &tenant, 1); + + reconcile_budgets_with_layouts(&ctx, &tenant, "default", true, async { Ok(layouts) }) + .await + .expect("reconcile"); + + assert_eq!( + requests(&calls), + vec![ + (Method::GET, BUDGETS_PATH.to_string()), + (Method::DELETE, budget_path("test-tenant-pool-0")), + ] + ); + } + + #[tokio::test] + async fn disabling_deletes_only_budgets_owned_by_the_tenant() { + let mut tenant = tenant_with_pools(4, &["pool-0", "pool-1"]); + tenant.spec.disable_pod_disruption_budget = Some(true); + let (ctx, calls) = mock_kube(vec![ + existing_budget(&tenant, "pool-0", true), + existing_budget(&tenant, "pool-1", false), + ]); + + let requeue = + reconcile_budgets_with_layouts(&ctx, &tenant, "default", true, layouts_never_read()) + .await + .expect("reconcile"); + + assert_eq!(requeue, None); + assert_eq!( + requests(&calls), + vec![ + (Method::GET, BUDGETS_PATH.to_string()), + (Method::DELETE, budget_path("test-tenant-pool-0")), + ] + ); + let calls = calls.lock().expect("calls lock"); + assert_eq!( + calls[1] + .body + .as_ref() + .map(|body| &body["preconditions"]["uid"]), + Some(&json!("pool-0-budget-uid")) + ); + } + + #[tokio::test] + async fn removed_pool_budget_is_deleted_while_pools_are_not_settled() { + let tenant = tenant_with_pools(4, &["pool-0"]); + let (ctx, calls) = mock_kube(vec![ + existing_budget(&tenant, "pool-0", true), + existing_budget(&tenant, "pool-1", true), + ]); + + let requeue = + reconcile_budgets_with_layouts(&ctx, &tenant, "default", false, layouts_never_read()) + .await + .expect("reconcile"); + + assert_eq!(requeue, None); + assert_eq!( + requests(&calls), + vec![ + (Method::GET, BUDGETS_PATH.to_string()), + (Method::DELETE, budget_path("test-tenant-pool-1")), + ] + ); + } + + #[tokio::test] + async fn unreadable_layout_keeps_existing_budgets_and_retries() { + let tenant = tenant_with_pools(4, &["pool-0"]); + let (ctx, calls) = mock_kube(vec![existing_budget(&tenant, "pool-0", true)]); + + let requeue = reconcile_budgets_with_layouts(&ctx, &tenant, "default", true, async { + Err("RustFS is not reachable".to_string()) + }) + .await + .expect("reconcile"); + + assert_eq!(requeue, Some(POD_DISRUPTION_BUDGET_RETRY_INTERVAL)); + assert_eq!( + requests(&calls), + vec![(Method::GET, BUDGETS_PATH.to_string())] + ); + } + + #[tokio::test] + async fn unhealthy_drive_keeps_the_pool_budget_and_retries() { + let tenant = tenant_with_pools(4, &["pool-0"]); + let (ctx, calls) = mock_kube(vec![existing_budget(&tenant, "pool-0", true)]); + let mut layouts = layouts_for(&ctx, &tenant, 2); + layouts.info.servers[2].drives[0].state = "offline".to_string(); + + let requeue = + reconcile_budgets_with_layouts(&ctx, &tenant, "default", true, async { Ok(layouts) }) + .await + .expect("reconcile"); + + assert_eq!(requeue, Some(POD_DISRUPTION_BUDGET_RETRY_INTERVAL)); + assert_eq!( + requests(&calls), + vec![(Method::GET, BUDGETS_PATH.to_string())] + ); + } }