Remove outdated vhost limits.

Author: MirahImage

controllers/vhost_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	"github.com/rabbitmq/messaging-topology-operator/internal"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -30,12 +31,25 @@ func (r *VhostReconciler) DeclareFunc(ctx context.Context, client rabbitmqclient
 		return err
 	}
 
-	vhostLimits := internal.GenerateVhostLimits(vhost.Spec.VhostLimits)
-	logger.Info("generated vhost limits", "vhost", vhost.Spec.Name, "limits", vhostLimits)
-	if len(vhostLimits) > 0 {
-		err = validateResponse(client.PutVhostLimits(vhost.Spec.Name, vhostLimits))
+	newVhostLimits := internal.GenerateVhostLimits(vhost.Spec.VhostLimits)
+	logger.Info("getting existing vhost limits", vhost, vhost.Spec.Name)
+	existingVhostLimits, err := r.getVhostLimits(client, vhost.Spec.Name)
+	if err != nil {
+		return err
+	}
+	limitsToDelete := r.vhostLimitsToDelete(existingVhostLimits, newVhostLimits)
+	if len(limitsToDelete) > 0 {
+		logger.Info("Deleting outdated vhost limits", "vhost", vhost.Spec.Name, "limits", limitsToDelete)
+		err = validateResponseForDeletion(client.DeleteVhostLimits(vhost.Spec.Name, limitsToDelete))
+		if err != nil && !errors.Is(err, NotFound) {
+			return err
+		}
+	}
+	if len(newVhostLimits) > 0 {
+		logger.Info("creating new vhost limits", "vhost", vhost.Spec.Name, "limits", newVhostLimits)
+		return validateResponse(client.PutVhostLimits(vhost.Spec.Name, newVhostLimits))
 	}
-	return err
+	return nil
 }
 
 // DeleteFunc deletes vhost from server
@@ -54,3 +68,28 @@ func (r *VhostReconciler) DeleteFunc(ctx context.Context, client rabbitmqclient.
 	}
 	return nil
 }
+
+func (r *VhostReconciler) vhostLimitsToDelete(existingVhostLimits, newVhostLimits rabbithole.VhostLimitsValues) (limitsToDelete rabbithole.VhostLimits) {
+	vhostLimitKeys := []string{"max-connections", "max-queues"}
+	for _, limit := range vhostLimitKeys {
+		_, oldExists := existingVhostLimits[limit]
+		_, newExists := newVhostLimits[limit]
+		if oldExists && !newExists {
+			limitsToDelete = append(limitsToDelete, limit)
+		}
+	}
+	return limitsToDelete
+}
+
+func (r *VhostReconciler) getVhostLimits(client rabbitmqclient.Client, vhost string) (rabbithole.VhostLimitsValues, error) {
+	vhostLimitsInfo, err := client.GetVhostLimits(vhost)
+	if errors.Is(err, error(rabbithole404)) {
+		return rabbithole.VhostLimitsValues{}, nil
+	} else if err != nil {
+		return rabbithole.VhostLimitsValues{}, err
+	}
+	if len(vhostLimitsInfo) == 0 {
+		return rabbithole.VhostLimitsValues{}, nil
+	}
+	return vhostLimitsInfo[0].Value, nil
+}

controllers/vhost_controller_test.go

@@ -14,6 +14,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"time"
 
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	. "github.com/onsi/gomega/gstruct"
@@ -159,6 +160,125 @@ var _ = Describe("vhost-controller", func() {
 					})))
 			})
 		})
+
+		Context("vhost limits", func() {
+			var connections, queues int32
+
+			When("vhost limits are provided", func() {
+				BeforeEach(func() {
+					connections = 708
+					queues = 509
+					vhostName = "vhost-with-limits"
+					vhostLimits = &topology.VhostLimits{
+						Connections: &connections,
+						Queues:      &queues,
+					}
+					fakeRabbitMQClient.PutVhostReturns(&http.Response{
+						Status:     "201 Created",
+						StatusCode: http.StatusCreated,
+					}, nil)
+					fakeRabbitMQClient.PutVhostLimitsReturns(&http.Response{
+						Status:     "200 OK",
+						StatusCode: http.StatusOK,
+					}, nil)
+					fakeRabbitMQClient.GetVhostLimitsReturns(nil, rabbithole.ErrorResponse{
+						StatusCode: 404,
+						Message:    "Object Not Found",
+						Reason:     "Not Found",
+					})
+				})
+
+				It("puts the vhost limits", func() {
+					Expect(k8sClient.Create(ctx, &vhost)).To(Succeed())
+					Eventually(func() []topology.Condition {
+						_ = k8sClient.Get(
+							ctx,
+							types.NamespacedName{Name: vhost.Name, Namespace: vhost.Namespace},
+							&vhost,
+						)
+
+						return vhost.Status.Conditions
+					}).
+						Within(statusEventsUpdateTimeout).
+						WithPolling(time.Second).
+						Should(ContainElement(MatchFields(IgnoreExtras, Fields{
+							"Type":   Equal(topology.ConditionType("Ready")),
+							"Reason": Equal("SuccessfulCreateOrUpdate"),
+							"Status": Equal(corev1.ConditionTrue),
+						})))
+
+					Expect(fakeRabbitMQClient.PutVhostLimitsCallCount()).To(BeNumerically(">", 0))
+					_, vhostLimitsValues := fakeRabbitMQClient.PutVhostLimitsArgsForCall(0)
+					Expect(len(vhostLimitsValues)).To(Equal(2))
+					Expect(vhostLimitsValues).To(HaveKeyWithValue("max-connections", int(connections)))
+					Expect(vhostLimitsValues).To(HaveKeyWithValue("max-queues", int(queues)))
+				})
+			})
+
+			When("vhost limits are not provided", func() {
+				BeforeEach(func() {
+					vhostName = "vhost-without-limits"
+					fakeRabbitMQClient.GetVhostLimitsReturns(nil, rabbithole.ErrorResponse{
+						StatusCode: 404,
+						Message:    "Object Not Found",
+						Reason:     "Not Found",
+					})
+				})
+
+				It("does not set vhost limits", func() {
+					Expect(k8sClient.Create(ctx, &vhost)).To(Succeed())
+					Expect(fakeRabbitMQClient.PutVhostLimitsCallCount()).To(Equal(0))
+				})
+			})
+
+			When("vhost limits are updated", func() {
+				BeforeEach(func() {
+					vhostName = "vhost-updated-limits"
+					queues = 613
+					vhostLimits = &topology.VhostLimits{
+						Connections: nil,
+						Queues:      &queues,
+					}
+
+					var vhostLimitsInfo []rabbithole.VhostLimitsInfo
+					vhostLimitsInfo = append(vhostLimitsInfo, rabbithole.VhostLimitsInfo{
+						Vhost: vhostName,
+						Value: rabbithole.VhostLimitsValues{"max-queues": 10, "max-connections": 300},
+					})
+
+					fakeRabbitMQClient.PutVhostReturns(&http.Response{
+						Status:     "201 Created",
+						StatusCode: http.StatusCreated,
+					}, nil)
+					fakeRabbitMQClient.PutVhostLimitsReturns(&http.Response{
+						Status:     "201 Created",
+						StatusCode: http.StatusCreated,
+					}, nil)
+					fakeRabbitMQClient.GetVhostLimitsReturns(vhostLimitsInfo, nil)
+					fakeRabbitMQClient.DeleteVhostLimitsReturns(&http.Response{
+						Status:     "204 No Content",
+						StatusCode: http.StatusNoContent,
+					}, nil)
+				})
+
+				It("updates the provided limits and removes unspecified limits", func() {
+					Expect(k8sClient.Create(ctx, &vhost)).To(Succeed())
+
+					By("deleting the outdated limits")
+					Expect(fakeRabbitMQClient.DeleteVhostLimitsCallCount()).To(BeNumerically(">", 0))
+					vhostname, limits := fakeRabbitMQClient.DeleteVhostLimitsArgsForCall(0)
+					Expect(vhostname).To(Equal(vhostName))
+					Expect(len(limits)).To(Equal(1))
+					Expect(limits).To(ContainElement("max-connections"))
+
+					By("updating the new limits")
+					Expect(fakeRabbitMQClient.PutVhostLimitsCallCount()).To(BeNumerically(">", 0))
+					_, vhostLimitsValues := fakeRabbitMQClient.PutVhostLimitsArgsForCall(0)
+					Expect(len(vhostLimitsValues)).To(Equal(1))
+					Expect(fakeRabbitMQClient).To(HaveKeyWithValue("max-queues", int(queues)))
+				})
+			})
+		})
 	})
 
 	Context("deletion", func() {
@@ -254,69 +374,4 @@ var _ = Describe("vhost-controller", func() {
 			Expect(fakeRabbitMQClient.DeleteVhostCallCount()).To(Equal(0))
 		})
 	})
-
-	Context("vhost limits", func() {
-		When("vhost limits are provided", func() {
-			var connections, queues int32
-
-			BeforeEach(func() {
-				connections = 708
-				queues = 509
-				vhostName = "vhost-with-limits"
-				vhostLimits = &topology.VhostLimits{
-					Connections: &connections,
-					Queues:      &queues,
-				}
-				fakeRabbitMQClient.PutVhostReturns(&http.Response{
-					Status:     "201 Created",
-					StatusCode: http.StatusCreated,
-				}, nil)
-				fakeRabbitMQClient.PutVhostLimitsReturns(&http.Response{
-					Status:     "200 OK",
-					StatusCode: http.StatusOK,
-				}, nil)
-			})
-
-			It("puts the vhost limits", func() {
-				Expect(k8sClient.Create(ctx, &vhost)).To(Succeed())
-				Eventually(func() []topology.Condition {
-					_ = k8sClient.Get(
-						ctx,
-						types.NamespacedName{Name: vhost.Name, Namespace: vhost.Namespace},
-						&vhost,
-					)
-
-					return vhost.Status.Conditions
-				}).
-					Within(statusEventsUpdateTimeout).
-					WithPolling(time.Second).
-					Should(ContainElement(MatchFields(IgnoreExtras, Fields{
-						"Type":   Equal(topology.ConditionType("Ready")),
-						"Reason": Equal("SuccessfulCreateOrUpdate"),
-						"Status": Equal(corev1.ConditionTrue),
-					})))
-
-				Expect(fakeRabbitMQClient.PutVhostLimitsCallCount()).To(BeNumerically(">", 0))
-				_, vhostLimitsValues := fakeRabbitMQClient.PutVhostLimitsArgsForCall(0)
-				Expect(len(vhostLimitsValues)).To(Equal(2))
-				connectionLimit, ok := vhostLimitsValues["max-connections"]
-				Expect(ok).To(BeTrue())
-				Expect(connectionLimit).To(Equal(int(connections)))
-				queueLimit, ok := vhostLimitsValues["max-queues"]
-				Expect(ok).To(BeTrue())
-				Expect(queueLimit).To(Equal(int(queues)))
-			})
-		})
-
-		When("vhost limits are not provided", func() {
-			BeforeEach(func() {
-				vhostName = "vhost-without-limits"
-			})
-
-			It("does not set vhost limits", func() {
-				Expect(k8sClient.Create(ctx, &vhost)).To(Succeed())
-				Expect(fakeRabbitMQClient.PutVhostLimitsCallCount()).To(Equal(0))
-			})
-		})
-	})
 })

rabbitmqclient/rabbitmq_client_factory.go

@@ -41,6 +41,7 @@ type Client interface {
 	PutVhost(string, rabbithole.VhostSettings) (*http.Response, error)
 	DeleteVhost(string) (*http.Response, error)
 	PutVhostLimits(string, rabbithole.VhostLimitsValues) (*http.Response, error)
+	GetVhostLimits(string) ([]rabbithole.VhostLimitsInfo, error)
 	DeleteVhostLimits(string, rabbithole.VhostLimits) (*http.Response, error)
 	PutGlobalParameter(name string, value interface{}) (*http.Response, error)
 	DeleteGlobalParameter(name string) (*http.Response, error)