diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 0d23fe037..593dcdbe6 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -1001,14 +1001,18 @@ func (c *Cluster) assignMons(mons []*monConfig, monsToSkipReconcile sets.Set[str result, err := waitForMonitorScheduling(c, deployment) if err != nil { log.NamespacedError(c.Namespace, logger, "failed to schedule mon %q. %v", mon.DaemonName, err) + resultLock.Lock() failedMonSchedule = true + resultLock.Unlock() return } nodeChoice := result.Node if nodeChoice == nil { log.NamespacedError(c.Namespace, logger, "failed to schedule monitor %q", mon.DaemonName) + resultLock.Lock() failedMonSchedule = true + resultLock.Unlock() return } @@ -1021,7 +1025,9 @@ func (c *Cluster) assignMons(mons []*monConfig, monsToSkipReconcile sets.Set[str schedule, err = getNodeInfoFromNode(*nodeChoice) if err != nil { log.NamespacedError(c.Namespace, logger, "failed to get node info for node %q. %v", nodeChoice.Name, err) + resultLock.Lock() failedMonSchedule = true + resultLock.Unlock() return } } else { diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index 8683b398f..44365a653 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/pkg/operator/ceph/cluster/mon/mon_test.go @@ -24,6 +24,7 @@ import ( "reflect" "strconv" "strings" + "sync" "testing" "time" @@ -1598,3 +1599,50 @@ func TestFloatingMonLabelsAndSchedulingSkip(t *testing.T) { assert.Contains(t, scheduled, "b") assert.NotContains(t, scheduled, "c") } + +// TestAssignMonsConcurrentSchedulingFailure ensures assignMons does not race on +// the shared failedMonSchedule flag when multiple mons fail to schedule at the +// same time. Each mon is scheduled in its own goroutine, so several goroutines +// can set failedMonSchedule concurrently. Run with -race to catch the write-write +// data race. +func TestAssignMonsConcurrentSchedulingFailure(t *testing.T) { + namespace := "ns" + c := newFloatingMonTestCluster(t, namespace) + + origWaitForMonitorScheduling := waitForMonitorScheduling + defer func() { waitForMonitorScheduling = origWaitForMonitorScheduling }() + + mons := []*monConfig{ + { + ResourceName: "rook-ceph-mon-a", DaemonName: "a", Port: DefaultMsgr1Port, + DataPathMap: config.NewStatefulDaemonDataPathMap("/var/lib/rook", dataDirRelativeHostPath("a"), config.MonType, "a", namespace), + }, + { + ResourceName: "rook-ceph-mon-b", DaemonName: "b", Port: DefaultMsgr1Port, + DataPathMap: config.NewStatefulDaemonDataPathMap("/var/lib/rook", dataDirRelativeHostPath("b"), config.MonType, "b", namespace), + }, + { + ResourceName: "rook-ceph-mon-d", DaemonName: "d", Port: DefaultMsgr1Port, + DataPathMap: config.NewStatefulDaemonDataPathMap("/var/lib/rook", dataDirRelativeHostPath("d"), config.MonType, "d", namespace), + }, + } + + // release all scheduling goroutines at the same time so their concurrent + // writes to failedMonSchedule overlap, making the data race detectable + var ready sync.WaitGroup + ready.Add(len(mons)) + start := make(chan struct{}) + waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) { + ready.Done() + <-start + return SchedulingResult{}, errors.New("mock scheduling failure") + } + go func() { + ready.Wait() + close(start) + }() + + err := c.assignMons(mons, sets.New[string]()) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to schedule mons") +}