diff --git a/pkg/operator/ceph/cluster/cephstatus.go b/pkg/operator/ceph/cluster/cephstatus.go index 293df7589..2b7ea1527 100644 --- a/pkg/operator/ceph/cluster/cephstatus.go +++ b/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) } } } diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index f8f2f1851..0340ab2c0 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/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 diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index a97224c19..0f240cc13 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/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) } } } diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index bf1700078..defb83734 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/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) } diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index c11ba8e8e..caaf05573 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/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) } } diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go index 27cac0ff6..e769f0755 100644 --- a/pkg/operator/ceph/cluster/osd/health.go +++ b/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 } } diff --git a/pkg/operator/ceph/cluster/osd/health_test.go b/pkg/operator/ceph/cluster/osd/health_test.go index 40ba8b7b2..e4c7c7ff3 100644 --- a/pkg/operator/ceph/cluster/osd/health_test.go +++ b/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() }