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