Reconcile logic and sys tests for topic permission

- issue: #191

Author: ChunyiLyu

controllers/common.go

@@ -23,6 +23,7 @@ const (
 	FederationControllerName        = "federation-controller"
 	ShovelControllerName            = "shovel-controller"
 	SuperStreamControllerName       = "super-stream-controller"
+	TopicPermissionControllerName   = "topic-permission-controller"
 )
 
 // names for environment variables

controllers/permission_controller.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
 	k8sApiErrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
@@ -33,7 +32,7 @@ func (r *PermissionReconciler) DeclareFunc(ctx context.Context, client rabbitmqc
 	username := permission.Spec.User
 	if permission.Spec.UserReference != nil {
 		var err error
-		if user, err = r.getUserFromReference(ctx, permission); err != nil {
+		if user, err = getUsernameFromUser(ctx, r.Client, permission.Namespace, permission.Spec.UserReference.Name); err != nil {
 			return err
 		} else if user != nil {
 			// User exist
@@ -56,26 +55,24 @@ func (r *PermissionReconciler) DeclareFunc(ctx context.Context, client rabbitmqc
 	return validateResponse(client.UpdatePermissionsIn(permission.Spec.Vhost, username, internal.GeneratePermissions(permission)))
 }
 
-func (r *PermissionReconciler) getUserFromReference(ctx context.Context, permission *topology.Permission) (*topology.User, error) {
+func getUsernameFromUser(ctx context.Context, client client.Client, namespace, name string) (*topology.User, error) {
 	logger := ctrl.LoggerFrom(ctx)
 
-	// get User from provided user reference
 	failureMsg := "failed to get User"
 	user := &topology.User{}
-	err := r.Get(ctx, types.NamespacedName{Name: permission.Spec.UserReference.Name, Namespace: permission.Namespace}, user)
-
+	err := client.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, user)
 	if err != nil && k8sApiErrors.IsNotFound(err) {
 		logger.Error(fmt.Errorf("user doesn't exist"), failureMsg)
 		return nil, nil
 	} else if err != nil {
-		logger.Error(err, failureMsg, "userReference", permission.Spec.UserReference.Name)
+		logger.Error(err, failureMsg, "userReference", name)
 		return nil, err
 	}
 
 	// get username from User status
 	if user.Status.Username == "" {
 		err := fmt.Errorf("this User does not have an username set in its status")
-		logger.Error(err, failureMsg, "userReference", permission.Spec.UserReference.Name)
+		logger.Error(err, failureMsg, "userReference", name)
 		return nil, err
 	}
 	return user, nil
@@ -87,7 +84,7 @@ func (r *PermissionReconciler) DeleteFunc(ctx context.Context, client rabbitmqcl
 
 	username := permission.Spec.User
 	if permission.Spec.UserReference != nil {
-		if user, err := r.getUserFromReference(ctx, permission); err != nil {
+		if user, err := getUsernameFromUser(ctx, r.Client, permission.Namespace, permission.Spec.UserReference.Name); err != nil {
 			return err
 		} else if user != nil {
 			// User exist

controllers/suite_test.go

@@ -14,10 +14,11 @@ import (
 	"crypto/x509"
 	"go/build"
 	"path/filepath"
-	"sigs.k8s.io/controller-runtime/pkg/envtest/komega"
 	"testing"
 	"time"
 
+	"sigs.k8s.io/controller-runtime/pkg/envtest/komega"
+
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient/rabbitmqclientfakes"
 
@@ -193,6 +194,14 @@ var _ = BeforeSuite(func() {
 			RabbitmqClientFactory: fakeRabbitMQClientFactory,
 			ReconcileFunc:         &controllers.ShovelReconciler{Client: mgr.GetClient()},
 		},
+		{
+			Client:                mgr.GetClient(),
+			Type:                  &topology.TopicPermission{},
+			Scheme:                mgr.GetScheme(),
+			Recorder:              fakeRecorder,
+			RabbitmqClientFactory: fakeRabbitMQClientFactory,
+			ReconcileFunc:         &controllers.TopicPermissionReconciler{Client: mgr.GetClient(), Scheme: mgr.GetScheme()},
+		},
 	}
 
 	for _, controller := range topologyReconcilers {

controllers/topicpermission_controller.go

@@ -0,0 +1,84 @@
+package controllers
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
+	"github.com/rabbitmq/messaging-topology-operator/internal"
+	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+)
+
+//+kubebuilder:rbac:groups=rabbitmq.com,resources=topicpermissions,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rabbitmq.com,resources=topicpermissions/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rabbitmq.com,resources=topicpermissions/finalizers,verbs=update
+
+type TopicPermissionReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
+}
+
+func (r *TopicPermissionReconciler) DeclareFunc(ctx context.Context, client rabbitmqclient.Client, obj topology.TopologyResource) error {
+	permission := obj.(*topology.TopicPermission)
+	user := &topology.User{}
+	username := permission.Spec.User
+	if permission.Spec.UserReference != nil {
+		var err error
+		if user, err = getUsernameFromUser(ctx, r.Client, permission.Namespace, permission.Spec.UserReference.Name); err != nil {
+			return err
+		} else if user != nil {
+			// User exist
+			username = user.Status.Username
+		}
+	}
+	if username == "" {
+		return fmt.Errorf("failed create Permission, missing User")
+	}
+
+	// user != nil, not working because user has always a name set
+	if user.Name != "" {
+		if err := controllerutil.SetControllerReference(user, permission, r.Scheme); err != nil {
+			return fmt.Errorf("failed set controller reference: %v", err)
+		}
+		if err := r.Client.Update(ctx, permission); err != nil {
+			return fmt.Errorf("failed to Update object with controller reference: %w", err)
+		}
+	}
+	return validateResponse(client.UpdateTopicPermissionsIn(permission.Spec.Vhost, username, internal.GenerateTopicPermissions(permission)))
+}
+
+func (r *TopicPermissionReconciler) DeleteFunc(ctx context.Context, client rabbitmqclient.Client, obj topology.TopologyResource) error {
+	logger := ctrl.LoggerFrom(ctx)
+	permission := obj.(*topology.TopicPermission)
+
+	username := permission.Spec.User
+	if permission.Spec.UserReference != nil {
+		if user, err := getUsernameFromUser(ctx, r.Client, permission.Namespace, permission.Spec.UserReference.Name); err != nil {
+			return err
+		} else if user != nil {
+			// User exist
+			username = user.Status.Username
+		}
+	}
+
+	if username == "" {
+		logger.Info("user already removed; no need to delete topic permission")
+	} else if err := r.clearTopicPermission(ctx, client, permission, username); err != nil {
+		return err
+	}
+	return removeFinalizer(ctx, r.Client, permission)
+}
+
+func (r *TopicPermissionReconciler) clearTopicPermission(ctx context.Context, client rabbitmqclient.Client, permission *topology.TopicPermission, user string) error {
+	logger := ctrl.LoggerFrom(ctx)
+	err := validateResponseForDeletion(client.DeleteTopicPermissionsIn(permission.Spec.Vhost, user, permission.Spec.Permissions.Exchange))
+	if errors.Is(err, NotFound) {
+		logger.Info("cannot find user or vhost in rabbitmq server; no need to delete permission", "user", user, "vhost", permission.Spec.Vhost)
+		return nil
+	}
+	return err
+}