Add user limits.

api/v1beta1/user_types.go

@@ -37,6 +37,11 @@ type UserSpec struct {
 	//
 	// Note that this import only occurs at creation time, and is ignored once a password has been set on a User.
 	ImportCredentialsSecret *corev1.LocalObjectReference `json:"importCredentialsSecret,omitempty"`
+	// Limits to apply to a user to restrict the number of connections and channels
+	// the user can create. These limits can be used as guard rails in environments
+	// where applications cannot be trusted and monitored in detail, for example,
+	// when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+	UserLimits *UserLimits `json:"limits,omitempty"`
 }
 
 // UserStatus defines the observed state of User.
@@ -56,6 +61,17 @@ type UserStatus struct {
 // +kubebuilder:validation:Enum=management;policymaker;monitoring;administrator
 type UserTag string
 
+// Limits to apply to a user to restrict the number of connections and channels
+// the user can create. These limits can be used as guard rails in environments
+// where applications cannot be trusted and monitored in detail, for example,
+// when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+type UserLimits struct {
+	// Limits how many connections the user can open.
+	Connections *int32 `json:"connections,omitempty"`
+	// Limits how many AMQP 0.9.1 channels the user can open.
+	Channels *int32 `json:"channels,omitempty"`
+}
+
 // +genclient
 // +kubebuilder:object:root=true
 // +kubebuilder:resource:categories=rabbitmq

api/v1beta1/user_types_test.go

@@ -92,4 +92,51 @@ var _ = Describe("user spec", func() {
 		})
 	})
 
+	When("creating a user with limits", func() {
+		var user User
+		var username string
+		var userLimits UserLimits
+		var connections, channels int32
+
+		JustBeforeEach(func() {
+			user = User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      username,
+					Namespace: namespace,
+				},
+				Spec: UserSpec{
+					RabbitmqClusterReference: RabbitmqClusterReference{
+						Name: "some-cluster",
+					},
+					UserLimits: &userLimits,
+				},
+			}
+		})
+
+		When("creating a user with valid limits", func() {
+			BeforeEach(func() {
+				username = "limits-user"
+				connections = 5
+				channels = 10
+				userLimits = UserLimits{
+					Connections: &connections,
+					Channels:    &channels,
+				}
+			})
+			It("successfully creates the user", func() {
+				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
+				fetchedUser := &User{}
+				Expect(k8sClient.Get(ctx, types.NamespacedName{
+					Name:      user.Name,
+					Namespace: user.Namespace,
+				}, fetchedUser)).To(Succeed())
+				Expect(fetchedUser.Spec.RabbitmqClusterReference).To(Equal(RabbitmqClusterReference{
+					Name: "some-cluster",
+				}))
+				Expect(fetchedUser.Spec.UserLimits).NotTo(BeNil())
+				Expect(*fetchedUser.Spec.UserLimits.Connections).To(Equal(connections))
+				Expect(*fetchedUser.Spec.UserLimits.Channels).To(Equal(channels))
+			})
+		})
+	})
 })

config/crd/bases/rabbitmq.com_users.yaml

@@ -64,6 +64,23 @@ spec:
                     type: string
                 type: object
                 x-kubernetes-map-type: atomic
+              limits:
+                description: |-
+                  Limits to apply to a user to restrict the number of connections and channels
+                  the user can create. These limits can be used as guard rails in environments
+                  where applications cannot be trusted and monitored in detail, for example,
+                  when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+                properties:
+                  channels:
+                    description: Limits how many AMQP 0.9.1 channels the user can
+                      open.
+                    format: int32
+                    type: integer
+                  connections:
+                    description: Limits how many connections the user can open.
+                    format: int32
+                    type: integer
+                type: object
               rabbitmqClusterReference:
                 description: |-
                   Reference to the RabbitmqCluster that the user will be created for. This cluster must

controllers/binding_controller_test.go

@@ -95,6 +95,10 @@ var _ = Describe("bindingController", func() {
 	})
 
 	When("creating a binding", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &binding)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				bindingName = "test-binding-http-error"

controllers/exchange_controller_test.go

@@ -97,6 +97,10 @@ var _ = Describe("exchange-controller", func() {
 	})
 
 	Context("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &exchange)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				exchangeName = "test-http-error"
@@ -150,6 +154,7 @@ var _ = Describe("exchange-controller", func() {
 			})
 		})
 	})
+
 	Context("LastTransitionTime", func() {
 		BeforeEach(func() {
 			exchangeName = "test-last-transition-time"
@@ -158,6 +163,11 @@ var _ = Describe("exchange-controller", func() {
 				StatusCode: http.StatusCreated,
 			}, nil)
 		})
+
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &exchange)).To(Succeed())
+		})
+
 		It("changes only if status changes", func() {
 			By("setting LastTransitionTime when transitioning to status Ready=true")
 			Expect(k8sClient.Create(ctx, &exchange)).To(Succeed())

controllers/federation_controller_test.go

@@ -98,6 +98,10 @@ var _ = Describe("federation-controller", func() {
 	})
 
 	When("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &federation)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				federationName = "test-federation-http-error"

controllers/operatorpolicy_controller_test.go

@@ -101,6 +101,10 @@ var _ = Describe("operatorpolicy-controller", func() {
 	})
 
 	Context("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &policy)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				policyName = "test-http-error"

controllers/permission_controller_test.go

@@ -101,6 +101,10 @@ var _ = Describe("permission-controller", func() {
 		})
 
 		Context("creation", func() {
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &permission)).To(Succeed())
+			})
+
 			When("the RabbitMQ Client returns a HTTP error response", func() {
 				BeforeEach(func() {
 					permissionName = "test-with-username-http-error"
@@ -261,9 +265,21 @@ var _ = Describe("permission-controller", func() {
 				Status:     "204 No Content",
 				StatusCode: http.StatusNoContent,
 			}, nil)
+			fakeRabbitMQClient.PutUserLimitsReturns(&http.Response{
+				Status:     "201 Created",
+				StatusCode: http.StatusCreated,
+			}, nil)
+			fakeRabbitMQClient.DeleteUserLimitsReturns(&http.Response{
+				Status:     "204 No Content",
+				StatusCode: http.StatusNoContent,
+			}, nil)
 		})
 
 		Context("creation", func() {
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &permission)).To(Succeed())
+			})
+
 			When("user not exist", func() {
 				BeforeEach(func() {
 					permissionName = "test-with-userref-create-not-exist"
@@ -407,12 +423,16 @@ var _ = Describe("permission-controller", func() {
 			})
 		})
 
-		Context("ownerref", func() {
+		When("the user already exists", func() {
 			BeforeEach(func() {
 				permissionName = "ownerref-with-userref-test"
 				userName = "example-ownerref"
 			})
 
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &user)).To(Succeed())
+			})
+
 			It("sets the correct deletion ownerref to the object", func() {
 				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
 				user.Status.Username = userName

controllers/policy_controller_test.go

@@ -100,6 +100,10 @@ var _ = Describe("policy-controller", func() {
 	})
 
 	Context("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &policy)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				policyName = "test-http-error"

controllers/queue_controller_test.go

@@ -96,6 +96,10 @@ var _ = Describe("queue-controller", func() {
 	})
 
 	Context("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &queue)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				queueName = "test-http-error"

controllers/schemareplication_controller_test.go

@@ -103,6 +103,10 @@ var _ = Describe("schema-replication-controller", func() {
 	})
 
 	When("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &replication)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				replicationName = "test-replication-http-error"
@@ -257,6 +261,7 @@ var _ = Describe("schema-replication-controller", func() {
 		})
 
 		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &replication)).To(Succeed())
 			rabbitmqclient.SecretStoreClientProvider = rabbitmqclient.GetSecretStoreClient
 		})
 

controllers/shovel_controller_test.go

@@ -100,6 +100,10 @@ var _ = Describe("shovel-controller", func() {
 	})
 
 	When("creation", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &shovel)).To(Succeed())
+		})
+
 		When("the RabbitMQ Client returns a HTTP error response", func() {
 			BeforeEach(func() {
 				shovelName = "test-shovel-http-error"

controllers/super_stream_controller_test.go

@@ -124,6 +124,10 @@ var _ = Describe("super-stream-controller", func() {
 		})
 
 		Context("creation", func() {
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &superStream)).To(Succeed())
+			})
+
 			When("an underlying resource is deleted", func() {
 				JustBeforeEach(func() {
 					Expect(k8sClient.Create(ctx, &superStream)).To(Succeed())

controllers/topicpermission_controller_test.go

@@ -103,6 +103,10 @@ var _ = Describe("topicpermission-controller", func() {
 		})
 
 		Context("creation", func() {
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &topicperm)).To(Succeed())
+			})
+
 			When("the RabbitMQ Client returns a HTTP error response", func() {
 				BeforeEach(func() {
 					name = "test-with-username-http-error"
@@ -284,6 +288,10 @@ var _ = Describe("topicpermission-controller", func() {
 		})
 
 		Context("creation", func() {
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &topicperm)).To(Succeed())
+			})
+
 			When("user not exist", func() {
 				BeforeEach(func() {
 					name = "test-with-userref-create-not-exist"
@@ -457,6 +465,11 @@ var _ = Describe("topicpermission-controller", func() {
 				userName = "topic-perm-topic-perm-user"
 			})
 
+			AfterEach(func() {
+				Expect(k8sClient.Delete(ctx, &user)).To(Succeed())
+				Expect(k8sClient.Delete(ctx, &topicperm)).To(Succeed())
+			})
+
 			It("sets the correct deletion ownerref to the object", func() {
 				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
 				user.Status.Username = userName