Fix the race condition that causes syncer missing in the syncerMap

pkg/neg/manager.go

@@ -199,7 +199,19 @@ func (manager *syncerManager) EnsureSyncers(namespace, name string, newPorts neg
 		}
 	}
 
-	for _, portInfo := range samePorts {
+	for svcPort, portInfo := range samePorts {
+		// Ensure syncer for the service port if no syncer found in the map.
+		// This is to fix the possible race condition that ensuring syncer
+		// while the old syncer is still shutting down.
+		syncerKey := manager.getSyncerKey(namespace, name, svcPort, portInfo)
+		if _, ok := manager.syncerMap[syncerKey]; !ok {
+			if err := manager.EnsureSyncer(namespace, name, svcPort, portInfo); err != nil {
+				errList = append(errList, err)
+				errorSyncers += 1
+			}
+			successfulSyncers += 1
+			continue
+		}
 		// To reduce the possibility of NEGs being leaked, ensure a SvcNeg CR exists for every
 		// desired port.
 		if err := manager.ensureSvcNegCR(key, portInfo); err != nil {

pkg/neg/manager_test.go

@@ -330,6 +330,38 @@ func TestEnsureAndStopSyncer(t *testing.T) {
 	manager.StopSyncer(svcNamespace2, svcName)
 }
 
+func TestEnsureSyncersWithMissingSyncer(t *testing.T) {
+	manager, _ := NewTestSyncerManager(fake.NewSimpleClientset())
+	namer := manager.namer
+	namespace := testServiceNamespace
+	name := testServiceName
+	portName := ""
+	svcPort := int32(3000)
+	targetPort := "80"
+	negName := namer.NEG(namespace, name, svcPort)
+	portMap := make(types.PortInfoMap)
+	portInfo := types.PortInfo{PortTuple: negtypes.SvcPortTuple{Name: portName, Port: svcPort, TargetPort: targetPort}, NegName: negName}
+
+	portMap[negtypes.PortInfoMapKey{ServicePort: svcPort}] = portInfo
+
+	manager.serviceLister.Add(&v1.Service{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}})
+
+	svcKey := serviceKey{namespace: namespace, name: name}
+	negSyncerKey := manager.getSyncerKey(namespace, name, negtypes.PortInfoMapKey{ServicePort: svcPort}, portInfo)
+	manager.svcPortMap[svcKey] = portMap
+
+	if _, _, err := manager.EnsureSyncers(namespace, name, portMap); err != nil {
+		t.Fatalf("Failed to ensure syncer: %v", err)
+	}
+	if syncer, ok := manager.syncerMap[negSyncerKey]; !ok {
+		t.Error("No syncer found for the service port")
+	} else if syncer.IsShuttingDown() || syncer.IsStopped() {
+		t.Errorf("Syner is not started, current status: IsShuttingDown: %v, IsStopped: %v", syncer.IsShuttingDown(), syncer.IsStopped())
+	}
+	// make sure there is no leaking go routine
+	manager.StopSyncer(namespace, name)
+}
+
 func TestGarbageCollectionSyncer(t *testing.T) {
 	t.Parallel()