Better shovel config validation

- accepted protocols for source and destination are amqp10 and amqp091. If empty the plugin defaults to amqp091: https://github.com/rabbitmq/rabbitmq-server/blob/f83ed85fedadc3dc29595d0b0d109db0b8c67f17/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl#L489-L500 - when any protocol is set to amqp10, address needs to be provided. Destination validation: https://github.com/rabbitmq/rabbitmq-server/blob/f83ed85fedadc3dc29595d0b0d109db0b8c67f17/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl#L153 Source validation: https://github.com/rabbitmq/rabbitmq-server/blob/main/deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl#L123
rabbitmq · May 24, 2023 · d72763f · d72763f
1 parent a1a5f32
commit d72763f
Show file tree

Hide file tree

Showing 6 changed files with 108 additions and 5 deletions.
diff --git a/api/v1beta1/shovel_types.go b/api/v1beta1/shovel_types.go
@@ -35,7 +35,8 @@ type ShovelSpec struct {
 	DestinationAddress            string `json:"destAddress,omitempty"`
 	DestinationExchange           string `json:"destExchange,omitempty"`
 	DestinationExchangeKey        string `json:"destExchangeKey,omitempty"`
-	DestinationProtocol           string `json:"destProtocol,omitempty"`
+	// +kubebuilder:validation:Enum=amqp091;amqp10
+	DestinationProtocol string `json:"destProtocol,omitempty"`
 	// +kubebuilder:validation:Type=object
 	// +kubebuilder:pruning:PreserveUnknownFields
 	DestinationApplicationProperties *runtime.RawExtension `json:"destApplicationProperties,omitempty"`
@@ -56,8 +57,9 @@ type ShovelSpec struct {
 	SourceExchange                string                `json:"srcExchange,omitempty"`
 	SourceExchangeKey             string                `json:"srcExchangeKey,omitempty"`
 	SourcePrefetchCount           int                   `json:"srcPrefetchCount,omitempty"`
-	SourceProtocol                string                `json:"srcProtocol,omitempty"`
-	SourceQueue                   string                `json:"srcQueue,omitempty"`
+	// +kubebuilder:validation:Enum=amqp091;amqp10
+	SourceProtocol string `json:"srcProtocol,omitempty"`
+	SourceQueue    string `json:"srcQueue,omitempty"`
 	// +kubebuilder:validation:Type=object
 	// +kubebuilder:pruning:PreserveUnknownFields
 	SourceConsumerArgs *runtime.RawExtension `json:"srcConsumerArgs,omitempty"`

diff --git a/api/v1beta1/shovel_types_test.go b/api/v1beta1/shovel_types_test.go
@@ -137,4 +137,48 @@ var _ = Describe("Shovel spec", func() {
 			Expect(k8sClient.Create(ctx, &shovel)).To(MatchError(`Shovel.rabbitmq.com "an-invalid-ackmode" is invalid: spec.ackMode: Unsupported value: "an-invalid-ackmode": supported values: "on-confirm", "on-publish", "no-ack"`))
 		})
 	})
+
+	When("creating a shovel with unsupported protocol", func() {
+		It("fails with validation errors", func() {
+			shovel := Shovel{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "an-invalid-destprotocol",
+					Namespace: namespace,
+				},
+				Spec: ShovelSpec{
+					Name: "an-invalid-destprotocol",
+					RabbitmqClusterReference: RabbitmqClusterReference{
+						Name: "some-cluster",
+					},
+					UriSecret: &corev1.LocalObjectReference{
+						Name: "a-secret",
+					},
+					SourceProtocol:      "amqp091",
+					DestinationProtocol: "stomp",
+				}}
+			Expect(k8sClient.Create(ctx, &shovel)).To(HaveOccurred())
+			Expect(k8sClient.Create(ctx, &shovel)).To(MatchError(`Shovel.rabbitmq.com "an-invalid-destprotocol" is invalid: spec.destProtocol: Unsupported value: "stomp": supported values: "amqp091", "amqp10"`))
+		})
+
+		It("fails with validation errors", func() {
+			shovel := Shovel{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "an-invalid-srcprotocol",
+					Namespace: namespace,
+				},
+				Spec: ShovelSpec{
+					Name: "an-invalid-srcprotocol",
+					RabbitmqClusterReference: RabbitmqClusterReference{
+						Name: "some-cluster",
+					},
+					UriSecret: &corev1.LocalObjectReference{
+						Name: "a-secret",
+					},
+					SourceProtocol:      "mqtt",
+					DestinationProtocol: "amqp10",
+				}}
+			Expect(k8sClient.Create(ctx, &shovel)).To(HaveOccurred())
+			Expect(k8sClient.Create(ctx, &shovel)).To(MatchError(`Shovel.rabbitmq.com "an-invalid-srcprotocol" is invalid: spec.srcProtocol: Unsupported value: "mqtt": supported values: "amqp091", "amqp10"`))
+		})
+	})
 })
diff --git a/api/v1beta1/shovel_webhook.go b/api/v1beta1/shovel_webhook.go
@@ -22,6 +22,9 @@ var _ webhook.Validator = &Shovel{}
 // 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 (s *Shovel) ValidateCreate() error {
+	if err := s.amqp10Validate(); err != nil {
+		return err
+	}
 	return s.Spec.RabbitmqClusterReference.ValidateOnCreate(s.GroupResource(), s.Name)
 }
 
@@ -32,6 +35,10 @@ func (s *Shovel) ValidateUpdate(old runtime.Object) error {
 		return apierrors.NewBadRequest(fmt.Sprintf("expected a shovel but got a %T", oldShovel))
 	}
 
+	if err := s.amqp10Validate(); err != nil {
+		return err
+	}
+
 	detailMsg := "updates on name, vhost and rabbitmqClusterReference are all forbidden"
 	if s.Spec.Name != oldShovel.Spec.Name {
 		return apierrors.NewForbidden(s.GroupResource(), s.Name,
@@ -53,3 +60,18 @@ func (s *Shovel) ValidateUpdate(old runtime.Object) error {
 func (s *Shovel) ValidateDelete() error {
 	return nil
 }
+
+func (s *Shovel) amqp10Validate() error {
+	var errorList field.ErrorList
+	if s.Spec.SourceProtocol == "amqp10" && s.Spec.SourceAddress == "" {
+		errorList = append(errorList, field.Required(field.NewPath("spec", "srcAddress"),
+			"must specify spec.srcAddress when spec.srcProtocol is amqp10"))
+		return apierrors.NewInvalid(GroupVersion.WithKind("Shovel").GroupKind(), s.Name, errorList)
+	}
+	if s.Spec.DestinationProtocol == "amqp10" && s.Spec.DestinationAddress == "" {
+		errorList = append(errorList, field.Required(field.NewPath("spec", "destAddress"),
+			"must specify spec.destAddress when spec.destProtocol is amqp10"))
+		return apierrors.NewInvalid(GroupVersion.WithKind("Shovel").GroupKind(), s.Name, errorList)
+	}
+	return nil
+}
diff --git a/api/v1beta1/shovel_webhook_test.go b/api/v1beta1/shovel_webhook_test.go
@@ -63,6 +63,20 @@ var _ = Describe("shovel webhook", func() {
 			notAllowed.Spec.RabbitmqClusterReference.ConnectionSecret = nil
 			Expect(apierrors.IsForbidden(notAllowed.ValidateCreate())).To(BeTrue())
 		})
+
+		It("spec.srcAddress must be set if spec.srcProtocol is amqp10", func() {
+			notValid := shovel.DeepCopy()
+			notValid.Spec.SourceProtocol = "amqp10"
+			notValid.Spec.SourceAddress = ""
+			Expect(apierrors.IsInvalid(notValid.ValidateCreate())).To(BeTrue())
+		})
+
+		It("spec.destAddress must be set if spec.destProtocol is amqp10", func() {
+			notValid := shovel.DeepCopy()
+			notValid.Spec.DestinationProtocol = "amqp10"
+			notValid.Spec.DestinationAddress = ""
+			Expect(apierrors.IsInvalid(notValid.ValidateCreate())).To(BeTrue())
+		})
 	})
 
 	Context("ValidateUpdate", func() {
@@ -86,6 +100,20 @@ var _ = Describe("shovel webhook", func() {
 			Expect(apierrors.IsForbidden(newShovel.ValidateUpdate(&shovel))).To(BeTrue())
 		})
 
+		It("spec.srcAddress must be set if spec.srcProtocol is amqp10", func() {
+			newShovel := shovel.DeepCopy()
+			newShovel.Spec.SourceProtocol = "amqp10"
+			newShovel.Spec.SourceAddress = ""
+			Expect(apierrors.IsInvalid(newShovel.ValidateUpdate(&shovel))).To(BeTrue())
+		})
+
+		It("spec.destAddress must be set if spec.destProtocol is amqp10", func() {
+			newShovel := shovel.DeepCopy()
+			newShovel.Spec.DestinationProtocol = "amqp10"
+			newShovel.Spec.DestinationAddress = ""
+			Expect(apierrors.IsInvalid(newShovel.ValidateUpdate(&shovel))).To(BeTrue())
+		})
+
 		It("does not allow updates on rabbitmqClusterReference.connectionSecret", func() {
 			connectionScr := Shovel{
 				ObjectMeta: metav1.ObjectMeta{
@@ -180,6 +208,7 @@ var _ = Describe("shovel webhook", func() {
 			newShovel.Spec.DestinationProperties = &runtime.RawExtension{Raw: []byte(`{"key": "new"}`)}
 			Expect(newShovel.ValidateUpdate(&shovel)).To(Succeed())
 		})
+
 		It("allows updates on DestinationProtocol", func() {
 			newShovel := shovel.DeepCopy()
 			newShovel.Spec.DestinationProtocol = "new"

diff --git a/config/crd/bases/rabbitmq.com_shovels.yaml b/config/crd/bases/rabbitmq.com_shovels.yaml
@@ -68,6 +68,9 @@ spec:
                 type: object
                 x-kubernetes-preserve-unknown-fields: true
               destProtocol:
+                enum:
+                - amqp091
+                - amqp10
                 type: string
               destPublishProperties:
                 type: object
@@ -120,6 +123,9 @@ spec:
               srcPrefetchCount:
                 type: integer
               srcProtocol:
+                enum:
+                - amqp091
+                - amqp10
                 type: string
               srcQueue:
                 type: string

diff --git a/system_tests/shovel_system_test.go b/system_tests/shovel_system_test.go
@@ -43,9 +43,9 @@ var _ = Describe("Shovel", func() {
 				Namespace: namespace,
 			},
 			Spec: topology.ShovelSpec{
-				UriSecret:   &corev1.LocalObjectReference{Name: shovelSecret.Name},
+				UriSecret:         &corev1.LocalObjectReference{Name: shovelSecret.Name},
 				SourceDeleteAfter: "never",
-				AckMode: "no-ack",
+				AckMode:           "no-ack",
 				RabbitmqClusterReference: topology.RabbitmqClusterReference{
 					Name: rmq.Name,
 				},