Merge pull request #110322 from cyclinder/fix_concurrent_map_1.23

ipvs: fix prevent concurrent map read and map write for 1.23

pkg/proxy/ipvs/graceful_termination.go

@@ -80,7 +80,17 @@ func (q *graceTerminateRSList) remove(rs *listItem) bool {
 	return false
 }
 
+// return the size of the list
+func (q *graceTerminateRSList) len() int {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	return len(q.list)
+}
+
 func (q *graceTerminateRSList) flushList(handler func(rsToDelete *listItem) (bool, error)) bool {
+	q.lock.Lock()
+	defer q.lock.Unlock()
 	success := true
 	for name, rs := range q.list {
 		deleted, err := handler(rs)
@@ -90,7 +100,7 @@ func (q *graceTerminateRSList) flushList(handler func(rsToDelete *listItem) (boo
 		}
 		if deleted {
 			klog.InfoS("Removed real server from graceful delete real server list", "realServer", name)
-			q.remove(rs)
+			delete(q.list, rs.String())
 		}
 	}
 	return success

pkg/proxy/ipvs/graceful_termination_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package ipvs
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
@@ -400,3 +401,46 @@ func Test_GracefulDeleteRS(t *testing.T) {
 		})
 	}
 }
+
+func Test_RaceTerminateRSList(t *testing.T) {
+	ipvs := utilipvstest.NewFake()
+	gracefulTerminationManager := NewGracefulTerminationManager(ipvs)
+
+	// run in parallel to cause the race
+	go func() {
+		for i := 1; i <= 10; i++ {
+			for j := 1; j <= 100; j++ {
+				item := makeListItem(i, j)
+				gracefulTerminationManager.rsList.add(item)
+			}
+		}
+	}()
+
+	// wait until the list has some elements
+	for gracefulTerminationManager.rsList.len() < 20 {
+	}
+
+	// fake the handler to avoid the check against the IPVS virtual servers
+	fakeHandler := func(rsToDelete *listItem) (bool, error) {
+		return true, nil
+	}
+	if !gracefulTerminationManager.rsList.flushList(fakeHandler) {
+		t.Error("failed to flush entries")
+	}
+}
+
+func makeListItem(i, j int) *listItem {
+	vs := fmt.Sprintf("%d.%d.%d.%d", 1, 1, i, i)
+	rs := fmt.Sprintf("%d.%d.%d.%d", 1, 1, i, j)
+	return &listItem{
+		VirtualServer: &utilipvs.VirtualServer{
+			Address:  netutils.ParseIPSloppy(vs),
+			Protocol: "tcp",
+			Port:     uint16(80),
+		},
+		RealServer: &utilipvs.RealServer{
+			Address: netutils.ParseIPSloppy(rs),
+			Port:    uint16(80),
+		},
+	}
+}