diff --git a/changelogs/unreleased/10635-shubham-pampattiwar b/changelogs/unreleased/10635-shubham-pampattiwar new file mode 100644 index 000000000..1d9a8c808 --- /dev/null +++ b/changelogs/unreleased/10635-shubham-pampattiwar @@ -0,0 +1 @@ +Fix concurrent map writes crash in pvcPvMap when multiple ItemBlock workers back up PVs at the same time diff --git a/internal/volume/volumes_information.go b/internal/volume/volumes_information.go index 67d5db4d5..56670aed1 100644 --- a/internal/volume/volumes_information.go +++ b/internal/volume/volumes_information.go @@ -411,6 +411,9 @@ type SkippedVolume struct { // BackupVolumesInformation contains the information needs by generating // the backup BackupVolumeInfo array. type BackupVolumesInformation struct { + // initLock guards the lazy initialization of pvMap, which may be triggered + // concurrently by multiple ItemBlock workers. + initLock sync.Mutex // A map contains the backup-included PV detail content. The key is PV name. pvMap *pvcPvMap volumeInfos []*BackupVolumeInfo @@ -440,11 +443,17 @@ func (v *BackupVolumesInformation) Init() { v.volumeInfos = make([]*BackupVolumeInfo, 0) } +// InsertPVMap records the PV and its bound PVC in the PV map. It is called from +// the ItemBlock worker goroutines, so it must stay safe for concurrent use. func (v *BackupVolumesInformation) InsertPVMap(pv corev1api.PersistentVolume, pvcName, pvcNamespace string) { + v.initLock.Lock() if v.pvMap == nil { v.Init() } - v.pvMap.insert(pv, pvcName, pvcNamespace) + pvMap := v.pvMap + v.initLock.Unlock() + + pvMap.insert(pv, pvcName, pvcNamespace) } func (v *BackupVolumesInformation) Result( @@ -808,11 +817,17 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() { v.volumeInfos = append(v.volumeInfos, tmpVolumeInfos...) } +// pvcPvMap is safe for concurrent use. On the backup path it is written by the +// ItemBlock worker goroutines via BackupVolumesInformation.InsertPVMap. type pvcPvMap struct { + lock sync.RWMutex data map[string]pvcPvInfo } func (m *pvcPvMap) insert(pv corev1api.PersistentVolume, pvcName, pvcNamespace string) { + m.lock.Lock() + defer m.lock.Unlock() + m.data[pv.Name] = pvcPvInfo{ PVCName: pvcName, PVCNamespace: pvcNamespace, @@ -821,6 +836,9 @@ func (m *pvcPvMap) insert(pv corev1api.PersistentVolume, pvcName, pvcNamespace s } func (m *pvcPvMap) retrieve(pvName, pvcName, pvcNS string) *pvcPvInfo { + m.lock.RLock() + defer m.lock.RUnlock() + if pvName != "" { if info, ok := m.data[pvName]; ok { return &info diff --git a/internal/volume/volumes_information_test.go b/internal/volume/volumes_information_test.go index 0c5c41bb7..9af9d6bfd 100644 --- a/internal/volume/volumes_information_test.go +++ b/internal/volume/volumes_information_test.go @@ -18,6 +18,7 @@ package volume import ( "encoding/json" + "fmt" "sync" "testing" @@ -1404,6 +1405,85 @@ func int64Ptr(val int) *int64 { return &i } +func TestInsertPVMapConcurrent(t *testing.T) { + // Velero backs up ItemBlocks on a pool of worker goroutines that all share + // the same BackupVolumesInformation, so InsertPVMap must be safe for + // concurrent use. Before the pvcPvMap lock was added this test reproduced + // the "fatal error: concurrent map writes" crash reported in #10634. + const ( + workers = 10 + pvsPerWork = 50 + ) + + volumesInfo := new(BackupVolumesInformation) + volumesInfo.Init() + + wg := new(sync.WaitGroup) + start := make(chan struct{}) + + for i := range workers { + wg.Add(1) + go func(worker int) { + defer wg.Done() + <-start + for j := range pvsPerWork { + pvName := fmt.Sprintf("pv-%d-%d", worker, j) + volumesInfo.InsertPVMap( + *builder.ForPersistentVolume(pvName).ClaimRef("ns", "pvc-"+pvName).Result(), + "pvc-"+pvName, + "ns", + ) + // Read concurrently as well, to exercise the RLock path. + volumesInfo.pvMap.retrieve(pvName, "", "") + } + }(i) + } + + close(start) + wg.Wait() + + require.Len(t, volumesInfo.pvMap.data, workers*pvsPerWork) + for i := range workers { + for j := range pvsPerWork { + pvName := fmt.Sprintf("pv-%d-%d", i, j) + info := volumesInfo.pvMap.retrieve(pvName, "", "") + require.NotNil(t, info, "PV %s missing from the map", pvName) + require.Equal(t, "pvc-"+pvName, info.PVCName) + require.Equal(t, "ns", info.PVCNamespace) + } + } +} + +func TestInsertPVMapLazyInitConcurrent(t *testing.T) { + // InsertPVMap lazily initializes pvMap when Init has not been called. + // Concurrent callers must not each install their own map and lose entries. + const workers = 20 + + volumesInfo := new(BackupVolumesInformation) + + wg := new(sync.WaitGroup) + start := make(chan struct{}) + + for i := range workers { + wg.Add(1) + go func(worker int) { + defer wg.Done() + <-start + pvName := fmt.Sprintf("pv-%d", worker) + volumesInfo.InsertPVMap( + *builder.ForPersistentVolume(pvName).ClaimRef("ns", "pvc-"+pvName).Result(), + "pvc-"+pvName, + "ns", + ) + }(i) + } + + close(start) + wg.Wait() + + require.Len(t, volumesInfo.pvMap.data, workers) +} + func TestGetVolumeSnapshotClasses(t *testing.T) { class := &snapshotv1api.VolumeSnapshotClass{ ObjectMeta: metav1.ObjectMeta{