Add user limits. (#962)

* Add user limits. * Clean up after tests. * Delete existing limits if they are removed from a user. * Only create user limits when limits exist to create.

Author: MirahImage

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))
+})
+})
+})
 })

api/v1beta1/zz_generated.deepcopy.go

@@ -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

config/crd/bases/rabbitmq.com_users.yaml

@@ -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/binding_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/exchange_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/federation_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/operatorpolicy_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/permission_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"