5555)
5656
5757const (
58- ownerKey = ".metadata.controller"
59- ownerKind = "RabbitmqCluster"
60- deletionFinalizer = "deletion.finalizers.rabbitmqclusters.rabbitmq.com"
58+ ownerKey = ".metadata.controller"
59+ ownerKind = "RabbitmqCluster"
60+ deletionFinalizer = "deletion.finalizers.rabbitmqclusters.rabbitmq.com"
61+ pluginsUpdateAnnotation = "rabbitmq.com/pluginsUpdatedAt"
6162)
6263
6364// RabbitmqClusterReconciler reconciles a RabbitmqCluster object
@@ -183,7 +184,6 @@ func (r *RabbitmqClusterReconciler) Reconcile(req ctrl.Request) (ctrl.Result, er
183184 operationResult , apiError = controllerutil .CreateOrUpdate (ctx , r , resource , func () error {
184185 return builder .Update (resource )
185186 })
186-
187187 return apiError
188188 })
189189 r .logAndRecordOperationResult (rabbitmqCluster , resource , operationResult , err )
@@ -194,10 +194,10 @@ func (r *RabbitmqClusterReconciler) Reconcile(req ctrl.Request) (ctrl.Result, er
194194 "namespace" , rabbitmqCluster .Namespace ,
195195 "name" , rabbitmqCluster .Name )
196196 }
197-
198197 return ctrl.Result {}, err
199198 }
200199
200+ r .annotatePluginsConfigMapIfUpdated (ctx , builder , operationResult , rabbitmqCluster )
201201 if restarted := r .restartStatefulSetIfNeeded (ctx , builder , operationResult , rabbitmqCluster ); restarted {
202202 return ctrl.Result {RequeueAfter : time .Second * 10 }, nil
203203 }
@@ -216,18 +216,13 @@ func (r *RabbitmqClusterReconciler) Reconcile(req ctrl.Request) (ctrl.Result, er
216216 return ctrl.Result {}, err
217217 }
218218
219- if ok , err := r .allReplicasReadyAndUpdated (ctx , rabbitmqCluster ); ! ok {
220- // only enable plugins when all pods of the StatefulSet become ready
221- // requeue request after 10 seconds without error
222- logger .Info ("Not all replicas ready yet; requeuing request to enable plugins on RabbitmqCluster" ,
223- "namespace" , rabbitmqCluster .Namespace ,
224- "name" , rabbitmqCluster .Name )
225- return ctrl.Result {RequeueAfter : time .Second * 10 }, err
226- }
227-
228- if err := r .enablePlugins (rabbitmqCluster ); err != nil {
219+ requeueAfter , err := r .setPluginsIfNeeded (ctx , rabbitmqCluster )
220+ if err != nil {
229221 return ctrl.Result {}, err
230222 }
223+ if requeueAfter > 0 {
224+ return ctrl.Result {RequeueAfter : requeueAfter }, nil
225+ }
231226
232227 logger .Info ("Finished reconciling RabbitmqCluster" ,
233228 "namespace" , rabbitmqCluster .Namespace ,
@@ -337,60 +332,131 @@ func (r *RabbitmqClusterReconciler) restartStatefulSetIfNeeded(
337332 sts .Spec .Template .ObjectMeta .Annotations ["rabbitmq.com/lastRestartAt" ] = time .Now ().Format (time .RFC3339 )
338333 return r .Update (ctx , sts )
339334 }); err != nil {
340- msg := fmt .Sprintf ("Failed to restart StatefulSet %s of Namespace %s; rabbitmq.conf configuration may be outdated" , rmq .ChildResourceName ("server" ), rmq .Namespace )
335+ msg := fmt .Sprintf ("failed to restart StatefulSet %s of Namespace %s; rabbitmq.conf configuration may be outdated" , rmq .ChildResourceName ("server" ), rmq .Namespace )
341336 r .Log .Error (err , msg )
342337 r .Recorder .Event (rmq , corev1 .EventTypeWarning , "FailedUpdate" , msg )
343338 return false
344339 }
345340
346- msg := fmt .Sprintf ("Restarted StatefulSet %s of Namespace %s" , rmq .ChildResourceName ("server" ), rmq .Namespace )
341+ msg := fmt .Sprintf ("restarted StatefulSet %s of Namespace %s" , rmq .ChildResourceName ("server" ), rmq .Namespace )
347342 r .Log .Info (msg )
348343 r .Recorder .Event (rmq , corev1 .EventTypeNormal , "SuccessfulUpdate" , msg )
349344 return true
350345}
351346
352- func (r * RabbitmqClusterReconciler ) allReplicasReadyAndUpdated (ctx context.Context , rmq * rabbitmqv1beta1.RabbitmqCluster ) (bool , error ) {
353- sts := & appsv1.StatefulSet {}
347+ // There are 2 paths how plugins are set:
348+ // 1. When SatefulSet is (re)started, the up-to-date plugins list (ConfigMap copied by the init container) is read by RabbitMQ nodes during node start up.
349+ // 2. When the plugins ConfigMap is changed, 'rabbitmq-plugins set' updates the plugins on every node (without the need to re-start the nodes).
350+ // This method implements the 2nd path.
351+ func (r * RabbitmqClusterReconciler ) setPluginsIfNeeded (ctx context.Context , rmq * rabbitmqv1beta1.RabbitmqCluster ) (requeueAfter time.Duration , err error ) {
352+ configMap := corev1.ConfigMap {}
353+ if err := r .Get (ctx , types.NamespacedName {Namespace : rmq .Namespace , Name : rmq .ChildResourceName (resource .PluginsConfig )}, & configMap ); err != nil {
354+ return 0 , client .IgnoreNotFound (err )
355+ }
354356
355- if err := r .Get (ctx , types.NamespacedName {Name : rmq .ChildResourceName ("server" ), Namespace : rmq .Namespace }, sts ); err != nil {
356- return false , client .IgnoreNotFound (err )
357+ pluginsUpdatedAt , ok := configMap .Annotations [pluginsUpdateAnnotation ]
358+ if ! ok {
359+ return 0 , nil // plugins configMap was not updated
357360 }
358361
359- desiredReplicas := * sts .Spec .Replicas
360- if sts .Status .ReadyReplicas < desiredReplicas ||
361- sts .Status .UpdatedReplicas < desiredReplicas { // StatefulSet rolling update is still ongoing (see https://github.com/rabbitmq/cluster-operator/issues/304)
362- return false , nil
362+ annotationTime , err := time .Parse (time .RFC3339 , pluginsUpdatedAt )
363+ if err != nil {
364+ return 0 , err
365+ }
366+ if time .Since (annotationTime ).Seconds () < 2 {
367+ // plugins configMap was updated very recently
368+ // give StatefulSet controller some time to trigger restart of StatefulSet if necessary
369+ // otherwise, there would be race conditions where we exec into containers losing the connection due to pods being terminated
370+ r .Log .Info ("requeuing request to set plugins on RabbitmqCluster" ,
371+ "namespace" , rmq .Namespace ,
372+ "name" , rmq .Name )
373+ return 2 * time .Second , nil
363374 }
364375
365- return true , nil
366- }
376+ ready , err := r .allReplicasReadyAndUpdated (ctx , rmq )
377+ if err != nil {
378+ return 0 , err
379+ }
380+ if ! ready {
381+ r .Log .Info ("not all replicas ready yet; requeuing request to set plugins on RabbitmqCluster" ,
382+ "namespace" , rmq .Namespace ,
383+ "name" , rmq .Name )
384+ return 15 * time .Second , err
385+ }
367386
368- // Helper function to set the list of enabled plugins in the given RabbitmqCluster pods.
369- // `rabbitmq-plugins set` disables plugins that are not in the provided list
370- func (r * RabbitmqClusterReconciler ) enablePlugins (rmq * rabbitmqv1beta1.RabbitmqCluster ) error {
371387 plugins := resource .NewRabbitmqPlugins (rmq .Spec .Rabbitmq .AdditionalPlugins )
372388 for i := int32 (0 ); i < * rmq .Spec .Replicas ; i ++ {
373389 podName := fmt .Sprintf ("%s-%d" , rmq .ChildResourceName ("server" ), i )
374390 rabbitCommand := fmt .Sprintf ("rabbitmq-plugins set %s" , plugins .AsString (" " ))
375-
376391 stdout , stderr , err := r .exec (rmq .Namespace , podName , "rabbitmq" , "sh" , "-c" , rabbitCommand )
377-
378392 if err != nil {
379- r .Log .Error (err , "Failed to enable plugins" ,
393+ r .Log .Error (err , "failed to set plugins" ,
380394 "namespace" , rmq .Namespace ,
381395 "name" , rmq .Name ,
382396 "pod" , podName ,
383397 "command" , rabbitCommand ,
384398 "stdout" , stdout ,
385399 "stderr" , stderr )
386- return err
400+ return 0 , err
387401 }
388402 }
389-
390- r .Log .Info ("Successfully enabled plugins on RabbitmqCluster" ,
403+ r .Log .Info ("successfully set plugins on RabbitmqCluster" ,
391404 "namespace" , rmq .Namespace ,
392405 "name" , rmq .Name )
393- return nil
406+
407+ delete (configMap .Annotations , pluginsUpdateAnnotation )
408+ if err := r .Update (ctx , & configMap ); err != nil {
409+ return 0 , client .IgnoreNotFound (err )
410+ }
411+
412+ return 0 , nil
413+ }
414+
415+ func (r * RabbitmqClusterReconciler ) allReplicasReadyAndUpdated (ctx context.Context , rmq * rabbitmqv1beta1.RabbitmqCluster ) (bool , error ) {
416+ sts := & appsv1.StatefulSet {}
417+
418+ if err := r .Get (ctx , types.NamespacedName {Name : rmq .ChildResourceName ("server" ), Namespace : rmq .Namespace }, sts ); err != nil {
419+ return false , client .IgnoreNotFound (err )
420+ }
421+
422+ desiredReplicas := * sts .Spec .Replicas
423+ if sts .Status .ReadyReplicas < desiredReplicas ||
424+ sts .Status .UpdatedReplicas < desiredReplicas { // StatefulSet rolling update is ongoing
425+ return false , nil
426+ }
427+
428+ return true , nil
429+ }
430+
431+ // Annotates the plugins ConfigMap if it was updated such that 'rabbitmq-plugins set' will be called on the RabbitMQ nodes at a later point in time
432+ func (r * RabbitmqClusterReconciler ) annotatePluginsConfigMapIfUpdated (
433+ ctx context.Context ,
434+ builder resource.ResourceBuilder ,
435+ operationResult controllerutil.OperationResult ,
436+ rmq * rabbitmqv1beta1.RabbitmqCluster ) {
437+
438+ if _ , ok := builder .(* resource.RabbitmqPluginsConfigMapBuilder ); ! ok {
439+ return
440+ }
441+ if operationResult != controllerutil .OperationResultUpdated {
442+ return
443+ }
444+
445+ if retryOnConflictErr := clientretry .RetryOnConflict (clientretry .DefaultRetry , func () error {
446+ configMap := corev1.ConfigMap {}
447+ if err := r .Get (ctx , types.NamespacedName {Namespace : rmq .Namespace , Name : rmq .ChildResourceName (resource .PluginsConfig )}, & configMap ); err != nil {
448+ return client .IgnoreNotFound (err )
449+ }
450+ if configMap .Annotations == nil {
451+ configMap .Annotations = make (map [string ]string )
452+ }
453+ configMap .Annotations [pluginsUpdateAnnotation ] = time .Now ().Format (time .RFC3339 )
454+ return r .Update (ctx , & configMap )
455+ }); retryOnConflictErr != nil {
456+ msg := fmt .Sprintf ("Failed to annotate ConfigMap %s of Namespace %s; enabled_plugins may be outdated" , rmq .ChildResourceName (resource .PluginsConfig ), rmq .Namespace )
457+ r .Log .Error (retryOnConflictErr , msg )
458+ r .Recorder .Event (rmq , corev1 .EventTypeWarning , "FailedUpdate" , msg )
459+ }
394460}
395461
396462func (r * RabbitmqClusterReconciler ) exec (namespace , podName , containerName string , command ... string ) (string , string , error ) {
0 commit comments