Merge pull request #9962 from leseb/fix-9911

core: reload go routine after CR is edited

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

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

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

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

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

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

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

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"

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

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