Create additional NEGs based on subnets in Node Topology CR. (#2702)

* Update namer interface to include NonDefaultSubnetNEG

* Exclude NodeTopologyInformer from hasSynced.

* Remove NodeTopologyInformer from controller context hasSynced so it
  won't block controllers from starting up in case of the CRD or CR does
  not exist.

* Add NodeTopologyInformer to zoneGetter.

* Fall back to only include default if nodeTopology CR isn't ready.

* Create additional NEGs based on subnets in Node Topology CR.

* Query Node Topology CR for the current set of NEGs in the cluster.
* When ensureNetworkEndpointGroups(), ensure NEGs are properly
  provisioned in the non-default subnets as well.

* Use correct namer based on the NEG type.

pkg/backends/ig_linker_test.go

@@ -40,7 +40,7 @@ import (
 
 const (
 	defaultTestZone      = "zone-a"
-	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 )
 
 func newTestIGLinker(fakeGCE *gce.Cloud, fakeInstancePool instancegroups.Manager) *instanceGroupLinker {
@@ -59,7 +59,7 @@ func TestLink(t *testing.T) {
 	fakeGCE := gce.NewFakeGCECloud(gce.DefaultTestClusterValues())
 
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.AddFakeNodes(fakeZoneGetter, defaultTestZone, "test-instance")
 
 	fakeNodePool := instancegroups.NewManager(&instancegroups.ManagerConfig{
@@ -101,7 +101,7 @@ func TestLinkWithCreationModeError(t *testing.T) {
 	fakeGCE := gce.NewFakeGCECloud(gce.DefaultTestClusterValues())
 
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.AddFakeNodes(fakeZoneGetter, defaultTestZone, "test-instance")
 
 	fakeNodePool := instancegroups.NewManager(&instancegroups.ManagerConfig{

pkg/backends/integration_test.go

@@ -53,7 +53,7 @@ func newTestJig(fakeGCE *gce.Cloud) *Jig {
 	fakeIGs := instancegroups.NewEmptyFakeInstanceGroups()
 
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.AddFakeNodes(fakeZoneGetter, defaultTestZone, "test-instance")
 
 	fakeInstancePool := instancegroups.NewManager(&instancegroups.ManagerConfig{

pkg/backends/regional_ig_linker_test.go

@@ -53,7 +53,7 @@ func newTestRegionalIgLinker(fakeGCE *gce.Cloud, backendPool *Pool, l4Namer *nam
 	fakeIGs := instancegroups.NewEmptyFakeInstanceGroups()
 
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.AddFakeNodes(fakeZoneGetter, usCentral1AZone, "test-instance1")
 	zonegetter.AddFakeNodes(fakeZoneGetter, "us-central1-c", "test-instance2")
 

pkg/context/context.go

@@ -250,7 +250,7 @@ func NewControllerContext(
 		logger,
 	)
 	// The subnet specified in gce.conf is considered as the default subnet.
-	context.ZoneGetter = zonegetter.NewZoneGetter(context.NodeInformer, context.Cloud.SubnetworkURL())
+	context.ZoneGetter = zonegetter.NewZoneGetter(context.NodeInformer, context.NodeTopologyInformer, context.Cloud.SubnetworkURL())
 	context.InstancePool = instancegroups.NewManager(&instancegroups.ManagerConfig{
 		Cloud:      context.Cloud,
 		Namer:      context.ClusterNamer,
@@ -330,10 +330,6 @@ func (ctx *ControllerContext) HasSynced() bool {
 		funcs = append(funcs, ctx.FirewallInformer.HasSynced)
 	}
 
-	if ctx.NodeTopologyInformer != nil {
-		funcs = append(funcs, ctx.NodeTopologyInformer.HasSynced)
-	}
-
 	for _, f := range funcs {
 		if !f() {
 			return false

pkg/controller/controller_test.go

@@ -58,7 +58,7 @@ import (
 	"k8s.io/ingress-gce/pkg/utils/zonegetter"
 )
 
-const defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+const defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 
 var (
 	nodePortCounter = 30000
@@ -71,9 +71,11 @@ func newLoadBalancerController() *LoadBalancerController {
 	kubeClient := fake.NewSimpleClientset()
 	backendConfigClient := backendconfigclient.NewSimpleClientset()
 	svcNegClient := svcnegclient.NewSimpleClientset()
-	fakeGCE := gce.NewFakeGCECloud(gce.DefaultTestClusterValues())
+	vals := gce.DefaultTestClusterValues()
+	vals.SubnetworkURL = defaultTestSubnetURL
+	fakeGCE := gce.NewFakeGCECloud(vals)
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.AddFakeNodes(fakeZoneGetter, fakeZone, "test-node")
 
 	(fakeGCE.Compute().(*cloud.MockGCE)).MockGlobalForwardingRules.InsertHook = loadbalancers.InsertGlobalForwardingRuleHook

pkg/instancegroups/controller_test.go

@@ -98,7 +98,7 @@ func TestSync(t *testing.T) {
 	config.HasSynced = func() bool {
 		return true
 	}
-	config.ZoneGetter = zonegetter.NewFakeZoneGetter(informer, defaultTestSubnetURL, false)
+	config.ZoneGetter = zonegetter.NewFakeZoneGetter(informer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 
 	controller := NewController(config, logr.Logger{})
 

pkg/instancegroups/manager_test.go

@@ -43,14 +43,14 @@ const (
 	testZoneC       = "dark-moon1-c"
 	basePath        = "/basepath/projects/project-id/"
 
-	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 )
 
 var defaultNamer = namer.NewNamer("uid1", "fw1", klog.TODO())
 
 func newNodePool(f Provider, maxIGSize int) Manager {
 	nodeInformer := zonegetter.FakeNodeInformer()
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 
 	pool := NewManager(&ManagerConfig{
 		Cloud:      f,

pkg/l4lb/l4controller_test.go

@@ -50,6 +50,7 @@ import (
 	"k8s.io/ingress-gce/pkg/test"
 	"k8s.io/ingress-gce/pkg/utils/common"
 	"k8s.io/ingress-gce/pkg/utils/namer"
+	"k8s.io/ingress-gce/pkg/utils/zonegetter"
 )
 
 const (
@@ -852,6 +853,7 @@ func newServiceController(t *testing.T, fakeGCE *gce.Cloud) *L4Controller {
 		NumL4Workers: 5,
 	}
 	ctx := context.NewControllerContext(kubeClient, nil, nil, nil, svcNegClient, nil, nil, nil, kubeClient /*kube client to be used for events*/, fakeGCE, namer, "" /*kubeSystemUID*/, ctxConfig, klog.TODO())
+	ctx.ZoneGetter = zonegetter.NewFakeZoneGetter(ctx.NodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	// Add some nodes so that NEG linker kicks in during ILB creation.
 	nodes, err := test.CreateAndInsertNodes(ctx.Cloud, []string{"instance-1"}, vals.ZoneName)
 	if err != nil {

pkg/l4lb/l4netlbcontroller_test.go

@@ -76,7 +76,7 @@ const (
 	shortSessionAffinityIdleTimeout = int32(20)     // 20 sec could be used for regular Session Affinity
 	longSessionAffinityIdleTimeout  = int32(2 * 60) // 2 min or 120 sec for Strong Session Affinity
 
-	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 )
 
 var (

pkg/neg/controller.go

@@ -49,7 +49,7 @@ import (
 	svcnegclient "k8s.io/ingress-gce/pkg/svcneg/client/clientset/versioned"
 	"k8s.io/ingress-gce/pkg/utils"
 	"k8s.io/ingress-gce/pkg/utils/endpointslices"
-	namer2 "k8s.io/ingress-gce/pkg/utils/namer"
+	"k8s.io/ingress-gce/pkg/utils/namer"
 	"k8s.io/ingress-gce/pkg/utils/patch"
 	"k8s.io/ingress-gce/pkg/utils/zonegetter"
 	"k8s.io/klog/v2"
@@ -68,7 +68,7 @@ type Controller struct {
 	gcPeriod        time.Duration
 	recorder        record.EventRecorder
 	namer           negtypes.NetworkEndpointGroupNamer
-	l4Namer         namer2.L4ResourcesNamer
+	l4Namer         namer.L4ResourcesNamer
 	zoneGetter      *zonegetter.ZoneGetter
 	networkResolver network.Resolver
 
@@ -135,7 +135,7 @@ func NewController(
 	gkeNetworkParamSetInformer cache.SharedIndexInformer,
 	nodeTopologyInformer cache.SharedIndexInformer,
 	hasSynced func() bool,
-	l4Namer namer2.L4ResourcesNamer,
+	l4Namer namer.L4ResourcesNamer,
 	defaultBackendService utils.ServicePort,
 	cloud negtypes.NetworkEndpointGroupCloud,
 	zoneGetter *zonegetter.ZoneGetter,
@@ -182,6 +182,7 @@ func NewController(
 	syncerMetrics := syncMetrics.NewNegMetricsCollector(flags.F.NegMetricsExportInterval, logger)
 	manager := newSyncerManager(
 		namer,
+		l4Namer,
 		recorder,
 		cloud,
 		zoneGetter,

pkg/neg/controller_test.go

@@ -124,7 +124,7 @@ func newTestControllerWithParamsAndContext(kubeClient kubernetes.Interface, test
 	}
 	nodeInformer := zonegetter.FakeNodeInformer()
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 
 	return NewController(
 		kubeClient,
@@ -1756,7 +1756,7 @@ func validateServiceAnnotationWithPortInfoMap(t *testing.T, svc *apiv1.Service,
 
 	nodeInformer := zonegetter.FakeNodeInformer()
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zones, _ := zoneGetter.ListZones(negtypes.NodeFilterForEndpointCalculatorMode(portInfoMap.EndpointsCalculatorMode()), klog.TODO())
 	if !sets.NewString(expectZones...).Equal(sets.NewString(zones...)) {
 		t.Errorf("Unexpected zones listed by the predicate function, got %v, want %v", zones, expectZones)
@@ -1837,7 +1837,7 @@ func validateServiceStateAnnotationExceptNames(t *testing.T, svc *apiv1.Service,
 	}
 	nodeInformer := zonegetter.FakeNodeInformer()
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	// This routine is called from tests verifying L7 NEGs.
 	zones, _ := zoneGetter.ListZones(negtypes.NodeFilterForEndpointCalculatorMode(negtypes.L7Mode), klog.TODO())
 
@@ -2115,6 +2115,7 @@ func newTestNode(name string, unschedulable bool) *apiv1.Node {
 			Name: name,
 		},
 		Spec: apiv1.NodeSpec{
+			PodCIDR:       "10.100.1.0/24",
 			Unschedulable: unschedulable,
 		},
 	}

pkg/neg/manager.go

@@ -46,6 +46,7 @@ import (
 	svcnegclient "k8s.io/ingress-gce/pkg/svcneg/client/clientset/versioned"
 	"k8s.io/ingress-gce/pkg/utils"
 	"k8s.io/ingress-gce/pkg/utils/common"
+	"k8s.io/ingress-gce/pkg/utils/namer"
 	"k8s.io/ingress-gce/pkg/utils/patch"
 	"k8s.io/ingress-gce/pkg/utils/zonegetter"
 	"k8s.io/klog/v2"
@@ -64,6 +65,7 @@ func (k serviceKey) Key() string {
 // syncerManager contains all the active syncer goroutines and manage their lifecycle.
 type syncerManager struct {
 	namer      negtypes.NetworkEndpointGroupNamer
+	l4Namer    namer.L4ResourcesNamer
 	recorder   record.EventRecorder
 	cloud      negtypes.NetworkEndpointGroupCloud
 	zoneGetter *zonegetter.ZoneGetter
@@ -118,6 +120,7 @@ type syncerManager struct {
 }
 
 func newSyncerManager(namer negtypes.NetworkEndpointGroupNamer,
+	l4Namer namer.L4ResourcesNamer,
 	recorder record.EventRecorder,
 	cloud negtypes.NetworkEndpointGroupCloud,
 	zoneGetter *zonegetter.ZoneGetter,
@@ -140,6 +143,7 @@ func newSyncerManager(namer negtypes.NetworkEndpointGroupNamer,
 
 	return &syncerManager{
 		namer:               namer,
+		l4Namer:             l4Namer,
 		recorder:            recorder,
 		cloud:               cloud,
 		zoneGetter:          zoneGetter,
@@ -228,6 +232,11 @@ func (manager *syncerManager) EnsureSyncers(namespace, name string, newPorts neg
 				&portInfo.NetworkInfo,
 				portInfo.L4LBType,
 			)
+			nonDefaultSubnetNEGNamer := manager.namer
+			if syncerKey.NegType == negtypes.VmIpEndpointType {
+				nonDefaultSubnetNEGNamer = manager.l4Namer
+			}
+
 			syncer = negsyncer.NewTransactionSyncer(
 				syncerKey,
 				manager.recorder,
@@ -243,11 +252,12 @@ func (manager *syncerManager) EnsureSyncers(namespace, name string, newPorts neg
 				string(manager.kubeSystemUID),
 				manager.svcNegClient,
 				manager.syncerMetrics,
-				!manager.namer.IsNEG(portInfo.NegName),
+				syncerKey.NegType == negtypes.VmIpPortEndpointType && !manager.namer.IsNEG(portInfo.NegName),
 				manager.logger,
 				manager.lpConfig,
 				manager.enableDualStackNEG,
 				portInfo.NetworkInfo,
+				nonDefaultSubnetNEGNamer,
 			)
 			manager.syncerMap[syncerKey] = syncer
 		}

pkg/neg/manager_test.go

@@ -82,16 +82,17 @@ const (
 	negName1    = "neg1"
 
 	defaultTestSubnet    = "default"
-	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 )
 
 func NewTestSyncerManager(kubeClient kubernetes.Interface) (*syncerManager, *gce.Cloud, *negtypes.TestContext) {
 	testContext := negtypes.NewTestContextWithKubeClient(kubeClient)
 	nodeInformer := zonegetter.FakeNodeInformer()
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	manager := newSyncerManager(
 		testContext.NegNamer,
+		testContext.L4Namer,
 		record.NewFakeRecorder(100),
 		negtypes.NewAdapter(testContext.Cloud),
 		zoneGetter,

pkg/neg/readiness/reflector_test.go

@@ -35,7 +35,7 @@ import (
 )
 
 const (
-	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/proj/regions/us-central1/subnetworks/default"
+	defaultTestSubnetURL = "https://www.googleapis.com/compute/v1/projects/mock-project/regions/test-region/subnetworks/default"
 
 	defaultTestSubnet    = "default"
 	nonDefaultTestSubnet = "non-default"
@@ -59,7 +59,7 @@ func (f *fakeLookUp) ReadinessGateEnabled(syncerKey negtypes.NegSyncerKey) bool
 }
 
 func newTestReadinessReflector(testContext *negtypes.TestContext, enableMultiSubnetCluster bool) *readinessReflector {
-	fakeZoneGetter := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, defaultTestSubnetURL, enableMultiSubnetCluster)
+	fakeZoneGetter := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, testContext.NodeTopologyInformer, defaultTestSubnetURL, enableMultiSubnetCluster)
 	reflector := NewReadinessReflector(
 		testContext.KubeClient,
 		testContext.KubeClient,

pkg/neg/syncers/endpoints_calculator_test.go

@@ -42,8 +42,9 @@ import (
 // The L7 implementation is tested in TestToZoneNetworkEndpointMapUtil.
 func TestLocalGetEndpointSet(t *testing.T) {
 	nodeInformer := zonegetter.FakeNodeInformer()
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
+	zonegetter.SetNodeTopologyHasSynced(zoneGetter, func() bool { return true })
 	defaultNetwork := network.NetworkInfo{IsDefault: true, K8sNetwork: "default", SubnetworkURL: defaultTestSubnetURL}
 	prevFlag := flags.F.EnableMultiSubnetCluster
 	defer func() { flags.F.EnableMultiSubnetCluster = prevFlag }()
@@ -196,8 +197,9 @@ func nodeInterfacesAnnotation(t *testing.T, network, ip string) string {
 // TestClusterGetEndpointSet verifies the GetEndpointSet method implemented by the ClusterL4EndpointsCalculator.
 func TestClusterGetEndpointSet(t *testing.T) {
 	nodeInformer := zonegetter.FakeNodeInformer()
-	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(nodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	zonegetter.PopulateFakeNodeInformer(nodeInformer, false)
+	zonegetter.SetNodeTopologyHasSynced(zoneGetter, func() bool { return true })
 	defaultNetwork := network.NetworkInfo{IsDefault: true, K8sNetwork: "default", SubnetworkURL: defaultTestSubnetURL}
 	prevFlag := flags.F.EnableMultiSubnetCluster
 	defer func() { flags.F.EnableMultiSubnetCluster = prevFlag }()
@@ -397,10 +399,10 @@ func TestValidateEndpoints(t *testing.T) {
 	nodeLister := testContext.NodeInformer.GetIndexer()
 	serviceLister := testContext.ServiceInformer.GetIndexer()
 	zonegetter.PopulateFakeNodeInformer(testContext.NodeInformer, true)
-	zoneGetter := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, defaultTestSubnetURL, false)
+	zoneGetter := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, false)
 	L7EndpointsCalculator := NewL7EndpointsCalculator(zoneGetter, podLister, nodeLister, serviceLister, svcPort, klog.TODO(), testContext.EnableDualStackNEG, metricscollector.FakeSyncerMetrics())
 
-	zoneGetterMSC := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, defaultTestSubnetURL, true)
+	zoneGetterMSC := zonegetter.NewFakeZoneGetter(testContext.NodeInformer, zonegetter.FakeNodeTopologyInformer(), defaultTestSubnetURL, true)
 	L7EndpointsCalculatorMSC := NewL7EndpointsCalculator(zoneGetterMSC, podLister, nodeLister, serviceLister, svcPort, klog.TODO(), testContext.EnableDualStackNEG, metricscollector.FakeSyncerMetrics())
 	L7EndpointsCalculatorMSC.enableMultiSubnetCluster = true
 	L4LocalEndpointCalculator := NewLocalL4EndpointsCalculator(listers.NewNodeLister(nodeLister), zoneGetter, fmt.Sprintf("%s/%s", testServiceName, testServiceNamespace), klog.TODO(), &network.NetworkInfo{SubnetworkURL: defaultTestSubnetURL}, negtypes.L4InternalLB)