Disallow updates of Queue arguments

Queue arguments can't be updated in RabbitMQ. The webhook now disallows updates on queue arguments, matching the behaviour of RabbitMQ. Signed-off-by: Aitor Perez Cedres <[email protected]>
rabbitmq · May 28, 2024 · 09c2f56 · 09c2f56
1 parent f6f3b41
commit 09c2f56
Show file tree

Hide file tree

Showing 2 changed files with 72 additions and 6 deletions.
diff --git a/api/v1beta1/queue_webhook.go b/api/v1beta1/queue_webhook.go
@@ -2,7 +2,9 @@ package v1beta1
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"maps"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -23,8 +25,8 @@ func (q *Queue) SetupWebhookWithManager(mgr ctrl.Manager) error {
 
 var _ webhook.CustomValidator = &Queue{}
 
-// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
-// either rabbitmqClusterReference.name or rabbitmqClusterReference.connectionSecret must be provided but not both
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type.
+// Either rabbitmqClusterReference.name or rabbitmqClusterReference.connectionSecret must be provided but not both
 func (q *Queue) ValidateCreate(_ context.Context, obj runtime.Object) (warnings admission.Warnings, err error) {
 	inQueue, ok := obj.(*Queue)
 	if !ok {
@@ -38,10 +40,11 @@ func (q *Queue) ValidateCreate(_ context.Context, obj runtime.Object) (warnings
 	return nil, q.Spec.RabbitmqClusterReference.validate(inQueue.RabbitReference())
 }
 
-// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
-// returns error type 'forbidden' for updates that the controller chooses to disallow: queue name/vhost/rabbitmqClusterReference
-// returns error type 'invalid' for updates that will be rejected by rabbitmq server: queue types/autoDelete/durable
-// queue arguments not handled because implementation couldn't change
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type.
+//
+// Returns error type 'forbidden' for updates that the controller chooses to disallow: queue name/vhost/rabbitmqClusterReference
+//
+// Returns error type 'invalid' for updates that will be rejected by rabbitmq server: queue types/autoDelete/durable
 func (q *Queue) ValidateUpdate(_ context.Context, oldObj, newObj runtime.Object) (warnings admission.Warnings, err error) {
 	oldQueue, ok := oldObj.(*Queue)
 	if !ok {
@@ -94,10 +97,49 @@ func (q *Queue) ValidateUpdate(_ context.Context, oldObj, newObj runtime.Object)
 		))
 	}
 
+	if oldQueue.Spec.Arguments != nil && newQueue.Spec.Arguments == nil {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "arguments"),
+			newQueue.Spec.Arguments,
+			"queue arguments cannot be updated",
+		))
+	}
+
+	if oldQueue.Spec.Arguments == nil && newQueue.Spec.Arguments != nil {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "arguments"),
+			newQueue.Spec.Arguments,
+			"queue arguments cannot be updated",
+		))
+	}
+
+	if oldQueue.Spec.Arguments != nil && newQueue.Spec.Arguments != nil {
+		previousArgs := make(map[string]any)
+		err := json.Unmarshal(oldQueue.Spec.Arguments.Raw, &previousArgs)
+		if err != nil {
+			return nil, apierrors.NewInternalError(fmt.Errorf("error unmarshalling previous Queue arguments: %w", err))
+		}
+
+		updatedArgs := make(map[string]any)
+		err = json.Unmarshal(newQueue.Spec.Arguments.Raw, &updatedArgs)
+		if err != nil {
+			return nil, apierrors.NewInternalError(fmt.Errorf("error unmarshalling current Queue arguments: %w", err))
+		}
+
+		if !maps.Equal(previousArgs, updatedArgs) {
+			allErrs = append(allErrs, field.Invalid(
+				field.NewPath("spec", "arguments"),
+				newQueue.Spec.Arguments,
+				"queue arguments cannot be updated",
+			))
+		}
+	}
+
 	if len(allErrs) == 0 {
 		return nil, nil
 	}
 
+	//goland:noinspection GoDfaNilDereference
 	return nil, allErrs.ToAggregate()
 }
 

diff --git a/api/v1beta1/queue_webhook_test.go b/api/v1beta1/queue_webhook_test.go
@@ -6,6 +6,7 @@ import (
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 )
 
 var _ = Describe("queue webhook", func() {
@@ -23,6 +24,7 @@ var _ = Describe("queue webhook", func() {
 				AutoDelete:     true,
 				DeleteIfEmpty:  true,
 				DeleteIfUnused: false,
+				Arguments:      &runtime.RawExtension{Raw: []byte(`{"this-argument": "cannot be updated"}`)},
 				RabbitmqClusterReference: RabbitmqClusterReference{
 					Name: "some-cluster",
 				},
@@ -130,5 +132,27 @@ var _ = Describe("queue webhook", func() {
 			_, err := newQueue.ValidateUpdate(rootCtx, &queue, newQueue)
 			Expect(err).To(MatchError(ContainSubstring("autoDelete cannot be updated")))
 		})
+
+		It("does not allow updates on queue arguments", func() {
+			By("not allowing value changes, nor adding new values")
+			newQueue := queue.DeepCopy()
+			newQueue.Spec.Arguments = &runtime.RawExtension{
+				Raw: []byte(`{"this-argument": "really cant be updated", "adding-args": "not possible"}`),
+			}
+			_, err := newQueue.ValidateUpdate(rootCtx, &queue, newQueue)
+			Expect(err).To(MatchError(ContainSubstring("queue arguments cannot be updated")))
+
+			By("not allowing removal")
+			newQueue.Spec.Arguments = nil
+			_, err = newQueue.ValidateUpdate(rootCtx, &queue, newQueue)
+			Expect(err).To(MatchError(ContainSubstring("queue arguments cannot be updated")))
+
+			By("not allowing emptying the arguments")
+			newQueue.Spec.Arguments = &runtime.RawExtension{
+				Raw: []byte(`{}`),
+			}
+			_, err = newQueue.ValidateUpdate(rootCtx, &queue, newQueue)
+			Expect(err).To(MatchError(ContainSubstring("queue arguments cannot be updated")))
+		})
 	})
 })