Vendor in latest k8s.io changes

These changes allow for the container's pid namespace to be set to the same
as the pod infra container's namespace if the pid namespace mode is set to POD

Signed-off-by: umohnani8 <umohnani@redhat.com>

vendor/k8s.io/kubernetes/pkg/proxy/endpoints.go

@@ -0,0 +1,314 @@
+/*
+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 (
+	"net"
+	"reflect"
+	"strconv"
+	"sync"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
+	api "k8s.io/kubernetes/pkg/apis/core"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
+)
+
+// BaseEndpointInfo contains base information that defines an endpoint.
+// This could be used directly by proxier while processing endpoints,
+// or can be used for constructing a more specific EndpointInfo struct
+// defined by the proxier if needed.
+type BaseEndpointInfo struct {
+	Endpoint string // TODO: should be an endpointString type
+	// IsLocal indicates whether the endpoint is running in same host as kube-proxy.
+	IsLocal bool
+}
+
+var _ Endpoint = &BaseEndpointInfo{}
+
+// String is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) String() string {
+	return info.Endpoint
+}
+
+// GetIsLocal is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) GetIsLocal() bool {
+	return info.IsLocal
+}
+
+// IP returns just the IP part of the endpoint, it's a part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) IP() string {
+	return utilproxy.IPPart(info.Endpoint)
+}
+
+// Port returns just the Port part of the endpoint.
+func (info *BaseEndpointInfo) Port() (int, error) {
+	return utilproxy.PortPart(info.Endpoint)
+}
+
+// Equal is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) Equal(other Endpoint) bool {
+	return info.String() == other.String() && info.GetIsLocal() == other.GetIsLocal()
+}
+
+func newBaseEndpointInfo(IP string, port int, isLocal bool) *BaseEndpointInfo {
+	return &BaseEndpointInfo{
+		Endpoint: net.JoinHostPort(IP, strconv.Itoa(port)),
+		IsLocal:  isLocal,
+	}
+}
+
+type makeEndpointFunc func(info *BaseEndpointInfo) Endpoint
+
+// 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
+	// makeEndpointInfo allows proxier to inject customized information when processing endpoint.
+	makeEndpointInfo makeEndpointFunc
+	// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
+	isIPv6Mode *bool
+	recorder   record.EventRecorder
+}
+
+// NewEndpointChangeTracker initializes an EndpointsChangeMap
+func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder) *EndpointChangeTracker {
+	return &EndpointChangeTracker{
+		hostname:         hostname,
+		items:            make(map[types.NamespacedName]*endpointsChange),
+		makeEndpointInfo: makeEndpointInfo,
+		isIPv6Mode:       isIPv6Mode,
+		recorder:         recorder,
+	}
+}
+
+// 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) 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 = ect.endpointsToEndpointsMap(previous)
+		ect.items[namespacedName] = change
+	}
+	change.current = ect.endpointsToEndpointsMap(current)
+	// 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 name to a list of all its Endpoints.
+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 (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *api.Endpoints) 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
+				}
+				// Filter out the incorrect IP version case.
+				// Any endpoint port that contains incorrect IP version will be ignored.
+				if ect.isIPv6Mode != nil && utilnet.IsIPv6String(addr.IP) != *ect.isIPv6Mode {
+					// Emit event on the corresponding service which had a different
+					// IP version than the endpoint.
+					utilproxy.LogAndEmitIncorrectIPVersionEvent(ect.recorder, "endpoints", addr.IP, endpoints.Name, endpoints.Namespace, "")
+					continue
+				}
+				isLocal := addr.NodeName != nil && *addr.NodeName == ect.hostname
+				baseEndpointInfo := newBaseEndpointInfo(addr.IP, int(port.Port), isLocal)
+				if ect.makeEndpointInfo != nil {
+					endpointsMap[svcPortName] = append(endpointsMap[svcPortName], ect.makeEndpointInfo(baseEndpointInfo))
+				} else {
+					endpointsMap[svcPortName] = append(endpointsMap[svcPortName], baseEndpointInfo)
+				}
+			}
+			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.GetIsLocal() {
+				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)
+		}
+	}
+}

vendor/k8s.io/kubernetes/pkg/proxy/service.go

@@ -0,0 +1,344 @@
+/*
+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 (
+	"fmt"
+	"net"
+	"reflect"
+	"strings"
+	"sync"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
+	apiservice "k8s.io/kubernetes/pkg/api/service"
+	api "k8s.io/kubernetes/pkg/apis/core"
+	"k8s.io/kubernetes/pkg/apis/core/helper"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
+)
+
+// BaseServiceInfo contains base information that defines a service.
+// This could be used directly by proxier while processing services,
+// or can be used for constructing a more specific ServiceInfo struct
+// defined by the proxier if needed.
+type BaseServiceInfo struct {
+	ClusterIP                net.IP
+	Port                     int
+	Protocol                 api.Protocol
+	NodePort                 int
+	LoadBalancerStatus       api.LoadBalancerStatus
+	SessionAffinityType      api.ServiceAffinity
+	StickyMaxAgeSeconds      int
+	ExternalIPs              []string
+	LoadBalancerSourceRanges []string
+	HealthCheckNodePort      int
+	OnlyNodeLocalEndpoints   bool
+}
+
+var _ ServicePort = &BaseServiceInfo{}
+
+// String is part of ServicePort interface.
+func (info *BaseServiceInfo) String() string {
+	return fmt.Sprintf("%s:%d/%s", info.ClusterIP, info.Port, info.Protocol)
+}
+
+// ClusterIPString is part of ServicePort interface.
+func (info *BaseServiceInfo) ClusterIPString() string {
+	return info.ClusterIP.String()
+}
+
+// GetProtocol is part of ServicePort interface.
+func (info *BaseServiceInfo) GetProtocol() api.Protocol {
+	return info.Protocol
+}
+
+// GetHealthCheckNodePort is part of ServicePort interface.
+func (info *BaseServiceInfo) GetHealthCheckNodePort() int {
+	return info.HealthCheckNodePort
+}
+
+func (sct *ServiceChangeTracker) newBaseServiceInfo(port *api.ServicePort, service *api.Service) *BaseServiceInfo {
+	onlyNodeLocalEndpoints := false
+	if apiservice.RequestsOnlyLocalTraffic(service) {
+		onlyNodeLocalEndpoints = true
+	}
+	var stickyMaxAgeSeconds int
+	if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
+		// Kube-apiserver side guarantees SessionAffinityConfig won't be nil when session affinity type is ClientIP
+		stickyMaxAgeSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
+	}
+	info := &BaseServiceInfo{
+		ClusterIP: net.ParseIP(service.Spec.ClusterIP),
+		Port:      int(port.Port),
+		Protocol:  port.Protocol,
+		NodePort:  int(port.NodePort),
+		// Deep-copy in case the service instance changes
+		LoadBalancerStatus:     *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer),
+		SessionAffinityType:    service.Spec.SessionAffinity,
+		StickyMaxAgeSeconds:    stickyMaxAgeSeconds,
+		OnlyNodeLocalEndpoints: onlyNodeLocalEndpoints,
+	}
+
+	if sct.isIPv6Mode == nil {
+		info.ExternalIPs = make([]string, len(service.Spec.ExternalIPs))
+		info.LoadBalancerSourceRanges = make([]string, len(service.Spec.LoadBalancerSourceRanges))
+		copy(info.LoadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
+		copy(info.ExternalIPs, service.Spec.ExternalIPs)
+	} else {
+		// Filter out the incorrect IP version case.
+		// If ExternalIPs and LoadBalancerSourceRanges on service contains incorrect IP versions,
+		// only filter out the incorrect ones.
+		var incorrectIPs []string
+		info.ExternalIPs, incorrectIPs = utilnet.FilterIncorrectIPVersion(service.Spec.ExternalIPs, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "externalIPs", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+		info.LoadBalancerSourceRanges, incorrectIPs = utilnet.FilterIncorrectCIDRVersion(service.Spec.LoadBalancerSourceRanges, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "loadBalancerSourceRanges", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+	}
+
+	if apiservice.NeedsHealthCheck(service) {
+		p := service.Spec.HealthCheckNodePort
+		if p == 0 {
+			glog.Errorf("Service %s/%s has no healthcheck nodeport", service.Namespace, service.Name)
+		} else {
+			info.HealthCheckNodePort = int(p)
+		}
+	}
+
+	return info
+}
+
+type makeServicePortFunc func(*api.ServicePort, *api.Service, *BaseServiceInfo) ServicePort
+
+// 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
+	// makeServiceInfo allows proxier to inject customized information when processing service.
+	makeServiceInfo makeServicePortFunc
+	// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
+	isIPv6Mode *bool
+	recorder   record.EventRecorder
+}
+
+// NewServiceChangeTracker initializes a ServiceChangeTracker
+func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
+	return &ServiceChangeTracker{
+		items:           make(map[types.NamespacedName]*serviceChange),
+		makeServiceInfo: makeServiceInfo,
+		isIPv6Mode:      isIPv6Mode,
+		recorder:        recorder,
+	}
+}
+
+// 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.
+func (sct *ServiceChangeTracker) Update(previous, current *api.Service) 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 = sct.serviceToServiceMap(previous)
+		sct.items[namespacedName] = change
+	}
+	change.current = sct.serviceToServiceMap(current)
+	// 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.GetHealthCheckNodePort() != 0 {
+			result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.GetHealthCheckNodePort())
+		}
+	}
+
+	return result
+}
+
+// ServiceMap maps a service to its ServicePort.
+type ServiceMap map[ServicePortName]ServicePort
+
+// serviceToServiceMap translates a single Service object to a ServiceMap.
+//
+// NOTE: service object should NOT be modified.
+func (sct *ServiceChangeTracker) serviceToServiceMap(service *api.Service) ServiceMap {
+	if service == nil {
+		return nil
+	}
+	svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if utilproxy.ShouldSkipService(svcName, service) {
+		return nil
+	}
+
+	if len(service.Spec.ClusterIP) != 0 {
+		// Filter out the incorrect IP version case.
+		// If ClusterIP on service has incorrect IP version, service itself will be ignored.
+		if sct.isIPv6Mode != nil && utilnet.IsIPv6String(service.Spec.ClusterIP) != *sct.isIPv6Mode {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "clusterIP", service.Spec.ClusterIP, service.Namespace, service.Name, service.UID)
+			return nil
+		}
+	}
+
+	serviceMap := make(ServiceMap)
+	for i := range service.Spec.Ports {
+		servicePort := &service.Spec.Ports[i]
+		svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name}
+		baseSvcInfo := sct.newBaseServiceInfo(servicePort, service)
+		if sct.makeServiceInfo != nil {
+			serviceMap[svcPortName] = sct.makeServiceInfo(servicePort, service, baseSvcInfo)
+		} else {
+			serviceMap[svcPortName] = baseSvcInfo
+		}
+	}
+	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.GetProtocol() == api.ProtocolUDP {
+				UDPStaleClusterIP.Insert(info.ClusterIPString())
+			}
+			delete(*sm, svcPortName)
+		} else {
+			glog.Errorf("Service port %q doesn't exists", svcPortName)
+		}
+	}
+}

vendor/k8s.io/kubernetes/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,37 @@ 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
+	// ClusterIPString returns service cluster IP in string format.
+	ClusterIPString() string
+	// GetProtocol returns service protocol.
+	GetProtocol() api.Protocol
+	// GetHealthCheckNodePort returns service health check node port if present.  If return 0, it means not present.
+	GetHealthCheckNodePort() int
+}
+
+// Endpoint in an interface which abstracts information about an endpoint.
+// TODO: Rename functions to be consistent with ServicePort.
+type Endpoint interface {
+	// String returns endpoint string.  An example format can be: `IP:Port`.
+	// We take the returned value as ServiceEndpoint.Endpoint.
+	String() string
+	// GetIsLocal returns true if the endpoint is running in same host as kube-proxy, otherwise returns false.
+	GetIsLocal() bool
+	// IP returns IP part of the endpoint.
+	IP() string
+	// Port returns the Port part of the endpoint.
+	Port() (int, error)
+	// 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
+}

vendor/k8s.io/kubernetes/pkg/proxy/util/conntrack.go

@@ -1,105 +0,0 @@
-/*
-Copyright 2016 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 util
-
-import (
-	"fmt"
-	"net"
-	"strconv"
-	"strings"
-
-	"k8s.io/utils/exec"
-)
-
-// Utilities for dealing with conntrack
-
-const NoConnectionToDelete = "0 flow entries have been deleted"
-
-func IsIPv6(netIP net.IP) bool {
-	return netIP != nil && netIP.To4() == nil
-}
-
-func IsIPv6String(ip string) bool {
-	netIP := net.ParseIP(ip)
-	return IsIPv6(netIP)
-}
-
-func parametersWithFamily(isIPv6 bool, parameters ...string) []string {
-	if isIPv6 {
-		parameters = append(parameters, "-f", "ipv6")
-	}
-	return parameters
-}
-
-// ClearUDPConntrackForIP uses the conntrack tool to delete the conntrack entries
-// for the UDP connections specified by the given service IP
-func ClearUDPConntrackForIP(execer exec.Interface, ip string) error {
-	parameters := parametersWithFamily(IsIPv6String(ip), "-D", "--orig-dst", ip, "-p", "udp")
-	err := ExecConntrackTool(execer, parameters...)
-	if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {
-		// TODO: Better handling for deletion failure. When failure occur, stale udp connection may not get flushed.
-		// These stale udp connection will keep black hole traffic. Making this a best effort operation for now, since it
-		// is expensive to baby-sit all udp connections to kubernetes services.
-		return fmt.Errorf("error deleting connection tracking state for UDP service IP: %s, error: %v", ip, err)
-	}
-	return nil
-}
-
-// ExecConntrackTool executes the conntrack tool using the given parameters
-func ExecConntrackTool(execer exec.Interface, parameters ...string) error {
-	conntrackPath, err := execer.LookPath("conntrack")
-	if err != nil {
-		return fmt.Errorf("error looking for path of conntrack: %v", err)
-	}
-	output, err := execer.Command(conntrackPath, parameters...).CombinedOutput()
-	if err != nil {
-		return fmt.Errorf("conntrack command returned: %q, error message: %s", string(output), err)
-	}
-	return nil
-}
-
-// ClearUDPConntrackForPort uses the conntrack tool to delete the conntrack entries
-// for the UDP connections specified by the port.
-// When a packet arrives, it will not go through NAT table again, because it is not "the first" packet.
-// The solution is clearing the conntrack. Known issues:
-// https://github.com/docker/docker/issues/8795
-// https://github.com/kubernetes/kubernetes/issues/31983
-func ClearUDPConntrackForPort(execer exec.Interface, port int, isIPv6 bool) error {
-	if port <= 0 {
-		return fmt.Errorf("Wrong port number. The port number must be greater than zero")
-	}
-	parameters := parametersWithFamily(isIPv6, "-D", "-p", "udp", "--dport", strconv.Itoa(port))
-	err := ExecConntrackTool(execer, parameters...)
-	if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {
-		return fmt.Errorf("error deleting conntrack entries for UDP port: %d, error: %v", port, err)
-	}
-	return nil
-}
-
-// ClearUDPConntrackForPeers uses the conntrack tool to delete the conntrack entries
-// for the UDP connections specified by the {origin, dest} IP pair.
-func ClearUDPConntrackForPeers(execer exec.Interface, origin, dest string) error {
-	parameters := parametersWithFamily(IsIPv6String(origin), "-D", "--orig-dst", origin, "--dst-nat", dest, "-p", "udp")
-	err := ExecConntrackTool(execer, parameters...)
-	if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {
-		// TODO: Better handling for deletion failure. When failure occur, stale udp connection may not get flushed.
-		// These stale udp connection will keep black hole traffic. Making this a best effort operation for now, since it
-		// is expensive to baby sit all udp connections to kubernetes services.
-		return fmt.Errorf("error deleting conntrack entries for UDP peer {%s, %s}, error: %v", origin, dest, err)
-	}
-	return nil
-}

vendor/k8s.io/kubernetes/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)

vendor/k8s.io/kubernetes/pkg/proxy/util/network.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2018 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 util
+
+import (
+	"net"
+)
+
+// NetworkInterfacer defines an interface for several net library functions. Production
+// code will forward to net library functions, and unit tests will override the methods
+// for testing purposes.
+type NetworkInterfacer interface {
+	Addrs(intf *net.Interface) ([]net.Addr, error)
+	Interfaces() ([]net.Interface, error)
+}
+
+// RealNetwork implements the NetworkInterfacer interface for production code, just
+// wrapping the underlying net library function calls.
+type RealNetwork struct{}
+
+// Addrs wraps net.Interface.Addrs(), it's a part of NetworkInterfacer interface.
+func (_ RealNetwork) Addrs(intf *net.Interface) ([]net.Addr, error) {
+	return intf.Addrs()
+}
+
+// Interfaces wraps net.Interfaces(), it's a part of NetworkInterfacer interface.
+func (_ RealNetwork) Interfaces() ([]net.Interface, error) {
+	return net.Interfaces()
+}
+
+var _ NetworkInterfacer = &RealNetwork{}

vendor/k8s.io/kubernetes/pkg/proxy/util/utils.go

@@ -17,15 +17,32 @@ limitations under the License.
 package util
 
 import (
+	"fmt"
 	"net"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 
 	"github.com/golang/glog"
 )
 
+const (
+	IPv4ZeroCIDR = "0.0.0.0/0"
+	IPv6ZeroCIDR = "::/0"
+)
+
+func IsZeroCIDR(cidr string) bool {
+	if cidr == IPv4ZeroCIDR || cidr == IPv6ZeroCIDR {
+		return true
+	}
+	return false
+}
+
 func IsLocalIP(ip string) (bool, error) {
 	addrs, err := net.InterfaceAddrs()
 	if err != nil {
@@ -56,3 +73,76 @@ func ShouldSkipService(svcName types.NamespacedName, service *api.Service) bool
 	}
 	return false
 }
+
+// GetNodeAddresses return all matched node IP addresses based on given cidr slice.
+// Some callers, e.g. IPVS proxier, need concrete IPs, not ranges, which is why this exists.
+// NetworkInterfacer is injected for test purpose.
+// We expect the cidrs passed in is already validated.
+// Given an empty input `[]`, it will return `0.0.0.0/0` and `::/0` directly.
+// If multiple cidrs is given, it will return the minimal IP sets, e.g. given input `[1.2.0.0/16, 0.0.0.0/0]`, it will
+// only return `0.0.0.0/0`.
+// NOTE: GetNodeAddresses only accepts CIDRs, if you want concrete IPs, e.g. 1.2.3.4, then the input should be 1.2.3.4/32.
+func GetNodeAddresses(cidrs []string, nw NetworkInterfacer) (sets.String, error) {
+	uniqueAddressList := sets.NewString()
+	if len(cidrs) == 0 {
+		uniqueAddressList.Insert(IPv4ZeroCIDR)
+		uniqueAddressList.Insert(IPv6ZeroCIDR)
+		return uniqueAddressList, nil
+	}
+	// First round of iteration to pick out `0.0.0.0/0` or `::/0` for the sake of excluding non-zero IPs.
+	for _, cidr := range cidrs {
+		if IsZeroCIDR(cidr) {
+			uniqueAddressList.Insert(cidr)
+		}
+	}
+	// Second round of iteration to parse IPs based on cidr.
+	for _, cidr := range cidrs {
+		if IsZeroCIDR(cidr) {
+			continue
+		}
+		_, ipNet, _ := net.ParseCIDR(cidr)
+		itfs, err := nw.Interfaces()
+		if err != nil {
+			return nil, fmt.Errorf("error listing all interfaces from host, error: %v", err)
+		}
+		for _, itf := range itfs {
+			addrs, err := nw.Addrs(&itf)
+			if err != nil {
+				return nil, fmt.Errorf("error getting address from interface %s, error: %v", itf.Name, err)
+			}
+			for _, addr := range addrs {
+				if addr == nil {
+					continue
+				}
+				ip, _, err := net.ParseCIDR(addr.String())
+				if err != nil {
+					return nil, fmt.Errorf("error parsing CIDR for interface %s, error: %v", itf.Name, err)
+				}
+				if ipNet.Contains(ip) {
+					if utilnet.IsIPv6(ip) && !uniqueAddressList.Has(IPv6ZeroCIDR) {
+						uniqueAddressList.Insert(ip.String())
+					}
+					if !utilnet.IsIPv6(ip) && !uniqueAddressList.Has(IPv4ZeroCIDR) {
+						uniqueAddressList.Insert(ip.String())
+					}
+				}
+			}
+		}
+	}
+	return uniqueAddressList, nil
+}
+
+// LogAndEmitIncorrectIPVersionEvent logs and emits incorrect IP version event.
+func LogAndEmitIncorrectIPVersionEvent(recorder record.EventRecorder, fieldName, fieldValue, svcNamespace, svcName string, svcUID types.UID) {
+	errMsg := fmt.Sprintf("%s in %s has incorrect IP version", fieldValue, fieldName)
+	glog.Errorf("%s (service %s/%s).", errMsg, svcNamespace, svcName)
+	if recorder != nil {
+		recorder.Eventf(
+			&v1.ObjectReference{
+				Kind:      "Service",
+				Name:      svcName,
+				Namespace: svcNamespace,
+				UID:       svcUID,
+			}, v1.EventTypeWarning, "KubeProxyIncorrectIPVersion", errMsg)
+	}
+}