Added new controller for L4 ILB services

Fixes including adding CustomSubnet support and self links for backend
service, healthcheck.
Removed the operation from queue key
Added connectiondraining to backend service.
Added a new flag to run L4 controller.
Used namer.PrimaryIPNEG() instead of namer.NEG()
Use a common description for all ILB resources.

cmd/glbc/main.go

@@ -218,6 +218,12 @@ func runControllers(ctx *ingctx.ControllerContext) {
 
 	fwc := firewalls.NewFirewallController(ctx, flags.F.NodePortRanges.Values())
 
+	if flags.F.RunL4Controller {
+		l4Controller := controller.NewL4Controller(ctx, stopCh)
+		go l4Controller.Run()
+		klog.V(0).Infof("L4 controller started")
+	}
+
 	// TODO: Refactor NEG to use cloud mocks so ctx.Cloud can be referenced within NewController.
 	negController := neg.NewController(negtypes.NewAdapter(ctx.Cloud), ctx, lbc.Translator, ctx.ClusterNamer, flags.F.ResyncPeriod, flags.F.NegGCPeriod, flags.F.EnableReadinessReflector, flags.F.RunIngressController, flags.F.RunL4Controller)
 

pkg/backends/backends.go

@@ -19,6 +19,8 @@ import (
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	"google.golang.org/api/compute/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/ingress-gce/pkg/backends/features"
 	"k8s.io/ingress-gce/pkg/composite"
 	"k8s.io/ingress-gce/pkg/utils"
@@ -27,6 +29,10 @@ import (
 	"k8s.io/legacy-cloud-providers/gce"
 )
 
+const (
+	DefaultConnectionDrainingTimeoutSeconds = 30
+)
+
 // Backends handles CRUD operations for backends.
 type Backends struct {
 	cloud *gce.Cloud
@@ -224,3 +230,91 @@ func (b *Backends) List(key *meta.Key, version meta.Version) ([]*composite.Backe
 	}
 	return clusterBackends, nil
 }
+
+// EnsureL4BackendService creates or updates the backend service with the given name.
+func (b *Backends) EnsureL4BackendService(name, hcLink, protocol, sessionAffinity, scheme string, nm types.NamespacedName, version meta.Version) (*composite.BackendService, error) {
+	klog.V(2).Infof("EnsureL4BackendService(%v, %v, %v): checking existing backend service", name, scheme, protocol)
+	key, err := composite.CreateKey(b.cloud, name, meta.Regional)
+	bs, err := composite.GetBackendService(b.cloud, key, meta.VersionGA)
+	if err != nil && !utils.IsNotFoundError(err) {
+		return nil, err
+	}
+	desc, err := utils.MakeL4ILBServiceDescription(nm.String(), "", meta.VersionGA)
+	if err != nil {
+		klog.Warningf("EnsureL4BackendService: Failed to generate description for BackendService %s, err %v",
+			name, err)
+	}
+	expectedBS := &composite.BackendService{
+		Name:                name,
+		Protocol:            string(protocol),
+		Description:         desc,
+		HealthChecks:        []string{hcLink},
+		SessionAffinity:     utils.TranslateAffinityType(sessionAffinity),
+		LoadBalancingScheme: string(scheme),
+		ConnectionDraining:  &composite.ConnectionDraining{DrainingTimeoutSec: DefaultConnectionDrainingTimeoutSeconds},
+	}
+
+	// Create backend service if none was found
+	if bs == nil {
+		klog.V(2).Infof("EnsureL4BackendService: creating backend service %v", name)
+		err := composite.CreateBackendService(b.cloud, key, expectedBS)
+		if err != nil {
+			return nil, err
+		}
+		klog.V(2).Infof("EnsureL4BackendService: created backend service %v successfully", name)
+		// We need to perform a GCE call to re-fetch the object we just created
+		// so that the "Fingerprint" field is filled in. This is needed to update the
+		// object without error. The lookup is also needed to populate the selfLink.
+		return composite.GetBackendService(b.cloud, key, meta.VersionGA)
+	}
+
+	if backendSvcEqual(expectedBS, bs) {
+		return bs, nil
+	}
+	if bs.ConnectionDraining != nil && bs.ConnectionDraining.DrainingTimeoutSec > 0 {
+		// if user overrides this value, continue using that.
+		expectedBS.ConnectionDraining.DrainingTimeoutSec = bs.ConnectionDraining.DrainingTimeoutSec
+	}
+	klog.V(2).Infof("EnsureL4BackendService: updating backend service %v", name)
+	// Set fingerprint for optimistic locking
+	expectedBS.Fingerprint = bs.Fingerprint
+	if err := composite.UpdateBackendService(b.cloud, key, expectedBS); err != nil {
+		return nil, err
+	}
+	klog.V(2).Infof("EnsureL4BackendService: updated backend service %v successfully", name)
+	return composite.GetBackendService(b.cloud, key, meta.VersionGA)
+}
+
+// backendsListEqual asserts that backend lists are equal by group link only
+func backendsListEqual(a, b []*composite.Backend) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	if len(a) == 0 {
+		return true
+	}
+
+	aSet := sets.NewString()
+	for _, v := range a {
+		aSet.Insert(v.Group)
+	}
+	bSet := sets.NewString()
+	for _, v := range b {
+		bSet.Insert(v.Group)
+	}
+
+	return aSet.Equal(bSet)
+}
+
+// backendSvcEqual returns true if the 2 BackendService objects are equal.
+// ConnectionDraining timeout is not checked for equality, if user changes
+// this timeout and no other backendService parameters change, the backend
+// service will not be updated.
+func backendSvcEqual(a, b *composite.BackendService) bool {
+	return a.Protocol == b.Protocol &&
+		a.Description == b.Description &&
+		a.SessionAffinity == b.SessionAffinity &&
+		a.LoadBalancingScheme == b.LoadBalancingScheme &&
+		utils.EqualStringSets(a.HealthChecks, b.HealthChecks) &&
+		backendsListEqual(a.Backends, b.Backends)
+}

pkg/backends/features/features.go

@@ -36,6 +36,8 @@ const (
 	// FeatureL7ILB defines the feature name of L7 Internal Load Balancer
 	// L7-ILB Resources are currently alpha and regional
 	FeatureL7ILB = "L7ILB"
+	//FeaturePrimaryVMIPNEG defines the feature name of GCE_PRIMARY_VM_IP NEGs which are used for L4 ILB.
+	FeaturePrimaryVMIPNEG = "PrimaryVMIPNEG"
 )
 
 var (
@@ -46,7 +48,7 @@ var (
 	}
 	// TODO: (shance) refactor all scope to be above the serviceport level
 	scopeToFeatures = map[meta.KeyType][]string{
-		meta.Regional: []string{FeatureL7ILB},
+		meta.Regional: []string{FeatureL7ILB, FeaturePrimaryVMIPNEG},
 	}
 )
 
@@ -68,6 +70,9 @@ func featuresFromServicePort(sp *utils.ServicePort) []string {
 	if sp.NEGEnabled {
 		features = append(features, FeatureNEG)
 	}
+	if sp.PrimaryIPNEGEnabled {
+		features = append(features, FeaturePrimaryVMIPNEG)
+	}
 	if sp.L7ILBEnabled {
 		features = append(features, FeatureL7ILB)
 	}

pkg/backends/syncer.go

@@ -179,6 +179,10 @@ func (s *backendSyncer) GC(svcPorts []utils.ServicePort) error {
 // gc deletes the provided backends
 func (s *backendSyncer) gc(backends []*composite.BackendService, knownPorts sets.String) error {
 	for _, be := range backends {
+		// Skip L4 LB backend services
+		if strings.Contains(be.Description, utils.L4ILBServiceDescKey) {
+			continue
+		}
 		var key *meta.Key
 		name := be.Name
 		scope, err := composite.ScopeFromSelfLink(be.SelfLink)
@@ -191,7 +195,6 @@ func (s *backendSyncer) gc(backends []*composite.BackendService, knownPorts sets
 		if knownPorts.Has(key.String()) {
 			continue
 		}
-
 		klog.V(2).Infof("GCing backendService for port %s", name)
 		err = s.backendPool.Delete(name, be.Version, scope)
 		if err != nil {