Delete existing limits if they are removed from a user.

Author: MirahImage

controllers/binding_controller.go

@@ -17,7 +17,7 @@ import (
 "reflect"
 
 "github.com/go-logr/logr"
-rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 "github.com/rabbitmq/messaging-topology-operator/internal"
 "github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"

controllers/user_controller.go

@@ -14,6 +14,7 @@ import (
 "errors"
 "fmt"
 
+rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 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/v3"
 . "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/v3"
 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 {

go.mod

@@ -9,9 +9,9 @@ require (
 github.com/go-logr/logr v1.4.2
 github.com/google/uuid v1.6.0
 github.com/hashicorp/vault/api v1.16.0
-github.com/michaelklishin/rabbit-hole/v2 v2.16.0
+github.com/michaelklishin/rabbit-hole/v3 v3.2.0
 github.com/onsi/ginkgo/v2 v2.23.3
-github.com/onsi/gomega v1.36.3
+github.com/onsi/gomega v1.37.0
 github.com/rabbitmq/cluster-operator/v2 v2.12.1
 gopkg.in/ini.v1 v1.67.0
 k8s.io/api v0.32.3