Merge pull request #10635 from shubham-pampattiwar/fix-concurrent-map-write-pvcpvmap

Fix concurrent map writes crash in pvcPvMap

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

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

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{