Populate State and Subnet for NEG reference in NEG CR.

* Populate Subnet in NEG CR based on the subnet on the NEG.
* Populate State as ACTIVE for new and existing NEGs. For NEGs in the zones
  which don't have any nodes, set their state as INACTIVE.

pkg/neg/syncers/transaction.go

@@ -769,8 +769,12 @@ func (s *transactionSyncer) logEndpoints(endpointMap map[string]negtypes.Network
 	s.logger.V(3).Info("Endpoints for NEG", "description", desc, "endpointMap", endpointMap)
 }
 
-// updateInitStatus queries the k8s api server for the current NEG CR and updates the Initialized condition and neg objects as appropriate.
-// If neg client is nil, will return immediately
+// updateInitStatus takes in the NEG refs based on the existing node zones,
+// then queries the k8s api server for the current NEG CR and updates the
+// Initialized condition and neg objects as appropriate.
+// Before patching the NEG CR, it also includes NEG refs for NEGs are no longer
+// needed and change status as INACTIVE.
+// If neg client is nil, will return immediately.
 func (s *transactionSyncer) updateInitStatus(negObjRefs []negv1beta1.NegObjectReference, errList []error) {
 	if s.svcNegClient == nil {
 		return
@@ -784,6 +788,8 @@ func (s *transactionSyncer) updateInitStatus(negObjRefs []negv1beta1.NegObjectRe
 	}
 
 	neg := origNeg.DeepCopy()
+	inactiveNegObjRefs := getInactiveNegRefs(origNeg.Status.NetworkEndpointGroups, negObjRefs, s.logger)
+	negObjRefs = append(negObjRefs, inactiveNegObjRefs...)
 	neg.Status.NetworkEndpointGroups = negObjRefs
 
 	initializedCondition := getInitializedCondition(utilerrors.NewAggregate(errList))
@@ -917,6 +923,41 @@ func ensureCondition(neg *negv1beta1.ServiceNetworkEndpointGroup, expectedCondit
 	return expectedCondition
 }
 
+// getInactiveNegRefs creates NEG references for NEGs in Inactive State.
+// Inactive NEG are NEGs that are no longer needed.
+func getInactiveNegRefs(oldNegRefs []negv1beta1.NegObjectReference, currentNegRefs []negv1beta1.NegObjectReference, logger klog.Logger) []negv1beta1.NegObjectReference {
+	activeNegs := make(map[negtypes.NegInfo]struct{})
+	for _, negRef := range currentNegRefs {
+		negInfo, err := negtypes.NegInfoFromNegRef(negRef)
+		if err != nil {
+			logger.Error(err, "Failed to extract name and zone information of a neg from the current snapshot", "negId", negRef.Id, "negSelfLink", negRef.SelfLink)
+			continue
+		}
+		activeNegs[negInfo] = struct{}{}
+	}
+
+	var inactiveNegRefs []negv1beta1.NegObjectReference
+	for _, origNegRef := range oldNegRefs {
+		negInfo, err := negtypes.NegInfoFromNegRef(origNegRef)
+		if err != nil {
+			logger.Error(err, "Failed to extract name and zone information of a neg from the previous snapshot, skipping validating if it is an Inactive NEG", "negId", origNegRef.Id, "negSelfLink", origNegRef.SelfLink)
+			continue
+		}
+
+		// NEGs are listed based on the current node zones. If a NEG no longer
+		// exists in the current list, it means there are no nodes/endpoints
+		// in that specific zone, and we mark it as INACTIVE.
+		// We use SelfLink as identifier since it contains the unique NEG zone
+		// and name pair.
+		if _, exists := activeNegs[negInfo]; !exists {
+			inactiveNegRef := origNegRef.DeepCopy()
+			inactiveNegRef.State = negv1beta1.InactiveState
+			inactiveNegRefs = append(inactiveNegRefs, *inactiveNegRef)
+		}
+	}
+	return inactiveNegRefs
+}
+
 // getSyncedCondition returns the expected synced condition based on given error
 func getSyncedCondition(err error) negv1beta1.Condition {
 	if err != nil {

pkg/neg/syncers/transaction_test.go

@@ -1200,6 +1200,7 @@ func TestTransactionSyncerWithNegCR(t *testing.T) {
 			expectZones := sets.NewString(negtypes.TestZone1, negtypes.TestZone2, negtypes.TestZone4)
 
 			var expectedNegRefs map[string]negv1beta1.NegObjectReference
+			var err error
 			if tc.negExists {
 				for zone := range expectZones {
 					fakeCloud.CreateNetworkEndpointGroup(&composite.NetworkEndpointGroup{
@@ -1211,8 +1212,10 @@ func TestTransactionSyncerWithNegCR(t *testing.T) {
 						Description:         tc.negDesc,
 					}, zone, klog.TODO())
 				}
-				ret, _ := fakeCloud.AggregatedListNetworkEndpointGroup(syncer.NegSyncerKey.GetAPIVersion(), klog.TODO())
-				expectedNegRefs = negObjectReferences(ret)
+				expectedNegRefs, err = negObjectReferences(fakeCloud, negv1beta1.ActiveState, expectZones)
+				if err != nil {
+					t.Errorf("Failed to get negObjRef from NEG CR: %v", err)
+				}
 			}
 			var refs []negv1beta1.NegObjectReference
 			if tc.crStatusPopulated {
@@ -1243,16 +1246,25 @@ func TestTransactionSyncerWithNegCR(t *testing.T) {
 			if err != nil {
 				t.Errorf("Failed to get NEG from neg client: %s", err)
 			}
-			ret, _ := fakeCloud.AggregatedListNetworkEndpointGroup(syncer.NegSyncerKey.GetAPIVersion(), klog.TODO())
 			if !tc.expectErr {
-				expectedNegRefs = negObjectReferences(ret)
+				expectedNegRefs, err = negObjectReferences(fakeCloud, negv1beta1.ActiveState, expectZones)
+				if err != nil {
+					t.Errorf("Failed to get negObjRef from NEG CR: %v", err)
+				}
 			}
 			// if error occurs, expect that neg object references are not populated
 			if tc.expectErr && !tc.crStatusPopulated {
 				expectedNegRefs = nil
 			}
 
-			checkNegCR(t, negCR, creationTS, expectZones, expectedNegRefs, false, tc.expectErr)
+			// NEG Object References should exist if:
+			//  1. ensureNetworkEndpointGroups() doesn't result in errors, which
+			//     should populate the NEG Object Reference for NEGs that have
+			//     been successfully ensured.
+			//  2. NEG CR is owned by a differ syncer, and the NEG object refs
+			//     have been populated.
+			expectPopulatedNegRefs := !tc.expectErr || (tc.crStatusPopulated && tc.expectNoopOnNegStatus)
+			checkNegCR(t, negCR, creationTS, expectZones, nil, expectPopulatedNegRefs, false, fakeCloud)
 			if tc.expectErr && tc.expectNoopOnNegStatus {
 				// If CR is populated, we should have initialized and synced condition
 				var expectedConditionLen int
@@ -1285,6 +1297,7 @@ func TestTransactionSyncerWithNegCR(t *testing.T) {
 			// Verify the NEGs are created as expected
 			retZones := sets.NewString()
 
+			ret, _ := fakeCloud.AggregatedListNetworkEndpointGroup(syncer.NegSyncerKey.GetAPIVersion(), klog.TODO())
 			for key, neg := range ret {
 				retZones.Insert(key.Zone)
 				if neg.Name != testNegName {
@@ -1309,6 +1322,129 @@ func TestTransactionSyncerWithNegCR(t *testing.T) {
 	}
 }
 
+// TestUpdateInitStatus iterates over different zone transition situation, and
+// check if NEG Object Reference in the corresponding zone has the expected State.
+func TestUpdateInitStatus(t *testing.T) {
+	t.Parallel()
+	testNetwork := cloud.ResourcePath("network", &meta.Key{Name: "test-network"})
+	testSubnetwork := cloud.ResourcePath("subnetwork", &meta.Key{Name: "test-subnetwork"})
+	testNegType := negtypes.VmIpPortEndpointType
+
+	// Active zones: zone1, zone2.
+	// Inactive zones: zone3
+	oldActiveZones := sets.NewString(negtypes.TestZone1, negtypes.TestZone2)
+	oldInactiveZones := sets.NewString(negtypes.TestZone3)
+
+	testCases := []struct {
+		desc             string
+		newActiveZones   sets.String
+		newInactiveZones sets.String
+	}{
+		{
+			desc:             "Add a new zone zone4, an additional NEG ref should be added to NEG CR with ACTIVE status",
+			newActiveZones:   sets.NewString(negtypes.TestZone1, negtypes.TestZone2, negtypes.TestZone4),
+			newInactiveZones: sets.NewString(negtypes.TestZone3),
+		},
+		{
+			desc:             "Removed an ACTIVE zone zone2, corresponding NEG ref should still in NEG CR but with INACTIVE status",
+			newActiveZones:   sets.NewString(negtypes.TestZone1),
+			newInactiveZones: sets.NewString(negtypes.TestZone2, negtypes.TestZone3),
+		},
+		{
+			desc:           "Add back an INACTIVE zone zone3, the NEG ref in this zone should become ACTIVE in NEG CR",
+			newActiveZones: sets.NewString(negtypes.TestZone1, negtypes.TestZone2, negtypes.TestZone3),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			t.Parallel()
+
+			fakeCloud := negtypes.NewFakeNetworkEndpointGroupCloud(testSubnetwork, testNetwork)
+			_, syncer := newTestTransactionSyncer(fakeCloud, testNegType, false)
+			svcNegClient := syncer.svcNegClient
+
+			// Create initial NEGs, and get their Object Ref to be used in NEG CR.
+			var initialNegRefs []negv1beta1.NegObjectReference
+			for zone := range oldActiveZones.Union(oldInactiveZones) {
+				err := fakeCloud.CreateNetworkEndpointGroup(&composite.NetworkEndpointGroup{
+					Version:             syncer.NegSyncerKey.GetAPIVersion(),
+					Name:                testNegName,
+					NetworkEndpointType: string(syncer.NegSyncerKey.NegType),
+					Network:             fakeCloud.NetworkURL(),
+					Subnetwork:          fakeCloud.SubnetworkURL(),
+				}, zone, klog.TODO())
+				if err != nil {
+					t.Fatalf("Failed to create NEG %s in zone %s: %v", testNegName, zone, err)
+				}
+				neg, err := fakeCloud.GetNetworkEndpointGroup(testNegName, zone, meta.VersionGA, klog.TODO())
+				if err != nil {
+					t.Fatalf("Failed to get NEG %s in zone %s: %v", testNegName, zone, err)
+				}
+				negRef := negv1beta1.NegObjectReference{
+					Id:                  fmt.Sprint(neg.Id),
+					SelfLink:            neg.SelfLink,
+					NetworkEndpointType: negv1beta1.NetworkEndpointType(neg.NetworkEndpointType),
+					State:               negv1beta1.ActiveState,
+					SubnetURL:           neg.Subnetwork,
+				}
+				if oldInactiveZones.Has(zone) {
+					negRef.State = negv1beta1.InactiveState
+				}
+				initialNegRefs = append(initialNegRefs, negRef)
+			}
+
+			// Create NEG CR.
+			creationTS := v1.Now()
+			origCR := createNegCR(testNegName, creationTS, true, true, initialNegRefs)
+			svcNeg, err := svcNegClient.NetworkingV1beta1().ServiceNetworkEndpointGroups(testServiceNamespace).Create(context.Background(), origCR, v1.CreateOptions{})
+			if err != nil {
+				t.Errorf("Failed to create test NEG CR: %s", err)
+			}
+			syncer.svcNegLister.Add(svcNeg)
+
+			// Create a NEG in a new zone if zone expanded.
+			addedZones := tc.newActiveZones.Difference(oldActiveZones.Union(oldInactiveZones))
+			if addedZones != nil {
+				for zone := range addedZones {
+					err := fakeCloud.CreateNetworkEndpointGroup(&composite.NetworkEndpointGroup{
+						Version:             syncer.NegSyncerKey.GetAPIVersion(),
+						Name:                testNegName,
+						NetworkEndpointType: string(syncer.NegSyncerKey.NegType),
+						Network:             fakeCloud.NetworkURL(),
+						Subnetwork:          fakeCloud.SubnetworkURL(),
+					}, zone, klog.TODO())
+					if err != nil {
+						t.Fatalf("Failed to create NEG %s in zone %s: %v", testNegName, zone, err)
+					}
+				}
+			}
+
+			// This is the input list to updateInitStatus().
+			// It should only include NEG ref in the active zones.
+			var activeNegList []negv1beta1.NegObjectReference
+			for zone := range tc.newActiveZones {
+				neg, err := fakeCloud.GetNetworkEndpointGroup(testNegName, zone, meta.VersionGA, klog.TODO())
+				if err != nil {
+					t.Fatalf("Failed to get NEG %s in zone %s: %v", testNegName, zone, err)
+				}
+				negRef := getNegObjectReference(neg, negv1beta1.ActiveState)
+				activeNegList = append(activeNegList, negRef)
+			}
+
+			// Inactive NEG refs should be added if there is any.
+			syncer.updateInitStatus(activeNegList, nil)
+
+			negCR, err := svcNegClient.NetworkingV1beta1().ServiceNetworkEndpointGroups(testServiceNamespace).Get(context.Background(), testNegName, v1.GetOptions{})
+			if err != nil {
+				t.Errorf("Failed to create test NEG CR: %s", err)
+			}
+
+			checkNegCR(t, negCR, creationTS, tc.newActiveZones, tc.newInactiveZones, true, false, fakeCloud)
+		})
+	}
+}
+
 func TestUpdateStatus(t *testing.T) {
 	testNetwork := cloud.ResourcePath("network", &meta.Key{Name: "test-network"})
 	testSubnetwork := cloud.ResourcePath("subnetwork", &meta.Key{Name: "test-subnetwork"})
@@ -1502,8 +1638,11 @@ func TestIsZoneChange(t *testing.T) {
 					Subnetwork:          fakeCloud.SubnetworkURL(),
 				}, zone, klog.TODO())
 			}
-			ret, _ := fakeCloud.AggregatedListNetworkEndpointGroup(syncer.NegSyncerKey.GetAPIVersion(), klog.TODO())
-			negRefMap := negObjectReferences(ret)
+			negRefMap, err := negObjectReferences(fakeCloud, negv1beta1.ActiveState, sets.NewString(origZones...))
+			if err != nil {
+				t.Errorf("Failed to get negObjRef from NEG CR: %v", err)
+			}
+
 			var refs []negv1beta1.NegObjectReference
 			for _, neg := range negRefMap {
 				refs = append(refs, neg)
@@ -2474,17 +2613,28 @@ func waitForTransactions(syncer *transactionSyncer) error {
 }
 
 // negObjectReferences returns objectReferences for NEG CRs from NEG Objects
-func negObjectReferences(negs map[*meta.Key]*composite.NetworkEndpointGroup) map[string]negv1beta1.NegObjectReference {
-
+func negObjectReferences(cloud negtypes.NetworkEndpointGroupCloud, state negv1beta1.NegState, zones sets.String) (map[string]negv1beta1.NegObjectReference, error) {
 	negObjs := make(map[string]negv1beta1.NegObjectReference)
-	for _, neg := range negs {
-		negObjs[neg.SelfLink] = negv1beta1.NegObjectReference{
-			Id:                  fmt.Sprint(neg.Id),
-			SelfLink:            neg.SelfLink,
-			NetworkEndpointType: negv1beta1.NetworkEndpointType(neg.NetworkEndpointType),
+	for zone := range zones {
+		neg, err := cloud.GetNetworkEndpointGroup(testNegName, zone, meta.VersionGA, klog.TODO())
+		if err != nil {
+			return nil, err
 		}
+		negRef := getNegObjectReference(neg, state)
+		negObjs[neg.SelfLink] = negRef
+	}
+	return negObjs, nil
+}
+
+// getNegObjectReference returns objectReference for NEG CRs from NEG Object
+func getNegObjectReference(neg *composite.NetworkEndpointGroup, negState negv1beta1.NegState) negv1beta1.NegObjectReference {
+	return negv1beta1.NegObjectReference{
+		Id:                  fmt.Sprint(neg.Id),
+		SelfLink:            neg.SelfLink,
+		NetworkEndpointType: negv1beta1.NetworkEndpointType(neg.NetworkEndpointType),
+		State:               negState,
+		SubnetURL:           neg.Subnetwork,
 	}
-	return negObjs
 }
 
 // checks the NEG Description on the cloud NEG Object and verifies with expected
@@ -2583,13 +2733,32 @@ func createNegCR(testNegName string, creationTS metav1.Time, populateInitialized
 }
 
 // checkNegCR validates the NegObjectReferences and the LastSyncTime. It will not validate the conditions fields but ensures at most 2 conditions exist
-func checkNegCR(t *testing.T, negCR *negv1beta1.ServiceNetworkEndpointGroup, previousLastSyncTime metav1.Time, expectZones sets.String, expectedNegRefs map[string]negv1beta1.NegObjectReference, expectSyncTimeUpdate, expectErr bool) {
+func checkNegCR(t *testing.T, negCR *negv1beta1.ServiceNetworkEndpointGroup, previousLastSyncTime metav1.Time, activeZones, inactiveZones sets.String, expectPopulatedNegRefs, expectSyncTimeUpdate bool, cloud negtypes.NetworkEndpointGroupCloud) {
 	if expectSyncTimeUpdate && !previousLastSyncTime.Before(&negCR.Status.LastSyncTime) {
 		t.Errorf("Expected Neg CR to have an updated LastSyncTime")
 	} else if !expectSyncTimeUpdate && !negCR.Status.LastSyncTime.IsZero() && !previousLastSyncTime.Equal(&negCR.Status.LastSyncTime) {
 		t.Errorf("Expected Neg CR to not have an updated LastSyncTime")
 	}
 
+	expectedNegRefs := make(map[string]negv1beta1.NegObjectReference)
+
+	if expectPopulatedNegRefs {
+		ret, err := negObjectReferences(cloud, negv1beta1.ActiveState, activeZones)
+		if err != nil {
+			t.Fatalf("Failed to get negObjRef: %v", err)
+		}
+		for k, v := range ret {
+			expectedNegRefs[k] = v
+		}
+		ret, err = negObjectReferences(cloud, negv1beta1.InactiveState, inactiveZones)
+		if err != nil {
+			t.Fatalf("Failed to get negObjRef: %v", err)
+		}
+		for k, v := range ret {
+			expectedNegRefs[k] = v
+		}
+	}
+
 	var foundNegObjs []string
 	if len(negCR.Status.NetworkEndpointGroups) != len(expectedNegRefs) {
 		t.Errorf("Expected Neg CR to have %d corresponding neg object references, but has %d", len(expectedNegRefs), len(negCR.Status.NetworkEndpointGroups))

pkg/neg/syncers/utils.go

@@ -227,6 +227,8 @@ func ensureNetworkEndpointGroup(svcNamespace, svcName, negName, zone, negService
 		Id:                  fmt.Sprint(neg.Id),
 		SelfLink:            neg.SelfLink,
 		NetworkEndpointType: negv1beta1.NetworkEndpointType(neg.NetworkEndpointType),
+		State:               negv1beta1.ActiveState,
+		SubnetURL:           neg.Subnetwork,
 	}
 	return negRef, nil
 }

pkg/neg/syncers/utils_test.go

@@ -1337,11 +1337,16 @@ func TestNegObjectCrd(t *testing.T) {
 			t.Errorf("Failed to find neg")
 		}
 
-		var expectedNegObj negv1beta1.NegObjectReference
-		expectedNegObj = negv1beta1.NegObjectReference{
+		var subnetURL string
+		if networkEndpointType != negtypes.NonGCPPrivateEndpointType {
+			subnetURL = testSubnetwork
+		}
+		expectedNegObj := negv1beta1.NegObjectReference{
 			Id:                  fmt.Sprint(neg.Id),
 			SelfLink:            neg.SelfLink,
 			NetworkEndpointType: negv1beta1.NetworkEndpointType(networkEndpointType),
+			State:               negv1beta1.ActiveState,
+			SubnetURL:           subnetURL,
 		}
 
 		if negObj != expectedNegObj {