unit test

pkg/neg/controller_test.go

@@ -295,28 +295,58 @@ func TestEnableNEGServiceWithIngress(t *testing.T) {
 }
 
 //TestEnableNEGSeviceWithL4ILB tests L4 ILB service with NEGs enabled.
+//Also verifies that modifying the TrafficPolicy on the service will
+//take effect.
 func TestEnableNEGServiceWithL4ILB(t *testing.T) {
 	controller := newTestController(fake.NewSimpleClientset())
 	controller.runL4 = true
 	defer controller.stop()
-	for _, randomize := range []bool{false, true} {
-		controller.serviceLister.Add(newTestILBService(controller, !randomize, 80))
-		svcClient := controller.client.CoreV1().Services(testServiceNamespace)
-		svcKey := utils.ServiceKeyFunc(testServiceNamespace, testServiceName)
-		err := controller.processService(svcKey)
-		if err != nil {
-			t.Fatalf("Failed to process service: %v", err)
-		}
-		svc, err := svcClient.Get(context2.TODO(), testServiceName, metav1.GetOptions{})
-		if err != nil {
-			t.Fatalf("Service was not created.(*apiv1.Service) successfully, err: %v", err)
-		}
-		validateSyncers(t, controller, 1, false)
-		expectedPortInfoMap := negtypes.NewPortInfoMapForVMIPNEG(testServiceNamespace, testServiceName,
-			controller.namer, randomize)
-		validateSyncerManagerWithPortInfoMap(t, controller, testServiceNamespace, testServiceName, expectedPortInfoMap)
-		validateServiceAnnotationWithPortInfoMap(t, svc, expectedPortInfoMap)
+	var prevSyncerKey, updatedSyncerKey negtypes.NegSyncerKey
+	localMode := false
+	t.Logf("Creating L4 ILB service with ExternalTrafficPolicy:Cluster")
+	controller.serviceLister.Add(newTestILBService(controller, localMode, 80))
+	svcClient := controller.client.CoreV1().Services(testServiceNamespace)
+	svcKey := utils.ServiceKeyFunc(testServiceNamespace, testServiceName)
+	err := controller.processService(svcKey)
+	if err != nil {
+		t.Fatalf("Failed to process service: %v", err)
 	}
+	svc, err := svcClient.Get(context2.TODO(), testServiceName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Service was not created.(*apiv1.Service) successfully, err: %v", err)
+	}
+	expectedPortInfoMap := negtypes.NewPortInfoMapForVMIPNEG(testServiceNamespace, testServiceName,
+		controller.namer, localMode)
+	// There will be only one entry in the map
+	for key, val := range expectedPortInfoMap {
+		prevSyncerKey = getSyncerKey(testServiceNamespace, testServiceName, key, val)
+	}
+	ValidateSyncerByKey(t, controller, 1, prevSyncerKey, false)
+	validateSyncerManagerWithPortInfoMap(t, controller, testServiceNamespace, testServiceName, expectedPortInfoMap)
+	validateServiceAnnotationWithPortInfoMap(t, svc, expectedPortInfoMap)
+	// Now Update the service to change the TrafficPolicy
+	t.Logf("Updating L4 ILB service from ExternalTrafficPolicy:Cluster to Local")
+	localMode = true
+	if err = controller.serviceLister.Update(updateTestILBService(controller, localMode, svc)); err != nil {
+		t.Fatalf("Failed to update test L4 ILB service: %v", err)
+	}
+	if err = controller.processService(svcKey); err != nil {
+		t.Fatalf("Failed to process updated L4 ILB srvice: %v", err)
+	}
+	expectedPortInfoMap = negtypes.NewPortInfoMapForVMIPNEG(testServiceNamespace, testServiceName,
+		controller.namer, localMode)
+	// There will be only one entry in the map
+	for key, val := range expectedPortInfoMap {
+		updatedSyncerKey = getSyncerKey(testServiceNamespace, testServiceName, key, val)
+	}
+	// there should only be 2 syncers - one stopped and one running.
+	ValidateSyncerByKey(t, controller, 2, updatedSyncerKey, false)
+	ValidateSyncerByKey(t, controller, 2, prevSyncerKey, true)
+	time.Sleep(1 * time.Second)
+	controller.manager.(*syncerManager).GC()
+	// check the port info map after all stale syncers have been deleted.
+	validateSyncerManagerWithPortInfoMap(t, controller, testServiceNamespace, testServiceName, expectedPortInfoMap)
+	validateServiceAnnotationWithPortInfoMap(t, svc, expectedPortInfoMap)
 }
 
 // TestEnableNEGServiceWithILBIngress tests ILB service with NEG enabled
@@ -1036,6 +1066,20 @@ func validateSyncers(t *testing.T, controller *Controller, num int, stopped bool
 	}
 }
 
+func ValidateSyncerByKey(t *testing.T, controller *Controller, num int, expectedKey negtypes.NegSyncerKey, stopped bool) {
+	t.Helper()
+	if len(controller.manager.(*syncerManager).syncerMap) != num {
+		t.Errorf("got %v syncer, want %v.", len(controller.manager.(*syncerManager).syncerMap), num)
+	}
+	if syncer, ok := controller.manager.(*syncerManager).syncerMap[expectedKey]; ok {
+		if syncer.IsStopped() == stopped {
+			return
+		}
+		t.Errorf("got syncer %q IsStopped() == %v, want %v.", expectedKey, syncer.IsStopped(), stopped)
+	}
+	t.Errorf("got %v syncer map, want syncer key %q", controller.manager.(*syncerManager).syncerMap, expectedKey)
+}
+
 func validateSyncerManagerWithPortInfoMap(t *testing.T, controller *Controller, namespace, svcName string, expectedPortInfoMap negtypes.PortInfoMap) {
 	t.Helper()
 	if len(controller.manager.(*syncerManager).syncerMap) != len(expectedPortInfoMap) {
@@ -1324,6 +1368,16 @@ func newTestILBService(c *Controller, onlyLocal bool, port int) *apiv1.Service {
 	return svc
 }
 
+func updateTestILBService(c *Controller, onlyLocal bool, svc *apiv1.Service) *apiv1.Service {
+	if onlyLocal {
+		svc.Spec.ExternalTrafficPolicy = apiv1.ServiceExternalTrafficPolicyTypeLocal
+	} else {
+		svc.Spec.ExternalTrafficPolicy = apiv1.ServiceExternalTrafficPolicyTypeCluster
+	}
+	c.client.CoreV1().Services(svc.Namespace).Update(context2.TODO(), svc, metav1.UpdateOptions{})
+	return svc
+}
+
 func newTestService(c *Controller, negIngress bool, negSvcPorts []int32) *apiv1.Service {
 	svcAnnotations := map[string]string{}
 	if negIngress || len(negSvcPorts) > 0 {

pkg/neg/syncers/endpoints_calculator_test.go

@@ -32,7 +32,8 @@ import (
 // The L7 implementation is tested in TestToZoneNetworkEndpointMapUtil.
 func TestLocalGetEndpointSet(t *testing.T) {
 	t.Parallel()
-	_, transactionSyncer := newL4ILBTestTransactionSyncer(negtypes.NewAdapter(gce.NewFakeGCECloud(gce.DefaultTestClusterValues())), false)
+	mode := negtypes.L4LocalMode
+	_, transactionSyncer := newL4ILBTestTransactionSyncer(negtypes.NewAdapter(gce.NewFakeGCECloud(gce.DefaultTestClusterValues())), mode)
 	nodeNames := []string{testInstance1, testInstance2, testInstance3, testInstance4, testInstance5, testInstance6}
 	for i := 0; i < len(nodeNames); i++ {
 		err := transactionSyncer.nodeLister.Add(&v1.Node{
@@ -102,7 +103,8 @@ func TestLocalGetEndpointSet(t *testing.T) {
 // TestClusterGetEndpointSet verifies the GetEndpointSet method implemented by the ClusterL4ILBEndpointsCalculator.
 func TestClusterGetEndpointSet(t *testing.T) {
 	t.Parallel()
-	_, transactionSyncer := newL4ILBTestTransactionSyncer(negtypes.NewAdapter(gce.NewFakeGCECloud(gce.DefaultTestClusterValues())), true)
+	mode := negtypes.L4ClusterMode
+	_, transactionSyncer := newL4ILBTestTransactionSyncer(negtypes.NewAdapter(gce.NewFakeGCECloud(gce.DefaultTestClusterValues())), mode)
 	nodeNames := []string{testInstance1, testInstance2, testInstance3, testInstance4, testInstance5, testInstance6}
 	for i := 0; i < len(nodeNames); i++ {
 		err := transactionSyncer.nodeLister.Add(&v1.Node{

pkg/neg/syncers/transaction_test.go

@@ -836,9 +836,9 @@ func TestCommitPods(t *testing.T) {
 	}
 }
 
-func newL4ILBTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, randomize bool) (negtypes.NegSyncer, *transactionSyncer) {
+func newL4ILBTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, mode negtypes.EndpointsCalculatorMode) (negtypes.NegSyncer, *transactionSyncer) {
 	negsyncer, ts := newTestTransactionSyncer(fakeGCE, negtypes.VmIpEndpointType)
-	ts.endpointsCalculator = GetEndpointsCalculator(ts.nodeLister, ts.podLister, ts.zoneGetter, ts.NegSyncerKey, randomize)
+	ts.endpointsCalculator = GetEndpointsCalculator(ts.nodeLister, ts.podLister, ts.zoneGetter, ts.NegSyncerKey, mode)
 	return negsyncer, ts
 }
 
@@ -861,10 +861,12 @@ func newTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, negTyp
 			TargetPort: "8080",
 		},
 	}
+	var mode negtypes.EndpointsCalculatorMode
 	if negType == negtypes.VmIpEndpointType {
 		svcPort.PortTuple.Port = 0
 		svcPort.PortTuple.TargetPort = ""
 		svcPort.PortTuple.Name = string(negtypes.VmIpEndpointType)
+		mode = negtypes.L4LocalMode
 	}
 
 	// TODO(freehan): use real readiness reflector
@@ -881,7 +883,7 @@ func newTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, negTyp
 		context.NodeInformer.GetIndexer(),
 		reflector,
 		GetEndpointsCalculator(context.NodeInformer.GetIndexer(), context.PodInformer.GetIndexer(), negtypes.NewFakeZoneGetter(),
-			svcPort, false))
+			svcPort, mode))
 	transactionSyncer := negsyncer.(*syncer).core.(*transactionSyncer)
 	return negsyncer, transactionSyncer
 }