UPSTREAM: 84466: gce: ensureInternalInstanceGroups: reuse instance-gr…

…oups for internal load balancers

UPSTREAM: 84466:  legacy-cloud-providers/gce/gce_fake.go: NewFakeGCECloud: make sure that the secondary zone is also part of managedZones
Origin-commit: 79d66e294a3906efd0351f125cefb4b9cc1c9ab4

UPSTREAM: 84466:  gce: ensureInternalInstanceGroups: reuse instance-groups for internal load balancers
Origin-commit: cfb25370a7c8f9bed9688cb334b4bc1c3342da0d

UPSTREAM: 84466: gce: add ExternalInstanceGroupsPrefix to filter instance groups that will be re-used for ILB backend
Origin-commit: e29c0b6ce3c068e02419a7b3cbc381b919981f50

UPSTREAM: 84466: gce: skip ensureInstanceGroup for a zone that has no remaining nodes for k8s managed IG
Origin-commit: 3915cef99ee4eedc9755d454abb7e4efa2a63bff

staging/src/k8s.io/legacy-cloud-providers/gce/gce.go

@@ -142,6 +142,9 @@ type Cloud struct {
 	useMetadataServer        bool
 	operationPollRateLimiter flowcontrol.RateLimiter
 	manager                  diskServiceManager
+
+	externalInstanceGroupsPrefix string // If non-"", finds prefixed instance groups for ILB.
+
 	// Lock for access to nodeZones
 	nodeZonesLock sync.Mutex
 	// nodeZones is a mapping from Zone to a sets.String of Node's names in the Zone
@@ -189,6 +192,9 @@ type ConfigGlobal struct {
 	NodeInstancePrefix string   `gcfg:"node-instance-prefix"`
 	Regional           bool     `gcfg:"regional"`
 	Multizone          bool     `gcfg:"multizone"`
+	// ExternalInstanceGroupsPrefix is the prefix that will be used to filter instance groups
+	// that be backend for ILB containing cluster nodes if not-empty.
+	ExternalInstanceGroupsPrefix string `gcfg:"external-instance-groups-prefix"`
 	// APIEndpoint is the GCE compute API endpoint to use. If this is blank,
 	// then the default endpoint is used.
 	APIEndpoint string `gcfg:"api-endpoint"`
@@ -229,12 +235,13 @@ type CloudConfig struct {
 	SubnetworkName       string
 	SubnetworkURL        string
 	// DEPRECATED: Do not rely on this value as it may be incorrect.
-	SecondaryRangeName string
-	NodeTags           []string
-	NodeInstancePrefix string
-	TokenSource        oauth2.TokenSource
-	UseMetadataServer  bool
-	AlphaFeatureGate   *AlphaFeatureGate
+	SecondaryRangeName           string
+	NodeTags                     []string
+	NodeInstancePrefix           string
+	ExternalInstanceGroupsPrefix string
+	TokenSource                  oauth2.TokenSource
+	UseMetadataServer            bool
+	AlphaFeatureGate             *AlphaFeatureGate
 }
 
 func init() {
@@ -324,6 +331,7 @@ func generateCloudConfig(configFile *ConfigFile) (cloudConfig *CloudConfig, err
 
 		cloudConfig.NodeTags = configFile.Global.NodeTags
 		cloudConfig.NodeInstancePrefix = configFile.Global.NodeInstancePrefix
+		cloudConfig.ExternalInstanceGroupsPrefix = configFile.Global.ExternalInstanceGroupsPrefix
 		cloudConfig.AlphaFeatureGate = NewAlphaFeatureGate(configFile.Global.AlphaFeatures)
 	}
 
@@ -501,29 +509,30 @@ func CreateGCECloud(config *CloudConfig) (*Cloud, error) {
 	operationPollRateLimiter := flowcontrol.NewTokenBucketRateLimiter(5, 5) // 5 qps, 5 burst.
 
 	gce := &Cloud{
-		service:                  service,
-		serviceAlpha:             serviceAlpha,
-		serviceBeta:              serviceBeta,
-		containerService:         containerService,
-		tpuService:               tpuService,
-		projectID:                projID,
-		networkProjectID:         netProjID,
-		onXPN:                    onXPN,
-		region:                   config.Region,
-		regional:                 config.Regional,
-		localZone:                config.Zone,
-		managedZones:             config.ManagedZones,
-		networkURL:               networkURL,
-		unsafeIsLegacyNetwork:    isLegacyNetwork,
-		unsafeSubnetworkURL:      subnetURL,
-		secondaryRangeName:       config.SecondaryRangeName,
-		nodeTags:                 config.NodeTags,
-		nodeInstancePrefix:       config.NodeInstancePrefix,
-		useMetadataServer:        config.UseMetadataServer,
-		operationPollRateLimiter: operationPollRateLimiter,
-		AlphaFeatureGate:         config.AlphaFeatureGate,
-		nodeZones:                map[string]sets.String{},
-		metricsCollector:         newLoadBalancerMetrics(),
+		service:                      service,
+		serviceAlpha:                 serviceAlpha,
+		serviceBeta:                  serviceBeta,
+		containerService:             containerService,
+		tpuService:                   tpuService,
+		projectID:                    projID,
+		networkProjectID:             netProjID,
+		onXPN:                        onXPN,
+		region:                       config.Region,
+		regional:                     config.Regional,
+		localZone:                    config.Zone,
+		managedZones:                 config.ManagedZones,
+		networkURL:                   networkURL,
+		unsafeIsLegacyNetwork:        isLegacyNetwork,
+		unsafeSubnetworkURL:          subnetURL,
+		secondaryRangeName:           config.SecondaryRangeName,
+		nodeTags:                     config.NodeTags,
+		nodeInstancePrefix:           config.NodeInstancePrefix,
+		externalInstanceGroupsPrefix: config.ExternalInstanceGroupsPrefix,
+		useMetadataServer:            config.UseMetadataServer,
+		operationPollRateLimiter:     operationPollRateLimiter,
+		AlphaFeatureGate:             config.AlphaFeatureGate,
+		nodeZones:                    map[string]sets.String{},
+		metricsCollector:             newLoadBalancerMetrics(),
 	}
 
 	gce.manager = &gceServiceManager{gce}

staging/src/k8s.io/legacy-cloud-providers/gce/gce_fake.go

@@ -71,7 +71,7 @@ func NewFakeGCECloud(vals TestClusterValues) *Cloud {
 	gce := &Cloud{
 		region:           vals.Region,
 		service:          service,
-		managedZones:     []string{vals.ZoneName},
+		managedZones:     []string{vals.ZoneName, vals.SecondaryZoneName},
 		projectID:        vals.ProjectID,
 		networkProjectID: vals.ProjectID,
 		ClusterID:        fakeClusterID(vals.ClusterID),

staging/src/k8s.io/legacy-cloud-providers/gce/gce_instancegroup.go

@@ -19,6 +19,8 @@ limitations under the License.
 package gce
 
 import (
+	"fmt"
+
 	compute "google.golang.org/api/compute/v1"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud"
@@ -60,6 +62,21 @@ func (g *Cloud) ListInstanceGroups(zone string) ([]*compute.InstanceGroup, error
 	return v, mc.Observe(err)
 }
 
+// ListInstanceGroupsWithPrefix lists all InstanceGroups in the project and
+// zone with given prefix.
+func (g *Cloud) ListInstanceGroupsWithPrefix(zone string, prefix string) ([]*compute.InstanceGroup, error) {
+	ctx, cancel := cloud.ContextWithCallTimeout()
+	defer cancel()
+
+	mc := newInstanceGroupMetricContext("list", zone)
+	f := filter.None
+	if prefix != "" {
+		f = filter.Regexp("name", fmt.Sprintf("%s.*", prefix))
+	}
+	v, err := g.c.InstanceGroups().List(ctx, zone, f)
+	return v, mc.Observe(err)
+}
+
 // ListInstancesInInstanceGroup lists all the instances in a given
 // instance group and state.
 func (g *Cloud) ListInstancesInInstanceGroup(name string, zone string, state string) ([]*compute.InstanceWithNamedPorts, error) {

staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal.go

@@ -556,17 +556,14 @@ func (g *Cloud) ensureInternalHealthCheck(name string, svcName types.NamespacedN
 	return hc, nil
 }
 
-func (g *Cloud) ensureInternalInstanceGroup(name, zone string, nodes []*v1.Node) (string, error) {
+func (g *Cloud) ensureInternalInstanceGroup(name, zone string, nodes []string) (string, error) {
 	klog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): checking group that it contains %v nodes", name, zone, len(nodes))
 	ig, err := g.GetInstanceGroup(name, zone)
 	if err != nil && !isNotFound(err) {
 		return "", err
 	}
 
-	kubeNodes := sets.NewString()
-	for _, n := range nodes {
-		kubeNodes.Insert(n.Name)
-	}
+	kubeNodes := sets.NewString(nodes...)
 
 	// Individual InstanceGroup has a limit for 1000 instances in it.
 	// As a result, it's not possible to add more to it.
@@ -632,8 +629,46 @@ func (g *Cloud) ensureInternalInstanceGroups(name string, nodes []*v1.Node) ([]s
 	zonedNodes := splitNodesByZone(nodes)
 	klog.V(2).Infof("ensureInternalInstanceGroups(%v): %d nodes over %d zones in region %v", name, len(nodes), len(zonedNodes), g.region)
 	var igLinks []string
-	for zone, nodes := range zonedNodes {
-		igLink, err := g.ensureInternalInstanceGroup(name, zone, nodes)
+	gceZonedNodes := map[string][]string{}
+	for zone, zNodes := range zonedNodes {
+		hosts, err := g.getFoundInstanceByNames(nodeNames(zNodes))
+		if err != nil {
+			return nil, err
+		}
+		names := sets.NewString()
+		for _, h := range hosts {
+			names.Insert(h.Name)
+		}
+		skip := sets.NewString()
+
+		igs, err := g.candidateExternalInstanceGroups(zone)
+		if err != nil {
+			return nil, err
+		}
+		for _, ig := range igs {
+			if strings.EqualFold(ig.Name, name) {
+				continue
+			}
+			instances, err := g.ListInstancesInInstanceGroup(ig.Name, zone, allInstances)
+			if err != nil {
+				return nil, err
+			}
+			groupInstances := sets.NewString()
+			for _, ins := range instances {
+				parts := strings.Split(ins.Instance, "/")
+				groupInstances.Insert(parts[len(parts)-1])
+			}
+			if names.HasAll(groupInstances.UnsortedList()...) {
+				igLinks = append(igLinks, ig.SelfLink)
+				skip.Insert(groupInstances.UnsortedList()...)
+			}
+		}
+		if remaining := names.Difference(skip).UnsortedList(); len(remaining) > 0 {
+			gceZonedNodes[zone] = remaining
+		}
+	}
+	for zone, gceNodes := range gceZonedNodes {
+		igLink, err := g.ensureInternalInstanceGroup(name, zone, gceNodes)
 		if err != nil {
 			return []string{}, err
 		}
@@ -643,6 +678,13 @@ func (g *Cloud) ensureInternalInstanceGroups(name string, nodes []*v1.Node) ([]s
 	return igLinks, nil
 }
 
+func (g *Cloud) candidateExternalInstanceGroups(zone string) ([]*compute.InstanceGroup, error) {
+	if g.externalInstanceGroupsPrefix == "" {
+		return nil, nil
+	}
+	return g.ListInstanceGroupsWithPrefix(zone, g.externalInstanceGroupsPrefix)
+}
+
 func (g *Cloud) ensureInternalInstanceGroupsDeleted(name string) error {
 	// List of nodes isn't available here - fetch all zones in region and try deleting this cluster's ig
 	zones, err := g.ListZonesInRegion(g.region)

staging/src/k8s.io/legacy-cloud-providers/gce/gce_loadbalancer_internal_test.go

@@ -807,6 +807,75 @@ func TestEnsureLoadBalancerDeletedSucceedsOnXPN(t *testing.T) {
 	checkEvent(t, recorder, FilewallChangeMsg, true)
 }
 
+func TestEnsureInternalInstanceGroupsReuseGroups(t *testing.T) {
+	vals := DefaultTestClusterValues()
+	gce, err := fakeGCECloud(vals)
+	require.NoError(t, err)
+	gce.externalInstanceGroupsPrefix = "pre-existing"
+
+	igName := makeInstanceGroupName(vals.ClusterID)
+	nodesA, err := createAndInsertNodes(gce, []string{"test-node-1", "test-node-2"}, vals.ZoneName)
+	require.NoError(t, err)
+	nodesB, err := createAndInsertNodes(gce, []string{"test-node-3"}, vals.SecondaryZoneName)
+	require.NoError(t, err)
+
+	preIGName := "pre-existing-ig"
+	err = gce.CreateInstanceGroup(&compute.InstanceGroup{Name: preIGName}, vals.ZoneName)
+	require.NoError(t, err)
+	err = gce.CreateInstanceGroup(&compute.InstanceGroup{Name: preIGName}, vals.SecondaryZoneName)
+	require.NoError(t, err)
+	err = gce.AddInstancesToInstanceGroup(preIGName, vals.ZoneName, gce.ToInstanceReferences(vals.ZoneName, []string{"test-node-1"}))
+	require.NoError(t, err)
+	err = gce.AddInstancesToInstanceGroup(preIGName, vals.SecondaryZoneName, gce.ToInstanceReferences(vals.SecondaryZoneName, []string{"test-node-3"}))
+	require.NoError(t, err)
+
+	anotherPreIGName := "another-existing-ig"
+	err = gce.CreateInstanceGroup(&compute.InstanceGroup{Name: anotherPreIGName}, vals.ZoneName)
+	require.NoError(t, err)
+	err = gce.AddInstancesToInstanceGroup(anotherPreIGName, vals.ZoneName, gce.ToInstanceReferences(vals.ZoneName, []string{"test-node-2"}))
+	require.NoError(t, err)
+
+	svc := fakeLoadbalancerService(string(LBTypeInternal))
+	svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	_, err = gce.ensureInternalLoadBalancer(
+		vals.ClusterName, vals.ClusterID,
+		svc,
+		nil,
+		append(nodesA, nodesB...),
+	)
+	assert.NoError(t, err)
+
+	backendServiceName := makeBackendServiceName(gce.GetLoadBalancerName(context.TODO(), "", svc), vals.ClusterID, shareBackendService(svc), cloud.SchemeInternal, "TCP", svc.Spec.SessionAffinity)
+	bs, err := gce.GetRegionBackendService(backendServiceName, gce.region)
+	require.NoError(t, err)
+	assert.Equal(t, 3, len(bs.Backends), "Want three backends referencing three instances groups")
+
+	igRef := func(zone, name string) string {
+		return fmt.Sprintf("zones/%s/instanceGroups/%s", zone, name)
+	}
+	for _, name := range []string{igRef(vals.ZoneName, preIGName), igRef(vals.SecondaryZoneName, preIGName), igRef(vals.ZoneName, igName)} {
+		var found bool
+		for _, be := range bs.Backends {
+			if strings.Contains(be.Group, name) {
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Expected list of backends to have group %q", name)
+	}
+
+	// Expect initial zone to have test-node-2
+	instances, err := gce.ListInstancesInInstanceGroup(igName, vals.ZoneName, "ALL")
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(instances))
+	assert.Contains(
+		t,
+		instances[0].Instance,
+		fmt.Sprintf("projects/%s/zones/%s/instances/%s", vals.ProjectID, vals.ZoneName, "test-node-2"),
+	)
+}
+
 func TestEnsureInternalInstanceGroupsDeleted(t *testing.T) {
 	vals := DefaultTestClusterValues()
 	gce, err := fakeGCECloud(vals)