unkin/cephrgw-operator
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Support adopting existing radosgw buckets and users' (#5) from benvin/safe-adoption into main
ci/woodpecker/tag/docker Pipeline was successful
Details

Browse Source 
Reviewed-on: #5
This commit was merged in pull request #5.
This commit is contained in:
BenVincent
2026-07-25 10:17:01 +10:00
parent 619aa6751d 54d3e38223
commit 9bbaa2b8ba
18 changed files with 578 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+11
View File
@@ -60,6 +60,17 @@ refine it (see `config/samples/04-access-fine-grained.yaml`):
RGW honours S3 bucket policy on **Reef 18.2+ / Squid**; condition-key support is
a subset of AWS, so validate exotic conditions against your cluster.
### Adopting existing buckets and users
The operator can take over buckets/users that already exist in radosgw and hand
them back without deleting them. In short: matching CRDs manage the resource in
place (no recreation, keys reused, `status.adopted: true`), the bucket policy is
**merged** so an existing hand-written policy is preserved (`spec.managePolicy:
false` opts out entirely), and `spec.retainOnDelete` on `ObjectStoreUser` /
`Bucket` / `BucketAccess` orphans the RGW object instead of deleting it. See
**[docs/adoption.md](docs/adoption.md)** and
`config/samples/05-adoption.yaml`.
The `Bucket` controller renders the policy as the **union of every ready
`BucketAccess`** that targets it, so the result is convergent regardless of the
order objects are created or deleted. It watches `BucketAccess` and
api/v1alpha1/bucket_types.go
+15
View File
@@ -64,6 +64,16 @@ type BucketSpec struct {
// +optional
Tags map[string]string `json:"tags,omitempty"`
// ManagePolicy controls whether the operator manages the bucket's S3 policy
// from BucketAccess grants. When true (the default) the operator reconciles
// its own statements while preserving any statements it does not own, so it
// is safe to adopt a bucket that already has a policy. Set to false to leave
// the bucket policy entirely untouched (BucketAccess grants then have no
// effect on this bucket).
// +kubebuilder:default=true
// +optional
ManagePolicy *bool `json:"managePolicy,omitempty"`
// RetainOnDelete keeps the RGW bucket (and its objects) when the Bucket
// resource is deleted. By default the operator removes the empty bucket;
// it never purges objects unless PurgeOnDelete is also set.
@@ -94,6 +104,10 @@ type BucketStatus struct {
// BucketAccess and reflected in the bucket policy.
// +optional
PolicyPrincipals int32 `json:"policyPrincipals,omitempty"`
// Adopted reports that the RGW bucket already existed when the operator
// first reconciled this resource (it was taken over, not created).
// +optional
Adopted bool `json:"adopted,omitempty"`
// +optional
ObservedGeneration int64 `json:"observedGeneration,omitempty"`
// +optional
@@ -108,6 +122,7 @@ type BucketStatus struct {
// +kubebuilder:printcolumn:name="Bucket",type=string,JSONPath=`.status.bucketName`
// +kubebuilder:printcolumn:name="Owner",type=string,JSONPath=`.status.owner`
// +kubebuilder:printcolumn:name="Grants",type=integer,JSONPath=`.status.policyPrincipals`
// +kubebuilder:printcolumn:name="Adopted",type=boolean,JSONPath=`.status.adopted`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// Bucket is a Ceph RGW S3 bucket.
api/v1alpha1/bucketaccess_types.go
+6
View File
@@ -46,6 +46,12 @@ type BucketAccessSpec struct {
// +optional
SecretName string `json:"secretName,omitempty"`
// RetainOnDelete keeps the dedicated RGW user (created when UserRef is empty)
// instead of deleting it when this BucketAccess is removed. Ignored when
// UserRef is set (that user is never managed here). Defaults to false.
// +optional
RetainOnDelete bool `json:"retainOnDelete,omitempty"`
// Paths optionally scopes object-level access to these key prefixes within
// the bucket; each becomes the resource "<bucket>/<prefix>*". Empty grants
// the whole bucket. The bucket-level ListBucket action always applies to the
api/v1alpha1/objectstoreuser_types.go
+15 -2
View File
@@ -27,9 +27,11 @@ type ObjectStoreUserSpec struct {
// +optional
MaxBuckets *int32 `json:"maxBuckets,omitempty"`
// Suspended, when true, suspends the user so its keys stop working.
// Suspended manages the user's suspended state: true suspends the user so
// its keys stop working, false resumes it. When unset the operator does not
// touch the suspended state (useful when adopting an existing user).
// +optional
Suspended bool `json:"suspended,omitempty"`
Suspended *bool `json:"suspended,omitempty"`
// Quota optionally applies a user-level quota.
// +optional
@@ -40,6 +42,12 @@ type ObjectStoreUserSpec struct {
// AWS_SECRET_ACCESS_KEY, BUCKET_HOST and the RGW uid.
// +optional
SecretName string `json:"secretName,omitempty"`
// RetainOnDelete keeps the RGW user (and its keys) when the ObjectStoreUser
// resource is deleted, instead of removing it. Set this before adopting an
// existing user you may later want to hand back. Defaults to false.
// +optional
RetainOnDelete bool `json:"retainOnDelete,omitempty"`
}
// ObjectStoreUserStatus reports observed user state.
@@ -53,6 +61,10 @@ type ObjectStoreUserStatus struct {
// SecretName is the Secret holding the user's credentials.
// +optional
SecretName string `json:"secretName,omitempty"`
// Adopted reports that the RGW user already existed when the operator first
// reconciled this resource (it was taken over, not created).
// +optional
Adopted bool `json:"adopted,omitempty"`
// +optional
ObservedGeneration int64 `json:"observedGeneration,omitempty"`
// +optional
@@ -66,6 +78,7 @@ type ObjectStoreUserStatus struct {
// +kubebuilder:resource:shortName=osu
// +kubebuilder:printcolumn:name="UID",type=string,JSONPath=`.status.uid`
// +kubebuilder:printcolumn:name="Secret",type=string,JSONPath=`.status.secretName`
// +kubebuilder:printcolumn:name="Adopted",type=boolean,JSONPath=`.status.adopted`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// ObjectStoreUser is a Ceph RGW S3 user whose keys are delivered into a Secret.
api/v1alpha1/zz_generated.deepcopy.go
+10
View File
@@ -226,6 +226,11 @@ func (in *BucketSpec) DeepCopyInto(out *BucketSpec) {
(*out)[key] = val
}
}
if in.ManagePolicy != nil {
in, out := &in.ManagePolicy, &out.ManagePolicy
*out = new(bool)
**out = **in
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BucketSpec.
@@ -352,6 +357,11 @@ func (in *ObjectStoreUserSpec) DeepCopyInto(out *ObjectStoreUserSpec) {
*out = new(int32)
**out = **in
}
if in.Suspended != nil {
in, out := &in.Suspended, &out.Suspended
*out = new(bool)
**out = **in
}
if in.Quota != nil {
in, out := &in.Quota, &out.Quota
*out = new(Quota)
config/crd/bases/ceph.unkin.net_bucketaccesses.yaml
+6
View File
@@ -156,6 +156,12 @@ spec:
- actions
type: object
type: array
retainOnDelete:
description: |-
RetainOnDelete keeps the dedicated RGW user (created when UserRef is empty)
instead of deleting it when this BucketAccess is removed. Ignored when
UserRef is set (that user is never managed here). Defaults to false.
type: boolean
secretName:
description: |-
SecretName is the Secret the operator writes credentials into for the
config/crd/bases/ceph.unkin.net_buckets.yaml
+18
View File
@@ -26,6 +26,9 @@ spec:
- jsonPath: .status.policyPrincipals
name: Grants
type: integer
- jsonPath: .status.adopted
name: Adopted
type: boolean
- jsonPath: .status.phase
name: Phase
type: string
@@ -58,6 +61,16 @@ spec:
description: BucketName is the S3 bucket name. Defaults to metadata.name.
Immutable.
type: string
managePolicy:
default: true
description: |-
ManagePolicy controls whether the operator manages the bucket's S3 policy
from BucketAccess grants. When true (the default) the operator reconciles
its own statements while preserving any statements it does not own, so it
is safe to adopt a bucket that already has a policy. Set to false to leave
the bucket policy entirely untouched (BucketAccess grants then have no
effect on this bucket).
type: boolean
objectLock:
description: ObjectLock configures S3 object lock. Enabling it forces
versioning on.
@@ -144,6 +157,11 @@ spec:
status:
description: BucketStatus reports observed bucket state.
properties:
adopted:
description: |-
Adopted reports that the RGW bucket already existed when the operator
first reconciled this resource (it was taken over, not created).
type: boolean
bucketID:
description: BucketID is the RGW internal bucket instance id.
type: string
config/crd/bases/ceph.unkin.net_objectstoreusers.yaml
+18 -2
View File
@@ -23,6 +23,9 @@ spec:
- jsonPath: .status.secretName
name: Secret
type: string
- jsonPath: .status.adopted
name: Adopted
type: boolean
- jsonPath: .status.phase
name: Phase
type: string
@@ -90,6 +93,12 @@ spec:
format: int64
type: integer
type: object
retainOnDelete:
description: |-
RetainOnDelete keeps the RGW user (and its keys) when the ObjectStoreUser
resource is deleted, instead of removing it. Set this before adopting an
existing user you may later want to hand back. Defaults to false.
type: boolean
secretName:
description: |-
SecretName is the Secret the operator writes the access/secret key into.
@@ -97,8 +106,10 @@ spec:
AWS_SECRET_ACCESS_KEY, BUCKET_HOST and the RGW uid.
type: string
suspended:
description: Suspended, when true, suspends the user so its keys stop
working.
description: |-
Suspended manages the user's suspended state: true suspends the user so
its keys stop working, false resumes it. When unset the operator does not
touch the suspended state (useful when adopting an existing user).
type: boolean
uid:
description: UID is the RGW user id. Defaults to metadata.name. Immutable
@@ -108,6 +119,11 @@ spec:
status:
description: ObjectStoreUserStatus reports observed user state.
properties:
adopted:
description: |-
Adopted reports that the RGW user already existed when the operator first
reconciled this resource (it was taken over, not created).
type: boolean
conditions:
items:
description: Condition contains details for one aspect of the current
config/crd/install.yaml
+42 -2
View File
@@ -157,6 +157,12 @@ spec:
- actions
type: object
type: array
retainOnDelete:
description: |-
RetainOnDelete keeps the dedicated RGW user (created when UserRef is empty)
instead of deleting it when this BucketAccess is removed. Ignored when
UserRef is set (that user is never managed here). Defaults to false.
type: boolean
secretName:
description: |-
SecretName is the Secret the operator writes credentials into for the
@@ -290,6 +296,9 @@ spec:
- jsonPath: .status.policyPrincipals
name: Grants
type: integer
- jsonPath: .status.adopted
name: Adopted
type: boolean
- jsonPath: .status.phase
name: Phase
type: string
@@ -322,6 +331,16 @@ spec:
description: BucketName is the S3 bucket name. Defaults to metadata.name.
Immutable.
type: string
managePolicy:
default: true
description: |-
ManagePolicy controls whether the operator manages the bucket's S3 policy
from BucketAccess grants. When true (the default) the operator reconciles
its own statements while preserving any statements it does not own, so it
is safe to adopt a bucket that already has a policy. Set to false to leave
the bucket policy entirely untouched (BucketAccess grants then have no
effect on this bucket).
type: boolean
objectLock:
description: ObjectLock configures S3 object lock. Enabling it forces
versioning on.
@@ -408,6 +427,11 @@ spec:
status:
description: BucketStatus reports observed bucket state.
properties:
adopted:
description: |-
Adopted reports that the RGW bucket already existed when the operator
first reconciled this resource (it was taken over, not created).
type: boolean
bucketID:
description: BucketID is the RGW internal bucket instance id.
type: string
@@ -519,6 +543,9 @@ spec:
- jsonPath: .status.secretName
name: Secret
type: string
- jsonPath: .status.adopted
name: Adopted
type: boolean
- jsonPath: .status.phase
name: Phase
type: string
@@ -586,6 +613,12 @@ spec:
format: int64
type: integer
type: object
retainOnDelete:
description: |-
RetainOnDelete keeps the RGW user (and its keys) when the ObjectStoreUser
resource is deleted, instead of removing it. Set this before adopting an
existing user you may later want to hand back. Defaults to false.
type: boolean
secretName:
description: |-
SecretName is the Secret the operator writes the access/secret key into.
@@ -593,8 +626,10 @@ spec:
AWS_SECRET_ACCESS_KEY, BUCKET_HOST and the RGW uid.
type: string
suspended:
description: Suspended, when true, suspends the user so its keys stop
working.
description: |-
Suspended manages the user's suspended state: true suspends the user so
its keys stop working, false resumes it. When unset the operator does not
touch the suspended state (useful when adopting an existing user).
type: boolean
uid:
description: UID is the RGW user id. Defaults to metadata.name. Immutable
@@ -604,6 +639,11 @@ spec:
status:
description: ObjectStoreUserStatus reports observed user state.
properties:
adopted:
description: |-
Adopted reports that the RGW user already existed when the operator first
reconciled this resource (it was taken over, not created).
type: boolean
conditions:
items:
description: Condition contains details for one aspect of the current
config/samples/05-adoption.yaml
+31
View File
@@ -0,0 +1,31 @@
# Adopting an existing radosgw user + bucket. The operator takes them over in
# place: no recreation, existing keys reused, existing bucket policy preserved.
# retainOnDelete keeps the RGW objects if these CRDs are later deleted.
# See docs/adoption.md.
apiVersion: ceph.unkin.net/v1alpha1
kind: ObjectStoreUser
metadata:
name: legacy-owner
namespace: default
spec:
# uid must match the existing RGW user id.
uid: legacy-owner
# Set maxBuckets to the existing user's limit (it otherwise defaults to 1000
# and would be applied). Leave displayName/suspended unset to keep them as-is.
maxBuckets: 1000
retainOnDelete: true
---
apiVersion: ceph.unkin.net/v1alpha1
kind: Bucket
metadata:
name: legacy-data
namespace: default
spec:
# bucketName must match the existing bucket.
bucketName: legacy-data
ownerRef: legacy-owner
retainOnDelete: true
# managePolicy defaults to true: the operator merges its BucketAccess grants
# into the existing policy, preserving statements it does not own. Set it to
# false to leave the bucket policy entirely under manual control.
managePolicy: true
docs/adoption.md
+68
View File
@@ -0,0 +1,68 @@
# Adopting existing radosgw buckets and users
The operator can **take over** buckets and users that already exist in radosgw:
write CRDs that match them and it manages them in place instead of recreating
them. It can also **hand them back** without deleting the underlying RGW objects.
## What happens when you create a CRD for an existing resource
| CRD | On adoption |
|-----|-------------|
| `ObjectStoreUser` | The user is **not** recreated and its **existing keys are reused** (never rotated). `status.adopted` becomes `true`. Attributes are only changed if the spec sets them (see below). |
| `Bucket` | The bucket is **not** recreated. `status.adopted` becomes `true`. Versioning/tags/quota are only touched if the spec sets them. The policy is **merged**, not overwritten (see below). |
| `BucketAccess` | Adds the grant's statement to the bucket policy and (for a dedicated user) reuses that user's keys. |
`status.adopted` is also shown in `kubectl get osu` / `kubectl get bkt` under
the **ADOPTED** column.
### User attributes
To avoid clobbering an adopted user, the operator only sends attributes you set:
- `displayName` — left unchanged when empty.
- `suspended` — left unchanged when unset (it is an optional `*bool`; set it
explicitly to `true`/`false` to manage it).
- `email` — left unchanged when empty.
- `maxBuckets` — **defaults to `1000`** and is always applied. If the existing
user has a different limit you want to keep, set `maxBuckets` to match (or to
the value you want).
### Bucket policy is merged, not replaced
The operator owns only the policy statements it writes — they carry a
`cephrgwop…` statement id. On every reconcile it **preserves statements it does
not own** and reconciles only its own. So adopting a bucket that already has a
hand-written policy keeps that policy; your `BucketAccess` grants are added
alongside it.
- Reserve the `cephrgwop` prefix (and the legacy prefixes `full`, `custom`,
`raw`, `readonly*`, `readwrite*`) for the operator — do not name your own
statements with them, or they will be treated as operator-owned and replaced.
- To have the operator **never touch** a bucket's policy, set
`spec.managePolicy: false`. `BucketAccess` grants then have no effect on that
bucket.
## What happens when you delete the CRD
By default, deleting a CRD deletes the underlying RGW resource. Opt out per
resource to **orphan** it instead (the finalizer is dropped, the RGW object is
left in place):
| CRD | Default on delete | Keep the RGW object |
|-----|-------------------|---------------------|
| `ObjectStoreUser` | deletes the RGW user + keys | `spec.retainOnDelete: true` |
| `Bucket` | deletes the (empty) bucket | `spec.retainOnDelete: true` (and never set `purgeOnDelete`) |
| `BucketAccess` (dedicated user) | deletes the dedicated user | `spec.retainOnDelete: true` |
| `BucketAccess` (`userRef`) | only drops the policy statement | n/a (never manages that user) |
## Recommended adoption procedure
1. Create an `ObjectStoreUser` for each owner, setting `retainOnDelete: true`
and `maxBuckets` to the value you want. Confirm `ADOPTED=true`.
2. Create the `Bucket` (with `retainOnDelete: true`) referencing that owner.
With `managePolicy: true` (default) the existing policy is preserved.
3. Optionally add `BucketAccess` objects to model existing grants; they merge
into the policy. If you would rather keep managing the policy by hand, set
`managePolicy: false` on the Bucket.
4. To hand a resource back, delete its CRD — with `retainOnDelete: true` the RGW
bucket/user is left untouched.
internal/ceph/buckets.go
+20
View File
@@ -125,6 +125,26 @@ func (c *Client) SetBucketPolicy(ctx context.Context, name, bucketID, ownerUID,
return err
}
// GetBucketPolicy returns the bucket's current S3 policy JSON, or "" when it has
// none. It is signed as the bucket owner.
func (c *Client) GetBucketPolicy(ctx context.Context, name, ownerUID string) (string, error) {
owner, err := c.asOwner(ctx, ownerUID)
if err != nil {
return "", err
}
out, err := c.s3.GetBucketPolicy(ctx, &s3.GetBucketPolicyInput{Bucket: aws.String(name)}, owner)
if err != nil {
if IsNotFound(err) {
return "", nil
}
return "", err
}
if out.Policy == nil {
return "", nil
}
return *out.Policy, nil
}
// SetBucketTags replaces the bucket tag set. tagsJSON is the JSON produced by
// BuildTagJSON (a list of {"Key","Value"} objects). bucketID is unused (kept for
// call-site stability).
internal/ceph/policy.go
+113 -12
View File
@@ -2,6 +2,7 @@ package ceph
import (
"encoding/json"
"fmt"
"sort"
"strconv"
"strings"
@@ -15,6 +16,19 @@ const (
LevelFull = "full"
)
// managedSidPrefix marks statement ids the operator owns, so it can reconcile
// its own statements while preserving foreign ones when adopting a bucket that
// already has a policy. Do not reuse this prefix (or the legacy prefixes below)
// for statements you manage yourself.
const managedSidPrefix = "cephrgwop"
// legacyManagedSidPrefixes are the statement-id prefixes the operator emitted
// before managedSidPrefix existed; they are still recognised as operator-owned
// so upgrading does not duplicate statements.
var legacyManagedSidPrefixes = []string{
"readonlybkt", "readonlyobj", "readwritebkt", "readwriteobj", "full", "custom", "raw",
}
// GrantConditions restricts when a grant's statements apply. The zero value adds
// no conditions.
type GrantConditions struct {
@@ -103,10 +117,98 @@ var objectActions = map[string][]string{
// principal its requested access. It returns "" when there are no grants so the
// caller can clear the policy.
func BuildBucketPolicy(bucket string, grants []Grant) (string, error) {
if len(grants) == 0 {
statements := buildStatements(bucket, grants)
if len(statements) == 0 {
return "", nil
}
b, err := json.Marshal(policyDocument{Version: "2012-10-17", Statement: statements})
if err != nil {
return "", err
}
return string(b), nil
}
// MergeBucketPolicy renders the operator's statements for grants and merges them
// into an existing policy, preserving any statement the operator does not own
// (identified by its Sid). It returns "" only when the merged policy would be
// empty, so a bucket adopted with a hand-written policy keeps that policy.
func MergeBucketPolicy(existing, bucket string, grants []Grant) (string, error) {
version := "2012-10-17"
var id string
var foreign []json.RawMessage
if strings.TrimSpace(existing) != "" {
var doc struct {
Version string `json:"Version"`
ID string `json:"Id,omitempty"`
Statement []json.RawMessage `json:"Statement"`
}
if err := json.Unmarshal([]byte(existing), &doc); err != nil {
return "", fmt.Errorf("parse existing bucket policy: %w", err)
}
if doc.Version != "" {
version = doc.Version
}
id = doc.ID
for _, raw := range doc.Statement {
var meta struct {
Sid string `json:"Sid"`
}
// Ignore unmarshal errors: a statement we cannot read the Sid of is
// treated as foreign and preserved verbatim.
_ = json.Unmarshal(raw, &meta)
if isManagedSid(meta.Sid) {
continue // operator-owned; re-rendered below
}
foreign = append(foreign, raw)
}
}
managed := buildStatements(bucket, grants)
if len(foreign) == 0 && len(managed) == 0 {
return "", nil
}
statements := make([]json.RawMessage, 0, len(foreign)+len(managed))
statements = append(statements, foreign...)
for _, st := range managed {
b, err := json.Marshal(st)
if err != nil {
return "", err
}
statements = append(statements, b)
}
b, err := json.Marshal(struct {
Version string `json:"Version"`
ID string `json:"Id,omitempty"`
Statement []json.RawMessage `json:"Statement"`
}{Version: version, ID: id, Statement: statements})
if err != nil {
return "", err
}
return string(b), nil
}
// isManagedSid reports whether a statement id was emitted by the operator.
func isManagedSid(s string) bool {
if strings.HasPrefix(s, managedSidPrefix) {
return true
}
for _, p := range legacyManagedSidPrefixes {
if strings.HasPrefix(s, p) {
return true
}
}
return false
}
// buildStatements renders the operator's statements for grants, sorted for
// deterministic output.
func buildStatements(bucket string, grants []Grant) []policyStatement {
if len(grants) == 0 {
return nil
}
sorted := make([]Grant, len(grants))
copy(sorted, grants)
sort.Slice(sorted, func(i, j int) bool {
@@ -117,17 +219,11 @@ func BuildBucketPolicy(bucket string, grants []Grant) (string, error) {
})
bucketARN := "arn:aws:s3:::" + bucket
doc := policyDocument{Version: "2012-10-17"}
var statements []policyStatement
for _, g := range sorted {
doc.Statement = append(doc.Statement, statementsForGrant(bucketARN, g)...)
statements = append(statements, statementsForGrant(bucketARN, g)...)
}
b, err := json.Marshal(doc)
if err != nil {
return "", err
}
return string(b), nil
return statements
}
// statementsForGrant renders the policy statements for a single grant.
@@ -138,7 +234,9 @@ func statementsForGrant(bucketARN string, g Grant) []policyStatement {
out := make([]policyStatement, 0, len(g.Raw))
for i, rs := range g.Raw {
st := policyStatement{
Sid: firstNonEmpty(rs.Sid, sid("raw", g.UID)+strconv.Itoa(i)),
// Always operator-owned so a merge re-renders (not duplicates)
// it; any user-supplied Sid is folded into the managed id.
Sid: sid(firstNonEmpty(rs.Sid, "raw"+strconv.Itoa(i)), g.UID),
Effect: firstNonEmpty(rs.Effect, "Allow"),
Principal: principal,
Action: rs.Actions,
@@ -257,9 +355,12 @@ func buildCondition(c *GrantConditions) map[string]map[string][]string {
return cond
}
// sid builds a policy statement id that only contains characters S3 accepts.
// sid builds a policy statement id that only contains characters S3 accepts,
// prefixed with managedSidPrefix so the operator can recognise its own
// statements when merging into an adopted bucket's policy.
func sid(prefix, uid string) string {
var b strings.Builder
b.WriteString(managedSidPrefix)
b.WriteString(strings.ReplaceAll(prefix, "-", ""))
for _, r := range uid {
switch {
internal/ceph/policy_test.go
+118
View File
@@ -223,6 +223,124 @@ func TestBuildBucketPolicyRawStatements(t *testing.T) {
}
}
func TestMergeBucketPolicyPreservesForeign(t *testing.T) {
// An existing policy with a foreign statement (unknown Sid, and a scalar
// condition value S3 allows but our typed struct does not model).
existing := `{"Version":"2012-10-17","Statement":[` +
`{"Sid":"AllowPublicRead","Effect":"Allow","Principal":"*","Action":["s3:GetObject"],` +
`"Resource":"arn:aws:s3:::data/public/*","Condition":{"Bool":{"aws:SecureTransport":"true"}}}]}`
merged, err := MergeBucketPolicy(existing, "data", []Grant{{UID: "reader", Level: LevelReadOnly}})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
doc := parsePolicyRaw(t, merged)
var sawForeign, sawManaged bool
for _, raw := range doc.Statement {
var s struct {
Sid string `json:"Sid"`
Condition map[string]map[string]any
}
if err := json.Unmarshal(raw, &s); err != nil {
t.Fatalf("statement not valid JSON: %v", err)
}
if s.Sid == "AllowPublicRead" {
sawForeign = true
// The scalar condition value must survive verbatim.
if v := s.Condition["Bool"]["aws:SecureTransport"]; v != "true" {
t.Fatalf("foreign scalar condition mangled: %v", s.Condition)
}
}
if isManagedSid(s.Sid) {
sawManaged = true
}
}
if !sawForeign {
t.Fatal("foreign statement was dropped")
}
if !sawManaged {
t.Fatal("operator statement missing from merge")
}
}
func TestMergeBucketPolicyReplacesManaged(t *testing.T) {
// Two rounds: an existing policy already carrying the operator's statements
// must not accumulate duplicates when re-merged.
first, err := MergeBucketPolicy("", "data", []Grant{{UID: "reader", Level: LevelReadOnly}})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
second, err := MergeBucketPolicy(first, "data", []Grant{{UID: "reader", Level: LevelReadOnly}})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if first != second {
t.Fatalf("re-merging its own policy was not idempotent:\n first=%s\nsecond=%s", first, second)
}
// A legacy (unprefixed) operator statement must also be recognised and
// replaced rather than preserved as foreign.
legacy := `{"Version":"2012-10-17","Statement":[` +
`{"Sid":"readonlybktreader","Effect":"Allow","Principal":{"AWS":["arn:aws:iam:::user/reader"]},` +
`"Action":["s3:ListBucket"],"Resource":["arn:aws:s3:::data"]}]}`
merged, err := MergeBucketPolicy(legacy, "data", []Grant{{UID: "reader", Level: LevelReadOnly}})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
for _, raw := range parsePolicyRaw(t, merged).Statement {
var s struct {
Sid string `json:"Sid"`
}
_ = json.Unmarshal(raw, &s)
if s.Sid == "readonlybktreader" {
t.Fatal("legacy operator statement was preserved instead of replaced")
}
}
}
func TestMergeBucketPolicyForeignOnlyKept(t *testing.T) {
existing := `{"Version":"2012-10-17","Statement":[` +
`{"Sid":"AllowPublicRead","Effect":"Allow","Principal":"*","Action":["s3:GetObject"],` +
`"Resource":"arn:aws:s3:::data/*"}]}`
// No grants: the operator adds nothing but must not wipe the foreign policy.
merged, err := MergeBucketPolicy(existing, "data", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if merged == "" {
t.Fatal("merge cleared a policy that had a foreign statement")
}
if len(parsePolicyRaw(t, merged).Statement) != 1 {
t.Fatalf("expected the single foreign statement, got %s", merged)
}
}
func TestMergeBucketPolicyEmpty(t *testing.T) {
merged, err := MergeBucketPolicy("", "data", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if merged != "" {
t.Fatalf("expected empty policy, got %q", merged)
}
}
// parsePolicyRaw parses a policy keeping statements as raw JSON.
func parsePolicyRaw(t *testing.T, raw string) struct {
Version string `json:"Version"`
Statement []json.RawMessage `json:"Statement"`
} {
t.Helper()
var doc struct {
Version string `json:"Version"`
Statement []json.RawMessage `json:"Statement"`
}
if err := json.Unmarshal([]byte(raw), &doc); err != nil {
t.Fatalf("policy is not valid JSON: %v\n%s", err, raw)
}
return doc
}
func TestBuildBucketPolicyFineGrainedDeterministic(t *testing.T) {
grants := []Grant{
{UID: "reader", Level: LevelReadOnly, Paths: []string{"a/", "b/"}},
internal/ceph/users.go
+24 -13
View File
@@ -31,13 +31,15 @@ func (u *User) S3Key() (UserKey, bool) {
return u.Keys[0], true
}
// UserSpec describes the desired state of an RGW user.
// UserSpec describes the desired state of an RGW user. A nil DisplayName/
// Suspended (empty string / nil pointer) leaves that attribute untouched on an
// existing user, so an adopted user is not mutated unless the fields are set.
type UserSpec struct {
UID string
DisplayName string
Email string
MaxBuckets *int32
Suspended bool
Suspended *bool
}
// fromAdminUser converts a go-ceph admin.User into the subset the operator uses.
@@ -73,7 +75,7 @@ func (c *Client) CreateUser(ctx context.Context, spec UserSpec) (*User, error) {
DisplayName: firstNonEmpty(spec.DisplayName, spec.UID),
Email: spec.Email,
MaxBuckets: int32PtrToIntPtr(spec.MaxBuckets),
Suspended: boolToIntPtr(spec.Suspended),
Suspended: boolPtrToIntPtr(spec.Suspended),
GenerateKey: boolPtr(true),
})
if err != nil {
@@ -83,15 +85,20 @@ func (c *Client) CreateUser(ctx context.Context, spec UserSpec) (*User, error) {
return fromAdminUser(u), nil
}
// UpdateUser reconciles the mutable attributes of an existing RGW user.
// UpdateUser reconciles the mutable attributes of an existing RGW user. It only
// sends attributes the spec sets: an empty DisplayName or nil Suspended is left
// as-is, so reconciling (or adopting) a user does not clobber those fields.
func (c *Client) UpdateUser(ctx context.Context, spec UserSpec) (*User, error) {
u, err := c.admin.ModifyUser(ctx, admin.User{
ID: spec.UID,
DisplayName: firstNonEmpty(spec.DisplayName, spec.UID),
Email: spec.Email,
MaxBuckets: int32PtrToIntPtr(spec.MaxBuckets),
Suspended: boolToIntPtr(spec.Suspended),
})
req := admin.User{
ID: spec.UID,
Email: spec.Email,
MaxBuckets: int32PtrToIntPtr(spec.MaxBuckets),
Suspended: boolPtrToIntPtr(spec.Suspended),
}
if spec.DisplayName != "" {
req.DisplayName = spec.DisplayName
}
u, err := c.admin.ModifyUser(ctx, req)
if err != nil {
return nil, err
}
@@ -156,9 +163,13 @@ func int32PtrToIntPtr(p *int32) *int {
return &v
}
func boolToIntPtr(b bool) *int {
// boolPtrToIntPtr renders an optional bool as RGW's 0/1 int, or nil to omit.
func boolPtrToIntPtr(b *bool) *int {
if b == nil {
return nil
}
v := 0
if b {
if *b {
v = 1
}
return &v
internal/controller/bucket_controller.go
+39 -12
View File
@@ -77,7 +77,11 @@ func (r *BucketReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
}
ownerUID := owner.Status.UID
// Ensure the bucket exists.
// Ensure the bucket exists. Record adoption once: whether the RGW bucket
// already existed the first time we reconciled this resource. status.BucketID
// is only set on a successful reconcile, so a Pending wait on the owner (or a
// transient failure) does not pollute the signal.
firstObserve := b.Status.BucketID == ""
info, err := r.Ceph.GetBucket(ctx, bucketName)
if ceph.IsNotFound(err) {
createSpec := ceph.CreateBucketSpec{
@@ -97,8 +101,14 @@ func (r *BucketReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
return r.fail(ctx, &b, "CreateFailed", err)
}
logger.Info("created bucket", "bucket", bucketName, "owner", ownerUID)
if firstObserve {
b.Status.Adopted = false
}
} else if err != nil {
return r.fail(ctx, &b, "LookupFailed", err)
} else if firstObserve {
b.Status.Adopted = true
logger.Info("adopted existing bucket", "bucket", bucketName, "owner", ownerUID)
}
bucketID := info.InstanceID()
@@ -129,17 +139,28 @@ func (r *BucketReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
}
}
// Render and apply the aggregate S3 policy from all BucketAccess grants.
grants, principals, err := r.collectGrants(ctx, b.Namespace, b.Name)
if err != nil {
return r.fail(ctx, &b, "GrantsFailed", err)
}
policy, err := ceph.BuildBucketPolicy(bucketName, grants)
if err != nil {
return r.fail(ctx, &b, "PolicyBuildFailed", err)
}
if err := r.Ceph.SetBucketPolicy(ctx, bucketName, bucketID, ownerUID, policy); err != nil {
return r.fail(ctx, &b, "PolicyFailed", err)
// Render and apply the aggregate S3 policy from all BucketAccess grants,
// unless the bucket opts out of policy management. The merge preserves any
// statements the operator does not own, so an adopted bucket keeps its
// existing policy.
principals := 0
if managePolicy(&b) {
grants, p, err := r.collectGrants(ctx, b.Namespace, b.Name)
if err != nil {
return r.fail(ctx, &b, "GrantsFailed", err)
}
existing, err := r.Ceph.GetBucketPolicy(ctx, bucketName, ownerUID)
if err != nil {
return r.fail(ctx, &b, "PolicyReadFailed", err)
}
policy, err := ceph.MergeBucketPolicy(existing, bucketName, grants)
if err != nil {
return r.fail(ctx, &b, "PolicyBuildFailed", err)
}
if err := r.Ceph.SetBucketPolicy(ctx, bucketName, bucketID, ownerUID, policy); err != nil {
return r.fail(ctx, &b, "PolicyFailed", err)
}
principals = p
}
b.Status.Phase = "Ready"
@@ -223,6 +244,12 @@ func grantKey(g ceph.Grant) string {
return string(b)
}
// managePolicy reports whether the operator should reconcile this bucket's S3
// policy. A nil ManagePolicy (the CRD default) is treated as true.
func managePolicy(b *v1alpha1.Bucket) bool {
return b.Spec.ManagePolicy == nil || *b.Spec.ManagePolicy
}
func (r *BucketReconciler) pending(ctx context.Context, b *v1alpha1.Bucket, reason, msg string) (ctrl.Result, error) {
b.Status.Phase = "Pending"
b.Status.ObservedGeneration = b.Generation
internal/controller/bucketaccess_controller.go
+3 -2
View File
@@ -49,8 +49,9 @@ func (r *BucketAccessReconciler) Reconcile(ctx context.Context, req ctrl.Request
if !ba.DeletionTimestamp.IsZero() {
if controllerutil.ContainsFinalizer(&ba, finalizer) {
// Only delete a user the operator created for this grant.
if managed && uid != "" {
// Only delete a user the operator created for this grant, and only
// when the grant does not ask to retain it.
if managed && uid != "" && !ba.Spec.RetainOnDelete {
if err := r.Ceph.DeleteUser(ctx, uid); err != nil {
return r.fail(ctx, &ba, "DeleteFailed", err)
}
internal/controller/objectstoreuser_controller.go
+21 -5
View File
@@ -40,7 +40,9 @@ func (r *ObjectStoreUserReconciler) Reconcile(ctx context.Context, req ctrl.Requ
if !osu.DeletionTimestamp.IsZero() {
if controllerutil.ContainsFinalizer(&osu, finalizer) {
if err := r.Ceph.DeleteUser(ctx, uid); err != nil {
if osu.Spec.RetainOnDelete {
logger.Info("retaining RGW user on delete", "uid", uid)
} else if err := r.Ceph.DeleteUser(ctx, uid); err != nil {
return r.fail(ctx, &osu, "DeleteFailed", err)
}
controllerutil.RemoveFinalizer(&osu, finalizer)
@@ -65,17 +67,31 @@ func (r *ObjectStoreUserReconciler) Reconcile(ctx context.Context, req ctrl.Requ
Suspended: osu.Spec.Suspended,
}
if _, err := r.Ceph.GetUser(ctx, uid); ceph.IsNotFound(err) {
// Record adoption once: whether the RGW user already existed the first time
// we reconciled this resource (taken over rather than created). status.UID is
// only set on a successful reconcile, so it is a clean "never provisioned"
// signal that transient failures do not pollute.
firstObserve := osu.Status.UID == ""
_, getErr := r.Ceph.GetUser(ctx, uid)
switch {
case ceph.IsNotFound(getErr):
if _, err := r.Ceph.CreateUser(ctx, spec); err != nil {
return r.fail(ctx, &osu, "CreateFailed", err)
}
logger.Info("created RGW user", "uid", uid)
} else if err != nil {
return r.fail(ctx, &osu, "LookupFailed", err)
} else {
if firstObserve {
osu.Status.Adopted = false
}
case getErr != nil:
return r.fail(ctx, &osu, "LookupFailed", getErr)
default:
if _, err := r.Ceph.UpdateUser(ctx, spec); err != nil {
return r.fail(ctx, &osu, "UpdateFailed", err)
}
if firstObserve {
osu.Status.Adopted = true
logger.Info("adopted existing RGW user", "uid", uid)
}
}
if q := osu.Spec.Quota; q != nil {