Support adopting existing radosgw buckets and users

The operator previously assumed it created every user and bucket it managed:
reconciling an existing resource could overwrite its user attributes or wipe its
bucket policy, and deleting a CRD always deleted the underlying RGW object (only
Bucket had retainOnDelete). That made taking over pre-existing radosgw state
unsafe. Make adoption first-class.

- add retainOnDelete to ObjectStoreUser and BucketAccess (dedicated users), so
  deleting the CRD orphans the RGW user instead of deleting it (symmetric with
  Bucket)
- merge bucket policy instead of replacing it: the operator marks its own
  statements with a cephrgwop* Sid and preserves any statement it does not own,
  so adopting a bucket with a hand-written policy keeps it; add Bucket
  managePolicy (default true) to opt out of policy management entirely
- only reconcile user attributes the spec sets: DisplayName when non-empty and
  Suspended is now an optional *bool, so adopting a user does not reset them
- record adoption: ObjectStoreUser/Bucket status.adopted (+ printcolumn) is true
  when the RGW object already existed on first reconcile
- add GetBucketPolicy + MergeBucketPolicy; keyed adoption detection off the
  status identity field so a Pending owner wait does not mislabel it
- regenerate CRDs/deepcopy; add docs/adoption.md and
  config/samples/05-adoption.yaml; cover the merge in policy_test.go

Claude-Session: https://claude.ai/code/session_016CEncETbf8cvy1PhsHfFHM

internal/ceph/buckets.go

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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 {