Removed helper import from l4lb

pkg/healthchecksl4/healthchecksl4.go

@@ -24,10 +24,10 @@ import (
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	"k8s.io/ingress-gce/pkg/composite"
 	"k8s.io/ingress-gce/pkg/firewalls"
@@ -117,11 +117,40 @@ func (l4hc *l4HealthChecks) EnsureHealthCheckWithFirewall(svc *corev1.Service, n
 	return l4hc.EnsureHealthCheckWithDualStackFirewalls(svc, namer, sharedHC, scope, l4Type, nodeNames /*create IPv4*/, true /*don't create IPv6*/, false, svcNetwork, svcLogger)
 }
 
+// GetServiceHealthCheckPathPort returns the path and nodePort programmed into the Cloud LB Health Check
+func getServiceHealthCheckPathPort(service *v1.Service) (string, int32) {
+	if !needsHealthCheck(service) {
+		return "", 0
+	}
+	port := service.Spec.HealthCheckNodePort
+	if port == 0 {
+		return "", 0
+	}
+	return "/healthz", port
+}
+
+// NeedsHealthCheck checks if service needs health check.
+func needsHealthCheck(service *v1.Service) bool {
+	if service.Spec.Type != v1.ServiceTypeLoadBalancer {
+		return false
+	}
+	return requestsOnlyLocalTraffic(service)
+}
+
+// RequestsOnlyLocalTraffic checks if service requests OnlyLocal traffic.
+func requestsOnlyLocalTraffic(service *v1.Service) bool {
+	if service.Spec.Type != v1.ServiceTypeLoadBalancer &&
+		service.Spec.Type != v1.ServiceTypeNodePort {
+		return false
+	}
+	return service.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyLocal
+}
+
 func (l4hc *l4HealthChecks) EnsureHealthCheckWithDualStackFirewalls(svc *corev1.Service, namer namer.L4ResourcesNamer, sharedHC bool, scope meta.KeyType, l4Type utils.L4LBType, nodeNames []string, needsIPv4 bool, needsIPv6 bool, svcNetwork network.NetworkInfo, svcLogger klog.Logger) *EnsureHealthCheckResult {
 	namespacedName := types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}
 
 	hcName := namer.L4HealthCheck(svc.Namespace, svc.Name, sharedHC)
-	hcPath, hcPort := helpers.GetServiceHealthCheckPathPort(svc)
+	hcPath, hcPort := getServiceHealthCheckPathPort(svc)
 	hcLogger := svcLogger.WithValues("healthcheckName", hcName)
 	hcLogger.V(3).Info("Ensuring L4 healthcheck with firewalls for service", "shared", sharedHC)
 

pkg/l4lb/l4lbcommon.go

@@ -24,7 +24,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/composite"
 	"k8s.io/ingress-gce/pkg/context"
 	l4metrics "k8s.io/ingress-gce/pkg/l4lb/metrics"
@@ -98,10 +97,37 @@ func deleteAnnotation(ctx *context.ControllerContext, svc *v1.Service, annotatio
 	return patch.PatchServiceObjectMetadata(ctx.KubeClient.CoreV1(), svc, *newObjectMeta)
 }
 
+// LoadBalancerStatusEqual checks if load balancer status are equal
+func loadBalancerStatusEqual(l, r *v1.LoadBalancerStatus) bool {
+	return ingressSliceEqual(l.Ingress, r.Ingress)
+}
+
+func ingressSliceEqual(lhs, rhs []v1.LoadBalancerIngress) bool {
+	if len(lhs) != len(rhs) {
+		return false
+	}
+	for i := range lhs {
+		if !ingressEqual(&lhs[i], &rhs[i]) {
+			return false
+		}
+	}
+	return true
+}
+
+func ingressEqual(lhs, rhs *v1.LoadBalancerIngress) bool {
+	if lhs.IP != rhs.IP {
+		return false
+	}
+	if lhs.Hostname != rhs.Hostname {
+		return false
+	}
+	return true
+}
+
 // updateServiceStatus this faction checks if LoadBalancer status changed and patch service if needed.
 func updateServiceStatus(ctx *context.ControllerContext, svc *v1.Service, newStatus *v1.LoadBalancerStatus, svcLogger klog.Logger) error {
 	svcLogger.V(2).Info("Updating service status", "newStatus", fmt.Sprintf("%+v", newStatus))
-	if helpers.LoadBalancerStatusEqual(&svc.Status.LoadBalancer, newStatus) {
+	if loadBalancerStatusEqual(&svc.Status.LoadBalancer, newStatus) {
 		svcLogger.V(2).Info("New and old statuses are equal, skipping patch")
 		return nil
 	}

pkg/l4lb/l4netlbcontroller_test.go

@@ -43,7 +43,6 @@ import (
 	networkv1 "k8s.io/cloud-provider-gcp/crd/apis/network/v1"
 	netfake "k8s.io/cloud-provider-gcp/crd/client/network/clientset/versioned/fake"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	"k8s.io/ingress-gce/pkg/backends"
 	"k8s.io/ingress-gce/pkg/composite"
@@ -60,12 +59,13 @@ import (
 )
 
 const (
-	FwIPAddress          = "10.0.0.1"
-	loadBalancerIP       = "10.0.0.10"
-	usersIP              = "35.10.211.60"
-	testServiceNamespace = "default"
-	hcNodePort           = int32(10111)
-	userAddrName         = "UserStaticAddress"
+	FwIPAddress                  = "10.0.0.1"
+	loadBalancerIP               = "10.0.0.10"
+	usersIP                      = "35.10.211.60"
+	testServiceNamespace         = "default"
+	hcNodePort                   = int32(10111)
+	userAddrName                 = "UserStaticAddress"
+	loadBalancerCleanupFinalizer = "service.kubernetes.io/load-balancer-cleanup"
 
 	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
@@ -1151,7 +1151,7 @@ func TestIsRBSBasedService(t *testing.T) {
 		},
 		{
 			desc:             "Legacy service should not be marked as RBS",
-			finalizers:       []string{helpers.LoadBalancerCleanupFinalizer},
+			finalizers:       []string{loadBalancerCleanupFinalizer},
 			expectRBSService: false,
 		},
 		{
@@ -1161,7 +1161,7 @@ func TestIsRBSBasedService(t *testing.T) {
 		},
 		{
 			desc:             "Should detect RBS by finalizer when service contains both legacy and NetLB finalizers",
-			finalizers:       []string{helpers.LoadBalancerCleanupFinalizer, common.NetLBFinalizerV2},
+			finalizers:       []string{loadBalancerCleanupFinalizer, common.NetLBFinalizerV2},
 			expectRBSService: true,
 		},
 		{

pkg/loadbalancers/l4.go

@@ -24,10 +24,10 @@ import (
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud"
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	"k8s.io/ingress-gce/pkg/backends"
 	"k8s.io/ingress-gce/pkg/composite"
@@ -111,6 +111,15 @@ type L4ILBParams struct {
 	EnableWeightedLB bool
 }
 
+// requestsOnlyLocalTraffic checks if service requests OnlyLocal traffic.
+func requestsOnlyLocalTraffic(service *v1.Service) bool {
+	if service.Spec.Type != v1.ServiceTypeLoadBalancer &&
+		service.Spec.Type != v1.ServiceTypeNodePort {
+		return false
+	}
+	return service.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyLocal
+}
+
 // NewL4Handler creates a new L4Handler for the given L4 service.
 func NewL4Handler(params *L4ILBParams, logger klog.Logger) *L4 {
 	logger = logger.WithName("L4Handler")
@@ -528,7 +537,7 @@ func (l4 *L4) provideHealthChecks(nodeNames []string, result *L4ILBSyncResult) s
 }
 
 func (l4 *L4) provideDualStackHealthChecks(nodeNames []string, result *L4ILBSyncResult) string {
-	sharedHC := !helpers.RequestsOnlyLocalTraffic(l4.Service)
+	sharedHC := !requestsOnlyLocalTraffic(l4.Service)
 	hcResult := l4.healthChecks.EnsureHealthCheckWithDualStackFirewalls(l4.Service, l4.namer, sharedHC, meta.Global, utils.ILB, nodeNames, utils.NeedsIPv4(l4.Service), utils.NeedsIPv6(l4.Service), l4.network, l4.svcLogger)
 	if hcResult.Err != nil {
 		result.GCEResourceInError = hcResult.GceResourceInError
@@ -547,7 +556,7 @@ func (l4 *L4) provideDualStackHealthChecks(nodeNames []string, result *L4ILBSync
 }
 
 func (l4 *L4) provideIPv4HealthChecks(nodeNames []string, result *L4ILBSyncResult) string {
-	sharedHC := !helpers.RequestsOnlyLocalTraffic(l4.Service)
+	sharedHC := !requestsOnlyLocalTraffic(l4.Service)
 	hcResult := l4.healthChecks.EnsureHealthCheckWithFirewall(l4.Service, l4.namer, sharedHC, meta.Global, utils.ILB, nodeNames, l4.network, l4.svcLogger)
 	if hcResult.Err != nil {
 		result.GCEResourceInError = hcResult.GceResourceInError

pkg/loadbalancers/l4_test.go

@@ -42,7 +42,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	servicehelper "k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/composite"
 	"k8s.io/ingress-gce/pkg/test"
 	namer_util "k8s.io/ingress-gce/pkg/utils/namer"
@@ -297,7 +296,7 @@ func TestEnsureInternalLoadBalancerWithExistingResources(t *testing.T) {
 	lbName := l4.namer.L4Backend(svc.Namespace, svc.Name)
 
 	// Create the expected resources necessary for an Internal Load Balancer
-	sharedHC := !servicehelper.RequestsOnlyLocalTraffic(svc)
+	sharedHC := !requestsOnlyLocalTraffic(svc)
 	defaultNetwork := network.DefaultNetwork(fakeGCE)
 	hcResult := l4.healthChecks.EnsureHealthCheckWithFirewall(l4.Service, l4.namer, sharedHC, meta.Global, utils.ILB, []string{}, *defaultNetwork, klog.TODO())
 
@@ -390,7 +389,7 @@ func TestEnsureInternalLoadBalancerClearPreviousResources(t *testing.T) {
 		t.Errorf("fakeGCE.CreateFirewall(%v) returned error %v", existingFirewall, err)
 	}
 
-	sharedHealthCheck := !servicehelper.RequestsOnlyLocalTraffic(svc)
+	sharedHealthCheck := !requestsOnlyLocalTraffic(svc)
 	hcName := l4.namer.L4HealthCheck(svc.Namespace, svc.Name, sharedHealthCheck)
 
 	// Create a healthcheck with an incomplete fields
@@ -576,7 +575,7 @@ func TestEnsureInternalLoadBalancerHealthCheckConfigurable(t *testing.T) {
 	if err != nil {
 		t.Errorf("Unexpected error when creating key - %v", err)
 	}
-	sharedHealthCheck := !servicehelper.RequestsOnlyLocalTraffic(svc)
+	sharedHealthCheck := !requestsOnlyLocalTraffic(svc)
 	hcName := l4.namer.L4HealthCheck(svc.Namespace, svc.Name, sharedHealthCheck)
 
 	// Create a healthcheck with an incorrect threshold, default value is 8s.
@@ -2414,7 +2413,7 @@ func assertDualStackILBResourcesWithCustomSubnet(t *testing.T, l4 *L4, nodeNames
 }
 
 func buildExpectedAnnotations(l4 *L4) map[string]string {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4.Service)
 	proto := utils.GetProtocol(l4.Service.Spec.Ports)
 
 	backendName := l4.namer.L4Backend(l4.Service.Namespace, l4.Service.Name)
@@ -2456,7 +2455,7 @@ func buildExpectedAnnotations(l4 *L4) map[string]string {
 }
 
 func getAndVerifyILBHealthCheck(l4 *L4) (*composite.HealthCheck, error) {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4.Service)
 	hcName := l4.namer.L4HealthCheck(l4.Service.Namespace, l4.Service.Name, isSharedHC)
 
 	healthcheck, err := composite.GetHealthCheck(l4.cloud, meta.GlobalKey(hcName), meta.VersionGA, klog.TODO())
@@ -2588,7 +2587,7 @@ func verifyILBIPv6NodesFirewall(l4 *L4, nodeNames []string) error {
 }
 
 func verifyILBIPv4HealthCheckFirewall(l4 *L4, nodeNames []string) error {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4.Service)
 
 	hcFwName := l4.namer.L4HealthCheckFirewall(l4.Service.Namespace, l4.Service.Name, isSharedHC)
 	hcFwDesc, err := utils.MakeL4LBFirewallDescription(utils.ServiceKeyFunc(l4.Service.Namespace, l4.Service.Name), "", meta.VersionGA, isSharedHC)
@@ -2600,7 +2599,7 @@ func verifyILBIPv4HealthCheckFirewall(l4 *L4, nodeNames []string) error {
 }
 
 func verifyILBIPv6HealthCheckFirewall(l4 *L4, nodeNames []string) error {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4.Service)
 
 	ipv6hcFwName := l4.namer.L4IPv6HealthCheckFirewall(l4.Service.Namespace, l4.Service.Name, isSharedHC)
 	hcFwDesc, err := utils.MakeL4LBFirewallDescription(utils.ServiceKeyFunc(l4.Service.Namespace, l4.Service.Name), "", meta.VersionGA, isSharedHC)

pkg/loadbalancers/l4netlb.go

@@ -26,7 +26,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	"k8s.io/ingress-gce/pkg/backends"
 	"k8s.io/ingress-gce/pkg/composite"
@@ -263,7 +262,7 @@ func (l4netlb *L4NetLB) provideHealthChecks(nodeNames []string, result *L4NetLBS
 }
 
 func (l4netlb *L4NetLB) provideDualStackHealthChecks(nodeNames []string, result *L4NetLBSyncResult) string {
-	sharedHC := !helpers.RequestsOnlyLocalTraffic(l4netlb.Service)
+	sharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 	hcResult := l4netlb.healthChecks.EnsureHealthCheckWithDualStackFirewalls(l4netlb.Service, l4netlb.namer, sharedHC, l4netlb.scope, utils.XLB, nodeNames, utils.NeedsIPv4(l4netlb.Service), utils.NeedsIPv6(l4netlb.Service), l4netlb.networkInfo, l4netlb.svcLogger)
 	if hcResult.Err != nil {
 		result.GCEResourceInError = hcResult.GceResourceInError
@@ -282,7 +281,7 @@ func (l4netlb *L4NetLB) provideDualStackHealthChecks(nodeNames []string, result
 }
 
 func (l4netlb *L4NetLB) provideIPv4HealthChecks(nodeNames []string, result *L4NetLBSyncResult) string {
-	sharedHC := !helpers.RequestsOnlyLocalTraffic(l4netlb.Service)
+	sharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 	hcResult := l4netlb.healthChecks.EnsureHealthCheckWithFirewall(l4netlb.Service, l4netlb.namer, sharedHC, l4netlb.scope, utils.XLB, nodeNames, l4netlb.networkInfo, l4netlb.svcLogger)
 	if hcResult.Err != nil {
 		result.GCEResourceInError = hcResult.GceResourceInError

pkg/loadbalancers/l4netlb_test.go

@@ -36,7 +36,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/cloud-provider-gcp/providers/gce"
-	servicehelper "k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	"k8s.io/ingress-gce/pkg/composite"
 	"k8s.io/ingress-gce/pkg/firewalls"
@@ -1361,7 +1360,7 @@ func verifyNetLBIPv6NodesFirewall(l4netlb *L4NetLB, nodeNames []string) error {
 }
 
 func verifyNetLBIPv4HealthCheckFirewall(l4netlb *L4NetLB, nodeNames []string) error {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4netlb.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 
 	hcFwName := l4netlb.namer.L4HealthCheckFirewall(l4netlb.Service.Namespace, l4netlb.Service.Name, isSharedHC)
 	hcFwDesc, err := utils.MakeL4LBFirewallDescription(utils.ServiceKeyFunc(l4netlb.Service.Namespace, l4netlb.Service.Name), "", meta.VersionGA, isSharedHC)
@@ -1373,7 +1372,7 @@ func verifyNetLBIPv4HealthCheckFirewall(l4netlb *L4NetLB, nodeNames []string) er
 }
 
 func verifyNetLBIPv6HealthCheckFirewall(l4netlb *L4NetLB, nodeNames []string) error {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4netlb.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 
 	ipv6hcFwName := l4netlb.namer.L4IPv6HealthCheckFirewall(l4netlb.Service.Namespace, l4netlb.Service.Name, isSharedHC)
 	hcFwDesc, err := utils.MakeL4LBFirewallDescription(utils.ServiceKeyFunc(l4netlb.Service.Namespace, l4netlb.Service.Name), "", meta.VersionGA, isSharedHC)
@@ -1385,7 +1384,7 @@ func verifyNetLBIPv6HealthCheckFirewall(l4netlb *L4NetLB, nodeNames []string) er
 }
 
 func getAndVerifyNetLBHealthCheck(l4netlb *L4NetLB) (*composite.HealthCheck, error) {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4netlb.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 	hcName := l4netlb.namer.L4HealthCheck(l4netlb.Service.Namespace, l4netlb.Service.Name, isSharedHC)
 
 	healthcheck, err := composite.GetHealthCheck(l4netlb.cloud, meta.RegionalKey(hcName, l4netlb.cloud.Region()), meta.VersionGA, klog.TODO())
@@ -1526,7 +1525,7 @@ func verifyNetLBIPv6ResourcesDeletedOnSync(l4netlb *L4NetLB) error {
 }
 
 func buildExpectedNetLBAnnotations(l4netlb *L4NetLB) map[string]string {
-	isSharedHC := !servicehelper.RequestsOnlyLocalTraffic(l4netlb.Service)
+	isSharedHC := !requestsOnlyLocalTraffic(l4netlb.Service)
 	proto := utils.GetProtocol(l4netlb.Service.Spec.Ports)
 
 	backendName := l4netlb.namer.L4Backend(l4netlb.Service.Namespace, l4netlb.Service.Name)

pkg/neg/controller.go

@@ -33,7 +33,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
-	"k8s.io/cloud-provider/service/helpers"
 	"k8s.io/ingress-gce/pkg/annotations"
 	svcnegv1beta1 "k8s.io/ingress-gce/pkg/apis/svcneg/v1beta1"
 	"k8s.io/ingress-gce/pkg/controller/translator"
@@ -110,6 +109,15 @@ type Controller struct {
 	logger klog.Logger
 }
 
+// requestsOnlyLocalTraffic checks if service requests OnlyLocal traffic.
+func requestsOnlyLocalTraffic(service *apiv1.Service) bool {
+	if service.Spec.Type != apiv1.ServiceTypeLoadBalancer &&
+		service.Spec.Type != apiv1.ServiceTypeNodePort {
+		return false
+	}
+	return service.Spec.ExternalTrafficPolicy == apiv1.ServiceExternalTrafficPolicyLocal
+}
+
 // NewController returns a network endpoint group controller.
 func NewController(
 	kubeClient kubernetes.Interface,
@@ -620,7 +628,7 @@ func (c *Controller) mergeVmIpNEGsPortInfo(service *apiv1.Service, name types.Na
 		return nil
 	}
 
-	onlyLocal := helpers.RequestsOnlyLocalTraffic(service)
+	onlyLocal := requestsOnlyLocalTraffic(service)
 	// Update usage metrics.
 	negUsage.VmIpNeg = metricscollector.NewVmIpNegType(onlyLocal)