Skip to content

Commit 31fbfbc

Browse files
ChunyiLyuChunyiLyu
authored
Merge pull request #621 from rabbitmq/parallel
Set Pod Management to 'Parallel' and disallow cluster scale down entirely
2 parents adc90a6 + b09b47a commit 31fbfbc

File tree

10 files changed

+167
-35
lines changed

10 files changed

+167
-35
lines changed

‎config/crd/bases/rabbitmq.com_rabbitmqclusters.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ apiVersion: apiextensions.k8s.io/v1
1010
kind: CustomResourceDefinition
1111
metadata:
1212
annotations:
13-
controller-gen.kubebuilder.io/version: v0.4.1
13+
controller-gen.kubebuilder.io/version: v0.5.0
1414
creationTimestamp: null
1515
name: rabbitmqclusters.rabbitmq.com
1616
spec:

‎controllers/rabbitmqcluster_controller.go‎

Lines changed: 30 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ import (
2222

2323
"github.com/rabbitmq/cluster-operator/internal/resource"
2424
"github.com/rabbitmq/cluster-operator/internal/status"
25-
"k8s.io/apimachinery/pkg/api/errors"
25+
k8serrors "k8s.io/apimachinery/pkg/api/errors"
2626
"k8s.io/client-go/kubernetes"
2727
"k8s.io/client-go/rest"
2828
"k8s.io/client-go/tools/record"
@@ -89,7 +89,7 @@ func (r *RabbitmqClusterReconciler) Reconcile(ctx context.Context, req ctrl.Requ
8989

9090
if client.IgnoreNotFound(err) != nil {
9191
return ctrl.Result{}, err
92-
} else if errors.IsNotFound(err) {
92+
} else if k8serrors.IsNotFound(err) {
9393
// No need to requeue if the resource no longer exists
9494
return ctrl.Result{}, nil
9595
}
@@ -163,13 +163,31 @@ func (r *RabbitmqClusterReconciler) Reconcile(ctx context.Context, req ctrl.Requ
163163

164164
// only StatefulSetBuilder returns true
165165
if builder.UpdateMayRequireStsRecreate() {
166-
if err = r.reconcilePVC(ctx, builder, rabbitmqCluster, resource); err != nil {
167-
rabbitmqCluster.Status.SetCondition(status.ReconcileSuccess, corev1.ConditionFalse, "FailedReconcilePVC", err.Error())
168-
if statusErr := r.Status().Update(ctx, rabbitmqCluster); statusErr != nil {
169-
logger.Error(statusErr, "Failed to update ReconcileSuccess condition state")
170-
}
166+
sts := resource.(*appsv1.StatefulSet)
167+
168+
current, err := r.statefulSet(ctx, rabbitmqCluster)
169+
if client.IgnoreNotFound(err) != nil {
171170
return ctrl.Result{}, err
172171
}
172+
173+
// only checks for PVC expansion and scale down if statefulSet is created
174+
// else continue to CreateOrUpdate()
175+
if !k8serrors.IsNotFound(err) {
176+
if err := builder.Update(sts); err != nil {
177+
return ctrl.Result{}, err
178+
}
179+
if err = r.reconcilePVC(ctx, rabbitmqCluster, current, sts); err != nil {
180+
rabbitmqCluster.Status.SetCondition(status.ReconcileSuccess, corev1.ConditionFalse, "FailedReconcilePVC", err.Error())
181+
if statusErr := r.Status().Update(ctx, rabbitmqCluster); statusErr != nil {
182+
logger.Error(statusErr, "Failed to update ReconcileSuccess condition state")
183+
}
184+
return ctrl.Result{}, err
185+
}
186+
if r.scaleDown(ctx, rabbitmqCluster, current, sts) {
187+
// return when cluster scale down detected; unsupported operation
188+
return ctrl.Result{}, nil
189+
}
190+
}
173191
}
174192

175193
var operationResult controllerutil.OperationResult
@@ -269,7 +287,7 @@ func (r *RabbitmqClusterReconciler) updateStatus(ctx context.Context, rmq *rabbi
269287

270288
if !reflect.DeepEqual(rmq.Status.Conditions, oldConditions) {
271289
if err = r.Status().Update(ctx, rmq); err != nil {
272-
if errors.IsConflict(err) {
290+
if k8serrors.IsConflict(err) {
273291
logger.Info("failed to update status because of conflict; requeueing...",
274292
"namespace", rmq.Namespace,
275293
"name", rmq.Name)
@@ -287,17 +305,17 @@ func (r *RabbitmqClusterReconciler) getChildResources(ctx context.Context, rmq *
287305

288306
if err := r.Client.Get(ctx,
289307
types.NamespacedName{Name: rmq.ChildResourceName("server"), Namespace: rmq.Namespace},
290-
sts); err != nil && !errors.IsNotFound(err) {
308+
sts); err != nil && !k8serrors.IsNotFound(err) {
291309
return nil, err
292-
} else if errors.IsNotFound(err) {
310+
} else if k8serrors.IsNotFound(err) {
293311
sts = nil
294312
}
295313

296314
if err := r.Client.Get(ctx,
297315
types.NamespacedName{Name: rmq.ChildResourceName(resource.ServiceSuffix), Namespace: rmq.Namespace},
298-
endPoints); err != nil && !errors.IsNotFound(err) {
316+
endPoints); err != nil && !k8serrors.IsNotFound(err) {
299317
return nil, err
300-
} else if errors.IsNotFound(err) {
318+
} else if k8serrors.IsNotFound(err) {
301319
endPoints = nil
302320
}
303321

‎controllers/rabbitmqcluster_controller_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -937,7 +937,7 @@ var _ = Describe("RabbitmqClusterController", func() {
937937

938938
It("updates", func() {
939939
Expect(updateWithRetry(cluster, func(r *rabbitmqv1beta1.RabbitmqCluster) {
940-
cluster.Spec.Override.StatefulSet.Spec.Replicas = pointer.Int32Ptr(5)
940+
cluster.Spec.Override.StatefulSet.Spec.Replicas = pointer.Int32Ptr(15)
941941
cluster.Spec.Override.StatefulSet.Spec.Template.Spec.Containers = []corev1.Container{
942942
{
943943
Name: "additional-container-2",
@@ -949,7 +949,7 @@ var _ = Describe("RabbitmqClusterController", func() {
949949
Eventually(func() int32 {
950950
sts := statefulSet(ctx, cluster)
951951
return *sts.Spec.Replicas
952-
}, 3).Should(Equal(int32(5)))
952+
}, 3).Should(Equal(int32(15)))
953953

954954
Eventually(func() string {
955955
sts := statefulSet(ctx, cluster)

‎controllers/reconcile_persistence.go‎

Lines changed: 1 addition & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"fmt"
77
"github.com/go-logr/logr"
88
rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
9-
"github.com/rabbitmq/cluster-operator/internal/resource"
109
appsv1 "k8s.io/api/apps/v1"
1110
corev1 "k8s.io/api/core/v1"
1211
k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -18,23 +17,7 @@ import (
1817
"time"
1918
)
2019

21-
func (r *RabbitmqClusterReconciler) reconcilePVC(ctx context.Context, builder resource.ResourceBuilder, cluster *rabbitmqv1beta1.RabbitmqCluster, resource client.Object) error {
22-
logger := ctrl.LoggerFrom(ctx)
23-
24-
sts := resource.(*appsv1.StatefulSet)
25-
current, err := r.statefulSet(ctx, cluster)
26-
27-
if client.IgnoreNotFound(err) != nil {
28-
return err
29-
} else if k8serrors.IsNotFound(err) {
30-
logger.Info("statefulSet not created yet, skipping checks to expand PersistentVolumeClaims")
31-
return nil
32-
}
33-
34-
if err := builder.Update(sts); err != nil {
35-
return err
36-
}
37-
20+
func (r *RabbitmqClusterReconciler) reconcilePVC(ctx context.Context, cluster *rabbitmqv1beta1.RabbitmqCluster, current, sts *appsv1.StatefulSet) error {
3821
resize, err := r.needsPVCResize(current, sts)
3922
if err != nil {
4023
return err

‎controllers/reconcile_scale_down.go‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
package controllers
2+
3+
import (
4+
"context"
5+
"errors"
6+
"github.com/go-logr/logr"
7+
"github.com/rabbitmq/cluster-operator/api/v1beta1"
8+
"github.com/rabbitmq/cluster-operator/internal/status"
9+
appsv1 "k8s.io/api/apps/v1"
10+
corev1 "k8s.io/api/core/v1"
11+
)
12+
13+
// cluster scale down not supported
14+
// log error, publish warning event, and set ReconcileSuccess to false when scale down request detected
15+
func (r *RabbitmqClusterReconciler) scaleDown(ctx context.Context, cluster *v1beta1.RabbitmqCluster, current, sts *appsv1.StatefulSet) bool {
16+
logger := logr.FromContext(ctx)
17+
18+
currentReplicas := *current.Spec.Replicas
19+
desiredReplicas := *sts.Spec.Replicas
20+
if currentReplicas > desiredReplicas {
21+
msg := "Cluster Scale down not supported"
22+
reason := "UnsupportedOperation"
23+
logger.Error(errors.New(reason), msg)
24+
r.Recorder.Event(cluster, corev1.EventTypeWarning, reason, msg)
25+
cluster.Status.SetCondition(status.ReconcileSuccess, corev1.ConditionFalse, reason, msg)
26+
if statusErr := r.Status().Update(ctx, cluster); statusErr != nil {
27+
logger.Error(statusErr, "Failed to update ReconcileSuccess condition state")
28+
}
29+
return true
30+
}
31+
return false
32+
}

‎controllers/reconcile_scale_down_test.go‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
package controllers_test
2+
3+
import (
4+
"context"
5+
"fmt"
6+
. "github.com/onsi/ginkgo"
7+
. "github.com/onsi/gomega"
8+
rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
9+
"github.com/rabbitmq/cluster-operator/internal/status"
10+
apierrors "k8s.io/apimachinery/pkg/api/errors"
11+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
12+
"k8s.io/apimachinery/pkg/types"
13+
"k8s.io/utils/pointer"
14+
runtimeClient "sigs.k8s.io/controller-runtime/pkg/client"
15+
)
16+
17+
var _ = Describe("Cluster scale down", func() {
18+
var (
19+
cluster *rabbitmqv1beta1.RabbitmqCluster
20+
defaultNamespace = "default"
21+
ctx = context.Background()
22+
)
23+
24+
AfterEach(func() {
25+
Expect(client.Delete(ctx, cluster)).To(Succeed())
26+
Eventually(func() bool {
27+
rmq := &rabbitmqv1beta1.RabbitmqCluster{}
28+
err := client.Get(ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, rmq)
29+
return apierrors.IsNotFound(err)
30+
}, 5).Should(BeTrue())
31+
})
32+
33+
It("does not allow cluster scale down", func() {
34+
By("not updating statefulSet replicas", func() {
35+
cluster = &rabbitmqv1beta1.RabbitmqCluster{
36+
ObjectMeta: metav1.ObjectMeta{
37+
Name: "rabbitmq-shrink",
38+
Namespace: defaultNamespace,
39+
},
40+
Spec: rabbitmqv1beta1.RabbitmqClusterSpec{
41+
Replicas: pointer.Int32Ptr(5),
42+
},
43+
}
44+
Expect(client.Create(ctx, cluster)).To(Succeed())
45+
waitForClusterCreation(ctx, cluster, client)
46+
47+
Expect(updateWithRetry(cluster, func(r *rabbitmqv1beta1.RabbitmqCluster) {
48+
r.Spec.Replicas = pointer.Int32Ptr(3)
49+
})).To(Succeed())
50+
Consistently(func() int32 {
51+
sts, err := clientSet.AppsV1().StatefulSets(defaultNamespace).Get(ctx, cluster.ChildResourceName("server"), metav1.GetOptions{})
52+
Expect(err).NotTo(HaveOccurred())
53+
return *sts.Spec.Replicas
54+
}, 10, 1).Should(Equal(int32(5)))
55+
})
56+
57+
By("setting 'Warning' events", func() {
58+
Expect(aggregateEventMsgs(ctx, cluster, "UnsupportedOperation")).To(
59+
ContainSubstring("Cluster Scale down not supported"))
60+
})
61+
62+
By("setting ReconcileSuccess to 'false'", func() {
63+
Eventually(func() string {
64+
rabbit := &rabbitmqv1beta1.RabbitmqCluster{}
65+
Expect(client.Get(ctx, runtimeClient.ObjectKey{
66+
Name: cluster.Name,
67+
Namespace: defaultNamespace,
68+
}, rabbit)).To(Succeed())
69+
70+
for i := range rabbit.Status.Conditions {
71+
if rabbit.Status.Conditions[i].Type == status.ReconcileSuccess {
72+
return fmt.Sprintf(
73+
"ReconcileSuccess status: %s, with reason: %s and message: %s",
74+
rabbit.Status.Conditions[i].Status,
75+
rabbit.Status.Conditions[i].Reason,
76+
rabbit.Status.Conditions[i].Message)
77+
}
78+
}
79+
return "ReconcileSuccess status: condition not present"
80+
}, 5).Should(Equal("ReconcileSuccess status: False, " +
81+
"with reason: UnsupportedOperation " +
82+
"and message: Cluster Scale down not supported"))
83+
})
84+
})
85+
})

‎internal/resource/configmap.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,9 @@ cluster_formation.k8s.host = kubernetes.default
3434
cluster_formation.k8s.address_type = hostname
3535
cluster_partition_handling = pause_minority
3636
queue_master_locator = min-masters
37-
disk_free_limit.absolute = 2GB`
37+
disk_free_limit.absolute = 2GB
38+
cluster_formation.randomized_startup_delay_range.min = 5
39+
cluster_formation.randomized_startup_delay_range.max = 30`
3840

3941
defaultTLSConf = `
4042
ssl_options.certfile = /etc/rabbitmq-tls/tls.crt

‎internal/resource/configmap_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@ cluster_formation.k8s.address_type = hostname
3333
cluster_partition_handling = pause_minority
3434
queue_master_locator = min-masters
3535
disk_free_limit.absolute = 2GB
36+
cluster_formation.randomized_startup_delay_range.min = 5
37+
cluster_formation.randomized_startup_delay_range.max = 30
3638
cluster_name = ` + instanceName)
3739
}
3840

‎internal/resource/statefulset.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ func (builder *StatefulSetBuilder) Build() (client.Object, error) {
6767
MatchLabels: metadata.LabelSelector(builder.Instance.Name),
6868
},
6969
VolumeClaimTemplates: pvc,
70+
PodManagementPolicy: appsv1.ParallelPodManagement,
7071
},
7172
}
7273

‎internal/resource/statefulset_test.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@ var _ = Describe("StatefulSet", func() {
6565

6666
Expect(statefulSet.Spec.ServiceName).To(Equal(instance.ChildResourceName("nodes")))
6767
})
68+
6869
It("adds the correct label selector", func() {
6970
obj, err := stsBuilder.Build()
7071
Expect(err).NotTo(HaveOccurred())
@@ -74,7 +75,15 @@ var _ = Describe("StatefulSet", func() {
7475
Expect(labels["app.kubernetes.io/name"]).To(Equal(instance.Name))
7576
})
7677

77-
It("references the storageclassname when specified", func() {
78+
It("sets pod management policy to 'Parallel' ", func() {
79+
obj, err := stsBuilder.Build()
80+
Expect(err).NotTo(HaveOccurred())
81+
statefulSet := obj.(*appsv1.StatefulSet)
82+
83+
Expect(statefulSet.Spec.PodManagementPolicy).To(Equal(appsv1.ParallelPodManagement))
84+
})
85+
86+
It("references the storage class name when specified", func() {
7887
storageClassName := "my-storage-class"
7988
builder.Instance.Spec.Persistence.StorageClassName = &storageClassName
8089

0 commit comments

Comments
 (0)