Add validating admission webhook for queues.rabbitmq.com

- not allowing updates on queue.spec, except spec.arguments
- we might change how controller handles queue.arguments in
the future

api/v1alpha1/queue_types.go

@@ -12,6 +12,7 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 // using runtime.RawExtension to represent queue arguments
@@ -77,6 +78,13 @@ type QueueList struct {
 	Items           []Queue `json:"items"`
 }
 
+func (q *Queue) GroupResource() schema.GroupResource {
+	return schema.GroupResource{
+		Group:    q.GroupVersionKind().Group,
+		Resource: q.GroupVersionKind().Kind,
+	}
+}
+
 func init() {
 	SchemeBuilder.Register(&Queue{}, &QueueList{})
 }

api/v1alpha1/queue_webhook.go

@@ -0,0 +1,88 @@
+package v1alpha1
+
+import (
+	"fmt"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+)
+
+func (r *Queue) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(r).
+		Complete()
+}
+
+// +kubebuilder:webhook:verbs=create;update,path=/validate-rabbitmq-com-v1alpha1-queue,mutating=false,failurePolicy=fail,groups=rabbitmq.com,resources=queues,versions=v1alpha1,name=vqueue.kb.io,sideEffects=none,admissionReviewVersions=v1sideEffects=none,admissionReviewVersions=v1
+
+var _ webhook.Validator = &Queue{}
+
+// no validation on create
+func (q *Queue) ValidateCreate() error {
+	return nil
+}
+
+// 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
+func (q *Queue) ValidateUpdate(old runtime.Object) error {
+	oldQueue, ok := old.(*Queue)
+	if !ok {
+		return apierrors.NewBadRequest(fmt.Sprintf("expected a queue but got a %T", old))
+	}
+
+	var allErrs field.ErrorList
+	detailMsg := "updates on name, vhost, and rabbitmqClusterReference are all forbidden"
+	if q.Spec.Name != oldQueue.Spec.Name {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "name"), detailMsg))
+	}
+
+	if q.Spec.Vhost != oldQueue.Spec.Vhost {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "vhost"), detailMsg))
+	}
+
+	if q.Spec.RabbitmqClusterReference != oldQueue.Spec.RabbitmqClusterReference {
+		return apierrors.NewForbidden(q.GroupResource(), q.Name,
+			field.Forbidden(field.NewPath("spec", "rabbitmqClusterReference"), detailMsg))
+	}
+
+	if q.Spec.Type != oldQueue.Spec.Type {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "type"),
+			q.Spec.Type,
+			"queue type cannot be updated",
+		))
+	}
+
+	if q.Spec.AutoDelete != oldQueue.Spec.AutoDelete {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "autoDelete"),
+			q.Spec.AutoDelete,
+			"autoDelete cannot be updated",
+		))
+	}
+
+	if q.Spec.Durable != oldQueue.Spec.Durable {
+		allErrs = append(allErrs, field.Invalid(
+			field.NewPath("spec", "durable"),
+			q.Spec.AutoDelete,
+			"durable cannot be updated",
+		))
+	}
+
+	if len(allErrs) == 0 {
+		return nil
+	}
+
+	return apierrors.NewInvalid(GroupVersion.WithKind("Queue").GroupKind(), q.Name, allErrs)
+}
+
+// no validation on delete
+func (q *Queue) ValidateDelete() error {
+	return nil
+}

api/v1alpha1/queue_webhook_test.go

@@ -0,0 +1,67 @@
+package v1alpha1
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = Describe("queue webhook", func() {
+
+	var queue = Queue{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "update-binding",
+		},
+		Spec: QueueSpec{
+			Name:       "test",
+			Vhost:      "/a-vhost",
+			Type:       "quorum",
+			Durable:    false,
+			AutoDelete: true,
+			RabbitmqClusterReference: RabbitmqClusterReference{
+				Name:      "some-cluster",
+				Namespace: "default",
+			},
+		},
+	}
+
+	It("does not allow updates on queue name", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Name = "new-name"
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on vhost", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Vhost = "/new-vhost"
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on RabbitmqClusterReference", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.RabbitmqClusterReference = RabbitmqClusterReference{
+			Name:      "new-cluster",
+			Namespace: "default",
+		}
+		Expect(apierrors.IsForbidden(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on queue type", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Type = "classic"
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on durable", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.Durable = true
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+
+	It("does not allow updates on autoDelete", func() {
+		newQueue := queue.DeepCopy()
+		newQueue.Spec.AutoDelete = false
+		Expect(apierrors.IsInvalid(newQueue.ValidateUpdate(&queue))).To(BeTrue())
+	})
+})

config/crd/kustomization.yaml

@@ -12,9 +12,11 @@ resources:
 
 patchesStrategicMerge:
 - patches/webhook_in_bindings.yaml
+- patches/webhook_in_queues.yaml
 # +kubebuilder:scaffold:crdkustomizewebhookpatch
 
 - patches/cainjection_in_bindings.yaml
+- patches/cainjection_in_queues.yaml
 # +kubebuilder:scaffold:crdkustomizecainjectionpatch
 
 configurations:

config/crd/patches/cainjection_in_queues.yaml

@@ -1,6 +1,4 @@
-# The following patch adds a directive for certmanager to inject CA into the CRD
-# CRD conversion requires k8s 1.13 or later.
-apiVersion: apiextensions.k8s.io/v1alpha1
+apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:

config/crd/patches/webhook_in_queues.yaml

@@ -1,17 +1,15 @@
-# The following patch enables conversion webhook for CRD
-# CRD conversion requires k8s 1.13 or later.
-apiVersion: apiextensions.k8s.io/v1alpha1
+apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   name: queues.rabbitmq.com
 spec:
   conversion:
     strategy: Webhook
-    webhookClientConfig:
-      # this is "\n" used as a placeholder, otherwise it will be rejected by the apiserver for being blank,
-      # but we're going to set it later using the cert-manager (or potentially a patch if not using cert-manager)
-      caBundle: Cg==
-      service:
-        namespace: system
-        name: webhook-service
-        path: /convert
+    webhook:
+      conversionReviewVersions: ["v1", "v1beta1"]
+      clientConfig:
+        caBundle: Cg==
+        service:
+          namespace: system
+          name: webhook-service
+          path: /convert

config/crd/patches/webhook_in_users.yaml

@@ -7,11 +7,11 @@ metadata:
 spec:
   conversion:
     strategy: Webhook
-    webhookClientConfig:
-      # this is "\n" used as a placeholder, otherwise it will be rejected by the apiserver for being blank,
-      # but we're going to set it later using the cert-manager (or potentially a patch if not using cert-manager)
-      caBundle: Cg==
-      service:
-        namespace: system
-        name: webhook-service
-        path: /convert
+    webhook:
+      conversionReviewVersions: ["v1", "v1beta1"]
+      clientConfig:
+        caBundle: Cg==
+        service:
+          namespace: system
+          name: webhook-service
+          path: /convert

config/webhook/manifests.yaml

@@ -26,3 +26,23 @@ webhooks:
     resources:
     - bindings
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /validate-rabbitmq-com-v1alpha1-queue
+  failurePolicy: Fail
+  name: vqueue.kb.io
+  rules:
+  - apiGroups:
+    - rabbitmq.com
+    apiVersions:
+    - v1alpha1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - queues
+  sideEffects: None

main.go

@@ -21,7 +21,6 @@ import (
 
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 
-	rabbitmqcomv1alpha1 "github.com/rabbitmq/messaging-topology-operator/api/v1alpha1"
 	topologyv1alpha1 "github.com/rabbitmq/messaging-topology-operator/api/v1alpha1"
 	"github.com/rabbitmq/messaging-topology-operator/controllers"
 	// +kubebuilder:scaffold:imports
@@ -119,10 +118,14 @@ func main() {
 		log.Error(err, "unable to create controller", "controller", policyControllerName)
 		os.Exit(1)
 	}
-	if err = (&rabbitmqcomv1alpha1.Binding{}).SetupWebhookWithManager(mgr); err != nil {
+	if err = (&topologyv1alpha1.Binding{}).SetupWebhookWithManager(mgr); err != nil {
 		log.Error(err, "unable to create webhook", "webhook", "Binding")
 		os.Exit(1)
 	}
+	if err = (&topologyv1alpha1.Queue{}).SetupWebhookWithManager(mgr); err != nil {
+		log.Error(err, "unable to create webhook", "webhook", "Queue")
+		os.Exit(1)
+	}
 	// +kubebuilder:scaffold:builder
 
 	log.Info("starting manager")

system_tests/queue_system_tests.go

@@ -78,6 +78,15 @@ var _ = Describe("Queue Controller", func() {
 		By("setting status.observedGeneration")
 		Expect(updatedQueue.Status.ObservedGeneration).To(Equal(updatedQueue.GetGeneration()))
 
+		By("not allowing certain updates")
+		updateQ := topologyv1alpha1.Queue{}
+		Expect(k8sClient.Get(ctx, types.NamespacedName{Name: q.Name, Namespace: q.Namespace}, &updateQ)).To(Succeed())
+		updateQ.Spec.Name = "a-new-name"
+		Expect(k8sClient.Update(ctx, &updateQ).Error()).To(ContainSubstring("spec.name: Forbidden: updates on name, vhost, and rabbitmqClusterReference are all forbidden"))
+		updateQ.Spec.Name = q.Spec.Name
+		updateQ.Spec.Type = "classic"
+		Expect(k8sClient.Update(ctx, &updateQ).Error()).To(ContainSubstring("spec.type: Invalid value: \"classic\": queue type cannot be updated"))
+
 		By("deleting queue")
 		Expect(k8sClient.Delete(ctx, q)).To(Succeed())
 		var err error