rgw: expand the reconcile loop for the Swift user management

For the specification see:
<https://github.com/rook/rook/blob/master/design/ceph/object/swift-and-keystone-integration.md>

Signed-off-by: Sebastian Riese <sebastian.riese@cloudandheat.com>
Signed-off-by: Silvio Ankermann <silvio.ankermann@cloudandheat.com>

PendingReleaseNotes.md

@@ -9,3 +9,4 @@ read affinity setting in cephCluster CR (CSIDriverOptions section) in [PR](https
 ## Features
 
 - Kubernetes versions **v1.24** through **v1.29** are supported.
+- Support creating swift sub-users through Kubernetes CRs (see [#9088](https://github.com/rook/rook/issues/9088)).

pkg/operator/ceph/object/controller.go

@@ -559,7 +559,7 @@ func (r *ReconcileCephObjectStore) reconcileCOSIUser(cephObjectStore *cephv1.Cep
 	}
 
 	// Create COSI user secret
-	return ReconcileCephUserSecret(r.opManagerContext, r.client, r.scheme, cephObjectStore, &user, objCtx.Endpoint, cephObjectStore.Namespace, cephObjectStore.Name, cephObjectStore.Spec.Gateway.SSLCertificateRef)
+	return ReconcileCephUserSecrets(r.opManagerContext, r.client, r.scheme, cephObjectStore, &user, objCtx.Endpoint, cephObjectStore.Namespace, cephObjectStore.Name, cephObjectStore.Spec.Gateway.SSLCertificateRef)
 }
 
 func generateCOSIUserConfig() *admin.User {

pkg/operator/ceph/object/user.go

@@ -269,7 +269,35 @@ func generateCephUserSecret(userConfig *admin.User, endpoint, namespace, storeNa
 	return secret
 }
 
-func ReconcileCephUserSecret(ctx context.Context, k8sclient client.Client, scheme *runtime.Scheme, ownerRef metav1.Object, userConfig *admin.User, endpoint, namespace, storeName, tlsSecretName string) (reconcile.Result, error) {
+func generateCephSubuserSecretName(store, username, subusername string) string {
+	return fmt.Sprintf("rook-ceph-object-subuser-%s-%s-%s", store, username, subusername)
+}
+
+func generateCephSubuserSecret(userConfig *admin.User, endpoint, namespace, storeName string, subuser *admin.SwiftKeySpec) *corev1.Secret {
+	secrets := map[string]string{
+		"SWIFT_USER":          subuser.User,
+		"SWIFT_SECRET_KEY":    subuser.SecretKey,
+		"SWIFT_AUTH_ENDPOINT": "TODO", // TODO
+	}
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      generateCephSubuserSecretName(storeName, userConfig.ID, subuser.User),
+			Namespace: namespace,
+			Labels: map[string]string{
+				"app":  AppName,
+				"user": userConfig.ID,
+				// XXX: should we add a subuser label?
+				"rook_cluster":      namespace,
+				"rook_object_store": storeName,
+			},
+		},
+		StringData: secrets,
+		Type:       k8sutil.RookType,
+	}
+	return secret
+}
+
+func ReconcileCephUserSecrets(ctx context.Context, k8sclient client.Client, scheme *runtime.Scheme, ownerRef metav1.Object, userConfig *admin.User, endpoint, namespace, storeName, tlsSecretName string) (reconcile.Result, error) {
 	// Generate Kubernetes Secret
 	secret := generateCephUserSecret(userConfig, endpoint, namespace, storeName, tlsSecretName)
 
@@ -284,5 +312,22 @@ func ReconcileCephUserSecret(ctx context.Context, k8sclient client.Client, schem
 	if err != nil {
 		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph object user %q secret", secret.Name)
 	}
+
+	for _, key := range userConfig.SwiftKeys {
+		key := key // To avoid memory aliasing. Won't be necessary in Go 1.22 anymore
+		secret = generateCephSubuserSecret(userConfig, endpoint, namespace, storeName, &key)
+
+		err = controllerutil.SetControllerReference(ownerRef, secret, scheme)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to set owner reference of ceph object subuser secret %q", secret.Name)
+		}
+
+		// Create Kubernetes Secret
+		err = opcontroller.CreateOrUpdateObject(ctx, k8sclient, secret)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph object subuser %q secret", secret.Name)
+		}
+	}
+
 	return reconcile.Result{}, nil
 }

pkg/operator/ceph/object/user/controller.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"sort"
 
 	"github.com/ceph/go-ceph/rgw/admin"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
@@ -263,7 +264,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
 	}
 
 	tlsSecretName := store.Spec.Gateway.SSLCertificateRef
-	reconcileResponse, err = object.ReconcileCephUserSecret(r.opManagerContext, r.client, r.scheme, cephObjectStoreUser, r.userConfig, r.objContext.Endpoint, cephObjectStoreUser.Namespace, cephObjectStoreUser.Spec.Store, tlsSecretName)
+	reconcileResponse, err = object.ReconcileCephUserSecrets(r.opManagerContext, r.client, r.scheme, cephObjectStoreUser, r.userConfig, r.objContext.Endpoint, cephObjectStoreUser.Namespace, cephObjectStoreUser.Spec.Store, tlsSecretName)
 	if err != nil {
 		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcileFailedStatus)
 		return reconcileResponse, *cephObjectStoreUser, err
@@ -362,17 +363,123 @@ func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectSt
 		return errors.Wrapf(err, "failed to set quotas for user %q", u.Name)
 	}
 
+	// Reconcile subusers
+	user, err = r.reconcileSubusers(user, u)
+	if err != nil {
+		return err
+	}
+
 	// Set access and secret key
 	if r.userConfig.Keys == nil {
 		r.userConfig.Keys = make([]admin.UserKeySpec, 1)
 	}
 	r.userConfig.Keys[0].AccessKey = user.Keys[0].AccessKey
 	r.userConfig.Keys[0].SecretKey = user.Keys[0].SecretKey
+
+	// Copy the subuser keys – XXX: should this be a deep copy for safety?
+	r.userConfig.SwiftKeys = append([]admin.SwiftKeySpec{}, user.SwiftKeys...)
+
 	logger.Info(logCreateOrUpdate)
 
 	return nil
 }
 
+var accessLevelMap = map[admin.SubuserAccess]admin.SubuserAccess{
+	admin.SubuserAccessReplyNone:      admin.SubuserAccessNone,
+	admin.SubuserAccessReplyRead:      admin.SubuserAccessRead,
+	admin.SubuserAccessReplyWrite:     admin.SubuserAccessWrite,
+	admin.SubuserAccessReplyReadWrite: admin.SubuserAccessReadWrite,
+	admin.SubuserAccessReplyFull:      admin.SubuserAccessFull,
+}
+
+func mapAccessLevel(outputAccessLevel admin.SubuserAccess) (admin.SubuserAccess, error) {
+	level, ok := accessLevelMap[outputAccessLevel]
+	if !ok {
+		return admin.SubuserAccessNone, fmt.Errorf("invalid access level returned by RGW %s", outputAccessLevel)
+	}
+	return level, nil
+}
+
+func (r *ReconcileObjectStoreUser) reconcileSubusers(userIs admin.User, userSpec *cephv1.CephObjectStoreUser) (admin.User, error) {
+	var err error
+
+	mustRefreshUser := false
+
+	subusersSpec := append([]cephv1.SubuserSpec{}, userSpec.Spec.Subusers...)
+	subusersIs := append([]admin.SubuserSpec{}, userIs.Subusers...)
+
+	sort.Slice(subusersSpec, func(i, j int) bool {
+		return subusersSpec[i].Name < subusersSpec[j].Name
+	})
+	sort.Slice(subusersIs, func(i, j int) bool {
+		return subusersIs[i].Name < subusersIs[j].Name
+	})
+
+	toCreate := make([]cephv1.SubuserSpec, 0)
+	toDelete := make([]admin.SubuserSpec, 0)
+	for len(subusersSpec) > 0 && len(subusersIs) > 0 {
+		subuserFullNameSpec := fmt.Sprintf("%s:%s", userIs.ID, subusersSpec[0].Name)
+
+		if subuserFullNameSpec < subusersIs[0].Name {
+			toCreate = append(toCreate, subusersSpec[0])
+			subusersSpec = subusersSpec[1:]
+		} else if subuserFullNameSpec > subusersIs[0].Name {
+			toDelete = append(toDelete, subusersIs[0])
+			subusersIs = subusersIs[1:]
+		} else { // subusersSpec[0].Name == subusersIs[0].Name
+			accessLevelIs, err := mapAccessLevel(subusersIs[0].Access)
+			if err != nil {
+				return userIs, errors.Wrapf(err, "failed to reconcile subusers")
+			}
+			if string(subusersSpec[0].Access) != string(accessLevelIs) {
+				modifiedSubuser := admin.SubuserSpec{
+					Name:   subuserFullNameSpec,
+					Access: admin.SubuserAccess(subusersSpec[0].Access),
+				}
+				mustRefreshUser = true
+				err = r.objContext.AdminOpsClient.ModifySubuser(r.opManagerContext, userIs, modifiedSubuser)
+				if err != nil {
+					return userIs, errors.Wrapf(err, "failed to modify subuser %q", modifiedSubuser.Name)
+				}
+			}
+
+			subusersSpec = subusersSpec[1:]
+			subusersIs = subusersIs[1:]
+		}
+	}
+	toCreate = append(toCreate, subusersSpec...)
+	toDelete = append(toDelete, subusersIs...)
+
+	for _, subuserToCreate := range toCreate {
+		err = r.objContext.AdminOpsClient.CreateSubuser(r.opManagerContext, userIs, admin.SubuserSpec{
+			Name:   fmt.Sprintf("%s:%s", userIs.ID, subuserToCreate.Name),
+			Access: admin.SubuserAccess(subuserToCreate.Access),
+		})
+		mustRefreshUser = true
+		if err != nil {
+			return userIs, errors.Wrapf(err, "failed to create subuser %q", subuserToCreate.Name)
+		}
+	}
+
+	for _, subuserToDelete := range toDelete {
+		mustRefreshUser = true
+		err = r.objContext.AdminOpsClient.RemoveSubuser(r.opManagerContext, userIs, subuserToDelete)
+		if err != nil {
+			return userIs, errors.Wrapf(err, "failed to delete subuser %q", subuserToDelete.Name)
+		}
+	}
+
+	if mustRefreshUser {
+		// As the subuser commands don't give us the full user object, we need to get it here to have an accurate representation of the user state after the reconcile
+		userIs, err = r.objContext.AdminOpsClient.GetUser(r.opManagerContext, userIs)
+		if err != nil {
+			return userIs, errors.Wrapf(err, "failed to reconcile subusers during refresh")
+		}
+	}
+
+	return userIs, nil
+}
+
 func (r *ReconcileObjectStoreUser) initializeObjectStoreContext(u *cephv1.CephObjectStoreUser) error {
 	err := r.objectStoreInitialized(u)
 	if err != nil {