From 05506e7a6836ab27d75552cc14174e6cea274cc8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Wed, 30 Mar 2022 11:10:13 +0200 Subject: [PATCH] core: reload go routine after CR is edited MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, the struct maintaining the list of cluster was still initialized with a cluster item. Then the monitoring check will see that the cluster is part of the struct already and thus won't run the monitoring go routine again. Now each time we cancel the context, we also remove the cluster item from the map so that when the controller runs again, the monitoring struct is re-populated and the go routine runs and statuses are updated. Closes: https://github.com/rook/rook/issues/9911 Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/cephstatus.go | 15 +++++++++++---- pkg/operator/ceph/cluster/cluster.go | 14 +++++--------- pkg/operator/ceph/cluster/controller.go | 8 ++++++-- pkg/operator/ceph/cluster/mon/health.go | 16 +++++++++++++--- pkg/operator/ceph/cluster/monitoring.go | 17 +++++++++-------- pkg/operator/ceph/cluster/osd/health.go | 13 ++++++++++--- pkg/operator/ceph/cluster/osd/health_test.go | 9 ++++++++- .../ceph/controller/controller_utils.go | 6 ++++++ pkg/operator/ceph/object/controller.go | 2 +- pkg/operator/ceph/object/health.go | 16 +++++++++++----- 10 files changed, 80 insertions(+), 36 deletions(-) diff --git a/pkg/operator/ceph/cluster/cephstatus.go b/pkg/operator/ceph/cluster/cephstatus.go index a7739bee3..e6ead5cdb 100644 --- a/pkg/operator/ceph/cluster/cephstatus.go +++ b/pkg/operator/ceph/cluster/cephstatus.go @@ -84,18 +84,25 @@ func newCephStatusChecker(context *clusterd.Context, clusterInfo *cephclient.Clu } // checkCephStatus periodically checks the health of the cluster -func (c *cephStatusChecker) checkCephStatus(context context.Context) { +func (c *cephStatusChecker) checkCephStatus(monitoringRoutines map[string]*opcontroller.ClusterHealth, daemon string) { // check the status immediately before starting the loop - c.checkStatus(context) + c.checkStatus(monitoringRoutines[daemon].InternalCtx) for { + // We must perform this check otherwise the case will check an index that does not exist anymore and + // we will get an invalid pointer error and the go routine will panic + if _, ok := monitoringRoutines[daemon]; !ok { + logger.Infof("ceph cluster %q has been deleted. stopping ceph status check", c.clusterInfo.Namespace) + return + } select { - case <-context.Done(): + case <-monitoringRoutines[daemon].InternalCtx.Done(): logger.Infof("stopping monitoring of ceph status") + delete(monitoringRoutines, daemon) return case <-time.After(*c.interval): - c.checkStatus(context) + c.checkStatus(monitoringRoutines[daemon].InternalCtx) } } } diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index abe7b57b1..80040c061 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -36,6 +36,7 @@ import ( "github.com/rook/rook/pkg/operator/ceph/cluster/osd" "github.com/rook/rook/pkg/operator/ceph/config" "github.com/rook/rook/pkg/operator/ceph/controller" + opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/ceph/csi" cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" @@ -58,15 +59,10 @@ type cluster struct { mons *mon.Cluster ownerInfo *k8sutil.OwnerInfo isUpgrade bool - monitoringRoutines map[string]*clusterHealth + monitoringRoutines map[string]*opcontroller.ClusterHealth observedGeneration int64 } -type clusterHealth struct { - internalCtx context.Context - internalCancel context.CancelFunc -} - func newCluster(ctx context.Context, c *cephv1.CephCluster, context *clusterd.Context, ownerInfo *k8sutil.OwnerInfo) *cluster { return &cluster{ // at this phase of the cluster creation process, the identity components of the cluster are @@ -77,11 +73,11 @@ func newCluster(ctx context.Context, c *cephv1.CephCluster, context *clusterd.Co Spec: &c.Spec, context: context, namespacedName: types.NamespacedName{Namespace: c.Namespace, Name: c.Name}, - monitoringRoutines: make(map[string]*clusterHealth), + monitoringRoutines: make(map[string]*opcontroller.ClusterHealth), ownerInfo: ownerInfo, mons: mon.New(ctx, context, c.Namespace, c.Spec, ownerInfo), // update observedGeneration with current generation value, - // because generation can be changed before reconile got completed + // because generation can be changed before reconcile got completed // CR status will be updated at end of reconcile, so to reflect the reconcile has finished observedGeneration: c.ObjectMeta.Generation, } @@ -459,7 +455,7 @@ func (c *cluster) replaceDefaultCrushMap(newRoot string) (err error) { func (c *cluster) preMonStartupActions(cephVersion cephver.CephVersion) error { // Disable the mds sanity checks for the mons due to a ceph upgrade issue // for the mds to Pacific if 16.2.7 or greater. We keep it more general for any - // Pacific upgrade greater than 16.2.7 in case they skip updrading directly to 16.2.7. + // Pacific upgrade greater than 16.2.7 in case they skip upgrading directly to 16.2.7. if c.isUpgrade && cephVersion.IsPacific() && cephVersion.IsAtLeast(cephver.CephVersion{Major: 16, Minor: 2, Extra: 7}) { if err := c.skipMDSSanityChecks(true); err != nil { // If there is an error, just print it and continue. Likely there is not a diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 254e82f97..f252226fe 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -377,8 +377,12 @@ func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (r // since the op manager context is cancelled. // close the goroutines watching the health of the cluster (mons, osds, ceph status) for _, daemon := range monitorDaemonList { - if monitoring, ok := cluster.monitoringRoutines[daemon]; ok && monitoring.internalCtx.Err() == nil { // if the context hasn't been cancelled - cluster.monitoringRoutines[daemon].internalCancel() + if monitoring, ok := cluster.monitoringRoutines[daemon]; ok && monitoring.InternalCtx.Err() == nil { // if the context hasn't been cancelled + // Stop the monitoring routine + cluster.monitoringRoutines[daemon].InternalCancel() + + // Remove the monitoring routine from the map + delete(cluster.monitoringRoutines, daemon) } } } diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index f0f2b0766..70779bdda 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/pkg/operator/ceph/cluster/mon/health.go @@ -112,20 +112,30 @@ func NewHealthChecker(monCluster *Cluster) *HealthChecker { } // Check periodically checks the health of the monitors -func (hc *HealthChecker) Check(context context.Context) { +func (hc *HealthChecker) Check(monitoringRoutines map[string]*controller.ClusterHealth, daemon string) { for { // Update Mon Timeout with CR details updateMonTimeout(hc.monCluster) + // Update Mon Interval with CR details updateMonInterval(hc.monCluster, hc) + + // We must perform this check otherwise the case will check an index that does not exist anymore and + // we will get an invalid pointer error and the go routine will panic + if _, ok := monitoringRoutines[daemon]; !ok { + logger.Infof("ceph cluster %q has been deleted. stopping monitoring of mons", hc.monCluster.Namespace) + return + } + select { - case <-context.Done(): + case <-monitoringRoutines[daemon].InternalCtx.Done(): logger.Infof("stopping monitoring of mons in namespace %q", hc.monCluster.Namespace) + delete(monitoringRoutines, daemon) return case <-time.After(hc.interval): logger.Debugf("checking health of mons") - err := hc.monCluster.checkHealth(context) + err := hc.monCluster.checkHealth(monitoringRoutines[daemon].InternalCtx) if err != nil { logger.Warningf("failed to check mon health. %v", err) } diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index 803ba47fa..71d39b3ef 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/pkg/operator/ceph/cluster/monitoring.go @@ -24,6 +24,7 @@ import ( cephclient "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/cluster/osd" + opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" ) var ( @@ -37,10 +38,10 @@ func (c *ClusterController) configureCephMonitoring(cluster *cluster, clusterInf isEnabled = isMonitoringEnabled(daemon, cluster.Spec) if health, ok := cluster.monitoringRoutines[daemon]; ok { // If the context Err() is nil this means it hasn't been cancelled yet - if health.internalCtx.Err() == nil { + if health.InternalCtx.Err() == nil { logger.Debugf("monitoring routine for %q is already running", daemon) if !isEnabled { - cluster.monitoringRoutines[daemon].internalCancel() + cluster.monitoringRoutines[daemon].InternalCancel() } } } else { @@ -49,9 +50,9 @@ func (c *ClusterController) configureCephMonitoring(cluster *cluster, clusterInf // They can individually be cancelled and will be cancelled when the parent context is cancelled internalCtx, internalCancel := context.WithCancel(c.OpManagerCtx) - cluster.monitoringRoutines[daemon] = &clusterHealth{ - internalCtx: internalCtx, - internalCancel: internalCancel, + cluster.monitoringRoutines[daemon] = &opcontroller.ClusterHealth{ + InternalCtx: internalCtx, + InternalCancel: internalCancel, } // Run the go routine @@ -81,18 +82,18 @@ func (c *ClusterController) startMonitoringCheck(cluster *cluster, clusterInfo * case "mon": healthChecker := mon.NewHealthChecker(cluster.mons) logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) - go healthChecker.Check(cluster.monitoringRoutines[daemon].internalCtx) + go healthChecker.Check(cluster.monitoringRoutines, daemon) case "osd": if !cluster.Spec.External.Enable { c.osdChecker = osd.NewOSDHealthMonitor(c.context, clusterInfo, cluster.Spec.RemoveOSDsIfOutAndSafeToRemove, cluster.Spec.HealthCheck) logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) - go c.osdChecker.Start(cluster.monitoringRoutines[daemon].internalCtx) + go c.osdChecker.Start(cluster.monitoringRoutines, daemon) } case "status": cephChecker := newCephStatusChecker(c.context, clusterInfo, cluster.Spec) logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) - go cephChecker.checkCephStatus(cluster.monitoringRoutines[daemon].internalCtx) + go cephChecker.checkCephStatus(cluster.monitoringRoutines, daemon) } } diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go index 821c146b7..ed18a3427 100644 --- a/pkg/operator/ceph/cluster/osd/health.go +++ b/pkg/operator/ceph/cluster/osd/health.go @@ -17,7 +17,6 @@ limitations under the License. package osd import ( - "context" "fmt" "reflect" "time" @@ -26,6 +25,7 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" + opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/ceph/reporting" "github.com/rook/rook/pkg/operator/k8sutil" kerrors "k8s.io/apimachinery/pkg/api/errors" @@ -69,16 +69,23 @@ func NewOSDHealthMonitor(context *clusterd.Context, clusterInfo *client.ClusterI } // Start runs monitoring logic for osds status at set intervals -func (m *OSDHealthMonitor) Start(context context.Context) { +func (m *OSDHealthMonitor) Start(monitoringRoutines map[string]*opcontroller.ClusterHealth, daemon string) { for { + // We must perform this check otherwise the case will check an index that does not exist anymore and + // we will get an invalid pointer error and the go routine will panic + if _, ok := monitoringRoutines[daemon]; !ok { + logger.Infof("ceph cluster %q has been deleted. stopping monitoring of OSDs", m.clusterInfo.Namespace) + return + } select { case <-time.After(*m.interval): logger.Debug("checking osd processes status.") m.checkOSDHealth() - case <-context.Done(): + case <-monitoringRoutines[daemon].InternalCtx.Done(): logger.Infof("stopping monitoring of OSDs in namespace %q", m.clusterInfo.Namespace) + delete(monitoringRoutines, daemon) return } } diff --git a/pkg/operator/ceph/cluster/osd/health_test.go b/pkg/operator/ceph/cluster/osd/health_test.go index f96ac9b7c..b260e773f 100644 --- a/pkg/operator/ceph/cluster/osd/health_test.go +++ b/pkg/operator/ceph/cluster/osd/health_test.go @@ -26,6 +26,7 @@ import ( "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/daemon/ceph/client" + opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/k8sutil" testexec "github.com/rook/rook/pkg/operator/test" exectest "github.com/rook/rook/pkg/util/exec/test" @@ -102,9 +103,15 @@ func TestOSDHealthCheck(t *testing.T) { func TestMonitorStart(t *testing.T) { context, cancel := context.WithCancel(context.TODO()) + monitoringRoutines := make(map[string]*opcontroller.ClusterHealth) + monitoringRoutines["osd"] = &opcontroller.ClusterHealth{ + InternalCtx: context, + InternalCancel: cancel, + } + osdMon := NewOSDHealthMonitor(&clusterd.Context{}, client.AdminTestClusterInfo("ns"), true, cephv1.CephClusterHealthCheckSpec{}) logger.Infof("starting osd monitor") - go osdMon.Start(context) + go osdMon.Start(monitoringRoutines, "osd") cancel() } diff --git a/pkg/operator/ceph/controller/controller_utils.go b/pkg/operator/ceph/controller/controller_utils.go index 5b23d596f..dcd096d3c 100644 --- a/pkg/operator/ceph/controller/controller_utils.go +++ b/pkg/operator/ceph/controller/controller_utils.go @@ -42,6 +42,12 @@ type OperatorConfig struct { Parameters map[string]string } +// ClusterHealth is passed to the various monitoring go routines to stop them when the context is cancelled +type ClusterHealth struct { + InternalCtx context.Context + InternalCancel context.CancelFunc +} + const ( // OperatorSettingConfigMapName refers to ConfigMap that configures rook ceph operator OperatorSettingConfigMapName string = "rook-ceph-operator-config" diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index 170974709..4c1a82df1 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -544,7 +544,7 @@ func (r *ReconcileCephObjectStore) startMonitoring(objectstore *cephv1.CephObjec } logger.Infof("starting rgw health checker for CephObjectStore %q", namespacedName.String()) - go rgwChecker.checkObjectStore(r.objectStoreContexts[channelKey].internalCtx) + go rgwChecker.checkObjectStore(r.objectStoreContexts, channelKey) // Set the monitoring flag so we don't start more than one go routine r.objectStoreContexts[channelKey].started = true diff --git a/pkg/operator/ceph/object/health.go b/pkg/operator/ceph/object/health.go index c24119982..f27a373e4 100644 --- a/pkg/operator/ceph/object/health.go +++ b/pkg/operator/ceph/object/health.go @@ -17,7 +17,6 @@ limitations under the License. package object import ( - "context" "fmt" "time" @@ -88,20 +87,27 @@ func newBucketChecker( } // checkObjectStore periodically checks the health of the cluster -func (c *bucketChecker) checkObjectStore(context context.Context) { +func (c *bucketChecker) checkObjectStore(objectStoreContexts map[string]*objectStoreHealth, channelKey string) { // check the object store health immediately before starting the loop err := c.checkObjectStoreHealth() if err != nil { - updateStatusBucket(context, c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) + updateStatusBucket(c.objContext.clusterInfo.Context, c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) logger.Debugf("failed to check rgw health for object store %q. %v", c.namespacedName.Name, err) } for { + // We must perform this check otherwise the case will check an index that does not exist anymore and + // we will get an invalid pointer error and the go routine will panic + if _, ok := objectStoreContexts[channelKey]; !ok { + logger.Infof("object store %q has been deleted. stopping monitoring of rgw endpoints", c.namespacedName.Name) + return + } select { - case <-context.Done(): + case <-objectStoreContexts[channelKey].internalCtx.Done(): // purge bucket and s3 user // Needed for external mode where in converged everything goes away with the CR deletion c.cleanupHealthCheck() + delete(objectStoreContexts, channelKey) logger.Infof("stopping monitoring of rgw endpoints for object store %q", c.namespacedName.Name) return @@ -109,7 +115,7 @@ func (c *bucketChecker) checkObjectStore(context context.Context) { logger.Debugf("checking rgw health of object store %q", c.namespacedName.Name) err := c.checkObjectStoreHealth() if err != nil { - updateStatusBucket(context, c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) + updateStatusBucket(c.objContext.clusterInfo.Context, c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) logger.Debugf("failed to check rgw health for object store %q. %v", c.namespacedName.Name, err) } }