Merge pull request #52528 from m1093782566/refactor-proxy

pkg/proxy/BUILD

@@ -3,16 +3,25 @@ package(default_visibility = ["//visibility:public"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
     name = "go_default_library",
     srcs = [
         "doc.go",
+        "endpoints.go",
+        "service.go",
         "types.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/proxy",
-    deps = ["//vendor/k8s.io/apimachinery/pkg/types:go_default_library"],
+    deps = [
+        "//pkg/apis/core:go_default_library",
+        "//pkg/proxy/util:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+    ],
 )
 
 filegroup(
@@ -39,3 +48,22 @@ filegroup(
     ],
     tags = ["automanaged"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "endpoints_test.go",
+        "service_test.go",
+    ],
+    embed = [":go_default_library"],
+    importpath = "k8s.io/kubernetes/pkg/proxy",
+    deps = [
+        "//pkg/api/service:go_default_library",
+        "//pkg/apis/core:go_default_library",
+        "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+    ],
+)

pkg/proxy/endpoints.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package proxy
+
+import (
+	"reflect"
+	"sync"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	api "k8s.io/kubernetes/pkg/apis/core"
+)
+
+// EndpointChangeTracker carries state about uncommitted changes to an arbitrary number of
+// Endpoints, keyed by their namespace and name.
+type EndpointChangeTracker struct {
+	// lock protects items.
+	lock sync.Mutex
+	// hostname is the host where kube-proxy is running.
+	hostname string
+	// items maps a service to is endpointsChange.
+	items map[types.NamespacedName]*endpointsChange
+}
+
+// NewEndpointChangeTracker initializes an EndpointsChangeMap
+func NewEndpointChangeTracker(hostname string) *EndpointChangeTracker {
+	return &EndpointChangeTracker{
+		hostname: hostname,
+		items:    make(map[types.NamespacedName]*endpointsChange),
+	}
+}
+
+// Update updates given service's endpoints change map based on the <previous, current> endpoints pair.  It returns true
+// if items changed, otherwise return false.  Update can be used to add/update/delete items of EndpointsChangeMap.  For example,
+// Add item
+//   - pass <nil, endpoints> as the <previous, current> pair.
+// Update item
+//   - pass <oldEndpoints, endpoints> as the <previous, current> pair.
+// Delete item
+//   - pass <endpoints, nil> as the <previous, current> pair.
+func (ect *EndpointChangeTracker) Update(previous, current *api.Endpoints, makeEndpoints func(IP string, port int, isLocal bool) Endpoint) bool {
+	endpoints := current
+	if endpoints == nil {
+		endpoints = previous
+	}
+	// previous == nil && current == nil is unexpected, we should return false directly.
+	if endpoints == nil {
+		return false
+	}
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+
+	ect.lock.Lock()
+	defer ect.lock.Unlock()
+
+	change, exists := ect.items[namespacedName]
+	if !exists {
+		change = &endpointsChange{}
+		change.previous = endpointsToEndpointsMap(previous, ect.hostname, makeEndpoints)
+		ect.items[namespacedName] = change
+	}
+	change.current = endpointsToEndpointsMap(current, ect.hostname, makeEndpoints)
+	// if change.previous equal to change.current, it means no change
+	if reflect.DeepEqual(change.previous, change.current) {
+		delete(ect.items, namespacedName)
+	}
+	return len(ect.items) > 0
+}
+
+// endpointsChange contains all changes to endpoints that happened since proxy rules were synced.  For a single object,
+// changes are accumulated, i.e. previous is state from before applying the changes,
+// current is state after applying the changes.
+type endpointsChange struct {
+	previous EndpointsMap
+	current  EndpointsMap
+}
+
+// UpdateEndpointMapResult is the updated results after applying endpoints changes.
+type UpdateEndpointMapResult struct {
+	// HCEndpointsLocalIPSize maps an endpoints name to the length of its local IPs.
+	HCEndpointsLocalIPSize map[types.NamespacedName]int
+	// StaleEndpoints identifies if an endpoints service pair is stale.
+	StaleEndpoints []ServiceEndpoint
+	// StaleServiceNames identifies if a service is stale.
+	StaleServiceNames []ServicePortName
+}
+
+// UpdateEndpointsMap updates endpointsMap base on the given changes.
+func UpdateEndpointsMap(endpointsMap EndpointsMap, changes *EndpointChangeTracker) (result UpdateEndpointMapResult) {
+	result.StaleEndpoints = make([]ServiceEndpoint, 0)
+	result.StaleServiceNames = make([]ServicePortName, 0)
+
+	endpointsMap.apply(changes, &result.StaleEndpoints, &result.StaleServiceNames)
+
+	// TODO: If this will appear to be computationally expensive, consider
+	// computing this incrementally similarly to endpointsMap.
+	result.HCEndpointsLocalIPSize = make(map[types.NamespacedName]int)
+	localIPs := GetLocalEndpointIPs(endpointsMap)
+	for nsn, ips := range localIPs {
+		result.HCEndpointsLocalIPSize[nsn] = len(ips)
+	}
+
+	return result
+}
+
+// EndpointsMap maps a service to one of its endpoint.
+type EndpointsMap map[ServicePortName][]Endpoint
+
+// endpointsToEndpointsMap translates single Endpoints object to EndpointsMap.
+// This function is used for incremental updated of endpointsMap.
+//
+// NOTE: endpoints object should NOT be modified.
+func endpointsToEndpointsMap(endpoints *api.Endpoints, hostname string, makeEndpoints func(IP string, port int, isLocal bool) Endpoint) EndpointsMap {
+	if endpoints == nil {
+		return nil
+	}
+
+	endpointsMap := make(EndpointsMap)
+	// We need to build a map of portname -> all ip:ports for that
+	// portname.  Explode Endpoints.Subsets[*] into this structure.
+	for i := range endpoints.Subsets {
+		ss := &endpoints.Subsets[i]
+		for i := range ss.Ports {
+			port := &ss.Ports[i]
+			if port.Port == 0 {
+				glog.Warningf("ignoring invalid endpoint port %s", port.Name)
+				continue
+			}
+			svcPortName := ServicePortName{
+				NamespacedName: types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name},
+				Port:           port.Name,
+			}
+			for i := range ss.Addresses {
+				addr := &ss.Addresses[i]
+				if addr.IP == "" {
+					glog.Warningf("ignoring invalid endpoint port %s with empty host", port.Name)
+					continue
+				}
+				isLocal := addr.NodeName != nil && *addr.NodeName == hostname
+				epInfo := makeEndpoints(addr.IP, int(port.Port), isLocal)
+				endpointsMap[svcPortName] = append(endpointsMap[svcPortName], epInfo)
+			}
+			if glog.V(3) {
+				newEPList := []string{}
+				for _, ep := range endpointsMap[svcPortName] {
+					newEPList = append(newEPList, ep.String())
+				}
+				glog.Infof("Setting endpoints for %q to %+v", svcPortName, newEPList)
+			}
+		}
+	}
+	return endpointsMap
+}
+
+// apply the changes to EndpointsMap and updates stale endpoints and service-endpoints pair. The `staleEndpoints` argument
+// is passed in to store the stale udp endpoints and `staleServiceNames` argument is passed in to store the stale udp service.
+// The changes map is cleared after applying them.
+func (endpointsMap EndpointsMap) apply(changes *EndpointChangeTracker, staleEndpoints *[]ServiceEndpoint, staleServiceNames *[]ServicePortName) {
+	if changes == nil {
+		return
+	}
+	changes.lock.Lock()
+	defer changes.lock.Unlock()
+	for _, change := range changes.items {
+		endpointsMap.Unmerge(change.previous)
+		endpointsMap.Merge(change.current)
+		detectStaleConnections(change.previous, change.current, staleEndpoints, staleServiceNames)
+	}
+	changes.items = make(map[types.NamespacedName]*endpointsChange)
+}
+
+// Merge ensures that the current EndpointsMap contains all <service, endpoints> pairs from the EndpointsMap passed in.
+func (em EndpointsMap) Merge(other EndpointsMap) {
+	for svcPortName := range other {
+		em[svcPortName] = other[svcPortName]
+	}
+}
+
+// Unmerge removes the <service, endpoints> pairs from the current EndpointsMap which are contained in the EndpointsMap passed in.
+func (em EndpointsMap) Unmerge(other EndpointsMap) {
+	for svcPortName := range other {
+		delete(em, svcPortName)
+	}
+}
+
+// GetLocalEndpointIPs returns endpoints IPs if given endpoint is local - local means the endpoint is running in same host as kube-proxy.
+func GetLocalEndpointIPs(endpointsMap EndpointsMap) map[types.NamespacedName]sets.String {
+	localIPs := make(map[types.NamespacedName]sets.String)
+	for svcPortName, epList := range endpointsMap {
+		for _, ep := range epList {
+			if ep.IsLocal() {
+				nsn := svcPortName.NamespacedName
+				if localIPs[nsn] == nil {
+					localIPs[nsn] = sets.NewString()
+				}
+				localIPs[nsn].Insert(ep.IP())
+			}
+		}
+	}
+	return localIPs
+}
+
+// detectStaleConnections modifies <staleEndpoints> and <staleServices> with detected stale connections. <staleServiceNames>
+// is used to store stale udp service in order to clear udp conntrack later.
+func detectStaleConnections(oldEndpointsMap, newEndpointsMap EndpointsMap, staleEndpoints *[]ServiceEndpoint, staleServiceNames *[]ServicePortName) {
+	for svcPortName, epList := range oldEndpointsMap {
+		for _, ep := range epList {
+			stale := true
+			for i := range newEndpointsMap[svcPortName] {
+				if newEndpointsMap[svcPortName][i].Equal(ep) {
+					stale = false
+					break
+				}
+			}
+			if stale {
+				glog.V(4).Infof("Stale endpoint %v -> %v", svcPortName, ep.String())
+				*staleEndpoints = append(*staleEndpoints, ServiceEndpoint{Endpoint: ep.String(), ServicePortName: svcPortName})
+			}
+		}
+	}
+
+	for svcPortName, epList := range newEndpointsMap {
+		// For udp service, if its backend changes from 0 to non-0. There may exist a conntrack entry that could blackhole traffic to the service.
+		if len(epList) > 0 && len(oldEndpointsMap[svcPortName]) == 0 {
+			*staleServiceNames = append(*staleServiceNames, svcPortName)
+		}
+	}
+}

pkg/proxy/iptables/BUILD

@@ -27,7 +27,6 @@ go_library(
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/client-go/tools/record:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
@@ -46,12 +45,10 @@ go_test(
         "//pkg/util/async:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/iptables/testing:go_default_library",
-        "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
         "//vendor/k8s.io/utils/exec/testing:go_default_library",
     ],

pkg/proxy/ipvs/BUILD

@@ -25,7 +25,6 @@ go_test(
         "//pkg/util/iptables/testing:go_default_library",
         "//pkg/util/ipvs:go_default_library",
         "//pkg/util/ipvs/testing:go_default_library",
-        "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/intstr:go_default_library",

pkg/proxy/service.go

+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package proxy
+
+import (
+	"reflect"
+	"sync"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	api "k8s.io/kubernetes/pkg/apis/core"
+	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
+)
+
+// serviceChange contains all changes to services that happened since proxy rules were synced.  For a single object,
+// changes are accumulated, i.e. previous is state from before applying the changes,
+// current is state after applying all of the changes.
+type serviceChange struct {
+	previous ServiceMap
+	current  ServiceMap
+}
+
+// ServiceChangeTracker carries state about uncommitted changes to an arbitrary number of
+// Services, keyed by their namespace and name.
+type ServiceChangeTracker struct {
+	// lock protects items.
+	lock sync.Mutex
+	// items maps a service to its serviceChange.
+	items map[types.NamespacedName]*serviceChange
+}
+
+// NewServiceChangeTracker initializes a ServiceChangeTracker
+func NewServiceChangeTracker() *ServiceChangeTracker {
+	return &ServiceChangeTracker{
+		items: make(map[types.NamespacedName]*serviceChange),
+	}
+}
+
+// Update updates given service's change map based on the <previous, current> service pair.  It returns true if items changed,
+// otherwise return false.  Update can be used to add/update/delete items of ServiceChangeMap.  For example,
+// Add item
+//   - pass <nil, service> as the <previous, current> pair.
+// Update item
+//   - pass <oldService, service> as the <previous, current> pair.
+// Delete item
+//   - pass <service, nil> as the <previous, current> pair.
+//
+// makeServicePort() return a proxy.ServicePort based on the given Service and its ServicePort.  We inject makeServicePort()
+// so that giving caller side a chance to initialize proxy.ServicePort interface.
+func (sct *ServiceChangeTracker) Update(previous, current *api.Service, makeServicePort func(servicePort *api.ServicePort, service *api.Service) ServicePort) bool {
+	svc := current
+	if svc == nil {
+		svc = previous
+	}
+	// previous == nil && current == nil is unexpected, we should return false directly.
+	if svc == nil {
+		return false
+	}
+	namespacedName := types.NamespacedName{Namespace: svc.Namespace, Name: svc.Name}
+
+	sct.lock.Lock()
+	defer sct.lock.Unlock()
+
+	change, exists := sct.items[namespacedName]
+	if !exists {
+		change = &serviceChange{}
+		change.previous = serviceToServiceMap(previous, makeServicePort)
+		sct.items[namespacedName] = change
+	}
+	change.current = serviceToServiceMap(current, makeServicePort)
+	// if change.previous equal to change.current, it means no change
+	if reflect.DeepEqual(change.previous, change.current) {
+		delete(sct.items, namespacedName)
+	}
+	return len(sct.items) > 0
+}
+
+// UpdateServiceMapResult is the updated results after applying service changes.
+type UpdateServiceMapResult struct {
+	// HCServiceNodePorts is a map of Service names to node port numbers which indicate the health of that Service on this Node.
+	// The value(uint16) of HCServices map is the service health check node port.
+	HCServiceNodePorts map[types.NamespacedName]uint16
+	// UDPStaleClusterIP holds stale (no longer assigned to a Service) Service IPs that had UDP ports.
+	// Callers can use this to abort timeout-waits or clear connection-tracking information.
+	UDPStaleClusterIP sets.String
+}
+
+// UpdateServiceMap updates ServiceMap based on the given changes.
+func UpdateServiceMap(serviceMap ServiceMap, changes *ServiceChangeTracker) (result UpdateServiceMapResult) {
+	result.UDPStaleClusterIP = sets.NewString()
+	serviceMap.apply(changes, result.UDPStaleClusterIP)
+
+	// TODO: If this will appear to be computationally expensive, consider
+	// computing this incrementally similarly to serviceMap.
+	result.HCServiceNodePorts = make(map[types.NamespacedName]uint16)
+	for svcPortName, info := range serviceMap {
+		if info.HealthCheckNodePort() != 0 {
+			result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.HealthCheckNodePort())
+		}
+	}
+
+	return result
+}
+
+// ServiceMap maps a service to its ServicePort information.
+type ServiceMap map[ServicePortName]ServicePort
+
+// serviceToServiceMap translates a single Service object to a ServiceMap.
+// makeServicePort() return a proxy.ServicePort based on the given Service and its ServicePort.  We inject makeServicePort()
+// so that giving caller side a chance to initialize proxy.ServicePort interface.
+//
+// NOTE: service object should NOT be modified.
+func serviceToServiceMap(service *api.Service, makeServicePort func(servicePort *api.ServicePort, service *api.Service) ServicePort) ServiceMap {
+	if service == nil {
+		return nil
+	}
+	svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if proxyutil.ShouldSkipService(svcName, service) {
+		return nil
+	}
+
+	serviceMap := make(ServiceMap)
+	for i := range service.Spec.Ports {
+		servicePort := &service.Spec.Ports[i]
+		svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name}
+		serviceMap[svcPortName] = makeServicePort(servicePort, service)
+	}
+	return serviceMap
+}
+
+// apply the changes to ServiceMap and update the stale udp cluster IP set. The UDPStaleClusterIP argument is passed in to store the
+// udp protocol service cluster ip when service is deleted from the ServiceMap.
+func (serviceMap *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
+	changes.lock.Lock()
+	defer changes.lock.Unlock()
+	for _, change := range changes.items {
+		serviceMap.merge(change.current)
+		// filter out the Update event of current changes from previous changes before calling unmerge() so that can
+		// skip deleting the Update events.
+		change.previous.filter(change.current)
+		serviceMap.unmerge(change.previous, UDPStaleClusterIP)
+	}
+	// clear changes after applying them to ServiceMap.
+	changes.items = make(map[types.NamespacedName]*serviceChange)
+	return
+}
+
+// merge adds other ServiceMap's elements to current ServiceMap.
+// If collision, other ALWAYS win. Otherwise add the other to current.
+// In other words, if some elements in current collisions with other, update the current by other.
+// It returns a string type set which stores all the newly merged services' identifier, ServicePortName.String(), to help users
+// tell if a service is deleted or updated.
+// The returned value is one of the arguments of ServiceMap.unmerge().
+// ServiceMap A Merge ServiceMap B will do following 2 things:
+//   * update ServiceMap A.
+//   * produce a string set which stores all other ServiceMap's ServicePortName.String().
+// For example,
+//   - A{}
+//   - B{{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
+//     - A updated to be {{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
+//     - produce string set {"ns/cluster-ip:http"}
+//   - A{{"ns", "cluster-ip", "http"}: {"172.16.55.10", 345, "UDP"}}
+//   - B{{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
+//     - A updated to be {{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
+//     - produce string set {"ns/cluster-ip:http"}
+func (sm *ServiceMap) merge(other ServiceMap) sets.String {
+	// existingPorts is going to store all identifiers of all services in `other` ServiceMap.
+	existingPorts := sets.NewString()
+	for svcPortName, info := range other {
+		// Take ServicePortName.String() as the newly merged service's identifier and put it into existingPorts.
+		existingPorts.Insert(svcPortName.String())
+		_, exists := (*sm)[svcPortName]
+		if !exists {
+			glog.V(1).Infof("Adding new service port %q at %s", svcPortName, info.String())
+		} else {
+			glog.V(1).Infof("Updating existing service port %q at %s", svcPortName, info.String())
+		}
+		(*sm)[svcPortName] = info
+	}
+	return existingPorts
+}
+
+// filter filters out elements from ServiceMap base on given ports string sets.
+func (sm *ServiceMap) filter(other ServiceMap) {
+	for svcPortName := range *sm {
+		// skip the delete for Update event.
+		if _, ok := other[svcPortName]; ok {
+			delete(*sm, svcPortName)
+		}
+	}
+}
+
+// unmerge deletes all other ServiceMap's elements from current ServiceMap.  We pass in the UDPStaleClusterIP strings sets
+// for storing the stale udp service cluster IPs. We will clear stale udp connection base on UDPStaleClusterIP later
+func (sm *ServiceMap) unmerge(other ServiceMap, UDPStaleClusterIP sets.String) {
+	for svcPortName := range other {
+		info, exists := (*sm)[svcPortName]
+		if exists {
+			glog.V(1).Infof("Removing service port %q", svcPortName)
+			if info.Protocol() == api.ProtocolUDP {
+				UDPStaleClusterIP.Insert(info.ClusterIP())
+			}
+			delete(*sm, svcPortName)
+		} else {
+			glog.Errorf("Service port %q doesn't exists", svcPortName)
+		}
+	}
+}

pkg/proxy/types.go

@@ -20,11 +20,12 @@ import (
 	"fmt"
 
 	"k8s.io/apimachinery/pkg/types"
+	api "k8s.io/kubernetes/pkg/apis/core"
 )
 
 // ProxyProvider is the interface provided by proxier implementations.
 type ProxyProvider interface {
-	// Sync immediately synchronizes the ProxyProvider's current state to iptables.
+	// Sync immediately synchronizes the ProxyProvider's current state to proxy rules.
 	Sync()
 	// SyncLoop runs periodic work.
 	// This is expected to run as a goroutine or as the main loop of the app.
@@ -33,7 +34,7 @@ type ProxyProvider interface {
 }
 
 // ServicePortName carries a namespace + name + portname.  This is the unique
-// identfier for a load-balanced service.
+// identifier for a load-balanced service.
 type ServicePortName struct {
 	types.NamespacedName
 	Port string
@@ -42,3 +43,34 @@ type ServicePortName struct {
 func (spn ServicePortName) String() string {
 	return fmt.Sprintf("%s:%s", spn.NamespacedName.String(), spn.Port)
 }
+
+// ServicePort is an interface which abstracts information about a service.
+type ServicePort interface {
+	// String returns service string.  An example format can be: `IP:Port/Protocol`.
+	String() string
+	// ClusterIP returns service cluster IP.
+	ClusterIP() string
+	// Protocol returns service protocol.
+	Protocol() api.Protocol
+	// HealthCheckNodePort returns service health check node port if present.  If return 0, it means not present.
+	HealthCheckNodePort() int
+}
+
+// Endpoint in an interface which abstracts information about an endpoint.
+type Endpoint interface {
+	// String returns endpoint string.  An example format can be: `IP:Port`.
+	// We take the returned value as ServiceEndpoint.Endpoint.
+	String() string
+	// IsLocal returns true if the endpoint is running in same host as kube-proxy, otherwise returns false.
+	IsLocal() bool
+	// IP returns IP part of endpoints.
+	IP() string
+	// Equal checks if two endpoints are equal.
+	Equal(Endpoint) bool
+}
+
+// ServiceEndpoint is used to identify a service and one of its endpoint pair.
+type ServiceEndpoint struct {
+	Endpoint        string
+	ServicePortName ServicePortName
+}

pkg/proxy/util/endpoints.go

@@ -47,6 +47,7 @@ func IPPart(s string) string {
 	return ""
 }
 
+// PortPart returns just the port part of an endpoint string.
 func PortPart(s string) (int, error) {
 	// Must be IP:port
 	_, port, err := net.SplitHostPort(s)