core: use sync.Map for cluster.monitoringRoutines

`cluster.monitoringRoutines` stores three types of
daemon health: "mon", "osd", and "status". These are
used by three goroutines respectively, resulting in:
fatal error: concurrent map reads and map writes.

Signed-off-by: Yite Gu <guyite@bytedance.com>

pkg/operator/ceph/cluster/cephstatus.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -82,25 +83,32 @@ func newCephStatusChecker(context *clusterd.Context, clusterInfo *cephclient.Clu
 }
 
 // checkCephStatus periodically checks the health of the cluster
-func (c *cephStatusChecker) checkCephStatus(monitoringRoutines map[string]*opcontroller.ClusterHealth, daemon string) {
+func (c *cephStatusChecker) checkCephStatus(monitoringRoutines *sync.Map, daemon string) {
 	// check the status immediately before starting the loop
-	c.checkStatus(monitoringRoutines[daemon].InternalCtx)
+	v, ok := monitoringRoutines.Load(daemon)
+	if !ok {
+		logger.Infof("ceph cluster %q has been deleted. stopping ceph status check", c.clusterInfo.Namespace)
+		return
+	}
+	c.checkStatus(v.(*opcontroller.ClusterHealth).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 {
+		v, ok := monitoringRoutines.Load(daemon)
+		if !ok {
 			logger.Infof("ceph cluster %q has been deleted. stopping ceph status check", c.clusterInfo.Namespace)
 			return
 		}
+		health := v.(*opcontroller.ClusterHealth)
 		select {
-		case <-monitoringRoutines[daemon].InternalCtx.Done():
+		case <-health.InternalCtx.Done():
 			logger.Infof("stopping monitoring of ceph status")
-			delete(monitoringRoutines, daemon)
+			monitoringRoutines.Delete(daemon)
 			return
 
 		case <-time.After(*c.interval):
-			c.checkStatus(monitoringRoutines[daemon].InternalCtx)
+			c.checkStatus(health.InternalCtx)
 		}
 	}
 }

pkg/operator/ceph/cluster/cluster.go

@@ -69,7 +69,7 @@ type cluster struct {
 	mons               *mon.Cluster
 	ownerInfo          *k8sutil.OwnerInfo
 	isUpgrade          bool
-	monitoringRoutines map[string]*controller.ClusterHealth
+	monitoringRoutines sync.Map
 	observedGeneration int64
 }
 
@@ -78,15 +78,14 @@ func newCluster(ctx context.Context, c *cephv1.CephCluster, context *clusterd.Co
 		// at this phase of the cluster creation process, the identity components of the cluster are
 		// not yet established. we reserve this struct which is filled in as soon as the cluster's
 		// identity can be established.
-		ClusterInfo:        client.AdminClusterInfo(ctx, c.Namespace, c.Name),
-		Namespace:          c.Namespace,
-		Spec:               &c.Spec,
-		clusterMetadata:    c.ObjectMeta,
-		context:            context,
-		namespacedName:     types.NamespacedName{Namespace: c.Namespace, Name: c.Name},
-		monitoringRoutines: make(map[string]*controller.ClusterHealth),
-		ownerInfo:          ownerInfo,
-		mons:               mon.New(ctx, context, c.Namespace, c.Spec, ownerInfo),
+		ClusterInfo:     client.AdminClusterInfo(ctx, c.Namespace, c.Name),
+		Namespace:       c.Namespace,
+		Spec:            &c.Spec,
+		clusterMetadata: c.ObjectMeta,
+		context:         context,
+		namespacedName:  types.NamespacedName{Namespace: c.Namespace, Name: c.Name},
+		ownerInfo:       ownerInfo,
+		mons:            mon.New(ctx, context, c.Namespace, c.Spec, ownerInfo),
 		// update observedGeneration with current generation value,
 		// 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

pkg/operator/ceph/cluster/controller.go

@@ -468,12 +468,13 @@ 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
+			// if the context hasn't been cancelled
+			if monitoring, ok := cluster.monitoringRoutines.Load(daemon); ok && monitoring.(*opcontroller.ClusterHealth).InternalCtx.Err() == nil {
 				// Stop the monitoring routine
-				cluster.monitoringRoutines[daemon].InternalCancel()
+				monitoring.(*opcontroller.ClusterHealth).InternalCancel()
 
 				// Remove the monitoring routine from the map
-				delete(cluster.monitoringRoutines, daemon)
+				cluster.monitoringRoutines.Delete(daemon)
 			}
 		}
 	}

pkg/operator/ceph/cluster/mon/health.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -111,7 +112,7 @@ func NewHealthChecker(monCluster *Cluster) *HealthChecker {
 }
 
 // Check periodically checks the health of the monitors
-func (hc *HealthChecker) Check(monitoringRoutines map[string]*controller.ClusterHealth, daemon string) {
+func (hc *HealthChecker) Check(monitoringRoutines *sync.Map, daemon string) {
 	for {
 		// Update Mon Timeout with CR details
 		updateMonTimeout(hc.monCluster)
@@ -121,26 +122,28 @@ func (hc *HealthChecker) Check(monitoringRoutines map[string]*controller.Cluster
 
 		// 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 {
+		v, ok := monitoringRoutines.Load(daemon)
+		if !ok {
 			logger.Infof("ceph cluster %q has been deleted. stopping monitoring of mons", hc.monCluster.Namespace)
 			return
 		}
 
+		health := v.(*controller.ClusterHealth)
 		select {
-		case <-monitoringRoutines[daemon].InternalCtx.Done():
+		case <-health.InternalCtx.Done():
 			logger.Infof("stopping monitoring of mons in namespace %q (internal context canceled)", hc.monCluster.Namespace)
-			delete(monitoringRoutines, daemon)
+			monitoringRoutines.Delete(daemon)
 			return
 
 		// Since c.ClusterInfo.IsInitialized() below uses a different context, we need to check if the context is done
 		case <-hc.monCluster.ClusterInfo.Context.Done():
 			logger.Infof("stopping monitoring of mons in namespace %q (mon cluster info context canceled)", hc.monCluster.Namespace)
-			delete(monitoringRoutines, daemon)
+			monitoringRoutines.Delete(daemon)
 			return
 
 		case <-time.After(hc.interval):
 			logger.Debugf("checking health of mons")
-			err := hc.monCluster.checkHealth(monitoringRoutines[daemon].InternalCtx)
+			err := hc.monCluster.checkHealth(health.InternalCtx)
 			if err != nil {
 				logger.Warningf("failed to check mon health. %v", err)
 			}

pkg/operator/ceph/cluster/monitoring.go

@@ -34,12 +34,12 @@ func (c *ClusterController) configureCephMonitoring(cluster *cluster, clusterInf
 	for _, daemon := range monitorDaemonList {
 		// Is the monitoring enabled for that daemon?
 		isEnabled = isMonitoringEnabled(daemon, cluster.Spec)
-		if health, ok := cluster.monitoringRoutines[daemon]; ok {
+		if health, ok := cluster.monitoringRoutines.Load(daemon); ok {
 			// If the context Err() is nil this means it hasn't been cancelled yet
-			if health.InternalCtx.Err() == nil {
+			if health.(*opcontroller.ClusterHealth).InternalCtx.Err() == nil {
 				logger.Debugf("monitoring routine for %q is already running", daemon)
 				if !isEnabled {
-					cluster.monitoringRoutines[daemon].InternalCancel()
+					health.(*opcontroller.ClusterHealth).InternalCancel()
 				}
 			}
 		} else {
@@ -48,10 +48,10 @@ 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] = &opcontroller.ClusterHealth{
+				cluster.monitoringRoutines.Store(daemon, &opcontroller.ClusterHealth{
 					InternalCtx:    internalCtx,
 					InternalCancel: internalCancel,
-				}
+				})
 
 				// Run the go routine
 				c.startMonitoringCheck(cluster, clusterInfo, daemon)
@@ -80,18 +80,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)
+		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)
+			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)
+		go cephChecker.checkCephStatus(&cluster.monitoringRoutines, daemon)
 	}
 }

pkg/operator/ceph/cluster/osd/health.go

@@ -18,6 +18,7 @@ package osd
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -65,22 +66,24 @@ func NewOSDHealthMonitor(context *clusterd.Context, clusterInfo *client.ClusterI
 }
 
 // Start runs monitoring logic for osds status at set intervals
-func (m *OSDHealthMonitor) Start(monitoringRoutines map[string]*opcontroller.ClusterHealth, daemon string) {
+func (m *OSDHealthMonitor) Start(monitoringRoutines *sync.Map, 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 {
+		v, ok := monitoringRoutines.Load(daemon)
+		if !ok {
 			logger.Infof("ceph cluster %q has been deleted. stopping monitoring of OSDs", m.clusterInfo.Namespace)
 			return
 		}
+		health := v.(*opcontroller.ClusterHealth)
 		select {
 		case <-time.After(*m.interval):
 			logger.Debug("checking osd processes status.")
 			m.checkOSDHealth()
 
-		case <-monitoringRoutines[daemon].InternalCtx.Done():
+		case <-health.InternalCtx.Done():
 			logger.Infof("stopping monitoring of OSDs in namespace %q", m.clusterInfo.Namespace)
-			delete(monitoringRoutines, daemon)
+			monitoringRoutines.Delete(daemon)
 			return
 		}
 	}

pkg/operator/ceph/cluster/osd/health_test.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -100,15 +101,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{
+	var monitoringRoutines sync.Map
+	monitoringRoutines.Store("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(monitoringRoutines, "osd")
+	go osdMon.Start(&monitoringRoutines, "osd")
 	cancel()
 }