Delete existing limits if they are removed from a user.

Author: MirahImage

controllers/user_controller.go

@@ -14,6 +14,7 @@ import (
 	"errors"
 	"fmt"
 
+	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"github.com/rabbitmq/messaging-topology-operator/internal"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
@@ -226,12 +227,46 @@ func (r *UserReconciler) DeclareFunc(ctx context.Context, client rabbitmqclient.
 		return err
 	}
 
-	userLimits := internal.GenerateUserLimits(user.Spec.UserLimits)
-	logger.Info("Generated user limits", "user", user.Name, "limits", userLimits)
-	if len(userLimits) > 0 {
-		return validateResponse(client.PutUserLimits(string(credentials.Data["username"]), userLimits))
+	newUserLimits := internal.GenerateUserLimits(user.Spec.UserLimits)
+	existingUserLimits, err := r.getUserLimits(client, string(credentials.Data["username"]))
+	if err != nil {
+		return err
 	}
-	return nil
+	logger.Info("Updating user limits", "user", user.Name, "existing limits", existingUserLimits, "new limits", newUserLimits)
+	limitsToDelete := r.userLimitsToDelete(existingUserLimits, newUserLimits)
+	if len(limitsToDelete) > 0 {
+		err = validateResponseForDeletion(client.DeleteUserLimits(string(credentials.Data["username"]), limitsToDelete))
+		if err != nil && !errors.Is(err, NotFound) {
+			return err
+		}
+		logger.Info("Deleted user limits", "user", user.Name, "limits", limitsToDelete)
+	}
+	return validateResponse(client.PutUserLimits(string(credentials.Data["username"]), newUserLimits))
+}
+
+func (r *UserReconciler) userLimitsToDelete(existingUserLimits, newUserLimits rabbithole.UserLimitsValues) (limitsToDelete rabbithole.UserLimits) {
+	userLimitKeys := []string{"max-connections", "max-channels"}
+	for _, limit := range userLimitKeys {
+		_, oldExists := existingUserLimits[limit]
+		_, newExists := newUserLimits[limit]
+		if oldExists && !newExists {
+			limitsToDelete = append(limitsToDelete, limit)
+		}
+	}
+	return limitsToDelete
+}
+
+func (r *UserReconciler) getUserLimits(client rabbitmqclient.Client, username string) (rabbithole.UserLimitsValues, error) {
+	userLimitsInfo, err := client.GetUserLimits(username)
+	if errors.Is(err, error(rabbithole404)) {
+		return rabbithole.UserLimitsValues{}, nil
+	} else if err != nil {
+		return rabbithole.UserLimitsValues{}, err
+	}
+	if len(userLimitsInfo) == 0 {
+		return rabbithole.UserLimitsValues{}, nil
+	}
+	return userLimitsInfo[0].Value, nil
 }
 
 func (r *UserReconciler) getUserCredentials(ctx context.Context, user *topology.User) (*corev1.Secret, error) {

controllers/user_controller_test.go

@@ -15,6 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
+	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	. "github.com/onsi/gomega/gstruct"
@@ -176,6 +177,11 @@ var _ = Describe("UserController", func() {
 					Status:     "201 Created",
 					StatusCode: http.StatusCreated,
 				}, nil)
+				fakeRabbitMQClient.GetUserLimitsReturns(nil, rabbithole.ErrorResponse{
+					StatusCode: 404,
+					Message:    "Object Not Found",
+					Reason:     "Not Found",
+				})
 			})
 
 			It("should create the user limits", func() {
@@ -203,6 +209,64 @@ var _ = Describe("UserController", func() {
 				Expect(userLimitsValues).To(HaveKeyWithValue("max-channels", (int(channels))))
 			})
 		})
+
+		When("the user already has existing limits that differ from the new limits", func() {
+			BeforeEach(func() {
+				userName = "test-changed-user-limits"
+				connections = 5
+				userLimits = topology.UserLimits{
+					Connections: &connections,
+					Channels:    nil,
+				}
+				var userLimitsInfo []rabbithole.UserLimitsInfo
+				userLimitsInfo = append(userLimitsInfo, rabbithole.UserLimitsInfo{
+					User:  userName,
+					Value: rabbithole.UserLimitsValues{"max-channels": 10, "max-connections": 3},
+				})
+				fakeRabbitMQClient.PutUserReturns(&http.Response{
+					Status:     "201 Created",
+					StatusCode: http.StatusCreated,
+				}, nil)
+				fakeRabbitMQClient.PutUserLimitsReturns(&http.Response{
+					Status:     "201 Created",
+					StatusCode: http.StatusCreated,
+				}, nil)
+				fakeRabbitMQClient.GetUserLimitsReturns(userLimitsInfo, nil)
+				fakeRabbitMQClient.DeleteUserLimitsReturns(&http.Response{
+					Status:     "204 No Content",
+					StatusCode: http.StatusNoContent,
+				}, nil)
+			})
+
+			It("should update the existing user limit and delete the unused old limit", func() {
+				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
+				Eventually(func() []topology.Condition {
+					_ = k8sClient.Get(
+						ctx,
+						types.NamespacedName{Name: user.Name, Namespace: user.Namespace},
+						&user,
+					)
+
+					return user.Status.Conditions
+				}).
+					Within(statusEventsUpdateTimeout).
+					WithPolling(time.Second).
+					Should(ContainElement(MatchFields(IgnoreExtras, Fields{
+						"Type":   Equal(topology.ConditionType("Ready")),
+						"Reason": Equal("SuccessfulCreateOrUpdate"),
+						"Status": Equal(corev1.ConditionTrue),
+					})))
+				By("calling DeleteUserLimits with the unused old user limits")
+				Expect(fakeRabbitMQClient.DeleteUserLimitsCallCount()).To(BeNumerically(">", 0))
+				_, userLimits := fakeRabbitMQClient.DeleteUserLimitsArgsForCall(0)
+				Expect(userLimits).To(HaveLen(1))
+				Expect(userLimits).To(ContainElement("max-channels"))
+				By("calling PutUserLimits with the correct new user limits")
+				Expect(fakeRabbitMQClient.PutUserLimitsCallCount()).To(BeNumerically(">", 0))
+				_, userLimitsValues := fakeRabbitMQClient.PutUserLimitsArgsForCall(0)
+				Expect(userLimitsValues).To(HaveKeyWithValue("max-connections", int(connections)))
+			})
+		})
 	})
 
 	When("deleting a user", func() {

controllers/utils.go

@@ -18,6 +18,7 @@ import (
 	"strings"
 	"time"
 
+	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
 	"k8s.io/client-go/tools/record"
@@ -30,6 +31,13 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
+// returned in some cases as an error when rabbithole encounters a 404 response
+var rabbithole404 = rabbithole.ErrorResponse{
+	StatusCode: 404,
+	Message:    "Object Not Found",
+	Reason:     "Not Found",
+}
+
 // TODO: check possible status code response from RabbitMQ
 // validate status code above 300 might not be all failure case
 func validateResponse(res *http.Response, err error) error {

rabbitmqclient/rabbitmq_client_factory.go

@@ -23,6 +23,7 @@ type Client interface {
 	PutUser(string, rabbithole.UserSettings) (*http.Response, error)
 	DeleteUser(string) (*http.Response, error)
 	PutUserLimits(string, rabbithole.UserLimitsValues) (*http.Response, error)
+	GetUserLimits(string) ([]rabbithole.UserLimitsInfo, error)
 	DeleteUserLimits(string, rabbithole.UserLimits) (*http.Response, error)
 	DeclareBinding(string, rabbithole.BindingInfo) (*http.Response, error)
 	DeleteBinding(string, rabbithole.BindingInfo) (*http.Response, error)