diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c55246..9fa213e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,23 @@ All notable changes to RustFS Operator are documented in this file. The format i ## [Unreleased] +### Added + +- Tenant `spec.network` for Service IP families and IPv6 listen addresses, plus dual-stack binds + for operator observability, STS, and Console sockets. +- Tenant `spec.hostUsers` and OpenShift `hostUsers: false` defaults for `restricted-v3`. +- Tenant bucket canned anonymous access and ConfigMap-sourced bucket policies. + +### Fixed + +- Provisioning now requeues transient RustFS admin/S3 and Kubernetes failures instead of leaving + policies, users, and buckets failed until an unrelated object change. + +### Changed + +- Documented that distinct-physical-disk erasure failures and a separate data-plane operator are + outside this controller's scope. + ## [0.0.6] - 2026-08-22 ### Added diff --git a/deploy/rustfs-operator/README.md b/deploy/rustfs-operator/README.md index 4ea11f8..87bd134 100755 --- a/deploy/rustfs-operator/README.md +++ b/deploy/rustfs-operator/README.md @@ -59,9 +59,10 @@ allowed runtime identity. Keep `openshift.enabled=false` and omit the Tenant fields on generic Kubernetes so the RustFS defaults remain in effect. This profile provides SCC-compatible manifests but does not by itself imply -OpenShift certification or OperatorHub distribution. Support is currently -limited to `restricted-v2`; the `restricted-v3` requirement to set -`spec.hostUsers: false` is not implemented. +OpenShift certification or OperatorHub distribution. Chart-managed Deployments +set `hostUsers: false` when `openshift.enabled=true`, and Tenant workloads do +the same for an explicit empty security-context pair or `spec.hostUsers: false`, +covering the OpenShift `restricted-v3` host-user-namespace control. The RustFS server image is an independent prerequisite. It must support an arbitrary SCC-assigned UID: writable image-layer directories, including @@ -109,6 +110,7 @@ The following table lists the configurable parameters of the RustFS Operator cha | `operator.prometheusRule.enabled` | Create Prometheus alert rules for operator and tenant storage health | `false` | | `operator.tenantMonitor.enabled` | Poll RustFS tenant storage health and capacity metrics | `true` | | `operator.tenantMonitor.intervalSeconds` | Tenant storage monitor interval | `300` | +| `operator.bindAddress` | Optional literal IPv4/IPv6 bind address for operator HTTP sockets; empty prefers `::` then `0.0.0.0` | `""` | | `clusterDomain` | Kubernetes cluster DNS domain used for Tenant peer URLs, generated TLS SANs, and operator STS auto TLS | `cluster.local` | | `operator.env` | Environment variables | `[{name: RUST_LOG, value: info}]` | | `operator.nodeSelector` | Node selector for pod placement | `{}` | @@ -236,7 +238,9 @@ The generated ClusterRole grants only `get`, `list`, and `watch` for Secrets and | Parameter | Description | Default | |-----------|-------------|---------| -| `openshift.enabled` | Omit chart-managed Deployment security contexts and delegate runtime identity to OpenShift SCC | `false` | +| `openshift.enabled` | Omit chart-managed Deployment security contexts, set `hostUsers: false`, and delegate runtime identity to OpenShift SCC | `false` | +| `network.ipFamilyPolicy` | Optional Service `ipFamilyPolicy` for chart-managed Services | `""` | +| `network.ipFamilies` | Optional Service `ipFamilies` for chart-managed Services | `[]` | | `namespace` | Namespace to deploy to | `""` (uses release namespace) | | `commonLabels` | Labels to add to all resources | `{}` | | `commonAnnotations` | Annotations to add to all resources | `{}` | diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index 6f28965..154febc 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -30,6 +30,14 @@ spec: description: Buckets that should exist in the RustFS tenant. items: properties: + anonymous: + description: Canned anonymous access for this bucket. Mutually exclusive with `policy`. + enum: + - Private + - Download + - Upload + - Public + type: string deletionPolicy: enum: - Retain @@ -45,12 +53,36 @@ spec: objectLock: nullable: true type: boolean + policy: + description: Custom bucket policy document sourced from a ConfigMap. Mutually exclusive with `anonymous`. + nullable: true + properties: + configMapKeyRef: + properties: + key: + maxLength: 253 + minLength: 1 + type: string + name: + maxLength: 253 + minLength: 1 + type: string + required: + - key + - name + type: object + required: + - configMapKeyRef + type: object region: nullable: true type: string required: - name type: object + x-kubernetes-validations: + - message: bucket policy and anonymous access are mutually exclusive + rule: '!(has(self.policy) && has(self.anonymous) && self.anonymous != ''Private'')' maxItems: 1024 type: array x-kubernetes-list-map-keys: @@ -357,6 +389,14 @@ spec: - name type: object type: array + hostUsers: + description: |- + Pod `hostUsers` for generated RustFS workloads. + + `false` isolates the user namespace and satisfies OpenShift `restricted-v3`. + When omitted, an OpenShift-style empty security-context pair also renders `hostUsers: false`. + nullable: true + type: boolean image: nullable: true type: string @@ -572,6 +612,32 @@ spec: default: /data nullable: true type: string + network: + description: |- + Tenant Service IP family policy and RustFS listen addresses. + When omitted, generated Services inherit the cluster default and RustFS listens on IPv4. + nullable: true + properties: + ipFamilies: + items: + description: Kubernetes Service IP family values. + enum: + - IPv4 + - IPv6 + type: string + maxItems: 2 + type: array + x-kubernetes-list-type: set + ipFamilyPolicy: + description: Kubernetes Service IP family policy values. + enum: + - SingleStack + - PreferDualStack + - RequireDualStack + - null + nullable: true + type: string + type: object podDeletionPolicyWhenNodeIsDown: description: |- Controls how the operator handles Pods when the node hosting them is down (NotReady/Unknown). diff --git a/deploy/rustfs-operator/templates/_helpers.tpl b/deploy/rustfs-operator/templates/_helpers.tpl index 6903f30..01c98d6 100755 --- a/deploy/rustfs-operator/templates/_helpers.tpl +++ b/deploy/rustfs-operator/templates/_helpers.tpl @@ -76,6 +76,22 @@ Name of the namespaced Role used by STS auto TLS. {{- printf "%s-sts-tls" (include "rustfs-operator.fullname" .) | trunc 63 | trimSuffix "-" }} {{- end }} +{{/* +Optional Service ipFamilyPolicy / ipFamilies from values.network. +*/}} +{{- define "rustfs-operator.serviceNetwork" -}} +{{- $network := default dict .Values.network -}} +{{- if $network.ipFamilyPolicy }} +ipFamilyPolicy: {{ $network.ipFamilyPolicy }} +{{- end }} +{{- if $network.ipFamilies }} +ipFamilies: +{{- range $network.ipFamilies }} +- {{ . }} +{{- end }} +{{- end }} +{{- end }} + {{/* Create the name of the console service account to use */}} diff --git a/deploy/rustfs-operator/templates/console-deployment.yaml b/deploy/rustfs-operator/templates/console-deployment.yaml index 78bb9ba..9de4cf3 100755 --- a/deploy/rustfs-operator/templates/console-deployment.yaml +++ b/deploy/rustfs-operator/templates/console-deployment.yaml @@ -7,6 +7,7 @@ "CONSOLE_LOGIN_ADMISSION_MAX_IN_FLIGHT" "console.loginAdmission.maxInFlight" "CONSOLE_LOGIN_ADMISSION_BODY_LIMIT_BYTES" "console.loginAdmission.bodyLimitBytes" "CONSOLE_LOGIN_ADMISSION_TIMEOUT_SECONDS" "console.loginAdmission.timeoutSeconds" + "CONSOLE_BIND_ADDRESS" "console.bindAddress" -}} {{- range $env := .Values.console.env }} {{- if hasKey $reservedConsoleEnv $env.name -}} @@ -48,6 +49,9 @@ spec: checksum/secret: {{ include (print $.Template.BasePath "/console-secret.yaml") . | sha256sum }} spec: serviceAccountName: {{ include "rustfs-operator.consoleServiceAccountName" . }} + {{- if $openShiftEnabled }} + hostUsers: false + {{- end }} {{- with .Values.console.imagePullSecrets }} imagePullSecrets: {{- toYaml . | nindent 8 }} @@ -71,6 +75,10 @@ spec: secretKeyRef: name: {{ include "rustfs-operator.fullname" . }}-console-secret key: jwt-secret + {{- with .Values.console.bindAddress }} + - name: CONSOLE_BIND_ADDRESS + value: {{ . | quote }} + {{- end }} {{- with $consoleLoginAdmission }} - name: CONSOLE_LOGIN_ADMISSION_REQUESTS_PER_SECOND value: {{ .requestsPerSecond | quote }} diff --git a/deploy/rustfs-operator/templates/console-frontend-deployment.yaml b/deploy/rustfs-operator/templates/console-frontend-deployment.yaml index 9022229..f9c210c 100755 --- a/deploy/rustfs-operator/templates/console-frontend-deployment.yaml +++ b/deploy/rustfs-operator/templates/console-frontend-deployment.yaml @@ -21,6 +21,9 @@ spec: {{- include "rustfs-operator.selectorLabels" . | nindent 8 }} app.kubernetes.io/component: console-frontend spec: + {{- if $openShiftEnabled }} + hostUsers: false + {{- end }} {{- with .Values.console.frontend.imagePullSecrets }} imagePullSecrets: {{- toYaml . | nindent 8 }} diff --git a/deploy/rustfs-operator/templates/console-frontend-service.yaml b/deploy/rustfs-operator/templates/console-frontend-service.yaml index 7cb3408..11fc766 100755 --- a/deploy/rustfs-operator/templates/console-frontend-service.yaml +++ b/deploy/rustfs-operator/templates/console-frontend-service.yaml @@ -9,6 +9,7 @@ metadata: app.kubernetes.io/component: console-frontend spec: type: ClusterIP + {{- include "rustfs-operator.serviceNetwork" . | nindent 2 }} ports: - port: {{ .Values.console.frontend.service.port }} targetPort: http diff --git a/deploy/rustfs-operator/templates/console-service.yaml b/deploy/rustfs-operator/templates/console-service.yaml index 2301380..2adbce3 100755 --- a/deploy/rustfs-operator/templates/console-service.yaml +++ b/deploy/rustfs-operator/templates/console-service.yaml @@ -13,6 +13,7 @@ metadata: {{- end }} spec: type: {{ .Values.console.service.type }} + {{- include "rustfs-operator.serviceNetwork" . | nindent 2 }} {{- if and (eq .Values.console.service.type "LoadBalancer") .Values.console.service.loadBalancerIP }} loadBalancerIP: {{ .Values.console.service.loadBalancerIP }} {{- end }} diff --git a/deploy/rustfs-operator/templates/deployment.yaml b/deploy/rustfs-operator/templates/deployment.yaml index e33c448..f0425c9 100755 --- a/deploy/rustfs-operator/templates/deployment.yaml +++ b/deploy/rustfs-operator/templates/deployment.yaml @@ -24,6 +24,7 @@ "OPERATOR_STS_TLS_AUTO" "sts.tls.auto" "OPERATOR_TENANT_MONITOR_ENABLED" "operator.tenantMonitor.enabled" "OPERATOR_TENANT_MONITOR_INTERVAL_SECONDS" "operator.tenantMonitor.intervalSeconds" + "OPERATOR_BIND_ADDRESS" "operator.bindAddress" "POD_NAME" "the Pod metadata.name field" -}} {{- range $env := .Values.operator.env }} @@ -56,6 +57,9 @@ spec: app.kubernetes.io/component: operator spec: serviceAccountName: {{ include "rustfs-operator.serviceAccountName" . }} + {{- if $openShiftEnabled }} + hostUsers: false + {{- end }} {{- with .Values.operator.imagePullSecrets }} imagePullSecrets: {{- toYaml . | nindent 8 }} @@ -137,6 +141,10 @@ spec: valueFrom: fieldRef: fieldPath: metadata.namespace + {{- with .Values.operator.bindAddress }} + - name: OPERATOR_BIND_ADDRESS + value: {{ . | quote }} + {{- end }} {{- if .Values.sts.enabled }} - name: OPERATOR_STS_PORT value: {{ .Values.sts.port | quote }} diff --git a/deploy/rustfs-operator/templates/operator-metrics-service.yaml b/deploy/rustfs-operator/templates/operator-metrics-service.yaml index 0f50c18..9d69aff 100644 --- a/deploy/rustfs-operator/templates/operator-metrics-service.yaml +++ b/deploy/rustfs-operator/templates/operator-metrics-service.yaml @@ -13,6 +13,7 @@ metadata: {{- end }} spec: type: {{ .Values.operator.metrics.service.type }} + {{- include "rustfs-operator.serviceNetwork" . | nindent 2 }} {{- if and (eq .Values.operator.metrics.service.type "ClusterIP") .Values.operator.metrics.service.clusterIP }} clusterIP: {{ .Values.operator.metrics.service.clusterIP }} {{- end }} diff --git a/deploy/rustfs-operator/templates/operator-sts-service.yaml b/deploy/rustfs-operator/templates/operator-sts-service.yaml index df4f953..302e623 100644 --- a/deploy/rustfs-operator/templates/operator-sts-service.yaml +++ b/deploy/rustfs-operator/templates/operator-sts-service.yaml @@ -16,6 +16,7 @@ metadata: {{- end }} spec: type: {{ .Values.sts.service.type }} + {{- include "rustfs-operator.serviceNetwork" . | nindent 2 }} {{- if and (eq .Values.sts.service.type "ClusterIP") .Values.sts.service.clusterIP }} clusterIP: {{ .Values.sts.service.clusterIP }} {{- end }} diff --git a/deploy/rustfs-operator/values.yaml b/deploy/rustfs-operator/values.yaml index ae52c8c..1c88bbd 100755 --- a/deploy/rustfs-operator/values.yaml +++ b/deploy/rustfs-operator/values.yaml @@ -5,10 +5,22 @@ clusterDomain: cluster.local # OpenShift installation compatibility. When enabled, the chart omits Pod and # container securityContext fields from its own Deployments so the namespace -# SecurityContextConstraints (SCC) can assign an allowed UID and FSGroup. +# SecurityContextConstraints (SCC) can assign an allowed UID and FSGroup, and +# sets hostUsers: false for restricted-v3. openshift: enabled: false +# Optional IP family policy for chart-managed Services. Empty inherits the +# cluster default. Example dual-stack: +# network: +# ipFamilyPolicy: PreferDualStack +# ipFamilies: +# - IPv4 +# - IPv6 +network: + ipFamilyPolicy: "" + ipFamilies: [] + # Operator deployment configuration operator: # Number of operator replicas @@ -61,6 +73,10 @@ operator: enabled: true intervalSeconds: 300 + # Empty binds the IPv6 unspecified address (::) then falls back to 0.0.0.0. + # Set to a literal IPv4 or IPv6 address to pin the listen address. + bindAddress: "" + # Basic process probes. Override these for stricter platform-specific checks. livenessProbe: httpGet: @@ -175,6 +191,9 @@ console: # Console server port port: 9090 + # Empty binds the IPv6 unspecified address (::) then falls back to 0.0.0.0. + bindAddress: "" + # Log level for console (trace, debug, info, warn, error) logLevel: info diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index 01ac356..ece7218 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -100,7 +100,10 @@ namespace SecurityContextConstraints (SCC) to assign an allowed UID and FSGroup. This is manifest compatibility, not an OpenShift certification claim. Generic Kubernetes installations must keep the default `openshift.enabled=false` behavior. The current target is `restricted-v2`; -`restricted-v3` also requires `spec.hostUsers: false`, which is not yet covered. +`restricted-v3` also requires `spec.hostUsers: false`. The Operator sets that +field on chart-managed Deployments when `openshift.enabled=true`, and on Tenant +StatefulSets when `spec.hostUsers` is set or when an OpenShift-style empty +security-context pair delegates identity to SCC. SCC-compatible manifests are insufficient when the server image assumes UID `10001`. Before deploying a Tenant, use an arbitrary-UID-compatible image whose @@ -544,6 +547,8 @@ Useful Tenant-level fields: | `podDeletionPolicyWhenNodeIsDown` | Node-down pod deletion behavior. | | `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`. | +| `network` | Optional Service `ipFamilyPolicy`/`ipFamilies` and RustFS listen addresses. Omitted Tenants keep IPv4 `0.0.0.0` listen addresses. IPv6 or dual-stack clusters should set `ipFamilies: [IPv6]` or `ipFamilyPolicy: PreferDualStack` so Services and `RUSTFS_ADDRESS` use `[::]`. | Both fields are also available on each `spec.pools[]` entry. Pool values are merged over Tenant values, which are merged over the Operator's defaults. By @@ -563,6 +568,7 @@ MinIO Operator contract: ```yaml spec: + hostUsers: false pools: - name: pool-0 securityContext: {} @@ -850,7 +856,9 @@ The operator can create RustFS policies, users, and buckets after the Tenant wor - `spec.credsSecret` for RustFS admin credentials. - `spec.policies` for canned policies sourced from ConfigMaps. - `spec.users` for regular users. Each user must have at least one direct policy mapping. -- `spec.buckets` for buckets and optional object lock. +- `spec.buckets` for buckets, optional object lock, canned anonymous access (`Private`, `Download`, `Upload`, `Public`), or a custom bucket policy ConfigMap. `anonymous` and `policy` are mutually exclusive. When both are omitted, the operator does not change a live bucket policy. + +Transient Kubernetes and RustFS admin/S3 failures (timeouts, 429, 5xx, connection errors, TLS not ready) leave provisioning items `Pending` and requeue instead of marking the Tenant `Failed`. Permanent 4xx configuration errors still fail and wait for a spec or object change. ConfigMaps and user Secrets must live in the Tenant namespace. The Operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. @@ -919,10 +927,13 @@ spec: buckets: - name: app-data objectLock: true + anonymous: Download ``` Deletion behavior is conservative: provisioned resources are retained when removed from the Tenant spec. +This operator provisions those objects onto the Tenant's RustFS cluster. A separate data-plane operator is out of scope. + ### 7.9 Pool Lifecycle `spec.poolLifecycle` controls explicit pool lifecycle requests. The current PVC retention policy is `Retain`. @@ -1175,6 +1186,15 @@ kubectl logs -n -l rustfs.tenant= Check PVC binding, StorageClass availability, image pull errors, node selectors, tolerations, and resource requests. +### Distinct physical disks + +RustFS requires each local erasure endpoint to map to a distinct physical disk. The Operator only +creates one PVC per `volumesPerServer` entry; if several PVCs land on the same node disk, the +server exits with `local erasure endpoints must use distinct physical disks`. That is a storage +topology / StorageClass issue, not an Operator provisioning bug. Fix the volume placement, or use +a StorageClass that provisions independent disks. `RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true` is a +RustFS data-plane escape hatch and is not applied by the Operator. + ### S3 API is not reachable Verify the Tenant S3 service and endpoints: diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index da61153..4f6347a 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -97,8 +97,10 @@ helm upgrade --install rustfs-operator deploy/rustfs-operator/ \ 该行为与 MinIO Operator 的安装方式一致:Chart 管理的 Deployment 不渲染 Pod 和容器 `securityContext`,由安装 namespace 的 SecurityContextConstraints(SCC) 分配合法 UID 和 FSGroup。这仅表示 manifest 与 SCC 兼容,不代表已获得 OpenShift -认证。普通 Kubernetes 安装必须保留默认的 `openshift.enabled=false`。当前支持目标 -限定为 `restricted-v2`;`restricted-v3` 还要求 `spec.hostUsers: false`,目前尚未覆盖。 +认证。普通 Kubernetes 安装必须保留默认的 `openshift.enabled=false`。当 +`openshift.enabled=true` 时,Chart 会为 Operator/Console Deployment 设置 +`hostUsers: false`;Tenant 在 `spec.hostUsers` 或 OpenShift 空 securityContext +对下同样会渲染该字段,以满足 `restricted-v3` 对用户命名空间的要求。 仅有 SCC 兼容 manifest 还不够,RustFS server 镜像也必须支持 SCC 分配的任意 UID。 部署 Tenant 前,应确认 `/data`、`/logs` 等镜像层可写目录属于 group `0`,并且 group @@ -517,6 +519,8 @@ spec: | `podDeletionPolicyWhenNodeIsDown` | 节点 NotReady/Unknown 时的 Pod 删除策略。 | | `securityContext` | 所有 RustFS Pool 的 Pod SecurityContext 覆盖。 | | `containerSecurityContext` | 所有 Pool 的 RustFS 容器 SecurityContext 覆盖。 | +| `hostUsers` | 可选的 Pod `hostUsers`。`false` 会隔离用户命名空间(`restricted-v3`)。OpenShift 空 securityContext 对也会默认渲染为 `false`。 | +| `network` | 可选的 Service `ipFamilyPolicy`/`ipFamilies` 以及 RustFS 监听地址。省略时保持 IPv4 `0.0.0.0`。IPv6 或双栈集群应设置 `ipFamilies: [IPv6]` 或 `ipFamilyPolicy: PreferDualStack`,使 Service 和 `RUSTFS_ADDRESS` 使用 `[::]`。 | 这两个字段也可配置在每个 `spec.pools[]` 条目上。Pool 级字段会按字段覆盖 Tenant 级字段,Tenant 级字段再覆盖 Operator 默认值。Operator 默认设置 @@ -532,6 +536,7 @@ namespace SCC;该契约与 MinIO Operator 保持一致: ```yaml spec: + hostUsers: false pools: - name: pool-0 securityContext: {} @@ -807,7 +812,9 @@ Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user - `spec.credsSecret`:RustFS 管理员凭据。 - `spec.policies`:从 ConfigMap 读取 policy document。 - `spec.users`:普通用户。每个 user 必须至少直接绑定一个 policy。 -- `spec.buckets`:bucket,可选择开启 object lock。 +- `spec.buckets`:bucket,以及可选的 object lock、匿名访问(`Private` / `Download` / `Upload` / `Public`)或来自 ConfigMap 的自定义 bucket policy。`anonymous` 与 `policy` 互斥。两者都省略时,Operator 不会改写已有 bucket policy。 + +Kubernetes 或 RustFS 管理/S3 API 的瞬时失败(超时、429、5xx、连接错误、TLS 未就绪)会把 provisioning 条目保持为 `Pending` 并重新入队,而不会把 Tenant 标为 `Failed`。永久性 4xx 配置错误仍会失败并等待 spec 或对象变更。 ConfigMap 和 user Secret 必须位于 Tenant namespace。Operator 会从 Tenant spec 建立反向引用索引,因此被引用资源的创建或更新会触发所有引用它的 Tenant reconcile;无需要求或修改资源标签,也不需要对这些资源拥有写权限。 @@ -876,10 +883,13 @@ spec: buckets: - name: app-data objectLock: true + anonymous: Download ``` 删除行为是保守的:从 Tenant spec 移除已 provisioning 的资源时,实际 RustFS 资源会保留。 +本 Operator 把这些对象 provisioning 到 Tenant 的 RustFS 集群中;独立的 data-plane operator 不在当前范围内。 + ### 7.9 Pool 生命周期 `spec.poolLifecycle` 用于显式 pool 生命周期请求。当前 PVC retention policy 为 `Retain`。 @@ -1131,6 +1141,15 @@ kubectl logs -n -l rustfs.tenant= 重点检查 PVC 绑定、StorageClass、镜像拉取、node selector、toleration 和资源 request。 +### 本地 erasure 盘必须使用不同物理磁盘 + +RustFS 要求每个本地 erasure endpoint 对应不同的物理磁盘。Operator 只会按 +`volumesPerServer` 为每个卷创建 PVC;如果多个 PVC 落到同一块节点磁盘上,server +会以 `local erasure endpoints must use distinct physical disks` 退出。这属于存储拓扑 +或 StorageClass 问题,不是 Operator provisioning 缺陷。应调整卷调度,或使用能提供 +独立磁盘的 StorageClass。`RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true` 是 RustFS 数据面的 +逃生开关,Operator 不会自动注入。 + ### S3 API 不可访问 检查 Tenant S3 Service 和 endpoints: diff --git a/e2e/tests/openshift_manifest.rs b/e2e/tests/openshift_manifest.rs index 06365c4..f4d8c73 100644 --- a/e2e/tests/openshift_manifest.rs +++ b/e2e/tests/openshift_manifest.rs @@ -66,6 +66,16 @@ fn openshift_chart_mode_delegates_deployment_security_contexts_to_scc() { .as_i64(), Some(101) ); + for name in [ + "rustfs-operator", + "rustfs-operator-console", + "rustfs-operator-console-frontend", + ] { + assert!( + deployment(&default_documents, name)["spec"]["template"]["spec"]["hostUsers"].is_null(), + "default chart must not pin hostUsers on {name}" + ); + } let openshift_render = helm_template(&[ "--set", @@ -93,6 +103,11 @@ fn openshift_chart_mode_delegates_deployment_security_contexts_to_scc() { deployment["spec"]["template"]["spec"]["securityContext"].is_null(), "OpenShift mode must omit the {name} Pod securityContext" ); + assert_eq!( + deployment["spec"]["template"]["spec"]["hostUsers"].as_bool(), + Some(false), + "OpenShift mode must set hostUsers: false on {name}" + ); for container in deployment["spec"]["template"]["spec"]["containers"] .as_sequence() .expect("Deployment containers are a sequence") @@ -116,6 +131,7 @@ fn openshift_tenant_example_uses_explicit_empty_pool_security_contexts() { .find(|document| document["kind"].as_str() == Some("Tenant")) .expect("example contains a Tenant"); let pool = &tenant["spec"]["pools"][0]; + assert_eq!(tenant["spec"]["hostUsers"].as_bool(), Some(false)); for field in ["securityContext", "containerSecurityContext"] { let value = pool[field] @@ -125,6 +141,53 @@ fn openshift_tenant_example_uses_explicit_empty_pool_security_contexts() { } } +#[test] +fn chart_network_values_are_copied_to_operator_services() { + let Some(render) = helm_template(&[ + "--set", + "network.ipFamilyPolicy=PreferDualStack", + "--set", + "network.ipFamilies={IPv4,IPv6}", + ]) else { + return; + }; + assert!( + render.status.success(), + "network chart render failed: {}", + String::from_utf8_lossy(&render.stderr) + ); + let output = String::from_utf8(render.stdout).expect("helm output is UTF-8"); + let documents = yaml_documents(&output, "network chart"); + for name in [ + "rustfs-operator-metrics", + "rustfs-operator-sts", + "rustfs-operator-console", + ] { + let service = documents + .iter() + .find(|document| { + document["kind"].as_str() == Some("Service") + && document["metadata"]["name"].as_str() == Some(name) + }) + .unwrap_or_else(|| panic!("missing Service {name}")); + assert_eq!( + service["spec"]["ipFamilyPolicy"].as_str(), + Some("PreferDualStack"), + "{name} should copy ipFamilyPolicy" + ); + let families = service["spec"]["ipFamilies"] + .as_sequence() + .unwrap_or_else(|| panic!("{name} ipFamilies")); + assert_eq!( + families + .iter() + .map(|item| item.as_str().unwrap_or_default()) + .collect::>(), + vec!["IPv4", "IPv6"] + ); + } +} + fn repository_root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() diff --git a/examples/openshift-tenant.yaml b/examples/openshift-tenant.yaml index 5e870c7..8d4a642 100644 --- a/examples/openshift-tenant.yaml +++ b/examples/openshift-tenant.yaml @@ -23,6 +23,7 @@ spec: # and grant the group the same permissions as the owner. Do not deploy this # example until such a RustFS image is available in your registry. image: registry.example.com/rustfs/rustfs:openshift-compatible + hostUsers: false pools: - name: pool-0 servers: 4 diff --git a/examples/provisioning-tenant.yaml b/examples/provisioning-tenant.yaml index c9abc47..e4784d3 100644 --- a/examples/provisioning-tenant.yaml +++ b/examples/provisioning-tenant.yaml @@ -85,3 +85,4 @@ spec: buckets: - name: provisioning-demo-data objectLock: true + anonymous: Download diff --git a/src/console/openapi.rs b/src/console/openapi.rs index 9a78706..97a8590 100644 --- a/src/console/openapi.rs +++ b/src/console/openapi.rs @@ -55,8 +55,8 @@ use crate::console::models::topology::{ TopologyOverviewResponse, TopologyPod, TopologyPool, TopologyTenant, TopologyTenantSummary, }; use crate::types::v1alpha1::provisioning::{ - ConfigMapKeyReference, PolicyDocumentSource, ProvisioningBucket, ProvisioningDeletionPolicy, - ProvisioningPolicy, ProvisioningUser, UserCredentialsSecretRef, + BucketAnonymousAccess, ConfigMapKeyReference, PolicyDocumentSource, ProvisioningBucket, + ProvisioningDeletionPolicy, ProvisioningPolicy, ProvisioningUser, UserCredentialsSecretRef, }; use crate::types::v1alpha1::status::provisioning::{ ProvisioningItemState, ProvisioningItemStatus, ProvisioningPhase, ProvisioningStatus, @@ -125,6 +125,7 @@ use crate::types::v1alpha1::status::provisioning::{ ProvisioningDeletionPolicy, PolicyDocumentSource, ConfigMapKeyReference, + BucketAnonymousAccess, CreateTenantRequest, CreatePoolRequest, PoolInfo, diff --git a/src/console/server.rs b/src/console/server.rs index 4b0d47f..2e68461 100755 --- a/src/console/server.rs +++ b/src/console/server.rs @@ -125,8 +125,12 @@ pub async fn run(port: u16) -> Result<(), Box> { .layer(middleware::from_fn(crate::metrics::record_console_http)); // Bind and serve - let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port)); - let listener = tokio::net::TcpListener::bind(addr).await?; + let listener = crate::utils::listen::bind_unspecified_listener( + port, + crate::utils::listen::CONSOLE_BIND_ADDRESS_ENV, + ) + .await?; + let addr = listener.local_addr()?; tracing::info!(%addr, "Console server listening"); tracing::info!("API endpoints:"); diff --git a/src/context.rs b/src/context.rs index 7204da6..1766131 100755 --- a/src/context.rs +++ b/src/context.rs @@ -411,6 +411,18 @@ pub(crate) fn is_kube_not_found(error: &Error) -> bool { ) } +pub(crate) fn is_transient_kube_error(error: &Error) -> bool { + match error { + Error::Kube { source } => match source { + kube::Error::Api(response) => { + response.code == 408 || response.code == 429 || response.code >= 500 + } + _ => true, + }, + _ => false, + } +} + pub(crate) fn map_secret_get_error( error: Error, name: String, @@ -1417,3 +1429,38 @@ mod validate_local_kms_tests { if message.contains("RUSTFS_KMS_LOCAL_MASTER_KEY"))); } } + +#[cfg(test)] +mod transient_kube_error_tests { + use super::{Error, is_transient_kube_error}; + + fn api_error(code: u16) -> Error { + Error::Kube { + source: kube::Error::Api(kube::error::ErrorResponse { + status: "Failure".to_string(), + message: format!("code {code}"), + reason: "Error".to_string(), + code, + }), + } + } + + #[test] + fn kube_api_5xx_429_and_408_are_transient() { + assert!(is_transient_kube_error(&api_error(408))); + assert!(is_transient_kube_error(&api_error(429))); + assert!(is_transient_kube_error(&api_error(500))); + assert!(is_transient_kube_error(&api_error(503))); + } + + #[test] + fn kube_api_404_and_4xx_semantic_errors_are_permanent() { + assert!(!is_transient_kube_error(&api_error(400))); + assert!(!is_transient_kube_error(&api_error(403))); + assert!(!is_transient_kube_error(&api_error(404))); + assert!(!is_transient_kube_error(&api_error(409))); + assert!(!is_transient_kube_error(&Error::CredentialSecretNotFound { + name: "creds".to_string(), + })); + } +} diff --git a/src/lib.rs b/src/lib.rs index e5cfdc9..9c608bf 100755 --- a/src/lib.rs +++ b/src/lib.rs @@ -672,8 +672,12 @@ async fn run_operator_observability_server( .with_state(state) .layer(middleware::from_fn(metrics::record_operator_http)); - let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port)); - let listener = tokio::net::TcpListener::bind(addr).await?; + let listener = crate::utils::listen::bind_unspecified_listener( + port, + crate::utils::listen::OPERATOR_BIND_ADDRESS_ENV, + ) + .await?; + let addr = listener.local_addr()?; info!(%addr, "operator observability server listening"); axum::serve(listener, app).await?; Ok(()) @@ -784,8 +788,12 @@ async fn bind_sts_listener( port: u16, tls_enabled: bool, ) -> Result> { - let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port)); - let listener = tokio::net::TcpListener::bind(addr).await?; + let listener = crate::utils::listen::bind_unspecified_listener( + port, + crate::utils::listen::OPERATOR_BIND_ADDRESS_ENV, + ) + .await?; + let addr = listener.local_addr()?; let scheme = if tls_enabled { "https" } else { "http" }; tracing::info!(%scheme, %addr, "Operator STS server listening"); Ok(listener) diff --git a/src/reconcile/phases.rs b/src/reconcile/phases.rs index eae7774..77e2eaf 100644 --- a/src/reconcile/phases.rs +++ b/src/reconcile/phases.rs @@ -1168,7 +1168,7 @@ pub(super) async fn finalize_tenant_status( ) -> Result { let mut builder = StatusBuilder::from_tenant(tenant); let pool_count = summary.pool_statuses.len(); - let requeue_after = reconcile_requeue_after(tenant, &summary, pod_cleanup_outcome); + let mut requeue_after = reconcile_requeue_after(tenant, &summary, pod_cleanup_outcome); builder.set_pool_statuses(summary.pool_statuses); if let Some(tls_status) = tls_plan.status { builder.set_tls_status(tls_status); @@ -1272,15 +1272,6 @@ pub(super) async fn finalize_tenant_status( ), ) } - ProvisioningOutcome::Pending { message } => { - builder.finish_provisioning_pending(message.clone()); - ( - ConditionType::ProvisioningReady, - Reason::ProvisioningPending, - EventType::Normal, - message, - ) - } ProvisioningOutcome::Failed { reason, message } => { builder.finish_provisioning_failed(reason, message.clone()); ( @@ -1293,16 +1284,38 @@ pub(super) async fn finalize_tenant_status( ProvisioningOutcome::Retry { message, retry_after, + persist_status: false, } => { warn!( tenant = %tenant.name(), namespace = %namespace, message = %message, retry_after_seconds = retry_after.as_secs(), - "retrying after RustFS user ownership checkpoint contention or transient failure" + "retrying after RustFS user ownership checkpoint contention" ); return Ok(Action::requeue(retry_after)); } + ProvisioningOutcome::Retry { + message, + retry_after, + persist_status: true, + } => { + warn!( + tenant = %tenant.name(), + namespace = %namespace, + message = %message, + retry_after_seconds = retry_after.as_secs(), + "retrying after a transient RustFS or Kubernetes provisioning failure" + ); + builder.finish_provisioning_pending(message.clone()); + requeue_after = earliest_requeue_after(requeue_after, Some(retry_after)); + ( + ConditionType::ProvisioningReady, + Reason::ProvisioningPending, + EventType::Normal, + message, + ) + } } } else { builder.finish_reconciling( diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 6fdf633..8c968a6 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -15,8 +15,8 @@ use crate::context::{self, Context}; use crate::sts::rustfs_client::{CreateBucketResult, RustfsAdminClient, RustfsClientError}; use crate::types::v1alpha1::provisioning::{ - ProvisioningBucket, ProvisioningPolicy, ProvisioningUser, - duplicate_user_credentials_secret_names, + BucketAnonymousAccess, PolicyDocumentSource, ProvisioningBucket, ProvisioningPolicy, + ProvisioningUser, duplicate_user_credentials_secret_names, }; use crate::types::v1alpha1::status::Reason; use crate::types::v1alpha1::status::provisioning::{ @@ -44,9 +44,6 @@ pub(super) struct ProvisioningReconcileResult { pub(super) enum ProvisioningOutcome { Ready, - Pending { - message: String, - }, Failed { reason: Reason, message: String, @@ -54,6 +51,7 @@ pub(super) enum ProvisioningOutcome { Retry { message: String, retry_after: Duration, + persist_status: bool, }, } @@ -69,6 +67,13 @@ enum CheckpointError { Retry(CheckpointRetry), } +#[derive(Clone, Debug)] +struct SpecLoadError { + reason: Reason, + message: String, + transient: bool, +} + struct ProvisioningRun<'a> { ctx: &'a Context, tenant: &'a Tenant, @@ -77,6 +82,7 @@ struct ProvisioningRun<'a> { now: String, status: ProvisioningStatus, failures: Vec<(Reason, String)>, + retry: Option, } #[derive(Clone)] @@ -91,7 +97,7 @@ enum UserCredentialsCheck { DuplicateSecret, Checked { policy_error: Option, - credentials: Result, + credentials: Result, }, } @@ -254,6 +260,99 @@ impl ProvisioningRun<'_> { item } + fn request_retry(&mut self, message: impl Into) { + if self.retry.is_none() { + self.retry = Some(CheckpointRetry { + message: message.into(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }); + } + } + + fn item_from_spec_error

( + &mut self, + previous: Option<&P>, + name: &str, + error: SpecLoadError, + ) -> ProvisioningItemStatus + where + P: AsRef + ?Sized, + { + if error.transient { + self.request_retry(error.message.clone()); + self.item( + previous, + name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + error.message, + ) + } else { + self.item( + previous, + name, + ProvisioningItemState::Failed, + error.reason, + error.message, + ) + } + } + + fn item_from_admin_error

( + &mut self, + previous: Option<&P>, + name: &str, + permanent_reason: Reason, + error: RustfsClientError, + context: impl Into, + ) -> ProvisioningItemStatus + where + P: AsRef + ?Sized, + { + let context = context.into(); + self.item_from_spec_error( + previous, + name, + SpecLoadError { + reason: permanent_reason, + message: format!("{context}: {error}"), + transient: error.is_transient(), + }, + ) + } + + fn has_pending_items(&self) -> bool { + let is_pending = |state: &str| state == ProvisioningItemState::Pending.as_str(); + self.status + .policies + .iter() + .any(|item| is_pending(&item.state)) + || self.status.users.iter().any(|item| is_pending(&item.state)) + || self + .status + .buckets + .iter() + .any(|item| is_pending(&item.state)) + } + + fn first_pending_message(&self) -> String { + self.status + .policies + .iter() + .chain(self.status.buckets.iter()) + .find(|item| item.state == ProvisioningItemState::Pending.as_str()) + .map(item_message) + .or_else(|| { + self.status.users.iter().find_map(|item| { + (item.state == ProvisioningItemState::Pending.as_str()) + .then(|| item_message(item)) + }) + }) + .unwrap_or_else(|| { + "provisioning is waiting for a transient RustFS or Kubernetes failure".to_string() + }) + } + fn retained_item(&self, previous: &ProvisioningItemStatus) -> ProvisioningItemStatus { let mut item = self.item( Some(previous), @@ -368,6 +467,22 @@ impl ProvisioningRun<'_> { } fn finish(mut self) -> ProvisioningReconcileResult { + let pending_items = self.has_pending_items(); + if self.retry.is_some() || pending_items { + let retry = self.retry.take().unwrap_or_else(|| CheckpointRetry { + message: self.first_pending_message(), + retry_after: CHECKPOINT_TRANSIENT_RETRY, + }); + self.prepare_status(ProvisioningPhase::Pending); + return ProvisioningReconcileResult { + status: self.status, + outcome: ProvisioningOutcome::Retry { + message: retry.message, + retry_after: retry.retry_after, + persist_status: true, + }, + }; + } let outcome = self .failures .first() @@ -378,7 +493,6 @@ impl ProvisioningRun<'_> { .unwrap_or(ProvisioningOutcome::Ready); let phase = match &outcome { ProvisioningOutcome::Ready => ProvisioningPhase::Ready, - ProvisioningOutcome::Pending { .. } => ProvisioningPhase::Pending, ProvisioningOutcome::Failed { .. } => ProvisioningPhase::Failed, ProvisioningOutcome::Retry { .. } => ProvisioningPhase::Pending, }; @@ -410,6 +524,7 @@ pub(super) async fn reconcile_provisioning( now, status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; if !has_active_spec(tenant) { @@ -434,46 +549,38 @@ pub(super) async fn reconcile_provisioning( ); if pending { run.mark_all_active(ProvisioningItemState::Pending, reason, &message); + run.request_retry(message); } else { run.fail_all_active(reason, &message); } - let phase = if pending { - ProvisioningPhase::Pending - } else { - ProvisioningPhase::Failed - }; - run.prepare_status(phase); - return ProvisioningReconcileResult { - status: run.status, - outcome: if pending { - ProvisioningOutcome::Pending { message } - } else { - ProvisioningOutcome::Failed { reason, message } - }, - }; + return run.finish(); } }; let user_credentials = preflight_user_credentials(&run).await; let mut live_policies = match load_live_policies(&client, tenant).await { Ok(policies) => policies, - Err(message) => { + Err(error) => { + let message = error.message.clone(); warn!( tenant = %tenant.name(), namespace = %namespace, - reason = Reason::PolicyApplyFailed.as_str(), + reason = error.reason.as_str(), + transient = error.transient, message = %message, "RustFS provisioning failed to load live policies" ); - run.fail_all_active(Reason::PolicyApplyFailed, &message); - run.prepare_status(ProvisioningPhase::Failed); - return ProvisioningReconcileResult { - status: run.status, - outcome: ProvisioningOutcome::Failed { - reason: Reason::PolicyApplyFailed, - message, - }, - }; + if error.transient { + run.mark_all_active( + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + &message, + ); + run.request_retry(message); + } else { + run.fail_all_active(error.reason, &message); + } + return run.finish(); } }; @@ -485,6 +592,7 @@ pub(super) async fn reconcile_provisioning( outcome: ProvisioningOutcome::Retry { message: retry.message, retry_after: retry.retry_after, + persist_status: false, }, }; } @@ -528,6 +636,11 @@ fn client_error_outcome(error: RustfsClientError) -> (Reason, String, bool) { "tenant TLS is not ready for provisioning".to_string(), true, ), + error if error.is_transient() => ( + Reason::ProvisioningPending, + format!("failed to create RustFS admin client: {error}"), + true, + ), error => ( Reason::ProvisioningFailed, format!("failed to create RustFS admin client: {error}"), @@ -539,7 +652,7 @@ fn client_error_outcome(error: RustfsClientError) -> (Reason, String, bool) { async fn load_live_policies( client: &RustfsAdminClient, tenant: &Tenant, -) -> Result, String> { +) -> Result, SpecLoadError> { if tenant.spec.policies.is_empty() && tenant .spec @@ -553,11 +666,18 @@ async fn load_live_policies( let mut policies = client .list_canned_policies() .await - .map_err(|error| format!("failed to list RustFS canned policies: {error}"))?; + .map_err(|error| SpecLoadError { + reason: Reason::PolicyApplyFailed, + message: format!("failed to list RustFS canned policies: {error}"), + transient: error.is_transient(), + })?; for (name, document) in &mut policies { - *document = normalize_policy_document(document) - .map_err(|error| format!("failed to normalize live RustFS policy '{name}': {error}"))?; + *document = normalize_policy_document(document).map_err(|error| SpecLoadError { + reason: Reason::PolicyApplyFailed, + message: format!("failed to normalize live RustFS policy '{name}': {error}"), + transient: false, + })?; } Ok(policies) @@ -568,29 +688,31 @@ async fn reconcile_policies( client: &RustfsAdminClient, live_policies: &mut BTreeMap, ) { - for policy in &run.tenant.spec.policies { + let policies = run.tenant.spec.policies.clone(); + for policy in &policies { let item = reconcile_policy(run, client, live_policies, policy).await; run.push_policy(item); } } async fn reconcile_policy( - run: &ProvisioningRun<'_>, + run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, live_policies: &mut BTreeMap, policy: &ProvisioningPolicy, ) -> ProvisioningItemStatus { - let previous = run.previous_policy(&policy.name); - let document = match load_policy_document(run, policy).await { + let previous = run.previous_policy(&policy.name).cloned(); + let document = match load_policy_source( + run, + &policy.document, + Reason::PolicyApplyFailed, + "policy", + ) + .await + { Ok(document) => document, - Err((reason, message)) => { - return run.item( - previous, - &policy.name, - ProvisioningItemState::Failed, - reason, - message, - ); + Err(error) => { + return run.item_from_spec_error(previous.as_ref(), &policy.name, error); } }; @@ -598,9 +720,14 @@ async fn reconcile_policy( let live_hash = live_policies .get(&policy.name) .map(|live_document| hash_document(live_document)); - let item = match policy_reconcile_action(previous, live_hash.as_deref(), &desired_hash) { + let item = match policy_reconcile_action( + previous.as_ref(), + live_hash.as_deref(), + &desired_hash, + Reason::PolicyConflict, + ) { PolicyReconcileAction::Ready(message) => run.item( - previous, + previous.as_ref(), &policy.name, ProvisioningItemState::Ready, Reason::ProvisioningConfigured, @@ -610,7 +737,7 @@ async fn reconcile_policy( match apply_policy(client, live_policies, &policy.name, &document.raw).await { Ok(applied_hash) => { let mut item = run.item( - previous, + previous.as_ref(), &policy.name, ProvisioningItemState::Ready, Reason::ProvisioningConfigured, @@ -619,17 +746,17 @@ async fn reconcile_policy( item.last_applied_hash = Some(applied_hash); item } - Err(message) => run.item( - previous, + Err(error) => run.item_from_admin_error( + previous.as_ref(), &policy.name, - ProvisioningItemState::Failed, Reason::PolicyApplyFailed, - message, + error, + format!("failed to apply RustFS policy '{}'", policy.name), ), } } PolicyReconcileAction::Failed(reason, message) => run.item( - previous, + previous.as_ref(), &policy.name, ProvisioningItemState::Failed, reason, @@ -639,7 +766,7 @@ async fn reconcile_policy( finalize_policy_item_status( item, - previous, + previous.as_ref(), &policy.name, desired_hash, live_policies, @@ -651,6 +778,7 @@ fn policy_reconcile_action( previous: Option<&ProvisioningItemStatus>, live_hash: Option<&str>, desired_hash: &str, + conflict: Reason, ) -> PolicyReconcileAction { let Some(live_hash) = live_hash else { return PolicyReconcileAction::Apply("RustFS policy was created"); @@ -661,7 +789,7 @@ fn policy_reconcile_action( PolicyReconcileAction::Ready("Existing RustFS policy matches spec and was adopted") } None => PolicyReconcileAction::Failed( - Reason::PolicyConflict, + conflict, "Live RustFS policy differs from spec and is not owned by this status", ), Some(last_applied_hash) if last_applied_hash == live_hash => { @@ -675,7 +803,7 @@ fn policy_reconcile_action( PolicyReconcileAction::Ready("RustFS policy matches spec") } Some(_) => PolicyReconcileAction::Failed( - Reason::PolicyConflict, + conflict, "Live RustFS policy changed since the operator last applied it", ), } @@ -712,29 +840,33 @@ fn finalize_policy_item_status( item } -async fn load_policy_document( +async fn load_policy_source( run: &ProvisioningRun<'_>, - policy: &ProvisioningPolicy, -) -> Result { - let reference = &policy.document.config_map_key_ref; + source: &PolicyDocumentSource, + apply_failed: Reason, + kind: &str, +) -> Result { + let reference = &source.config_map_key_ref; let config_map: ConfigMap = run.ctx .get(&reference.name, run.namespace) .await .map_err(|error| { if context::is_kube_not_found(&error) { - ( - Reason::PolicyDocumentConfigMapNotFound, - format!("policy ConfigMap '{}' was not found", reference.name), - ) + SpecLoadError { + reason: Reason::PolicyDocumentConfigMapNotFound, + message: format!("{kind} ConfigMap '{}' was not found", reference.name), + transient: false, + } } else { - ( - Reason::PolicyApplyFailed, - format!( - "failed to read policy ConfigMap '{}': {error}", + SpecLoadError { + reason: apply_failed, + message: format!( + "failed to read {kind} ConfigMap '{}': {error}", reference.name ), - ) + transient: context::is_transient_kube_error(&error), + } } })?; @@ -742,17 +874,20 @@ async fn load_policy_document( .data .as_ref() .and_then(|data| data.get(&reference.key)) - .ok_or_else(|| { - ( - Reason::PolicyDocumentKeyNotFound, - format!( - "policy ConfigMap '{}' is missing key '{}'", - reference.name, reference.key - ), - ) + .ok_or_else(|| SpecLoadError { + reason: Reason::PolicyDocumentKeyNotFound, + message: format!( + "{kind} ConfigMap '{}' is missing key '{}'", + reference.name, reference.key + ), + transient: false, })?; - PolicyDocument::parse(raw).map_err(|message| (Reason::PolicyApplyFailed, message)) + PolicyDocument::parse(raw).map_err(|message| SpecLoadError { + reason: apply_failed, + message, + transient: false, + }) } async fn apply_policy( @@ -760,17 +895,12 @@ async fn apply_policy( live_policies: &mut BTreeMap, name: &str, document: &str, -) -> Result { - client - .add_canned_policy(name, document) - .await - .map_err(|error| format!("failed to apply RustFS policy '{name}': {error}"))?; +) -> Result { + client.add_canned_policy(name, document).await?; - let live_document = client - .get_canned_policy(name) - .await - .map_err(|error| format!("failed to read RustFS policy '{name}' after apply: {error}"))?; - let live_document = normalize_policy_document(&live_document)?; + let live_document = client.get_canned_policy(name).await?; + let live_document = normalize_policy_document(&live_document) + .map_err(|_| RustfsClientError::InvalidPolicyDocument)?; let live_hash = hash_document(&live_document); live_policies.insert(name.to_string(), live_document); Ok(live_hash) @@ -789,15 +919,17 @@ async fn reconcile_users( .filter(|item| item.state == ProvisioningItemState::Failed.as_str()) .map(|item| item.name.clone()) .collect::>(); - let mut plans = Vec::with_capacity(run.tenant.spec.users.len()); - - for (user, preflight) in run - .tenant - .spec - .users + let pending_spec_policies = run + .status + .policies .iter() - .zip(credentials_preflight.checks.iter()) - { + .filter(|item| item.state == ProvisioningItemState::Pending.as_str()) + .map(|item| item.name.clone()) + .collect::>(); + let users = run.tenant.spec.users.clone(); + let mut plans = Vec::with_capacity(users.len()); + + for (user, preflight) in users.iter().zip(credentials_preflight.checks.iter()) { let (policy_error, credentials) = match preflight { UserCredentialsCheck::DuplicateSecret => { let previous = run.previous_user(&user.name); @@ -855,16 +987,10 @@ async fn reconcile_users( } let credentials = match credentials { Ok(credentials) => credentials, - Err(message) => { - let previous = run.previous_user(&user.name); - let item = run.item( - previous, - &user.name, - ProvisioningItemState::Failed, - Reason::UserSecretInvalid, - message, - ); - let item = annotate_user_item(item, user, previous, None); + Err(error) => { + let previous = run.previous_user(&user.name).cloned(); + let item = run.item_from_spec_error(previous.as_ref(), &user.name, error.clone()); + let item = annotate_user_item(item, user, previous.as_ref(), None); plans.push(UserReconcilePlan::Complete(Box::new(item))); continue; } @@ -876,6 +1002,7 @@ async fn reconcile_users( client, live_policies, &failed_spec_policies, + &pending_spec_policies, user, credentials, ) @@ -981,24 +1108,28 @@ fn duplicate_user_access_key_hashes( } async fn prepare_user_reconcile( - run: &ProvisioningRun<'_>, + run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, live_policies: &BTreeMap, failed_spec_policies: &BTreeSet, + pending_spec_policies: &BTreeSet, user: &ProvisioningUser, credentials: &UserCredentials, ) -> UserReconcilePlan { - let previous = run.previous_user(&user.name); - if user_access_key_changed(previous, credentials) { + let previous = run.previous_user(&user.name).cloned(); + if user_access_key_changed(previous.as_ref(), credentials) { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::ImmutableFieldModified, "user access key is immutable after provisioning; create a new user entry to migrate it", ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } @@ -1008,14 +1139,39 @@ async fn prepare_user_reconcile( .find(|policy_name| failed_spec_policies.contains(*policy_name)) { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserPolicySetFailed, format!("referenced policy '{policy_name}' is not ready"), ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, + ))); + } + + if let Some(policy_name) = user + .policies + .iter() + .find(|policy_name| pending_spec_policies.contains(*policy_name)) + { + let message = format!("referenced policy '{policy_name}' is not ready"); + run.request_retry(message.clone()); + let item = run.item( + previous.as_ref(), + &user.name, + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + message, + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, + user, + previous.as_ref(), + None, ))); } @@ -1025,52 +1181,62 @@ async fn prepare_user_reconcile( .find(|policy_name| !live_policies.contains_key(*policy_name)) { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserPolicyNotFound, format!("referenced policy '{policy_name}' does not exist"), ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } let exists = match client.user_exists(&credentials.access_key).await { Ok(exists) => exists, Err(error) => { - let item = run.item( - previous, + let item = run.item_from_admin_error( + previous.as_ref(), &user.name, - ProvisioningItemState::Failed, Reason::UserSecretInvalid, - format!("failed to query RustFS user: {error}"), + error, + "failed to query RustFS user", ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } }; - let mut ownership = match matching_user_ownership(previous, run.tenant, user, credentials) { - Ok(ownership) => ownership, - Err(message) => { - let item = run.item( - previous, - &user.name, - ProvisioningItemState::Failed, - Reason::UserOwnershipConflict, - message, - ); - return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, - ))); - } - }; + let mut ownership = + match matching_user_ownership(previous.as_ref(), run.tenant, user, credentials) { + Ok(ownership) => ownership, + Err(message) => { + let item = run.item( + previous.as_ref(), + &user.name, + ProvisioningItemState::Failed, + Reason::UserOwnershipConflict, + message, + ); + return UserReconcilePlan::Complete(Box::new(annotate_user_item( + item, + user, + previous.as_ref(), + None, + ))); + } + }; let mut checkpoint_update = None; if exists && ownership.is_none() { - if legacy_user_status_can_migrate(previous, user, credentials) { + if legacy_user_status_can_migrate(previous.as_ref(), user, credentials) { let managed_ownership = match user_ownership( run.tenant, user, @@ -1080,19 +1246,22 @@ async fn prepare_user_reconcile( Ok(ownership) => ownership, Err(message) => { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserOwnershipCheckpointFailed, message, ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } }; let managed_checkpoint = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Ready, Reason::ProvisioningConfigured, @@ -1101,20 +1270,23 @@ async fn prepare_user_reconcile( // Preserve the legacy observed Secret version so a concurrently rotated Secret is // still applied after the ownership checkpoint has been persisted. let mut managed_checkpoint = - annotate_user_item(managed_checkpoint, user, previous, None); + annotate_user_item(managed_checkpoint, user, previous.as_ref(), None); managed_checkpoint.ownership = Some(managed_ownership.clone()); checkpoint_update = Some(managed_checkpoint); ownership = Some(managed_ownership); } else { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserOwnershipConflict, "RustFS user already exists without a matching operator ownership checkpoint; choose a different access key or remove the unmanaged user", ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } } @@ -1129,26 +1301,33 @@ async fn prepare_user_reconcile( Ok(ownership) => ownership, Err(message) => { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserOwnershipCheckpointFailed, message, ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); } }; let pending_checkpoint = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator ownership checkpoint was persisted before creating the RustFS user", ); - let mut pending_checkpoint = - annotate_user_item(pending_checkpoint, user, previous, Some(credentials)); + let mut pending_checkpoint = annotate_user_item( + pending_checkpoint, + user, + previous.as_ref(), + Some(credentials), + ); pending_checkpoint.ownership = Some(pending_ownership.clone()); checkpoint_update = Some(pending_checkpoint); ownership = Some(pending_ownership); @@ -1161,28 +1340,35 @@ async fn prepare_user_reconcile( // This recovery relies on per-Tenant controller serialization; it does not provide // exactly-once delivery across Kubernetes and independent RustFS actors. let pending_checkpoint = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Pending, Reason::ProvisioningPending, "Operator is resuming a pending RustFS user creation", ); - let mut pending_checkpoint = - annotate_user_item(pending_checkpoint, user, previous, Some(credentials)); + let mut pending_checkpoint = annotate_user_item( + pending_checkpoint, + user, + previous.as_ref(), + Some(credentials), + ); pending_checkpoint.ownership = ownership.clone(); checkpoint_update = Some(pending_checkpoint); } let Some(ownership) = ownership else { let item = run.item( - previous, + previous.as_ref(), &user.name, ProvisioningItemState::Failed, Reason::UserOwnershipCheckpointFailed, "Operator ownership checkpoint is required before synchronizing RustFS user credentials", ); return UserReconcilePlan::Complete(Box::new(annotate_user_item( - item, user, previous, None, + item, + user, + previous.as_ref(), + None, ))); }; @@ -1196,7 +1382,7 @@ async fn prepare_user_reconcile( } async fn execute_prepared_user( - run: &ProvisioningRun<'_>, + run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, prepared: PreparedUserReconcile, ) -> ProvisioningUserStatus { @@ -1207,20 +1393,20 @@ async fn execute_prepared_user( mut ownership, .. } = prepared; - let previous = run.previous_user(&user.name); + let previous = run.previous_user(&user.name).cloned(); let credentials_applied = - match sync_user_credentials(client, previous, &credentials, exists).await { + match sync_user_credentials(client, previous.as_ref(), &credentials, exists).await { Ok(applied) => applied, Err(error) => { - let item = run.item( - previous, + let item = run.item_from_admin_error( + previous.as_ref(), &user.name, - ProvisioningItemState::Failed, Reason::UserSecretInvalid, - format!("failed to update RustFS user credentials: {error}"), + error, + "failed to update RustFS user credentials", ); - let mut item = annotate_user_item(item, &user, previous, None); + let mut item = annotate_user_item(item, &user, previous.as_ref(), None); item.ownership = Some(ownership); return item; } @@ -1232,14 +1418,14 @@ async fn execute_prepared_user( .set_user_policy(&credentials.access_key, &user.policies) .await { - let item = run.item( - previous, + let item = run.item_from_admin_error( + previous.as_ref(), &user.name, - ProvisioningItemState::Failed, Reason::UserPolicySetFailed, - format!("failed to set RustFS user policy mapping: {error}"), + error, + "failed to set RustFS user policy mapping", ); - let mut item = annotate_user_item(item, &user, previous, Some(&credentials)); + let mut item = annotate_user_item(item, &user, previous.as_ref(), Some(&credentials)); item.ownership = Some(ownership); return item; } @@ -1257,7 +1443,7 @@ async fn execute_prepared_user( Reason::ProvisioningConfigured.as_str(), ); item.message = Some(message.to_string()); - item.last_transition_time = match previous { + item.last_transition_time = match previous.as_ref() { Some(previous) if previous.state == item.state && previous.reason == item.reason @@ -1267,14 +1453,14 @@ async fn execute_prepared_user( } _ => Some(run.now.clone()), }; - let mut item = annotate_user_item(item, &user, previous, Some(&credentials)); + let mut item = annotate_user_item(item, &user, previous.as_ref(), Some(&credentials)); item.ownership = Some(ownership); item } #[cfg(test)] async fn reconcile_user( - run: &ProvisioningRun<'_>, + run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, live_policies: &BTreeMap, failed_spec_policies: &BTreeSet, @@ -1286,6 +1472,7 @@ async fn reconcile_user( client, live_policies, failed_spec_policies, + &BTreeSet::new(), user, credentials, ) @@ -1297,15 +1484,15 @@ async fn reconcile_user( && let Err(error) = persist_user_ownership_checkpoints(run, std::slice::from_ref(checkpoint)).await { - let previous = run.previous_user(&prepared.user.name); + let previous = run.previous_user(&prepared.user.name).cloned(); let item = run.item( - previous, + previous.as_ref(), &prepared.user.name, ProvisioningItemState::Failed, Reason::UserOwnershipCheckpointFailed, checkpoint_error_message(error), ); - return annotate_user_item(item, &prepared.user, previous, None); + return annotate_user_item(item, &prepared.user, previous.as_ref(), None); } execute_prepared_user(run, client, *prepared).await } @@ -1641,7 +1828,7 @@ fn access_key_hash(access_key: &str) -> String { async fn load_user_secret( run: &ProvisioningRun<'_>, user: &ProvisioningUser, -) -> Result { +) -> Result { let secret_name = user.credentials_secret_name(); let secret: Secret = run .ctx @@ -1649,15 +1836,24 @@ async fn load_user_secret( .await .map_err(|error| { if context::is_kube_not_found(&error) { - format!("user Secret '{secret_name}' was not found") + SpecLoadError { + reason: Reason::UserSecretInvalid, + message: format!("user Secret '{secret_name}' was not found"), + transient: false, + } } else { - format!("failed to read user Secret '{secret_name}': {error}") + SpecLoadError { + reason: Reason::UserSecretInvalid, + message: format!("failed to read user Secret '{secret_name}': {error}"), + transient: context::is_transient_kube_error(&error), + } } })?; - let data = secret - .data - .as_ref() - .ok_or_else(|| format!("user Secret '{secret_name}' has no data"))?; + let data = secret.data.as_ref().ok_or_else(|| SpecLoadError { + reason: Reason::UserSecretInvalid, + message: format!("user Secret '{secret_name}' has no data"), + transient: false, + })?; let access_key = read_compatible_secret_value( data, @@ -1665,17 +1861,35 @@ async fn load_user_secret( "CONSOLE_ACCESS_KEY", secret_name, "access key", - )?; + ) + .map_err(|message| SpecLoadError { + reason: Reason::UserSecretInvalid, + message, + transient: false, + })?; let secret_key = read_compatible_secret_value( data, "secretkey", "CONSOLE_SECRET_KEY", secret_name, "secret key", - )?; + ) + .map_err(|message| SpecLoadError { + reason: Reason::UserSecretInvalid, + message, + transient: false, + })?; - validate_user_access_key(&access_key)?; - validate_user_secret_key(&secret_key)?; + validate_user_access_key(&access_key).map_err(|message| SpecLoadError { + reason: Reason::UserSecretInvalid, + message, + transient: false, + })?; + validate_user_secret_key(&secret_key).map_err(|message| SpecLoadError { + reason: Reason::UserSecretInvalid, + message, + transient: false, + })?; Ok(UserCredentials { access_key, @@ -1748,21 +1962,22 @@ fn validate_user_secret_key(secret_key: &str) -> Result<(), String> { } async fn reconcile_buckets(run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient) { - for bucket in &run.tenant.spec.buckets { + let buckets = run.tenant.spec.buckets.clone(); + for bucket in &buckets { let item = reconcile_bucket(run, client, bucket).await; run.push_bucket(item); } } async fn reconcile_bucket( - run: &ProvisioningRun<'_>, + run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, bucket: &ProvisioningBucket, ) -> ProvisioningItemStatus { - let previous = run.previous_bucket(&bucket.name); + let previous = run.previous_bucket(&bucket.name).cloned(); if let Err(message) = validate_bucket_name(&bucket.name) { let item = run.item( - previous, + previous.as_ref(), &bucket.name, ProvisioningItemState::Failed, Reason::BucketCreateFailed, @@ -1771,6 +1986,17 @@ async fn reconcile_bucket( return annotate_bucket_item(item, bucket); } + if bucket.has_custom_policy() && bucket.has_anonymous_access() { + let item = run.item( + previous.as_ref(), + &bucket.name, + ProvisioningItemState::Failed, + Reason::BucketPolicyConflict, + "bucket policy and anonymous access are mutually exclusive", + ); + return annotate_bucket_item(item, bucket); + } + let create_result = match client .create_bucket( &bucket.name, @@ -1781,12 +2007,12 @@ async fn reconcile_bucket( { Ok(result) => result, Err(error) => { - let item = run.item( - previous, + let item = run.item_from_admin_error( + previous.as_ref(), &bucket.name, - ProvisioningItemState::Failed, Reason::BucketCreateFailed, - format!("failed to create RustFS bucket: {error}"), + error, + "failed to create RustFS bucket", ); return annotate_bucket_item(item, bucket); } @@ -1794,24 +2020,7 @@ async fn reconcile_bucket( if bucket.object_lock_enabled() { match client.bucket_object_lock_enabled(&bucket.name).await { - Ok(true) => { - let message = match create_result { - CreateBucketResult::Created => { - "RustFS bucket was created with object lock enabled" - } - CreateBucketResult::AlreadyExists => { - "Bucket already existed with object lock enabled" - } - }; - let item = run.item( - previous, - &bucket.name, - ProvisioningItemState::Ready, - Reason::ProvisioningConfigured, - message, - ); - return annotate_bucket_item(item, bucket); - } + Ok(true) => {} Ok(false) => { let message = match create_result { CreateBucketResult::Created => { @@ -1822,7 +2031,7 @@ async fn reconcile_bucket( } }; let item = run.item( - previous, + previous.as_ref(), &bucket.name, ProvisioningItemState::Failed, Reason::BucketObjectLockConflict, @@ -1831,38 +2040,242 @@ async fn reconcile_bucket( return annotate_bucket_item(item, bucket); } Err(error) => { - let message = match create_result { - CreateBucketResult::Created => { - format!("failed to verify created bucket object lock: {error}") - } + let context = match create_result { + CreateBucketResult::Created => "failed to verify created bucket object lock", CreateBucketResult::AlreadyExists => { - format!("failed to verify existing bucket object lock: {error}") + "failed to verify existing bucket object lock" } }; - let item = run.item( - previous, + let item = run.item_from_admin_error( + previous.as_ref(), &bucket.name, - ProvisioningItemState::Failed, Reason::BucketObjectLockConflict, - message, + error, + context, ); return annotate_bucket_item(item, bucket); } } - } + } + + let created_message = match create_result { + CreateBucketResult::Created => { + if bucket.object_lock_enabled() { + "RustFS bucket was created with object lock enabled" + } else { + "RustFS bucket was created" + } + } + CreateBucketResult::AlreadyExists => { + if bucket.object_lock_enabled() { + "Bucket already existed with object lock enabled" + } else { + "RustFS bucket already exists" + } + } + }; + + let Some(desired) = desired_bucket_policy(run, bucket).await else { + let item = run.item( + previous.as_ref(), + &bucket.name, + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured, + created_message, + ); + return annotate_bucket_item(item, bucket); + }; + let desired = match desired { + Ok(document) => document, + Err(error) => { + return annotate_bucket_item( + run.item_from_spec_error(previous.as_ref(), &bucket.name, error), + bucket, + ); + } + }; + + sync_bucket_policy( + run, + client, + bucket, + previous.as_ref(), + &desired, + created_message, + ) + .await +} + +async fn desired_bucket_policy( + run: &ProvisioningRun<'_>, + bucket: &ProvisioningBucket, +) -> Option> { + if let Some(source) = bucket.policy.as_ref() { + return Some( + load_policy_source( + run, + source, + Reason::BucketPolicyApplyFailed, + "bucket policy", + ) + .await, + ); + } + canned_anonymous_bucket_policy(bucket.anonymous, &bucket.name) + .map(|raw| PolicyDocument::parse(&raw)) + .map(|result| { + result.map_err(|message| SpecLoadError { + reason: Reason::BucketPolicyApplyFailed, + message, + transient: false, + }) + }) +} + +async fn sync_bucket_policy( + run: &mut ProvisioningRun<'_>, + client: &RustfsAdminClient, + bucket: &ProvisioningBucket, + previous: Option<&ProvisioningItemStatus>, + desired: &PolicyDocument, + created_message: &str, +) -> ProvisioningItemStatus { + let desired_hash = desired.hash(); + let live_document = match client.get_bucket_policy(&bucket.name).await { + Ok(document) => document, + Err(error) => { + return finalize_bucket_policy_item( + run.item_from_admin_error( + previous, + &bucket.name, + Reason::BucketPolicyApplyFailed, + error, + "failed to read RustFS bucket policy", + ), + bucket, + Some(desired_hash), + ); + } + }; + let live_hash = match live_document.as_deref() { + None => None, + Some(document) => match normalize_policy_document(document) { + Ok(normalized) => Some(hash_document(&normalized)), + Err(message) => { + return finalize_bucket_policy_item( + run.item( + previous, + &bucket.name, + ProvisioningItemState::Failed, + Reason::BucketPolicyApplyFailed, + format!("failed to normalize live RustFS bucket policy: {message}"), + ), + bucket, + Some(desired_hash), + ); + } + }, + }; + + let item = match policy_reconcile_action( + previous, + live_hash.as_deref(), + &desired_hash, + Reason::BucketPolicyConflict, + ) { + PolicyReconcileAction::Ready(message) => run.item( + previous, + &bucket.name, + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured, + format!("{created_message}; {message}"), + ), + PolicyReconcileAction::Apply(_) => { + match client.put_bucket_policy(&bucket.name, &desired.raw).await { + Ok(()) => { + let mut item = run.item( + previous, + &bucket.name, + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured, + format!("{created_message}; RustFS bucket policy was applied"), + ); + item.last_applied_hash = Some(desired_hash.clone()); + item + } + Err(error) => run.item_from_admin_error( + previous, + &bucket.name, + Reason::BucketPolicyApplyFailed, + error, + "failed to apply RustFS bucket policy", + ), + } + } + PolicyReconcileAction::Failed(reason, message) => run.item( + previous, + &bucket.name, + ProvisioningItemState::Failed, + reason, + message, + ), + }; + + finalize_bucket_policy_item(item, bucket, Some(desired_hash)) +} - let message = match create_result { - CreateBucketResult::Created => "RustFS bucket was created", - CreateBucketResult::AlreadyExists => "RustFS bucket already exists", +fn canned_anonymous_bucket_policy(access: BucketAnonymousAccess, bucket: &str) -> Option { + let bucket_arn = format!("arn:aws:s3:::{bucket}"); + let object_arn = format!("arn:aws:s3:::{bucket}/*"); + let (bucket_actions, object_actions): (&[&str], &[&str]) = match access { + BucketAnonymousAccess::Private => return None, + BucketAnonymousAccess::Download => ( + &["s3:GetBucketLocation", "s3:ListBucket"], + &["s3:GetObject"], + ), + BucketAnonymousAccess::Upload => ( + &["s3:ListBucketMultipartUploads"], + &[ + "s3:AbortMultipartUpload", + "s3:ListMultipartUploadParts", + "s3:PutObject", + ], + ), + BucketAnonymousAccess::Public => ( + &[ + "s3:GetBucketLocation", + "s3:ListBucket", + "s3:ListBucketMultipartUploads", + ], + &[ + "s3:AbortMultipartUpload", + "s3:DeleteObject", + "s3:GetObject", + "s3:ListMultipartUploadParts", + "s3:PutObject", + ], + ), }; - let item = run.item( - previous, - &bucket.name, - ProvisioningItemState::Ready, - Reason::ProvisioningConfigured, - message, - ); - annotate_bucket_item(item, bucket) + Some( + serde_json::json!({ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": ["*"]}, + "Action": bucket_actions, + "Resource": [bucket_arn] + }, + { + "Effect": "Allow", + "Principal": {"AWS": ["*"]}, + "Action": object_actions, + "Resource": [object_arn] + } + ] + }) + .to_string(), + ) } fn annotate_bucket_item( @@ -1874,6 +2287,19 @@ fn annotate_bucket_item( item } +fn finalize_bucket_policy_item( + mut item: ProvisioningItemStatus, + bucket: &ProvisioningBucket, + desired_hash: Option, +) -> ProvisioningItemStatus { + item = annotate_bucket_item(item, bucket); + item.desired_hash = desired_hash.clone(); + if item.last_applied_hash.is_none() && item.state == ProvisioningItemState::Ready.as_str() { + item.last_applied_hash = desired_hash; + } + item +} + fn has_active_spec(tenant: &Tenant) -> bool { !tenant.spec.policies.is_empty() || !tenant.spec.users.is_empty() @@ -2062,6 +2488,8 @@ fn reason_from_str(reason: &str) -> Reason { "UserOwnershipCheckpointFailed" => Reason::UserOwnershipCheckpointFailed, "BucketCreateFailed" => Reason::BucketCreateFailed, "BucketObjectLockConflict" => Reason::BucketObjectLockConflict, + "BucketPolicyApplyFailed" => Reason::BucketPolicyApplyFailed, + "BucketPolicyConflict" => Reason::BucketPolicyConflict, _ => Reason::ProvisioningFailed, } } @@ -2069,6 +2497,7 @@ fn reason_from_str(reason: &str) -> Reason { #[cfg(test)] mod tests { use super::*; + use crate::types::v1alpha1::provisioning::ConfigMapKeyReference; use axum::{ Router, body::Body, @@ -2196,6 +2625,7 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let request_count = Arc::new(AtomicUsize::new(0)); @@ -2282,6 +2712,7 @@ mod tests { now: "2026-07-18T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let client = RustfsAdminClient::new_with_base_url( @@ -2503,6 +2934,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let winner = make_run(); let loser = make_run(); @@ -2604,8 +3036,8 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; - persist_user_ownership_checkpoints(&run, &[first, second]) .await .expect("all checkpoints should be persisted together"); @@ -2672,7 +3104,7 @@ mod tests { let ctx = Context::new(Client::new(kube_service, "default")); let user = provisioning_user("app-user", "app-user-secret", "readwrite"); let tenant = provisioning_test_tenant(user.clone(), ProvisioningStatus::default()); - let run = ProvisioningRun { + let mut run = ProvisioningRun { ctx: &ctx, tenant: &tenant, namespace: "storage", @@ -2680,6 +3112,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let write_requests = Arc::new(AtomicUsize::new(0)); @@ -2724,7 +3157,7 @@ mod tests { let credentials = user_credentials("1"); let item = reconcile_user( - &run, + &mut run, &client, &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), &BTreeSet::new(), @@ -2808,7 +3241,7 @@ mod tests { } }); let ctx = Context::new(Client::new(kube_service, "default")); - let run = ProvisioningRun { + let mut run = ProvisioningRun { ctx: &ctx, tenant: &tenant, namespace: "storage", @@ -2816,6 +3249,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let get_sequence = sequence.clone(); @@ -2869,7 +3303,7 @@ mod tests { let credentials = user_credentials("5"); let item = reconcile_user( - &run, + &mut run, &client, &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), &BTreeSet::new(), @@ -2952,7 +3386,7 @@ mod tests { } }); let ctx = Context::new(Client::new(kube_service, "default")); - let run = ProvisioningRun { + let mut run = ProvisioningRun { ctx: &ctx, tenant: &tenant, namespace: "storage", @@ -2960,6 +3394,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let write_requests = Arc::new(AtomicUsize::new(0)); @@ -3003,7 +3438,7 @@ mod tests { RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); let item = reconcile_user( - &run, + &mut run, &client, &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), &BTreeSet::new(), @@ -3081,6 +3516,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let error = persist_user_ownership_checkpoints(&run, std::slice::from_ref(&checkpoint)) @@ -3115,7 +3551,7 @@ mod tests { ..Default::default() }; let tenant = provisioning_test_tenant(user.clone(), previous.clone()); - let run = ProvisioningRun { + let mut run = ProvisioningRun { ctx: &ctx, tenant: &tenant, namespace: "storage", @@ -3123,6 +3559,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let add_requests = Arc::new(AtomicUsize::new(0)); @@ -3167,7 +3604,7 @@ mod tests { RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); let item = reconcile_user( - &run, + &mut run, &client, &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), &BTreeSet::new(), @@ -3237,7 +3674,7 @@ mod tests { } }); let ctx = Context::new(Client::new(kube_service, "default")); - let run = ProvisioningRun { + let mut run = ProvisioningRun { ctx: &ctx, tenant: &tenant, namespace: "storage", @@ -3245,6 +3682,7 @@ mod tests { now: "2026-08-02T00:00:00Z".to_string(), status: ProvisioningStatus::default(), failures: Vec::new(), + retry: None, }; let add_requests = Arc::new(AtomicUsize::new(0)); @@ -3305,7 +3743,7 @@ mod tests { RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); let item = reconcile_user( - &run, + &mut run, &client, &BTreeMap::from([("readwrite".to_string(), "{}".to_string())]), &BTreeSet::new(), @@ -3464,7 +3902,7 @@ mod tests { .expect_err("RustFS policy parse error should fail provisioning"); assert_eq!( - error, + format!("failed to apply RustFS policy 'tenant-policy': {error}"), r#"failed to apply RustFS policy 'tenant-policy': upstream returned 400 Bad Request: InvalidRequest: invalid resource: unknown "*""# ); server.abort(); @@ -3679,7 +4117,12 @@ mod tests { ); previous.last_applied_hash = Some("sha256:old".to_string()); - let action = policy_reconcile_action(Some(&previous), Some("sha256:new"), "sha256:new"); + let action = policy_reconcile_action( + Some(&previous), + Some("sha256:new"), + "sha256:new", + Reason::PolicyConflict, + ); assert_eq!( action, @@ -3801,4 +4244,377 @@ mod tests { ); } } + + fn empty_kube_context() -> Context { + let kube_service = service_fn(|_request: http::Request| async { + Ok::<_, Infallible>( + http::Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(KubeBody::empty()) + .expect("response should build"), + ) + }); + Context::new(Client::new(kube_service, "default")) + } + + fn empty_run<'a>(ctx: &'a Context, tenant: &'a Tenant) -> ProvisioningRun<'a> { + ProvisioningRun { + ctx, + tenant, + namespace: "storage", + previous: ProvisioningStatus::default(), + now: "2026-09-04T00:00:00Z".to_string(), + status: ProvisioningStatus::default(), + failures: Vec::new(), + retry: None, + } + } + + #[test] + fn canned_anonymous_policies_cover_download_upload_and_public() { + assert!( + canned_anonymous_bucket_policy(BucketAnonymousAccess::Private, "app-data").is_none() + ); + + let download = canned_anonymous_bucket_policy(BucketAnonymousAccess::Download, "app-data") + .expect("download policy"); + assert!(download.contains("s3:GetObject")); + assert!(download.contains("s3:ListBucket")); + assert!(!download.contains("s3:PutObject")); + PolicyDocument::parse(&download).expect("download policy should parse"); + + let upload = canned_anonymous_bucket_policy(BucketAnonymousAccess::Upload, "app-data") + .expect("upload policy"); + assert!(upload.contains("s3:PutObject")); + assert!(!upload.contains("s3:GetObject")); + PolicyDocument::parse(&upload).expect("upload policy should parse"); + + let public = canned_anonymous_bucket_policy(BucketAnonymousAccess::Public, "logs") + .expect("public policy"); + assert!(public.contains("s3:GetObject")); + assert!(public.contains("s3:PutObject")); + assert!(public.contains("arn:aws:s3:::logs/*")); + PolicyDocument::parse(&public).expect("public policy should parse"); + } + + #[tokio::test] + async fn finish_prefers_retry_over_existing_failures() { + 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); + run.failures.push(( + Reason::BucketCreateFailed, + "bucket name is invalid".to_string(), + )); + run.request_retry("upstream returned 503"); + + match run.finish().outcome { + ProvisioningOutcome::Retry { + message, + persist_status, + retry_after, + } => { + assert!(message.contains("503")); + assert!(persist_status); + assert_eq!(retry_after, CHECKPOINT_TRANSIENT_RETRY); + } + _ => panic!("expected retry, got a different outcome"), + } + } + + #[tokio::test] + async fn finish_retries_pending_items_even_without_request_retry() { + 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); + run.status.buckets.push(run.item( + None::<&ProvisioningItemStatus>, + "app-data", + ProvisioningItemState::Pending, + Reason::ProvisioningPending, + "failed to create RustFS bucket: upstream returned 503", + )); + + match run.finish().outcome { + ProvisioningOutcome::Retry { persist_status, .. } => { + assert!(persist_status); + } + _ => panic!("pending items must requeue even if request_retry was skipped"), + } + } + + #[tokio::test] + async fn client_error_outcome_retries_transient_admin_failures() { + let (_, _, pending) = client_error_outcome(RustfsClientError::RequestFailed); + assert!(pending); + let (_, _, pending) = client_error_outcome(RustfsClientError::TenantTlsNotReady); + assert!(pending); + let (reason, _, pending) = client_error_outcome(RustfsClientError::MissingCredsSecret); + assert!(!pending); + assert_eq!(reason, Reason::ProvisioningUnsupported); + let (_, _, pending) = client_error_outcome(RustfsClientError::UnexpectedStatus { + status: reqwest::StatusCode::BAD_REQUEST, + detail: None, + }); + assert!(!pending); + let (_, _, pending) = client_error_outcome(RustfsClientError::UnexpectedStatus { + status: reqwest::StatusCode::SERVICE_UNAVAILABLE, + detail: None, + }); + assert!(pending); + } + + #[tokio::test] + async fn create_bucket_503_marks_pending_and_retries() { + let router = Router::new().route( + "/app-data", + 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 = reconcile_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Pending.as_str()); + assert_eq!(item.reason, Reason::ProvisioningPending.as_str()); + run.push_bucket(item); + + match run.finish().outcome { + ProvisioningOutcome::Retry { persist_status, .. } => assert!(persist_status), + _ => panic!("transient bucket create must retry"), + } + server.abort(); + } + + #[tokio::test] + async fn create_bucket_400_fails_without_retry() { + let router = Router::new().route("/app-data", put(|| async { StatusCode::BAD_REQUEST })); + 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 = reconcile_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketCreateFailed.as_str()); + run.push_bucket(item); + + match run.finish().outcome { + ProvisioningOutcome::Failed { reason, .. } => { + assert_eq!(reason, Reason::BucketCreateFailed); + } + _ => panic!("permanent bucket create errors must not retry"), + } + server.abort(); + } + + #[tokio::test] + async fn anonymous_download_policy_is_applied_when_live_policy_is_missing() { + let capture = PolicyApplyCapture::default(); + let route_capture = capture.clone(); + let router = Router::new().route( + "/app-data", + put({ + let capture = route_capture.clone(); + move |req: Request| { + let capture = capture.clone(); + async move { + if req.uri().query().unwrap_or("").contains("policy") { + let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) + .await + .expect("policy body"); + *capture.body.lock().await = + String::from_utf8(body_bytes.to_vec()).expect("utf8"); + } + StatusCode::OK + } + } + }) + .get(|req: Request| async move { + if req.uri().query().unwrap_or("").contains("policy") { + ( + StatusCode::NOT_FOUND, + r#"NoSuchBucketPolicy"#, + ) + } else { + (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(), + anonymous: BucketAnonymousAccess::Download, + ..Default::default() + }; + + let item = reconcile_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!(item.last_applied_hash.is_some()); + let applied = capture.body.lock().await; + assert!(applied.contains("s3:GetObject")); + assert!(!applied.contains("s3:PutObject")); + server.abort(); + } + + #[tokio::test] + async fn mixed_anonymous_and_custom_policy_fails_before_rustfs_calls() { + 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 client = RustfsAdminClient::new_with_base_url("http://127.0.0.1:1", "access", "secret"); + let bucket = ProvisioningBucket { + name: "app-data".to_string(), + anonymous: BucketAnonymousAccess::Public, + policy: Some(PolicyDocumentSource { + config_map_key_ref: ConfigMapKeyReference { + name: "bucket-policy".to_string(), + key: "policy.json".to_string(), + }, + }), + ..Default::default() + }; + + let item = reconcile_bucket(&mut run, &client, &bucket).await; + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketPolicyConflict.as_str()); + } + + #[tokio::test] + async fn list_canned_policies_503_is_transient() { + let router = Router::new().route( + "/rustfs/admin/v3/list-canned-policies", + get(|| 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 tenant = Tenant { + metadata: ObjectMeta { + name: Some("tenant-a".to_string()), + namespace: Some("storage".to_string()), + ..Default::default() + }, + spec: crate::types::v1alpha1::tenant::TenantSpec { + policies: vec![ProvisioningPolicy { + name: "app-readwrite".to_string(), + document: PolicyDocumentSource { + config_map_key_ref: ConfigMapKeyReference { + name: "app-policy".to_string(), + key: "policy.json".to_string(), + }, + }, + ..Default::default() + }], + ..Default::default() + }, + status: None, + }; + + let error = load_live_policies(&client, &tenant) + .await + .expect_err("503 should fail the live policy load"); + assert!(error.transient); + assert!( + error.message.contains("503") + || error.message.contains("unavailable") + || error.message.contains("failed to list") + ); + server.abort(); + } + + #[tokio::test] + async fn user_stays_pending_when_referenced_policy_is_pending() { + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user.clone(), ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + let client = RustfsAdminClient::new_with_base_url("http://127.0.0.1:1", "access", "secret"); + let pending = BTreeSet::from(["readwrite".to_string()]); + let credentials = user_credentials("1"); + + let plan = prepare_user_reconcile( + &mut run, + &client, + &BTreeMap::new(), + &BTreeSet::new(), + &pending, + &user, + &credentials, + ) + .await; + + let UserReconcilePlan::Complete(item) = plan else { + panic!("pending referenced policy must not call RustFS"); + }; + assert_eq!(item.state, ProvisioningItemState::Pending.as_str()); + assert_eq!(item.reason, Reason::ProvisioningPending.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("readwrite")) + ); + run.push_user(*item); + + match run.finish().outcome { + ProvisioningOutcome::Retry { persist_status, .. } => assert!(persist_status), + _ => panic!("pending referenced policy must requeue instead of failing the user"), + } + } } diff --git a/src/sts/admin_ops.rs b/src/sts/admin_ops.rs index 593d1a5..59e5937 100644 --- a/src/sts/admin_ops.rs +++ b/src/sts/admin_ops.rs @@ -18,15 +18,12 @@ use std::collections::BTreeMap; -use super::helpers::{ - body_mentions_not_found, build_canonical_query, extract_canned_policy_document, -}; +use super::helpers::{build_canonical_query, extract_canned_policy_document, is_absent_resource}; use super::{ ADD_CANNED_POLICY_PATH, ADD_USER_PATH, ADMIN_SIGNING_SERVICE, INFO_CANNED_POLICY_PATH, JSON_CONTENT_TYPE, LIST_CANNED_POLICIES_PATH, RustfsAdminClient, RustfsClientError, RustfsServerInfo, RustfsServerInfoResponse, SERVER_INFO_PATH, SET_POLICY_PATH, USER_INFO_PATH, }; -use reqwest::StatusCode; use serde_json::Value; impl RustfsAdminClient { @@ -183,7 +180,7 @@ impl RustfsAdminClient { let status = response.status(); let (body, truncated) = RustfsClientError::limited_response_body(response).await; - if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) { + if is_absent_resource(status, &body) { return Ok(false); } diff --git a/src/sts/helpers.rs b/src/sts/helpers.rs index b0975bd..17f6f36 100644 --- a/src/sts/helpers.rs +++ b/src/sts/helpers.rs @@ -148,9 +148,26 @@ pub(super) fn body_mentions_not_found(body: &str) -> bool { || body.contains("nosuchpolicy") || body.contains("no such policy") || body.contains("objectlockconfigurationnotfound") + || body.contains("nosuchbucketpolicy") + || body.contains("no such bucket policy") || body.contains("not found") } +/// True when an upstream status means the queried object is absent. +/// +/// 5xx, 408, 429, and 425 must not be treated as absence even if a proxy error page contains +/// "Not Found"; those are transient and should retry. +pub(super) fn is_absent_resource(status: StatusCode, body: &str) -> bool { + if status.is_server_error() + || status == StatusCode::REQUEST_TIMEOUT + || status == StatusCode::TOO_MANY_REQUESTS + || status == StatusCode::TOO_EARLY + { + return false; + } + status == StatusCode::NOT_FOUND || body_mentions_not_found(body) +} + pub(super) fn bucket_already_exists(status: StatusCode, body: &str) -> bool { if status == StatusCode::CONFLICT { let body = body.to_ascii_lowercase(); @@ -224,3 +241,38 @@ pub(super) fn extract_xml_tag(document: &str, tag: &str) -> Option { Some(rest[..end].trim().to_string()) } + +#[cfg(test)] +mod tests { + use super::is_absent_resource; + use reqwest::StatusCode; + + #[test] + fn absent_resource_accepts_not_found_and_semantic_4xx() { + assert!(is_absent_resource(StatusCode::NOT_FOUND, "")); + assert!(is_absent_resource( + StatusCode::NOT_FOUND, + "NoSuchBucketPolicy" + )); + assert!(is_absent_resource( + StatusCode::BAD_REQUEST, + "NoSuchUser" + )); + } + + #[test] + fn absent_resource_rejects_transient_status_even_with_not_found_body() { + let proxy_page = "Not Found"; + assert!(!is_absent_resource( + StatusCode::SERVICE_UNAVAILABLE, + proxy_page + )); + assert!(!is_absent_resource(StatusCode::BAD_GATEWAY, proxy_page)); + assert!(!is_absent_resource( + StatusCode::TOO_MANY_REQUESTS, + proxy_page + )); + assert!(!is_absent_resource(StatusCode::REQUEST_TIMEOUT, proxy_page)); + assert!(!is_absent_resource(StatusCode::TOO_EARLY, proxy_page)); + } +} diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index 865c27f..c0a19a3 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -262,6 +262,38 @@ impl std::fmt::Display for RustfsClientError { impl std::error::Error for RustfsClientError {} impl RustfsClientError { + /// Network, timeout, and retryable upstream failures. Permanent configuration and + /// 4xx semantic errors are not transient. + pub(crate) fn is_transient(&self) -> bool { + match self { + Self::RequestFailed + | Self::TenantTlsNotReady + | Self::TenantSecretLookupFailed + | Self::TenantTlsCaSecretLookupFailed { .. } + | Self::ParseResponseFailed => true, + Self::UnexpectedStatus { status, .. } => { + *status == StatusCode::REQUEST_TIMEOUT + || *status == StatusCode::TOO_MANY_REQUESTS + || *status == StatusCode::TOO_EARLY + || status.is_server_error() + } + Self::MissingTenantNamespace + | Self::MissingCredsSecret + | Self::MissingCredentialKey { .. } + | Self::EmptyCredentialValue { .. } + | Self::InvalidCredentialValue { .. } + | Self::InvalidPolicyName + | Self::InvalidPolicyDocument + | Self::TenantTlsRequired + | Self::TenantTlsClientCertificateRequired + | Self::MissingTenantTlsCaKey { .. } + | Self::InvalidTenantTlsCa + | Self::TlsClientBuildFailed + | Self::RequestBuildFailed + | Self::SigningFailed => false, + } + } + pub(super) async fn unexpected_response(response: Response) -> Self { let status = response.status(); let (body, truncated) = read_limited_response_body(response).await; diff --git a/src/sts/s3_ops.rs b/src/sts/s3_ops.rs index 0df217f..f48568c 100644 --- a/src/sts/s3_ops.rs +++ b/src/sts/s3_ops.rs @@ -16,10 +16,8 @@ //! - bucket lifecycle methods (create/lookup features) //! - request semantics for S3-style object storage operations. -use reqwest::StatusCode; - use super::helpers::{ - body_mentions_not_found, bucket_already_exists, build_canonical_query, create_bucket_body, + bucket_already_exists, build_canonical_query, create_bucket_body, is_absent_resource, }; use super::{ADMIN_SIGNING_SERVICE, CreateBucketResult, RustfsAdminClient, RustfsClientError}; @@ -122,7 +120,7 @@ impl RustfsAdminClient { if !response.status().is_success() { let status = response.status(); let (body, truncated) = RustfsClientError::limited_response_body(response).await; - if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) { + if is_absent_resource(status, &body) { return Ok(false); } return Err(RustfsClientError::unexpected_status_with_limited_body( @@ -136,4 +134,99 @@ impl RustfsAdminClient { .map_err(|_| RustfsClientError::RequestFailed)?; Ok(body.contains("Enabled")) } + + pub async fn put_bucket_policy( + &self, + bucket: &str, + policy: &str, + ) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() || policy.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("policy", "")]); + let signed = self.sign_request( + "PUT", + &path, + &query, + policy, + Some("application/json"), + ADMIN_SIGNING_SERVICE, + )?; + let host = self.host()?; + + let response = self + .http_client + .put(format!( + "{}{}?{query}", + 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) + .header("content-type", "application/json") + .body(policy.to_string()) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + return Ok(()); + } + + Err(RustfsClientError::unexpected_response(response).await) + } + + pub async fn get_bucket_policy( + &self, + bucket: &str, + ) -> Result, RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("policy", "")]); + let signed = self.sign_request("GET", &path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + + let response = self + .http_client + .get(format!( + "{}{}?{query}", + 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)?; + + if response.status().is_success() { + let body = response + .text() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + let trimmed = body.trim(); + if trimmed.is_empty() { + return Ok(None); + } + return Ok(Some(body)); + } + + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if is_absent_resource(status, &body) { + return Ok(None); + } + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } } diff --git a/src/sts/tests.rs b/src/sts/tests.rs index 502a92a..7001315 100644 --- a/src/sts/tests.rs +++ b/src/sts/tests.rs @@ -1208,3 +1208,112 @@ fn extract_canned_policy_document_accepts_raw_policy_document() { assert_eq!(policy_value["Version"], "2012-10-17"); assert_eq!(policy_value["Statement"][0]["Sid"], "raw"); } + +#[test] +fn rustfs_client_error_classifies_transient_failures() { + assert!(RustfsClientError::RequestFailed.is_transient()); + assert!(RustfsClientError::TenantTlsNotReady.is_transient()); + assert!(RustfsClientError::ParseResponseFailed.is_transient()); + assert!(RustfsClientError::TenantSecretLookupFailed.is_transient()); + assert!( + RustfsClientError::UnexpectedStatus { + status: StatusCode::SERVICE_UNAVAILABLE, + detail: None, + } + .is_transient() + ); + assert!( + RustfsClientError::UnexpectedStatus { + status: StatusCode::TOO_MANY_REQUESTS, + detail: None, + } + .is_transient() + ); + assert!( + !RustfsClientError::UnexpectedStatus { + status: StatusCode::BAD_REQUEST, + detail: None, + } + .is_transient() + ); + assert!(!RustfsClientError::MissingCredsSecret.is_transient()); + assert!(!RustfsClientError::InvalidPolicyDocument.is_transient()); + assert!(!RustfsClientError::InvalidPolicyName.is_transient()); +} + +#[tokio::test] +async fn get_bucket_policy_treats_nosuchbucketpolicy_as_absent() { + let router = Router::new().route( + "/app-data", + get(|req: Request| async move { + assert_eq!(req.uri().query().unwrap_or(""), "policy="); + ( + StatusCode::NOT_FOUND, + r#"NoSuchBucketPolicyThe bucket policy does not exist"#, + ) + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + + assert_eq!(client.get_bucket_policy("app-data").await.unwrap(), None); + server.abort(); +} + +#[tokio::test] +async fn get_bucket_policy_treats_503_as_transient_even_with_not_found_body() { + let router = Router::new().route( + "/app-data", + get(|req: Request| async move { + assert_eq!(req.uri().query().unwrap_or(""), "policy="); + ( + StatusCode::SERVICE_UNAVAILABLE, + "Not Foundupstream unavailable", + ) + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + + let error = client.get_bucket_policy("app-data").await.unwrap_err(); + assert!(error.is_transient()); + server.abort(); +} + +#[tokio::test] +async fn put_bucket_policy_sends_json_document() { + let capture = Arc::new(Mutex::new(String::new())); + let route_capture = capture.clone(); + let router = Router::new().route( + "/app-data", + put(move |req: Request| { + let capture = route_capture.clone(); + async move { + assert_eq!(req.uri().query().unwrap_or(""), "policy="); + let body = axum::body::to_bytes(req.into_body(), usize::MAX) + .await + .unwrap(); + *capture.lock().await = String::from_utf8(body.to_vec()).unwrap(); + StatusCode::NO_CONTENT + } + }), + ); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let policy = r#"{"Version":"2012-10-17","Statement":[]}"#; + client.put_bucket_policy("app-data", policy).await.unwrap(); + assert_eq!(&*capture.lock().await, policy); + server.abort(); +} diff --git a/src/tenant_reference_index.rs b/src/tenant_reference_index.rs index 4d605bc..c68eda3 100644 --- a/src/tenant_reference_index.rs +++ b/src/tenant_reference_index.rs @@ -52,6 +52,11 @@ impl TenantReferences { for policy in &tenant.spec.policies { references.insert_config_map(namespace, &policy.document.config_map_key_ref.name); } + for bucket in &tenant.spec.buckets { + if let Some(source) = bucket.policy.as_ref() { + references.insert_config_map(namespace, &source.config_map_key_ref.name); + } + } for env in &tenant.spec.env { if let Some(config_map) = env .value_from @@ -281,6 +286,19 @@ mod tests { }, ..Default::default() }); + tenant + .spec + .buckets + .push(crate::types::v1alpha1::provisioning::ProvisioningBucket { + name: "app-data".to_string(), + policy: Some(PolicyDocumentSource { + config_map_key_ref: ConfigMapKeyReference { + name: "bucket-policy".to_string(), + key: "policy.json".to_string(), + }, + }), + ..Default::default() + }); tenant.spec.creds_secret = Some(local_ref("credentials")); tenant.spec.image_pull_secret = Some(local_ref("image-pull")); tenant.spec.rpc_secret = Some(RpcSecretRef { @@ -362,6 +380,11 @@ mod tests { "tenant-a", "storage", ); + assert_single_ref( + &index.refs_for_config_map(Some("storage"), Some("bucket-policy")), + "tenant-a", + "storage", + ); assert_single_ref( &index.refs_for_config_map(Some("storage"), Some("runtime-settings")), "tenant-a", diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs index f2c7300..202525a 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -15,6 +15,7 @@ pub mod encryption; pub mod k8s; pub mod logging; +pub mod network; pub mod persistence; pub mod policy_binding; pub mod pool; @@ -338,6 +339,23 @@ mod tenant_provisioning_crd_tests { [0]["message"], json!("bucket name must be a valid RustFS/S3 bucket name") ); + assert_eq!(spec["properties"]["hostUsers"]["type"], json!("boolean")); + assert_eq!( + spec["properties"]["network"]["properties"]["ipFamilyPolicy"]["enum"], + json!(["SingleStack", "PreferDualStack", "RequireDualStack", null]) + ); + assert_eq!( + spec["properties"]["network"]["properties"]["ipFamilies"]["items"]["enum"], + json!(["IPv4", "IPv6"]) + ); + assert_eq!( + spec["properties"]["buckets"]["items"]["properties"]["anonymous"]["enum"], + json!(["Private", "Download", "Upload", "Public"]) + ); + assert_eq!( + spec["properties"]["buckets"]["items"]["x-kubernetes-validations"][0]["message"], + json!("bucket policy and anonymous access are mutually exclusive") + ); } } diff --git a/src/types/v1alpha1/network.rs b/src/types/v1alpha1/network.rs new file mode 100644 index 0000000..4d8b309 --- /dev/null +++ b/src/types/v1alpha1/network.rs @@ -0,0 +1,176 @@ +// 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. + +use kube::KubeSchema; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +/// Kubernetes Service IP family policy values. +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +pub enum IpFamilyPolicy { + SingleStack, + PreferDualStack, + RequireDualStack, +} + +impl IpFamilyPolicy { + pub(crate) fn as_str(&self) -> &'static str { + match self { + Self::SingleStack => "SingleStack", + Self::PreferDualStack => "PreferDualStack", + Self::RequireDualStack => "RequireDualStack", + } + } +} + +/// Kubernetes Service IP family values. +#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +pub enum IpFamily { + #[serde(rename = "IPv4")] + #[schemars(rename = "IPv4")] + IPv4, + #[serde(rename = "IPv6")] + #[schemars(rename = "IPv6")] + IPv6, +} + +impl IpFamily { + pub(crate) fn as_str(&self) -> &'static str { + match self { + Self::IPv4 => "IPv4", + Self::IPv6 => "IPv6", + } + } +} + +/// Tenant Service and listen-address networking. +/// +/// When omitted, generated Services inherit the cluster default IP family policy and RustFS +/// listens on `0.0.0.0`. Set `ipFamilies: [IPv6]` or a dual-stack policy for IPv6-only and +/// dual-stack clusters. +#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct NetworkConfig { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ip_family_policy: Option, + + #[schemars(length(max = 2), extend("x-kubernetes-list-type" = "set"))] + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub ip_families: Vec, +} + +impl NetworkConfig { + pub(crate) fn uses_ipv6(&self) -> bool { + self.ip_families + .iter() + .any(|family| matches!(family, IpFamily::IPv6)) + || matches!( + self.ip_family_policy, + Some(IpFamilyPolicy::PreferDualStack | IpFamilyPolicy::RequireDualStack) + ) + } + + pub(crate) fn rustfs_listen_address(&self, port: u16) -> String { + rustfs_listen_address(Some(self), port) + } + + pub(crate) fn service_ip_family_policy(&self) -> Option { + self.ip_family_policy + .as_ref() + .map(|policy| policy.as_str().to_string()) + } + + pub(crate) fn service_ip_families(&self) -> Option> { + if self.ip_families.is_empty() { + None + } else { + Some( + self.ip_families + .iter() + .map(|family| family.as_str().to_string()) + .collect(), + ) + } + } +} + +pub(crate) fn rustfs_listen_address(network: Option<&NetworkConfig>, port: u16) -> String { + if network.is_some_and(NetworkConfig::uses_ipv6) { + format!("[::]:{port}") + } else { + format!("0.0.0.0:{port}") + } +} + +#[cfg(test)] +mod tests { + use super::{IpFamily, IpFamilyPolicy, NetworkConfig, rustfs_listen_address}; + + #[test] + fn omitted_network_keeps_ipv4_listen_addresses() { + assert_eq!(rustfs_listen_address(None, 9000), "0.0.0.0:9000"); + assert_eq!( + NetworkConfig::default().rustfs_listen_address(9001), + "0.0.0.0:9001" + ); + } + + #[test] + fn ipv6_family_and_dual_stack_policy_listen_on_unspecified_v6() { + let ipv6 = NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::SingleStack), + ip_families: vec![IpFamily::IPv6], + }; + assert_eq!(ipv6.rustfs_listen_address(9000), "[::]:9000"); + + let dual = NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::PreferDualStack), + ip_families: vec![IpFamily::IPv4, IpFamily::IPv6], + }; + assert_eq!(dual.rustfs_listen_address(9001), "[::]:9001"); + assert_eq!( + dual.service_ip_families().as_deref(), + Some(["IPv4".to_string(), "IPv6".to_string()].as_slice()) + ); + assert_eq!( + dual.service_ip_family_policy().as_deref(), + Some("PreferDualStack") + ); + } + + #[test] + fn ipv4_single_stack_keeps_ipv4_listen_address() { + let ipv4 = NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::SingleStack), + ip_families: vec![IpFamily::IPv4], + }; + assert!(!ipv4.uses_ipv6()); + assert_eq!(ipv4.rustfs_listen_address(9000), "0.0.0.0:9000"); + } + + #[test] + fn ipv6_first_dual_stack_preserves_family_order_and_listens_on_v6() { + let dual = NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::RequireDualStack), + ip_families: vec![IpFamily::IPv6, IpFamily::IPv4], + }; + assert!(dual.uses_ipv6()); + assert_eq!(dual.rustfs_listen_address(9000), "[::]:9000"); + assert_eq!( + dual.service_ip_families().as_deref(), + Some(["IPv6".to_string(), "IPv4".to_string()].as_slice()) + ); + } +} diff --git a/src/types/v1alpha1/provisioning.rs b/src/types/v1alpha1/provisioning.rs index 1921a4e..8cdf1b1 100644 --- a/src/types/v1alpha1/provisioning.rs +++ b/src/types/v1alpha1/provisioning.rs @@ -121,8 +121,31 @@ pub(crate) fn duplicate_user_credentials_secret_names( .collect() } +#[derive( + Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, Default, PartialEq, Eq, +)] +#[serde(rename_all = "PascalCase")] +pub enum BucketAnonymousAccess { + #[default] + Private, + Download, + Upload, + Public, +} + +impl BucketAnonymousAccess { + pub(crate) fn is_private(&self) -> bool { + matches!(self, Self::Private) + } +} + +fn skip_private_anonymous(value: &BucketAnonymousAccess) -> bool { + value.is_private() +} + #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] #[serde(rename_all = "camelCase")] +#[x_kube(validation = Rule::new("!(has(self.policy) && has(self.anonymous) && self.anonymous != 'Private')").message("bucket policy and anonymous access are mutually exclusive"))] pub struct ProvisioningBucket { #[schemars( length(min = MIN_BUCKET_NAME_LENGTH, max = MAX_BUCKET_NAME_LENGTH), @@ -137,6 +160,14 @@ pub struct ProvisioningBucket { #[serde(default, skip_serializing_if = "Option::is_none")] pub object_lock: Option, + /// Canned anonymous access for this bucket. Mutually exclusive with `policy`. + #[serde(default, skip_serializing_if = "skip_private_anonymous")] + pub anonymous: BucketAnonymousAccess, + + /// Custom bucket policy document sourced from a ConfigMap. Mutually exclusive with `anonymous`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub policy: Option, + #[serde(default, skip_serializing_if = "is_retain")] pub deletion_policy: ProvisioningDeletionPolicy, } @@ -145,6 +176,14 @@ impl ProvisioningBucket { pub fn object_lock_enabled(&self) -> bool { self.object_lock.unwrap_or(false) } + + pub(crate) fn has_custom_policy(&self) -> bool { + self.policy.is_some() + } + + pub(crate) fn has_anonymous_access(&self) -> bool { + !self.anonymous.is_private() + } } #[cfg(test)] @@ -208,4 +247,23 @@ mod tests { BTreeSet::from(["shared-secret"]) ); } + + #[test] + fn private_anonymous_access_is_omitted_from_serialized_bucket() { + let bucket = super::ProvisioningBucket { + name: "app-data".to_string(), + ..Default::default() + }; + let value = serde_json::to_value(&bucket).expect("bucket serializes"); + assert!(value.get("anonymous").is_none()); + assert!(value.get("policy").is_none()); + + let public: super::ProvisioningBucket = serde_json::from_value(serde_json::json!({ + "name": "app-data", + "anonymous": "Public" + })) + .expect("public anonymous deserializes"); + assert!(public.has_anonymous_access()); + assert!(!public.has_custom_policy()); + } } diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs index c7860bf..7d01c4f 100755 --- a/src/types/v1alpha1/status.rs +++ b/src/types/v1alpha1/status.rs @@ -175,6 +175,8 @@ pub enum Reason { UserOwnershipCheckpointFailed, BucketCreateFailed, BucketObjectLockConflict, + BucketPolicyApplyFailed, + BucketPolicyConflict, KubernetesApiError, StatusPatchFailed, ObservedGenerationStale, @@ -246,6 +248,8 @@ impl Reason { Self::UserOwnershipCheckpointFailed => "UserOwnershipCheckpointFailed", Self::BucketCreateFailed => "BucketCreateFailed", Self::BucketObjectLockConflict => "BucketObjectLockConflict", + Self::BucketPolicyApplyFailed => "BucketPolicyApplyFailed", + Self::BucketPolicyConflict => "BucketPolicyConflict", Self::KubernetesApiError => "KubernetesApiError", Self::StatusPatchFailed => "StatusPatchFailed", Self::ObservedGenerationStale => "ObservedGenerationStale", @@ -520,6 +524,8 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "UserOwnershipCheckpointFailed" | "BucketCreateFailed" | "BucketObjectLockConflict" + | "BucketPolicyApplyFailed" + | "BucketPolicyConflict" ) } @@ -616,6 +622,8 @@ pub fn next_actions_for_reason(reason: &str) -> Vec<&'static str> { } "BucketCreateFailed" => vec!["inspectBucket", "inspectOperatorLogs"], "BucketObjectLockConflict" => vec!["createObjectLockBucket", "fixBucketSpec"], + "BucketPolicyApplyFailed" => vec!["fixBucketPolicy", "inspectOperatorLogs"], + "BucketPolicyConflict" => vec!["inspectLiveBucketPolicy", "updateBucketSpec"], "KubernetesApiError" => vec!["retry", "inspectOperatorLogs"], "ObservedGenerationStale" => vec!["waitForReconcile"], _ => Vec::new(), @@ -707,6 +715,14 @@ mod tests { "inspectOperatorLogs" ] ); + assert_eq!( + next_actions_for_reason("BucketPolicyApplyFailed"), + vec!["fixBucketPolicy", "inspectOperatorLogs"] + ); + assert_eq!( + next_actions_for_reason("BucketPolicyConflict"), + vec!["inspectLiveBucketPolicy", "updateBucketSpec"] + ); } #[test] diff --git a/src/types/v1alpha1/tenant.rs b/src/types/v1alpha1/tenant.rs index c62a22d..fb87d51 100755 --- a/src/types/v1alpha1/tenant.rs +++ b/src/types/v1alpha1/tenant.rs @@ -15,6 +15,7 @@ use crate::types::v1alpha1::encryption::EncryptionConfig; use crate::types::v1alpha1::k8s; use crate::types::v1alpha1::logging::LoggingConfig; +use crate::types::v1alpha1::network::NetworkConfig; use crate::types::v1alpha1::pool::{Pool, validate_pool_collection}; use crate::types::v1alpha1::pool_lifecycle::PoolLifecycleSpec; use crate::types::v1alpha1::provisioning::{ @@ -123,6 +124,18 @@ pub struct TenantSpec { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub env: Vec, + /// Tenant Service IP family policy and RustFS listen addresses. + /// When omitted, generated Services inherit the cluster default and RustFS listens on IPv4. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub network: Option, + + /// Pod `hostUsers` for generated RustFS workloads. + /// + /// `false` isolates the user namespace and satisfies OpenShift `restricted-v3`. + /// When omitted, an OpenShift-style empty security-context pair also renders `hostUsers: false`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host_users: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub tls: Option, diff --git a/src/types/v1alpha1/tenant/services.rs b/src/types/v1alpha1/tenant/services.rs index 21d5487..76b75fd 100755 --- a/src/types/v1alpha1/tenant/services.rs +++ b/src/types/v1alpha1/tenant/services.rs @@ -26,6 +26,19 @@ fn console_service_name(tenant: &Tenant) -> String { format!("{}-console", tenant.name()) } +fn tenant_service_network(tenant: &Tenant) -> (Option, Option>) { + tenant + .spec + .network + .as_ref() + .map_or((None, None), |network| { + ( + network.service_ip_family_policy(), + network.service_ip_families(), + ) + }) +} + impl Tenant { /// a new io Service for tenant pub fn new_io_service(&self) -> corev1::Service { @@ -41,16 +54,21 @@ impl Tenant { labels: Some(self.common_labels()), ..Default::default() }, - spec: Some(corev1::ServiceSpec { - type_: Some("ClusterIP".to_owned()), - selector: Some(self.selector_labels()), - ports: Some(vec![corev1::ServicePort { - port: 9000, - target_port: Some(intstr::IntOrString::Int(9000)), - name: Some(rustfs_service_port_name(tls_plan).to_owned()), + spec: Some({ + let (ip_family_policy, ip_families) = tenant_service_network(self); + corev1::ServiceSpec { + type_: Some("ClusterIP".to_owned()), + selector: Some(self.selector_labels()), + ip_family_policy, + ip_families, + ports: Some(vec![corev1::ServicePort { + port: 9000, + target_port: Some(intstr::IntOrString::Int(9000)), + name: Some(rustfs_service_port_name(tls_plan).to_owned()), + ..Default::default() + }]), ..Default::default() - }]), - ..Default::default() + } }), ..Default::default() } @@ -66,16 +84,21 @@ impl Tenant { labels: Some(self.common_labels()), ..Default::default() }, - spec: Some(corev1::ServiceSpec { - type_: Some("ClusterIP".to_owned()), - selector: Some(self.selector_labels()), - ports: Some(vec![corev1::ServicePort { - port: 9001, - target_port: Some(intstr::IntOrString::Int(9001)), - name: Some("http-console".to_owned()), + spec: Some({ + let (ip_family_policy, ip_families) = tenant_service_network(self); + corev1::ServiceSpec { + type_: Some("ClusterIP".to_owned()), + selector: Some(self.selector_labels()), + ip_family_policy, + ip_families, + ports: Some(vec![corev1::ServicePort { + port: 9001, + target_port: Some(intstr::IntOrString::Int(9001)), + name: Some("http-console".to_owned()), + ..Default::default() + }]), ..Default::default() - }]), - ..Default::default() + } }), ..Default::default() } @@ -95,17 +118,22 @@ impl Tenant { labels: Some(self.common_labels()), ..Default::default() }, - spec: Some(corev1::ServiceSpec { - type_: Some("ClusterIP".to_owned()), - cluster_ip: Some("None".to_owned()), - publish_not_ready_addresses: Some(true), - selector: Some(self.selector_labels()), - ports: Some(vec![corev1::ServicePort { - port: 9000, - name: Some(rustfs_service_port_name(tls_plan).to_owned()), + spec: Some({ + let (ip_family_policy, ip_families) = tenant_service_network(self); + corev1::ServiceSpec { + type_: Some("ClusterIP".to_owned()), + cluster_ip: Some("None".to_owned()), + publish_not_ready_addresses: Some(true), + selector: Some(self.selector_labels()), + ip_family_policy, + ip_families, + ports: Some(vec![corev1::ServicePort { + port: 9000, + name: Some(rustfs_service_port_name(tls_plan).to_owned()), + ..Default::default() + }]), ..Default::default() - }]), - ..Default::default() + } }), ..Default::default() } @@ -164,4 +192,28 @@ mod tests { Some("https-rustfs") ); } + + #[test] + fn ipv6_network_config_is_copied_onto_tenant_services() { + use crate::types::v1alpha1::network::{IpFamily, IpFamilyPolicy, NetworkConfig}; + + let mut tenant = crate::tests::create_test_tenant(None, None); + tenant.spec.network = Some(NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::SingleStack), + ip_families: vec![IpFamily::IPv6], + }); + + for service in [ + tenant.new_io_service(), + tenant.new_console_service(), + tenant.new_headless_service(), + ] { + let spec = service.spec.expect("Service should have spec"); + assert_eq!(spec.ip_family_policy.as_deref(), Some("SingleStack")); + assert_eq!( + spec.ip_families.as_deref(), + Some(["IPv6".to_string()].as_slice()) + ); + } + } } diff --git a/src/types/v1alpha1/tenant/workloads.rs b/src/types/v1alpha1/tenant/workloads.rs index 0b8a15b..b8cf09a 100755 --- a/src/types/v1alpha1/tenant/workloads.rs +++ b/src/types/v1alpha1/tenant/workloads.rs @@ -16,6 +16,7 @@ use super::Tenant; use crate::cluster_dns; use crate::types; use crate::types::v1alpha1::encryption::KmsBackendType; +use crate::types::v1alpha1::network::rustfs_listen_address; use crate::types::v1alpha1::persistence::{ DEFAULT_PERSISTENCE_PATH, LEGACY_LOCAL_KMS_KEY_DIR, data_volume_mount_path, default_local_kms_key_directory, @@ -528,6 +529,10 @@ fn effective_workload_security_context( } } +fn effective_host_users(explicit: Option, platform_delegated: bool) -> Option { + explicit.or_else(|| platform_delegated.then_some(false)) +} + const TLS_OPERATOR_MANAGED_ENV_VARS: &[&str] = &[ "RUSTFS_VOLUMES", "RUSTFS_TLS_PATH", @@ -1319,13 +1324,19 @@ impl Tenant { // Add required RustFS environment variables env_vars.push(corev1::EnvVar { name: "RUSTFS_ADDRESS".to_owned(), - value: Some("0.0.0.0:9000".to_owned()), + value: Some(self.spec.network.as_ref().map_or_else( + || rustfs_listen_address(None, 9000), + |network| network.rustfs_listen_address(9000), + )), ..Default::default() }); env_vars.push(corev1::EnvVar { name: "RUSTFS_CONSOLE_ADDRESS".to_owned(), - value: Some("0.0.0.0:9001".to_owned()), + value: Some(self.spec.network.as_ref().map_or_else( + || rustfs_listen_address(None, 9001), + |network| network.rustfs_listen_address(9001), + )), ..Default::default() }); @@ -1424,6 +1435,10 @@ impl Tenant { pool.container_security_context.as_ref(), ); self.validate_effective_workload_identity(pool, &security)?; + let host_users = effective_host_users( + self.spec.host_users, + security.pod_operator_defaults_delegated, + ); let EffectiveWorkloadSecurityContext { pod: pod_security_context, container: container_security_context, @@ -1517,6 +1532,7 @@ impl Tenant { .service_account_name .is_none() .then_some(false), + host_users, containers: vec![container], security_context: Some(pod_security_context), volumes: Some(pod_volumes), @@ -1664,6 +1680,10 @@ impl Tenant { return Ok(true); } + if existing_pod_spec.host_users != desired_pod_spec.host_users { + return Ok(true); + } + // Operator-created ServiceAccounts do not require Kubernetes API access. Compare this // field only for the default ServiceAccount so custom workload identity webhooks remain // free to manage token projection without causing a reconcile loop. @@ -3242,6 +3262,10 @@ mod tests { container_security_context.read_only_root_filesystem, None, "readOnlyRootFilesystem is configurable but not required by restricted" ); + assert_eq!( + pod_spec.host_users, None, + "default tenants keep the Kubernetes host user namespace" + ); } #[test] @@ -3272,12 +3296,79 @@ mod tests { pod_context.fs_group_change_policy.as_deref(), Some("OnRootMismatch") ); + assert_eq!(pod_spec.host_users, Some(false)); assert_eq!( pod_spec.containers[0].security_context, Some(corev1::SecurityContext::default()) ); } + #[test] + fn explicit_host_users_overrides_platform_delegation_default() { + let mut tenant = crate::tests::create_test_tenant(None, None); + tenant.spec.security_context = Some(PodSecurityContextOverride::default()); + tenant.spec.container_security_context = Some(corev1::SecurityContext::default()); + tenant.spec.host_users = Some(true); + + let pod_spec = tenant + .new_statefulset(&tenant.spec.pools[0]) + .expect("hostUsers override should render") + .spec + .expect("StatefulSet should have spec") + .template + .spec + .expect("Pod template should have spec"); + assert_eq!(pod_spec.host_users, Some(true)); + } + + #[test] + fn ipv6_network_config_listens_on_unspecified_v6() { + use crate::types::v1alpha1::network::{IpFamily, IpFamilyPolicy, NetworkConfig}; + + let mut tenant = crate::tests::create_test_tenant(None, None); + tenant.spec.network = Some(NetworkConfig { + ip_family_policy: Some(IpFamilyPolicy::SingleStack), + ip_families: vec![IpFamily::IPv6], + }); + + let env = tenant + .new_statefulset(&tenant.spec.pools[0]) + .expect("IPv6 Tenant should render") + .spec + .expect("StatefulSet should have spec") + .template + .spec + .expect("Pod template should have spec") + .containers[0] + .env + .clone() + .unwrap_or_default(); + let value = |name: &str| { + env.iter() + .find(|var| var.name == name) + .and_then(|var| var.value.clone()) + }; + assert_eq!(value("RUSTFS_ADDRESS").as_deref(), Some("[::]:9000")); + assert_eq!( + value("RUSTFS_CONSOLE_ADDRESS").as_deref(), + Some("[::]:9001") + ); + } + + #[test] + fn host_users_change_marks_statefulset_for_update() { + let mut tenant = crate::tests::create_test_tenant(None, None); + let existing = tenant + .new_statefulset(&tenant.spec.pools[0]) + .expect("existing StatefulSet"); + tenant.spec.host_users = Some(false); + assert!( + tenant + .statefulset_needs_update(&existing, &tenant.spec.pools[0]) + .expect("compare should succeed") + ); + } + #[test] fn lone_empty_tenant_pod_context_retains_operator_defaults() { let mut tenant = crate::tests::create_test_tenant(None, None); diff --git a/src/utils.rs b/src/utils.rs index c227851..17b9dfa 100755 --- a/src/utils.rs +++ b/src/utils.rs @@ -12,5 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(crate) mod listen; pub(crate) mod sanitize; pub mod tls; diff --git a/src/utils/listen.rs b/src/utils/listen.rs new file mode 100644 index 0000000..7771686 --- /dev/null +++ b/src/utils/listen.rs @@ -0,0 +1,175 @@ +// 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. + +use std::io; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use tokio::net::TcpListener; + +pub(crate) const OPERATOR_BIND_ADDRESS_ENV: &str = "OPERATOR_BIND_ADDRESS"; +pub(crate) const CONSOLE_BIND_ADDRESS_ENV: &str = "CONSOLE_BIND_ADDRESS"; + +/// Bind an HTTP listener for operator-owned process sockets. +/// +/// An explicit IPv4 or IPv6 address in `bind_address_env` always wins. Otherwise the listener +/// prefers the IPv6 unspecified address (`::`), which is dual-stack on typical Linux kernels, +/// and falls back to IPv4 (`0.0.0.0`) on IPv4-only nodes. +pub(crate) async fn bind_unspecified_listener( + port: u16, + bind_address_env: &str, +) -> io::Result { + let mut last_error = None; + for addr in listen_addrs(port, bind_address_env)? { + match TcpListener::bind(addr).await { + Ok(listener) => return Ok(listener), + Err(error) => last_error = Some(error), + } + } + Err(last_error.unwrap_or_else(|| { + io::Error::new( + io::ErrorKind::AddrNotAvailable, + format!("no listen address available for port {port}"), + ) + })) +} + +fn listen_addrs(port: u16, bind_address_env: &str) -> io::Result> { + if let Some(ip) = explicit_bind_ip(bind_address_env)? { + return Ok(vec![SocketAddr::from((ip, port))]); + } + Ok(vec![ + SocketAddr::from((Ipv6Addr::UNSPECIFIED, port)), + SocketAddr::from((Ipv4Addr::UNSPECIFIED, port)), + ]) +} + +fn explicit_bind_ip(bind_address_env: &str) -> io::Result> { + match std::env::var(bind_address_env) { + Ok(raw) => { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Ok(None); + } + trimmed.parse::().map(Some).map_err(|error| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("invalid {bind_address_env} value '{trimmed}': {error}"), + ) + }) + } + Err(_) => Ok(None), + } +} + +#[cfg(test)] +mod tests { + use super::{ + CONSOLE_BIND_ADDRESS_ENV, OPERATOR_BIND_ADDRESS_ENV, bind_unspecified_listener, + listen_addrs, + }; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; + use std::sync::{Mutex, MutexGuard, OnceLock}; + + static ENV_LOCK: OnceLock> = OnceLock::new(); + + fn env_lock() -> MutexGuard<'static, ()> { + ENV_LOCK + .get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|error| error.into_inner()) + } + + fn restore_env(name: &str, previous: Option) { + match previous { + Some(value) => unsafe { std::env::set_var(name, value) }, + None => unsafe { std::env::remove_var(name) }, + } + } + + #[test] + fn auto_listen_addrs_prefer_ipv6_unspecified_then_ipv4() { + let _guard = env_lock(); + let previous = std::env::var(OPERATOR_BIND_ADDRESS_ENV).ok(); + unsafe { std::env::remove_var(OPERATOR_BIND_ADDRESS_ENV) }; + + let addrs = listen_addrs(8080, OPERATOR_BIND_ADDRESS_ENV).expect("auto addrs"); + restore_env(OPERATOR_BIND_ADDRESS_ENV, previous); + + assert_eq!( + addrs, + vec![ + SocketAddr::from((Ipv6Addr::UNSPECIFIED, 8080)), + SocketAddr::from((Ipv4Addr::UNSPECIFIED, 8080)), + ] + ); + } + + #[test] + fn explicit_ipv4_bind_address_is_the_only_candidate() { + let _guard = env_lock(); + let previous = std::env::var(OPERATOR_BIND_ADDRESS_ENV).ok(); + unsafe { std::env::set_var(OPERATOR_BIND_ADDRESS_ENV, "127.0.0.1") }; + + let addrs = listen_addrs(9090, OPERATOR_BIND_ADDRESS_ENV).expect("explicit addrs"); + restore_env(OPERATOR_BIND_ADDRESS_ENV, previous); + + assert_eq!( + addrs, + vec![SocketAddr::from((IpAddr::V4(Ipv4Addr::LOCALHOST), 9090))] + ); + } + + #[test] + fn explicit_ipv6_bind_address_is_the_only_candidate() { + let _guard = env_lock(); + let previous = std::env::var(CONSOLE_BIND_ADDRESS_ENV).ok(); + unsafe { std::env::set_var(CONSOLE_BIND_ADDRESS_ENV, "::1") }; + + let addrs = listen_addrs(4223, CONSOLE_BIND_ADDRESS_ENV).expect("explicit addrs"); + restore_env(CONSOLE_BIND_ADDRESS_ENV, previous); + + assert_eq!( + addrs, + vec![SocketAddr::from((IpAddr::V6(Ipv6Addr::LOCALHOST), 4223))] + ); + } + + #[test] + fn invalid_bind_address_is_rejected_before_listen() { + let _guard = env_lock(); + let previous = std::env::var(OPERATOR_BIND_ADDRESS_ENV).ok(); + unsafe { std::env::set_var(OPERATOR_BIND_ADDRESS_ENV, "not-an-ip") }; + + let error = listen_addrs(80, OPERATOR_BIND_ADDRESS_ENV).expect_err("invalid IP"); + restore_env(OPERATOR_BIND_ADDRESS_ENV, previous); + + assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput); + assert!(error.to_string().contains("OPERATOR_BIND_ADDRESS")); + } + + #[tokio::test] + async fn bind_unspecified_listener_accepts_ipv4_loopback() { + let _guard = env_lock(); + let previous = std::env::var(OPERATOR_BIND_ADDRESS_ENV).ok(); + unsafe { std::env::set_var(OPERATOR_BIND_ADDRESS_ENV, "127.0.0.1") }; + + let listener = bind_unspecified_listener(0, OPERATOR_BIND_ADDRESS_ENV) + .await + .expect("loopback listener"); + restore_env(OPERATOR_BIND_ADDRESS_ENV, previous); + + let addr = listener.local_addr().expect("bound address"); + assert_eq!(addr.ip(), IpAddr::V4(Ipv4Addr::LOCALHOST)); + assert_ne!(addr.port(), 0); + } +}