Initial checkin from CRI-O repo

Signed-off-by: Matthew Heon <matthew.heon@gmail.com>

7 files changed, 2253 insertions, 0 deletions

diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/doc.go b/vendor/k8s.io/kubernetes/pkg/proxy/doc.go
new file mode 100644
index 000000000..3bed0fa39
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2014 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 implements the layer-3 network proxy.
+package proxy // import "k8s.io/kubernetes/pkg/proxy"
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/doc.go b/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/doc.go
new file mode 100644
index 000000000..0a9ea0944
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/doc.go
@@ -0,0 +1,18 @@
+/*
+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 healthcheck provides tools for serving kube-proxy healthchecks.
+package healthcheck // import "k8s.io/kubernetes/pkg/proxy/healthcheck"
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/healthcheck.go b/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/healthcheck.go
new file mode 100644
index 000000000..39f10f71a
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/healthcheck/healthcheck.go
@@ -0,0 +1,333 @@
+/*
+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 healthcheck
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/golang/glog"
+	"github.com/renstrom/dedent"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/clock"
+	clientv1 "k8s.io/client-go/pkg/api/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+// Server serves HTTP endpoints for each service name, with results
+// based on the endpoints.  If there are 0 endpoints for a service, it returns a
+// 503 "Service Unavailable" error (telling LBs not to use this node).  If there
+// are 1 or more endpoints, it returns a 200 "OK".
+type Server interface {
+	// Make the new set of services be active.  Services that were open before
+	// will be closed.  Services that are new will be opened.  Service that
+	// existed and are in the new set will be left alone.  The value of the map
+	// is the healthcheck-port to listen on.
+	SyncServices(newServices map[types.NamespacedName]uint16) error
+	// Make the new set of endpoints be active.  Endpoints for services that do
+	// not exist will be dropped.  The value of the map is the number of
+	// endpoints the service has on this node.
+	SyncEndpoints(newEndpoints map[types.NamespacedName]int) error
+}
+
+// Listener allows for testing of Server.  If the Listener argument
+// to NewServer() is nil, the real net.Listen function will be used.
+type Listener interface {
+	// Listen is very much like net.Listen, except the first arg (network) is
+	// fixed to be "tcp".
+	Listen(addr string) (net.Listener, error)
+}
+
+// HTTPServerFactory allows for testing of Server.  If the
+// HTTPServerFactory argument to NewServer() is nil, the real
+// http.Server type will be used.
+type HTTPServerFactory interface {
+	// New creates an instance of a type satisfying HTTPServer.  This is
+	// designed to include http.Server.
+	New(addr string, handler http.Handler) HTTPServer
+}
+
+// HTTPServer allows for testing of Server.
+type HTTPServer interface {
+	// Server is designed so that http.Server satifies this interface,
+	Serve(listener net.Listener) error
+}
+
+// NewServer allocates a new healthcheck server manager.  If either
+// of the injected arguments are nil, defaults will be used.
+func NewServer(hostname string, recorder record.EventRecorder, listener Listener, httpServerFactory HTTPServerFactory) Server {
+	if listener == nil {
+		listener = stdNetListener{}
+	}
+	if httpServerFactory == nil {
+		httpServerFactory = stdHTTPServerFactory{}
+	}
+	return &server{
+		hostname:    hostname,
+		recorder:    recorder,
+		listener:    listener,
+		httpFactory: httpServerFactory,
+		services:    map[types.NamespacedName]*hcInstance{},
+	}
+}
+
+// Implement Listener in terms of net.Listen.
+type stdNetListener struct{}
+
+func (stdNetListener) Listen(addr string) (net.Listener, error) {
+	return net.Listen("tcp", addr)
+}
+
+var _ Listener = stdNetListener{}
+
+// Implement HTTPServerFactory in terms of http.Server.
+type stdHTTPServerFactory struct{}
+
+func (stdHTTPServerFactory) New(addr string, handler http.Handler) HTTPServer {
+	return &http.Server{
+		Addr:    addr,
+		Handler: handler,
+	}
+}
+
+var _ HTTPServerFactory = stdHTTPServerFactory{}
+
+type server struct {
+	hostname    string
+	recorder    record.EventRecorder // can be nil
+	listener    Listener
+	httpFactory HTTPServerFactory
+
+	lock     sync.Mutex
+	services map[types.NamespacedName]*hcInstance
+}
+
+func (hcs *server) SyncServices(newServices map[types.NamespacedName]uint16) error {
+	hcs.lock.Lock()
+	defer hcs.lock.Unlock()
+
+	// Remove any that are not needed any more.
+	for nsn, svc := range hcs.services {
+		if port, found := newServices[nsn]; !found || port != svc.port {
+			glog.V(2).Infof("Closing healthcheck %q on port %d", nsn.String(), svc.port)
+			if err := svc.listener.Close(); err != nil {
+				glog.Errorf("Close(%v): %v", svc.listener.Addr(), err)
+			}
+			delete(hcs.services, nsn)
+		}
+	}
+
+	// Add any that are needed.
+	for nsn, port := range newServices {
+		if hcs.services[nsn] != nil {
+			glog.V(3).Infof("Existing healthcheck %q on port %d", nsn.String(), port)
+			continue
+		}
+
+		glog.V(2).Infof("Opening healthcheck %q on port %d", nsn.String(), port)
+		svc := &hcInstance{port: port}
+		addr := fmt.Sprintf(":%d", port)
+		svc.server = hcs.httpFactory.New(addr, hcHandler{name: nsn, hcs: hcs})
+		var err error
+		svc.listener, err = hcs.listener.Listen(addr)
+		if err != nil {
+			msg := fmt.Sprintf("node %s failed to start healthcheck %q on port %d: %v", hcs.hostname, nsn.String(), port, err)
+
+			if hcs.recorder != nil {
+				hcs.recorder.Eventf(
+					&clientv1.ObjectReference{
+						Kind:      "Service",
+						Namespace: nsn.Namespace,
+						Name:      nsn.Name,
+						UID:       types.UID(nsn.String()),
+					}, api.EventTypeWarning, "FailedToStartHealthcheck", msg)
+			}
+			glog.Error(msg)
+			continue
+		}
+		hcs.services[nsn] = svc
+
+		go func(nsn types.NamespacedName, svc *hcInstance) {
+			// Serve() will exit when the listener is closed.
+			glog.V(3).Infof("Starting goroutine for healthcheck %q on port %d", nsn.String(), svc.port)
+			if err := svc.server.Serve(svc.listener); err != nil {
+				glog.V(3).Infof("Healthcheck %q closed: %v", nsn.String(), err)
+				return
+			}
+			glog.V(3).Infof("Healthcheck %q closed", nsn.String())
+		}(nsn, svc)
+	}
+	return nil
+}
+
+type hcInstance struct {
+	port      uint16
+	listener  net.Listener
+	server    HTTPServer
+	endpoints int // number of local endpoints for a service
+}
+
+type hcHandler struct {
+	name types.NamespacedName
+	hcs  *server
+}
+
+var _ http.Handler = hcHandler{}
+
+func (h hcHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+	h.hcs.lock.Lock()
+	svc, ok := h.hcs.services[h.name]
+	if !ok || svc == nil {
+		h.hcs.lock.Unlock()
+		glog.Errorf("Received request for closed healthcheck %q", h.name.String())
+		return
+	}
+	count := svc.endpoints
+	h.hcs.lock.Unlock()
+
+	resp.Header().Set("Content-Type", "application/json")
+	if count == 0 {
+		resp.WriteHeader(http.StatusServiceUnavailable)
+	} else {
+		resp.WriteHeader(http.StatusOK)
+	}
+	fmt.Fprintf(resp, strings.Trim(dedent.Dedent(fmt.Sprintf(`
+		{
+			"service": {
+				"namespace": %q,
+				"name": %q
+			},
+			"localEndpoints": %d
+		}
+		`, h.name.Namespace, h.name.Name, count)), "\n"))
+}
+
+func (hcs *server) SyncEndpoints(newEndpoints map[types.NamespacedName]int) error {
+	hcs.lock.Lock()
+	defer hcs.lock.Unlock()
+
+	for nsn, count := range newEndpoints {
+		if hcs.services[nsn] == nil {
+			glog.V(3).Infof("Not saving endpoints for unknown healthcheck %q", nsn.String())
+			continue
+		}
+		glog.V(3).Infof("Reporting %d endpoints for healthcheck %q", count, nsn.String())
+		hcs.services[nsn].endpoints = count
+	}
+	for nsn, hci := range hcs.services {
+		if _, found := newEndpoints[nsn]; !found {
+			hci.endpoints = 0
+		}
+	}
+	return nil
+}
+
+// HealthzUpdater allows callers to update healthz timestamp only.
+type HealthzUpdater interface {
+	UpdateTimestamp()
+}
+
+// HealthzServer returns 200 "OK" by default. Once timestamp has been
+// updated, it verifies we don't exceed max no respond duration since
+// last update.
+type HealthzServer struct {
+	listener    Listener
+	httpFactory HTTPServerFactory
+	clock       clock.Clock
+
+	addr          string
+	port          int32
+	healthTimeout time.Duration
+
+	lastUpdated atomic.Value
+}
+
+// NewDefaultHealthzServer returns a default healthz http server.
+func NewDefaultHealthzServer(addr string, healthTimeout time.Duration) *HealthzServer {
+	return newHealthzServer(nil, nil, nil, addr, healthTimeout)
+}
+
+func newHealthzServer(listener Listener, httpServerFactory HTTPServerFactory, c clock.Clock, addr string, healthTimeout time.Duration) *HealthzServer {
+	if listener == nil {
+		listener = stdNetListener{}
+	}
+	if httpServerFactory == nil {
+		httpServerFactory = stdHTTPServerFactory{}
+	}
+	if c == nil {
+		c = clock.RealClock{}
+	}
+	return &HealthzServer{
+		listener:      listener,
+		httpFactory:   httpServerFactory,
+		clock:         c,
+		addr:          addr,
+		healthTimeout: healthTimeout,
+	}
+}
+
+// UpdateTimestamp updates the lastUpdated timestamp.
+func (hs *HealthzServer) UpdateTimestamp() {
+	hs.lastUpdated.Store(hs.clock.Now())
+}
+
+// Run starts the healthz http server and returns.
+func (hs *HealthzServer) Run() {
+	serveMux := http.NewServeMux()
+	serveMux.Handle("/healthz", healthzHandler{hs: hs})
+	server := hs.httpFactory.New(hs.addr, serveMux)
+	listener, err := hs.listener.Listen(hs.addr)
+	if err != nil {
+		glog.Errorf("Failed to start healthz on %s: %v", hs.addr, err)
+		return
+	}
+	go func() {
+		glog.V(3).Infof("Starting goroutine for healthz on %s", hs.addr)
+		if err := server.Serve(listener); err != nil {
+			glog.Errorf("Healhz closed: %v", err)
+			return
+		}
+		glog.Errorf("Unexpected healhz closed.")
+	}()
+}
+
+type healthzHandler struct {
+	hs *HealthzServer
+}
+
+func (h healthzHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
+	lastUpdated := time.Time{}
+	if val := h.hs.lastUpdated.Load(); val != nil {
+		lastUpdated = val.(time.Time)
+	}
+	currentTime := h.hs.clock.Now()
+
+	resp.Header().Set("Content-Type", "application/json")
+	if !lastUpdated.IsZero() && currentTime.After(lastUpdated.Add(h.hs.healthTimeout)) {
+		resp.WriteHeader(http.StatusServiceUnavailable)
+	} else {
+		resp.WriteHeader(http.StatusOK)
+	}
+	fmt.Fprintf(resp, fmt.Sprintf(`{"lastUpdated": %q,"currentTime": %q}`, lastUpdated, currentTime))
+}
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/iptables/metrics.go b/vendor/k8s.io/kubernetes/pkg/proxy/iptables/metrics.go
new file mode 100644
index 000000000..fabe6a595
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/iptables/metrics.go
@@ -0,0 +1,50 @@
+/*
+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 iptables
+
+import (
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const kubeProxySubsystem = "kubeproxy"
+
+var (
+	SyncProxyRulesLatency = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Subsystem: kubeProxySubsystem,
+			Name:      "sync_proxy_rules_latency_microseconds",
+			Help:      "SyncProxyRules latency",
+			Buckets:   prometheus.ExponentialBuckets(1000, 2, 15),
+		},
+	)
+)
+
+var registerMetricsOnce sync.Once
+
+func RegisterMetrics() {
+	registerMetricsOnce.Do(func() {
+		prometheus.MustRegister(SyncProxyRulesLatency)
+	})
+}
+
+// Gets the time since the specified start in microseconds.
+func sinceInMicroseconds(start time.Time) float64 {
+	return float64(time.Since(start).Nanoseconds() / time.Microsecond.Nanoseconds())
+}
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/iptables/proxier.go b/vendor/k8s.io/kubernetes/pkg/proxy/iptables/proxier.go
new file mode 100644
index 000000000..9d29d7b42
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/iptables/proxier.go
@@ -0,0 +1,1732 @@
+/*
+Copyright 2015 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 iptables
+
+//
+// NOTE: this needs to be tested in e2e since it uses iptables for everything.
+//
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/base32"
+	"fmt"
+	"net"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	clientv1 "k8s.io/client-go/pkg/api/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/helper"
+	apiservice "k8s.io/kubernetes/pkg/api/service"
+	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/proxy"
+	"k8s.io/kubernetes/pkg/proxy/healthcheck"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	"k8s.io/kubernetes/pkg/util/async"
+	utilexec "k8s.io/kubernetes/pkg/util/exec"
+	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
+	utilsysctl "k8s.io/kubernetes/pkg/util/sysctl"
+	utilversion "k8s.io/kubernetes/pkg/util/version"
+)
+
+const (
+	// iptablesMinVersion is the minimum version of iptables for which we will use the Proxier
+	// from this package instead of the userspace Proxier.  While most of the
+	// features we need were available earlier, the '-C' flag was added more
+	// recently.  We use that indirectly in Ensure* functions, and if we don't
+	// have it, we have to be extra careful about the exact args we feed in being
+	// the same as the args we read back (iptables itself normalizes some args).
+	// This is the "new" Proxier, so we require "new" versions of tools.
+	iptablesMinVersion = utiliptables.MinCheckVersion
+
+	// the services chain
+	kubeServicesChain utiliptables.Chain = "KUBE-SERVICES"
+
+	// the nodeports chain
+	kubeNodePortsChain utiliptables.Chain = "KUBE-NODEPORTS"
+
+	// the kubernetes postrouting chain
+	kubePostroutingChain utiliptables.Chain = "KUBE-POSTROUTING"
+
+	// the mark-for-masquerade chain
+	KubeMarkMasqChain utiliptables.Chain = "KUBE-MARK-MASQ"
+
+	// the mark-for-drop chain
+	KubeMarkDropChain utiliptables.Chain = "KUBE-MARK-DROP"
+)
+
+// IPTablesVersioner can query the current iptables version.
+type IPTablesVersioner interface {
+	// returns "X.Y.Z"
+	GetVersion() (string, error)
+}
+
+// KernelCompatTester tests whether the required kernel capabilities are
+// present to run the iptables proxier.
+type KernelCompatTester interface {
+	IsCompatible() error
+}
+
+// CanUseIPTablesProxier returns true if we should use the iptables Proxier
+// instead of the "classic" userspace Proxier.  This is determined by checking
+// the iptables version and for the existence of kernel features. It may return
+// an error if it fails to get the iptables version without error, in which
+// case it will also return false.
+func CanUseIPTablesProxier(iptver IPTablesVersioner, kcompat KernelCompatTester) (bool, error) {
+	minVersion, err := utilversion.ParseGeneric(iptablesMinVersion)
+	if err != nil {
+		return false, err
+	}
+	versionString, err := iptver.GetVersion()
+	if err != nil {
+		return false, err
+	}
+	version, err := utilversion.ParseGeneric(versionString)
+	if err != nil {
+		return false, err
+	}
+	if version.LessThan(minVersion) {
+		return false, nil
+	}
+
+	// Check that the kernel supports what we need.
+	if err := kcompat.IsCompatible(); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+type LinuxKernelCompatTester struct{}
+
+func (lkct LinuxKernelCompatTester) IsCompatible() error {
+	// Check for the required sysctls.  We don't care about the value, just
+	// that it exists.  If this Proxier is chosen, we'll initialize it as we
+	// need.
+	_, err := utilsysctl.New().GetSysctl(sysctlRouteLocalnet)
+	return err
+}
+
+const sysctlRouteLocalnet = "net/ipv4/conf/all/route_localnet"
+const sysctlBridgeCallIPTables = "net/bridge/bridge-nf-call-iptables"
+
+// internal struct for string service information
+type serviceInfo struct {
+	clusterIP                net.IP
+	port                     int
+	protocol                 api.Protocol
+	nodePort                 int
+	loadBalancerStatus       api.LoadBalancerStatus
+	sessionAffinityType      api.ServiceAffinity
+	stickyMaxAgeMinutes      int
+	externalIPs              []string
+	loadBalancerSourceRanges []string
+	onlyNodeLocalEndpoints   bool
+	healthCheckNodePort      int
+	// The following fields are computed and stored for performance reasons.
+	serviceNameString        string
+	servicePortChainName     utiliptables.Chain
+	serviceFirewallChainName utiliptables.Chain
+	serviceLBChainName       utiliptables.Chain
+}
+
+// internal struct for endpoints information
+type endpointsInfo struct {
+	endpoint string // TODO: should be an endpointString type
+	isLocal  bool
+	// The following fields we lazily compute and store here for performance
+	// reasons. If the protocol is the same as you expect it to be, then the
+	// chainName can be reused, otherwise it should be recomputed.
+	protocol  string
+	chainName utiliptables.Chain
+}
+
+// Returns just the IP part of the endpoint.
+func (e *endpointsInfo) IPPart() string {
+	if index := strings.Index(e.endpoint, ":"); index != -1 {
+		return e.endpoint[0:index]
+	}
+	return e.endpoint
+}
+
+// Returns the endpoint chain name for a given endpointsInfo.
+func (e *endpointsInfo) endpointChain(svcNameString, protocol string) utiliptables.Chain {
+	if e.protocol != protocol {
+		e.protocol = protocol
+		e.chainName = servicePortEndpointChainName(svcNameString, protocol, e.endpoint)
+	}
+	return e.chainName
+}
+
+func (e *endpointsInfo) String() string {
+	return fmt.Sprintf("%v", *e)
+}
+
+// returns a new serviceInfo struct
+func newServiceInfo(svcPortName proxy.ServicePortName, port *api.ServicePort, service *api.Service) *serviceInfo {
+	onlyNodeLocalEndpoints := false
+	if utilfeature.DefaultFeatureGate.Enabled(features.ExternalTrafficLocalOnly) &&
+		apiservice.RequestsOnlyLocalTraffic(service) {
+		onlyNodeLocalEndpoints = true
+	}
+	info := &serviceInfo{
+		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,
+		stickyMaxAgeMinutes:      180, // TODO: paramaterize this in the API.
+		externalIPs:              make([]string, len(service.Spec.ExternalIPs)),
+		loadBalancerSourceRanges: make([]string, len(service.Spec.LoadBalancerSourceRanges)),
+		onlyNodeLocalEndpoints:   onlyNodeLocalEndpoints,
+	}
+	copy(info.loadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
+	copy(info.externalIPs, service.Spec.ExternalIPs)
+
+	if apiservice.NeedsHealthCheck(service) {
+		p := apiservice.GetServiceHealthCheckNodePort(service)
+		if p == 0 {
+			glog.Errorf("Service %q has no healthcheck nodeport", svcPortName.NamespacedName.String())
+		} else {
+			info.healthCheckNodePort = int(p)
+		}
+	}
+
+	// Store the following for performance reasons.
+	protocol := strings.ToLower(string(info.protocol))
+	info.serviceNameString = svcPortName.String()
+	info.servicePortChainName = servicePortChainName(info.serviceNameString, protocol)
+	info.serviceFirewallChainName = serviceFirewallChainName(info.serviceNameString, protocol)
+	info.serviceLBChainName = serviceLBChainName(info.serviceNameString, protocol)
+
+	return info
+}
+
+type endpointsChange struct {
+	previous proxyEndpointsMap
+	current  proxyEndpointsMap
+}
+
+type endpointsChangeMap struct {
+	lock     sync.Mutex
+	hostname string
+	items    map[types.NamespacedName]*endpointsChange
+}
+
+type serviceChange struct {
+	previous proxyServiceMap
+	current  proxyServiceMap
+}
+
+type serviceChangeMap struct {
+	lock  sync.Mutex
+	items map[types.NamespacedName]*serviceChange
+}
+
+type updateEndpointMapResult struct {
+	hcEndpoints       map[types.NamespacedName]int
+	staleEndpoints    map[endpointServicePair]bool
+	staleServiceNames map[proxy.ServicePortName]bool
+}
+
+type updateServiceMapResult struct {
+	hcServices    map[types.NamespacedName]uint16
+	staleServices sets.String
+}
+
+type proxyServiceMap map[proxy.ServicePortName]*serviceInfo
+type proxyEndpointsMap map[proxy.ServicePortName][]*endpointsInfo
+
+func newEndpointsChangeMap(hostname string) endpointsChangeMap {
+	return endpointsChangeMap{
+		hostname: hostname,
+		items:    make(map[types.NamespacedName]*endpointsChange),
+	}
+}
+
+func (ecm *endpointsChangeMap) update(namespacedName *types.NamespacedName, previous, current *api.Endpoints) bool {
+	ecm.lock.Lock()
+	defer ecm.lock.Unlock()
+
+	change, exists := ecm.items[*namespacedName]
+	if !exists {
+		change = &endpointsChange{}
+		change.previous = endpointsToEndpointsMap(previous, ecm.hostname)
+		ecm.items[*namespacedName] = change
+	}
+	change.current = endpointsToEndpointsMap(current, ecm.hostname)
+	if reflect.DeepEqual(change.previous, change.current) {
+		delete(ecm.items, *namespacedName)
+	}
+	return len(ecm.items) > 0
+}
+
+func newServiceChangeMap() serviceChangeMap {
+	return serviceChangeMap{
+		items: make(map[types.NamespacedName]*serviceChange),
+	}
+}
+
+func (scm *serviceChangeMap) update(namespacedName *types.NamespacedName, previous, current *api.Service) bool {
+	scm.lock.Lock()
+	defer scm.lock.Unlock()
+
+	change, exists := scm.items[*namespacedName]
+	if !exists {
+		change = &serviceChange{}
+		change.previous = serviceToServiceMap(previous)
+		scm.items[*namespacedName] = change
+	}
+	change.current = serviceToServiceMap(current)
+	if reflect.DeepEqual(change.previous, change.current) {
+		delete(scm.items, *namespacedName)
+	}
+	return len(scm.items) > 0
+}
+
+func (sm *proxyServiceMap) merge(other proxyServiceMap) sets.String {
+	existingPorts := sets.NewString()
+	for svcPortName, info := range other {
+		existingPorts.Insert(svcPortName.Port)
+		_, exists := (*sm)[svcPortName]
+		if !exists {
+			glog.V(1).Infof("Adding new service port %q at %s:%d/%s", svcPortName, info.clusterIP, info.port, info.protocol)
+		} else {
+			glog.V(1).Infof("Updating existing service port %q at %s:%d/%s", svcPortName, info.clusterIP, info.port, info.protocol)
+		}
+		(*sm)[svcPortName] = info
+	}
+	return existingPorts
+}
+
+func (sm *proxyServiceMap) unmerge(other proxyServiceMap, existingPorts, staleServices sets.String) {
+	for svcPortName := range other {
+		if existingPorts.Has(svcPortName.Port) {
+			continue
+		}
+		info, exists := (*sm)[svcPortName]
+		if exists {
+			glog.V(1).Infof("Removing service port %q", svcPortName)
+			if info.protocol == api.ProtocolUDP {
+				staleServices.Insert(info.clusterIP.String())
+			}
+			delete(*sm, svcPortName)
+		} else {
+			glog.Errorf("Service port %q removed, but doesn't exists", svcPortName)
+		}
+	}
+}
+
+func (em proxyEndpointsMap) merge(other proxyEndpointsMap) {
+	for svcPortName := range other {
+		em[svcPortName] = other[svcPortName]
+	}
+}
+
+func (em proxyEndpointsMap) unmerge(other proxyEndpointsMap) {
+	for svcPortName := range other {
+		delete(em, svcPortName)
+	}
+}
+
+// Proxier is an iptables based proxy for connections between a localhost:lport
+// and services that provide the actual backends.
+type Proxier struct {
+	// endpointsChanges and serviceChanges contains all changes to endpoints and
+	// services that happened since iptables was synced. For a single object,
+	// changes are accumulated, i.e. previous is state from before all of them,
+	// current is state after applying all of those.
+	endpointsChanges endpointsChangeMap
+	serviceChanges   serviceChangeMap
+
+	mu           sync.Mutex // protects the following fields
+	serviceMap   proxyServiceMap
+	endpointsMap proxyEndpointsMap
+	portsMap     map[localPort]closeable
+	// endpointsSynced and servicesSynced are set to true when corresponding
+	// objects are synced after startup. This is used to avoid updating iptables
+	// with some partial data after kube-proxy restart.
+	endpointsSynced bool
+	servicesSynced  bool
+	initialized     int32
+	syncRunner      *async.BoundedFrequencyRunner // governs calls to syncProxyRules
+
+	// These are effectively const and do not need the mutex to be held.
+	iptables       utiliptables.Interface
+	masqueradeAll  bool
+	masqueradeMark string
+	exec           utilexec.Interface
+	clusterCIDR    string
+	hostname       string
+	nodeIP         net.IP
+	portMapper     portOpener
+	recorder       record.EventRecorder
+	healthChecker  healthcheck.Server
+	healthzServer  healthcheck.HealthzUpdater
+
+	// Since converting probabilities (floats) to strings is expensive
+	// and we are using only probabilities in the format of 1/n, we are
+	// precomputing some number of those and cache for future reuse.
+	precomputedProbabilities []string
+
+	// The following buffers are used to reuse memory and avoid allocations
+	// that are significantly impacting performance.
+	iptablesData *bytes.Buffer
+	filterChains *bytes.Buffer
+	filterRules  *bytes.Buffer
+	natChains    *bytes.Buffer
+	natRules     *bytes.Buffer
+}
+
+type localPort struct {
+	desc     string
+	ip       string
+	port     int
+	protocol string
+}
+
+func (lp *localPort) String() string {
+	return fmt.Sprintf("%q (%s:%d/%s)", lp.desc, lp.ip, lp.port, lp.protocol)
+}
+
+type closeable interface {
+	Close() error
+}
+
+// portOpener is an interface around port opening/closing.
+// Abstracted out for testing.
+type portOpener interface {
+	OpenLocalPort(lp *localPort) (closeable, error)
+}
+
+// listenPortOpener opens ports by calling bind() and listen().
+type listenPortOpener struct{}
+
+// OpenLocalPort holds the given local port open.
+func (l *listenPortOpener) OpenLocalPort(lp *localPort) (closeable, error) {
+	return openLocalPort(lp)
+}
+
+// Proxier implements ProxyProvider
+var _ proxy.ProxyProvider = &Proxier{}
+
+// NewProxier returns a new Proxier given an iptables Interface instance.
+// Because of the iptables logic, it is assumed that there is only a single Proxier active on a machine.
+// An error will be returned if iptables fails to update or acquire the initial lock.
+// Once a proxier is created, it will keep iptables up to date in the background and
+// will not terminate if a particular iptables call fails.
+func NewProxier(ipt utiliptables.Interface,
+	sysctl utilsysctl.Interface,
+	exec utilexec.Interface,
+	syncPeriod time.Duration,
+	minSyncPeriod time.Duration,
+	masqueradeAll bool,
+	masqueradeBit int,
+	clusterCIDR string,
+	hostname string,
+	nodeIP net.IP,
+	recorder record.EventRecorder,
+	healthzServer healthcheck.HealthzUpdater,
+) (*Proxier, error) {
+	// check valid user input
+	if minSyncPeriod > syncPeriod {
+		return nil, fmt.Errorf("minSyncPeriod (%v) must be <= syncPeriod (%v)", minSyncPeriod, syncPeriod)
+	}
+
+	// Set the route_localnet sysctl we need for
+	if err := sysctl.SetSysctl(sysctlRouteLocalnet, 1); err != nil {
+		return nil, fmt.Errorf("can't set sysctl %s: %v", sysctlRouteLocalnet, err)
+	}
+
+	// Proxy needs br_netfilter and bridge-nf-call-iptables=1 when containers
+	// are connected to a Linux bridge (but not SDN bridges).  Until most
+	// plugins handle this, log when config is missing
+	if val, err := sysctl.GetSysctl(sysctlBridgeCallIPTables); err == nil && val != 1 {
+		glog.Warningf("missing br-netfilter module or unset sysctl br-nf-call-iptables; proxy may not work as intended")
+	}
+
+	// Generate the masquerade mark to use for SNAT rules.
+	if masqueradeBit < 0 || masqueradeBit > 31 {
+		return nil, fmt.Errorf("invalid iptables-masquerade-bit %v not in [0, 31]", masqueradeBit)
+	}
+	masqueradeValue := 1 << uint(masqueradeBit)
+	masqueradeMark := fmt.Sprintf("%#08x/%#08x", masqueradeValue, masqueradeValue)
+
+	if nodeIP == nil {
+		glog.Warningf("invalid nodeIP, initializing kube-proxy with 127.0.0.1 as nodeIP")
+		nodeIP = net.ParseIP("127.0.0.1")
+	}
+
+	if len(clusterCIDR) == 0 {
+		glog.Warningf("clusterCIDR not specified, unable to distinguish between internal and external traffic")
+	}
+
+	healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) // use default implementations of deps
+
+	proxier := &Proxier{
+		portsMap:                 make(map[localPort]closeable),
+		serviceMap:               make(proxyServiceMap),
+		serviceChanges:           newServiceChangeMap(),
+		endpointsMap:             make(proxyEndpointsMap),
+		endpointsChanges:         newEndpointsChangeMap(hostname),
+		iptables:                 ipt,
+		masqueradeAll:            masqueradeAll,
+		masqueradeMark:           masqueradeMark,
+		exec:                     exec,
+		clusterCIDR:              clusterCIDR,
+		hostname:                 hostname,
+		nodeIP:                   nodeIP,
+		portMapper:               &listenPortOpener{},
+		recorder:                 recorder,
+		healthChecker:            healthChecker,
+		healthzServer:            healthzServer,
+		precomputedProbabilities: make([]string, 0, 1001),
+		iptablesData:             bytes.NewBuffer(nil),
+		filterChains:             bytes.NewBuffer(nil),
+		filterRules:              bytes.NewBuffer(nil),
+		natChains:                bytes.NewBuffer(nil),
+		natRules:                 bytes.NewBuffer(nil),
+	}
+	burstSyncs := 2
+	glog.V(3).Infof("minSyncPeriod: %v, syncPeriod: %v, burstSyncs: %d", minSyncPeriod, syncPeriod, burstSyncs)
+	proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
+	return proxier, nil
+}
+
+// CleanupLeftovers removes all iptables rules and chains created by the Proxier
+// It returns true if an error was encountered. Errors are logged.
+func CleanupLeftovers(ipt utiliptables.Interface) (encounteredError bool) {
+	// Unlink the services chain.
+	args := []string{
+		"-m", "comment", "--comment", "kubernetes service portals",
+		"-j", string(kubeServicesChain),
+	}
+	tableChainsWithJumpServices := []struct {
+		table utiliptables.Table
+		chain utiliptables.Chain
+	}{
+		{utiliptables.TableFilter, utiliptables.ChainInput},
+		{utiliptables.TableFilter, utiliptables.ChainOutput},
+		{utiliptables.TableNAT, utiliptables.ChainOutput},
+		{utiliptables.TableNAT, utiliptables.ChainPrerouting},
+	}
+	for _, tc := range tableChainsWithJumpServices {
+		if err := ipt.DeleteRule(tc.table, tc.chain, args...); err != nil {
+			if !utiliptables.IsNotFoundError(err) {
+				glog.Errorf("Error removing pure-iptables proxy rule: %v", err)
+				encounteredError = true
+			}
+		}
+	}
+
+	// Unlink the postrouting chain.
+	args = []string{
+		"-m", "comment", "--comment", "kubernetes postrouting rules",
+		"-j", string(kubePostroutingChain),
+	}
+	if err := ipt.DeleteRule(utiliptables.TableNAT, utiliptables.ChainPostrouting, args...); err != nil {
+		if !utiliptables.IsNotFoundError(err) {
+			glog.Errorf("Error removing pure-iptables proxy rule: %v", err)
+			encounteredError = true
+		}
+	}
+
+	// Flush and remove all of our chains.
+	iptablesData := bytes.NewBuffer(nil)
+	if err := ipt.SaveInto(utiliptables.TableNAT, iptablesData); err != nil {
+		glog.Errorf("Failed to execute iptables-save for %s: %v", utiliptables.TableNAT, err)
+		encounteredError = true
+	} else {
+		existingNATChains := utiliptables.GetChainLines(utiliptables.TableNAT, iptablesData.Bytes())
+		natChains := bytes.NewBuffer(nil)
+		natRules := bytes.NewBuffer(nil)
+		writeLine(natChains, "*nat")
+		// Start with chains we know we need to remove.
+		for _, chain := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} {
+			if _, found := existingNATChains[chain]; found {
+				chainString := string(chain)
+				writeLine(natChains, existingNATChains[chain]) // flush
+				writeLine(natRules, "-X", chainString)         // delete
+			}
+		}
+		// Hunt for service and endpoint chains.
+		for chain := range existingNATChains {
+			chainString := string(chain)
+			if strings.HasPrefix(chainString, "KUBE-SVC-") || strings.HasPrefix(chainString, "KUBE-SEP-") || strings.HasPrefix(chainString, "KUBE-FW-") || strings.HasPrefix(chainString, "KUBE-XLB-") {
+				writeLine(natChains, existingNATChains[chain]) // flush
+				writeLine(natRules, "-X", chainString)         // delete
+			}
+		}
+		writeLine(natRules, "COMMIT")
+		natLines := append(natChains.Bytes(), natRules.Bytes()...)
+		// Write it.
+		err = ipt.Restore(utiliptables.TableNAT, natLines, utiliptables.NoFlushTables, utiliptables.RestoreCounters)
+		if err != nil {
+			glog.Errorf("Failed to execute iptables-restore for %s: %v", utiliptables.TableNAT, err)
+			encounteredError = true
+		}
+	}
+	{
+		filterBuf := bytes.NewBuffer(nil)
+		writeLine(filterBuf, "*filter")
+		writeLine(filterBuf, fmt.Sprintf(":%s - [0:0]", kubeServicesChain))
+		writeLine(filterBuf, fmt.Sprintf("-X %s", kubeServicesChain))
+		writeLine(filterBuf, "COMMIT")
+		// Write it.
+		if err := ipt.Restore(utiliptables.TableFilter, filterBuf.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters); err != nil {
+			glog.Errorf("Failed to execute iptables-restore for %s: %v", utiliptables.TableFilter, err)
+			encounteredError = true
+		}
+	}
+	return encounteredError
+}
+
+func computeProbability(n int) string {
+	return fmt.Sprintf("%0.5f", 1.0/float64(n))
+}
+
+// This assumes proxier.mu is held
+func (proxier *Proxier) precomputeProbabilities(numberOfPrecomputed int) {
+	if len(proxier.precomputedProbabilities) == 0 {
+		proxier.precomputedProbabilities = append(proxier.precomputedProbabilities, "<bad value>")
+	}
+	for i := len(proxier.precomputedProbabilities); i <= numberOfPrecomputed; i++ {
+		proxier.precomputedProbabilities = append(proxier.precomputedProbabilities, computeProbability(i))
+	}
+}
+
+// This assumes proxier.mu is held
+func (proxier *Proxier) probability(n int) string {
+	if n >= len(proxier.precomputedProbabilities) {
+		proxier.precomputeProbabilities(n)
+	}
+	return proxier.precomputedProbabilities[n]
+}
+
+// Sync is called to synchronize the proxier state to iptables as soon as possible.
+func (proxier *Proxier) Sync() {
+	proxier.syncRunner.Run()
+}
+
+// SyncLoop runs periodic work.  This is expected to run as a goroutine or as the main loop of the app.  It does not return.
+func (proxier *Proxier) SyncLoop() {
+	// Update healthz timestamp at beginning in case Sync() never succeeds.
+	if proxier.healthzServer != nil {
+		proxier.healthzServer.UpdateTimestamp()
+	}
+	proxier.syncRunner.Loop(wait.NeverStop)
+}
+
+func (proxier *Proxier) setInitialized(value bool) {
+	var initialized int32
+	if value {
+		initialized = 1
+	}
+	atomic.StoreInt32(&proxier.initialized, initialized)
+}
+
+func (proxier *Proxier) isInitialized() bool {
+	return atomic.LoadInt32(&proxier.initialized) > 0
+}
+
+func (proxier *Proxier) OnServiceAdd(service *api.Service) {
+	namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if proxier.serviceChanges.update(&namespacedName, nil, service) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnServiceUpdate(oldService, service *api.Service) {
+	namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if proxier.serviceChanges.update(&namespacedName, oldService, service) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnServiceDelete(service *api.Service) {
+	namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if proxier.serviceChanges.update(&namespacedName, service, nil) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnServiceSynced() {
+	proxier.mu.Lock()
+	proxier.servicesSynced = true
+	proxier.setInitialized(proxier.servicesSynced && proxier.endpointsSynced)
+	proxier.mu.Unlock()
+
+	// Sync unconditionally - this is called once per lifetime.
+	proxier.syncProxyRules()
+}
+
+func shouldSkipService(svcName types.NamespacedName, service *api.Service) bool {
+	// if ClusterIP is "None" or empty, skip proxying
+	if !helper.IsServiceIPSet(service) {
+		glog.V(3).Infof("Skipping service %s due to clusterIP = %q", svcName, service.Spec.ClusterIP)
+		return true
+	}
+	// Even if ClusterIP is set, ServiceTypeExternalName services don't get proxied
+	if service.Spec.Type == api.ServiceTypeExternalName {
+		glog.V(3).Infof("Skipping service %s due to Type=ExternalName", svcName)
+		return true
+	}
+	return false
+}
+
+// <serviceMap> is updated by this function (based on the given changes).
+// <changes> map is cleared after applying them.
+func updateServiceMap(
+	serviceMap proxyServiceMap,
+	changes *serviceChangeMap) (result updateServiceMapResult) {
+	result.staleServices = sets.NewString()
+
+	func() {
+		changes.lock.Lock()
+		defer changes.lock.Unlock()
+		for _, change := range changes.items {
+			existingPorts := serviceMap.merge(change.current)
+			serviceMap.unmerge(change.previous, existingPorts, result.staleServices)
+		}
+		changes.items = make(map[types.NamespacedName]*serviceChange)
+	}()
+
+	// TODO: If this will appear to be computationally expensive, consider
+	// computing this incrementally similarly to serviceMap.
+	result.hcServices = make(map[types.NamespacedName]uint16)
+	for svcPortName, info := range serviceMap {
+		if info.healthCheckNodePort != 0 {
+			result.hcServices[svcPortName.NamespacedName] = uint16(info.healthCheckNodePort)
+		}
+	}
+
+	return result
+}
+
+func (proxier *Proxier) OnEndpointsAdd(endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+	if proxier.endpointsChanges.update(&namespacedName, nil, endpoints) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+	if proxier.endpointsChanges.update(&namespacedName, oldEndpoints, endpoints) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnEndpointsDelete(endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+	if proxier.endpointsChanges.update(&namespacedName, endpoints, nil) && proxier.isInitialized() {
+		proxier.syncRunner.Run()
+	}
+}
+
+func (proxier *Proxier) OnEndpointsSynced() {
+	proxier.mu.Lock()
+	proxier.endpointsSynced = true
+	proxier.setInitialized(proxier.servicesSynced && proxier.endpointsSynced)
+	proxier.mu.Unlock()
+
+	// Sync unconditionally - this is called once per lifetime.
+	proxier.syncProxyRules()
+}
+
+// <endpointsMap> is updated by this function (based on the given changes).
+// <changes> map is cleared after applying them.
+func updateEndpointsMap(
+	endpointsMap proxyEndpointsMap,
+	changes *endpointsChangeMap,
+	hostname string) (result updateEndpointMapResult) {
+	result.staleEndpoints = make(map[endpointServicePair]bool)
+	result.staleServiceNames = make(map[proxy.ServicePortName]bool)
+
+	func() {
+		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, result.staleEndpoints, result.staleServiceNames)
+		}
+		changes.items = make(map[types.NamespacedName]*endpointsChange)
+	}()
+
+	if !utilfeature.DefaultFeatureGate.Enabled(features.ExternalTrafficLocalOnly) {
+		return
+	}
+
+	// TODO: If this will appear to be computationally expensive, consider
+	// computing this incrementally similarly to endpointsMap.
+	result.hcEndpoints = make(map[types.NamespacedName]int)
+	localIPs := getLocalIPs(endpointsMap)
+	for nsn, ips := range localIPs {
+		result.hcEndpoints[nsn] = len(ips)
+	}
+
+	return result
+}
+
+// <staleEndpoints> and <staleServices> are modified by this function with detected stale connections.
+func detectStaleConnections(oldEndpointsMap, newEndpointsMap proxyEndpointsMap, staleEndpoints map[endpointServicePair]bool, staleServiceNames map[proxy.ServicePortName]bool) {
+	for svcPortName, epList := range oldEndpointsMap {
+		for _, ep := range epList {
+			stale := true
+			for i := range newEndpointsMap[svcPortName] {
+				if *newEndpointsMap[svcPortName][i] == *ep {
+					stale = false
+					break
+				}
+			}
+			if stale {
+				glog.V(4).Infof("Stale endpoint %v -> %v", svcPortName, ep.endpoint)
+				staleEndpoints[endpointServicePair{endpoint: ep.endpoint, servicePortName: svcPortName}] = true
+			}
+		}
+	}
+
+	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[svcPortName] = true
+		}
+	}
+}
+
+func getLocalIPs(endpointsMap proxyEndpointsMap) map[types.NamespacedName]sets.String {
+	localIPs := make(map[types.NamespacedName]sets.String)
+	for svcPortName := range endpointsMap {
+		for _, ep := range endpointsMap[svcPortName] {
+			if ep.isLocal {
+				nsn := svcPortName.NamespacedName
+				if localIPs[nsn] == nil {
+					localIPs[nsn] = sets.NewString()
+				}
+				localIPs[nsn].Insert(ep.IPPart()) // just the IP part
+			}
+		}
+	}
+	return localIPs
+}
+
+// Translates single Endpoints object to proxyEndpointsMap.
+// This function is used for incremental updated of endpointsMap.
+//
+// NOTE: endpoints object should NOT be modified.
+func endpointsToEndpointsMap(endpoints *api.Endpoints, hostname string) proxyEndpointsMap {
+	if endpoints == nil {
+		return nil
+	}
+
+	endpointsMap := make(proxyEndpointsMap)
+	// 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 := proxy.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
+				}
+				epInfo := &endpointsInfo{
+					endpoint: net.JoinHostPort(addr.IP, strconv.Itoa(int(port.Port))),
+					isLocal:  addr.NodeName != nil && *addr.NodeName == hostname,
+				}
+				endpointsMap[svcPortName] = append(endpointsMap[svcPortName], epInfo)
+			}
+			if glog.V(3) {
+				newEPList := []string{}
+				for _, ep := range endpointsMap[svcPortName] {
+					newEPList = append(newEPList, ep.endpoint)
+				}
+				glog.Infof("Setting endpoints for %q to %+v", svcPortName, newEPList)
+			}
+		}
+	}
+	return endpointsMap
+}
+
+// Translates single Service object to proxyServiceMap.
+//
+// NOTE: service object should NOT be modified.
+func serviceToServiceMap(service *api.Service) proxyServiceMap {
+	if service == nil {
+		return nil
+	}
+	svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
+	if shouldSkipService(svcName, service) {
+		return nil
+	}
+
+	serviceMap := make(proxyServiceMap)
+	for i := range service.Spec.Ports {
+		servicePort := &service.Spec.Ports[i]
+		svcPortName := proxy.ServicePortName{NamespacedName: svcName, Port: servicePort.Name}
+		serviceMap[svcPortName] = newServiceInfo(svcPortName, servicePort, service)
+	}
+	return serviceMap
+}
+
+// portProtoHash takes the ServicePortName and protocol for a service
+// returns the associated 16 character hash. This is computed by hashing (sha256)
+// then encoding to base32 and truncating to 16 chars. We do this because IPTables
+// Chain Names must be <= 28 chars long, and the longer they are the harder they are to read.
+func portProtoHash(servicePortName string, protocol string) string {
+	hash := sha256.Sum256([]byte(servicePortName + protocol))
+	encoded := base32.StdEncoding.EncodeToString(hash[:])
+	return encoded[:16]
+}
+
+// servicePortChainName takes the ServicePortName for a service and
+// returns the associated iptables chain.  This is computed by hashing (sha256)
+// then encoding to base32 and truncating with the prefix "KUBE-SVC-".
+func servicePortChainName(servicePortName string, protocol string) utiliptables.Chain {
+	return utiliptables.Chain("KUBE-SVC-" + portProtoHash(servicePortName, protocol))
+}
+
+// serviceFirewallChainName takes the ServicePortName for a service and
+// returns the associated iptables chain.  This is computed by hashing (sha256)
+// then encoding to base32 and truncating with the prefix "KUBE-FW-".
+func serviceFirewallChainName(servicePortName string, protocol string) utiliptables.Chain {
+	return utiliptables.Chain("KUBE-FW-" + portProtoHash(servicePortName, protocol))
+}
+
+// serviceLBPortChainName takes the ServicePortName for a service and
+// returns the associated iptables chain.  This is computed by hashing (sha256)
+// then encoding to base32 and truncating with the prefix "KUBE-XLB-".  We do
+// this because IPTables Chain Names must be <= 28 chars long, and the longer
+// they are the harder they are to read.
+func serviceLBChainName(servicePortName string, protocol string) utiliptables.Chain {
+	return utiliptables.Chain("KUBE-XLB-" + portProtoHash(servicePortName, protocol))
+}
+
+// This is the same as servicePortChainName but with the endpoint included.
+func servicePortEndpointChainName(servicePortName string, protocol string, endpoint string) utiliptables.Chain {
+	hash := sha256.Sum256([]byte(servicePortName + protocol + endpoint))
+	encoded := base32.StdEncoding.EncodeToString(hash[:])
+	return utiliptables.Chain("KUBE-SEP-" + encoded[:16])
+}
+
+type endpointServicePair struct {
+	endpoint        string
+	servicePortName proxy.ServicePortName
+}
+
+func (esp *endpointServicePair) IPPart() string {
+	if index := strings.Index(esp.endpoint, ":"); index != -1 {
+		return esp.endpoint[0:index]
+	}
+	return esp.endpoint
+}
+
+const noConnectionToDelete = "0 flow entries have been deleted"
+
+// After a UDP endpoint has been removed, we must flush any pending conntrack entries to it, or else we
+// risk sending more traffic to it, all of which will be lost (because UDP).
+// This assumes the proxier mutex is held
+func (proxier *Proxier) deleteEndpointConnections(connectionMap map[endpointServicePair]bool) {
+	for epSvcPair := range connectionMap {
+		if svcInfo, ok := proxier.serviceMap[epSvcPair.servicePortName]; ok && svcInfo.protocol == api.ProtocolUDP {
+			endpointIP := epSvcPair.endpoint[0:strings.Index(epSvcPair.endpoint, ":")]
+			glog.V(2).Infof("Deleting connection tracking state for service IP %s, endpoint IP %s", svcInfo.clusterIP.String(), endpointIP)
+			err := utilproxy.ExecConntrackTool(proxier.exec, "-D", "--orig-dst", svcInfo.clusterIP.String(), "--dst-nat", endpointIP, "-p", "udp")
+			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.
+				glog.Errorf("conntrack return with error: %v", err)
+			}
+		}
+	}
+}
+
+// This is where all of the iptables-save/restore calls happen.
+// The only other iptables rules are those that are setup in iptablesInit()
+// This assumes proxier.mu is NOT held
+func (proxier *Proxier) syncProxyRules() {
+	proxier.mu.Lock()
+	defer proxier.mu.Unlock()
+
+	start := time.Now()
+	defer func() {
+		SyncProxyRulesLatency.Observe(sinceInMicroseconds(start))
+		glog.V(4).Infof("syncProxyRules took %v", time.Since(start))
+	}()
+	// don't sync rules till we've received services and endpoints
+	if !proxier.endpointsSynced || !proxier.servicesSynced {
+		glog.V(2).Info("Not syncing iptables until Services and Endpoints have been received from master")
+		return
+	}
+
+	// We assume that if this was called, we really want to sync them,
+	// even if nothing changed in the meantime. In other words, callers are
+	// responsible for detecting no-op changes and not calling this function.
+	serviceUpdateResult := updateServiceMap(
+		proxier.serviceMap, &proxier.serviceChanges)
+	endpointUpdateResult := updateEndpointsMap(
+		proxier.endpointsMap, &proxier.endpointsChanges, proxier.hostname)
+
+	staleServices := serviceUpdateResult.staleServices
+	// merge stale services gathered from updateEndpointsMap
+	for svcPortName := range endpointUpdateResult.staleServiceNames {
+		if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && svcInfo.protocol == api.ProtocolUDP {
+			glog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.clusterIP.String())
+			staleServices.Insert(svcInfo.clusterIP.String())
+		}
+	}
+
+	glog.V(3).Infof("Syncing iptables rules")
+
+	// Create and link the kube services chain.
+	{
+		tablesNeedServicesChain := []utiliptables.Table{utiliptables.TableFilter, utiliptables.TableNAT}
+		for _, table := range tablesNeedServicesChain {
+			if _, err := proxier.iptables.EnsureChain(table, kubeServicesChain); err != nil {
+				glog.Errorf("Failed to ensure that %s chain %s exists: %v", table, kubeServicesChain, err)
+				return
+			}
+		}
+
+		tableChainsNeedJumpServices := []struct {
+			table utiliptables.Table
+			chain utiliptables.Chain
+		}{
+			{utiliptables.TableFilter, utiliptables.ChainInput},
+			{utiliptables.TableFilter, utiliptables.ChainOutput},
+			{utiliptables.TableNAT, utiliptables.ChainOutput},
+			{utiliptables.TableNAT, utiliptables.ChainPrerouting},
+		}
+		comment := "kubernetes service portals"
+		args := []string{"-m", "comment", "--comment", comment, "-j", string(kubeServicesChain)}
+		for _, tc := range tableChainsNeedJumpServices {
+			if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, tc.table, tc.chain, args...); err != nil {
+				glog.Errorf("Failed to ensure that %s chain %s jumps to %s: %v", tc.table, tc.chain, kubeServicesChain, err)
+				return
+			}
+		}
+	}
+
+	// Create and link the kube postrouting chain.
+	{
+		if _, err := proxier.iptables.EnsureChain(utiliptables.TableNAT, kubePostroutingChain); err != nil {
+			glog.Errorf("Failed to ensure that %s chain %s exists: %v", utiliptables.TableNAT, kubePostroutingChain, err)
+			return
+		}
+
+		comment := "kubernetes postrouting rules"
+		args := []string{"-m", "comment", "--comment", comment, "-j", string(kubePostroutingChain)}
+		if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, utiliptables.TableNAT, utiliptables.ChainPostrouting, args...); err != nil {
+			glog.Errorf("Failed to ensure that %s chain %s jumps to %s: %v", utiliptables.TableNAT, utiliptables.ChainPostrouting, kubePostroutingChain, err)
+			return
+		}
+	}
+
+	//
+	// Below this point we will not return until we try to write the iptables rules.
+	//
+
+	// Get iptables-save output so we can check for existing chains and rules.
+	// This will be a map of chain name to chain with rules as stored in iptables-save/iptables-restore
+	existingFilterChains := make(map[utiliptables.Chain]string)
+	proxier.iptablesData.Reset()
+	err := proxier.iptables.SaveInto(utiliptables.TableFilter, proxier.iptablesData)
+	if err != nil { // if we failed to get any rules
+		glog.Errorf("Failed to execute iptables-save, syncing all rules: %v", err)
+	} else { // otherwise parse the output
+		existingFilterChains = utiliptables.GetChainLines(utiliptables.TableFilter, proxier.iptablesData.Bytes())
+	}
+
+	existingNATChains := make(map[utiliptables.Chain]string)
+	proxier.iptablesData.Reset()
+	err = proxier.iptables.SaveInto(utiliptables.TableNAT, proxier.iptablesData)
+	if err != nil { // if we failed to get any rules
+		glog.Errorf("Failed to execute iptables-save, syncing all rules: %v", err)
+	} else { // otherwise parse the output
+		existingNATChains = utiliptables.GetChainLines(utiliptables.TableNAT, proxier.iptablesData.Bytes())
+	}
+
+	// Reset all buffers used later.
+	// This is to avoid memory reallocations and thus improve performance.
+	proxier.filterChains.Reset()
+	proxier.filterRules.Reset()
+	proxier.natChains.Reset()
+	proxier.natRules.Reset()
+
+	// Write table headers.
+	writeLine(proxier.filterChains, "*filter")
+	writeLine(proxier.natChains, "*nat")
+
+	// Make sure we keep stats for the top-level chains, if they existed
+	// (which most should have because we created them above).
+	if chain, ok := existingFilterChains[kubeServicesChain]; ok {
+		writeLine(proxier.filterChains, chain)
+	} else {
+		writeLine(proxier.filterChains, utiliptables.MakeChainLine(kubeServicesChain))
+	}
+	if chain, ok := existingNATChains[kubeServicesChain]; ok {
+		writeLine(proxier.natChains, chain)
+	} else {
+		writeLine(proxier.natChains, utiliptables.MakeChainLine(kubeServicesChain))
+	}
+	if chain, ok := existingNATChains[kubeNodePortsChain]; ok {
+		writeLine(proxier.natChains, chain)
+	} else {
+		writeLine(proxier.natChains, utiliptables.MakeChainLine(kubeNodePortsChain))
+	}
+	if chain, ok := existingNATChains[kubePostroutingChain]; ok {
+		writeLine(proxier.natChains, chain)
+	} else {
+		writeLine(proxier.natChains, utiliptables.MakeChainLine(kubePostroutingChain))
+	}
+	if chain, ok := existingNATChains[KubeMarkMasqChain]; ok {
+		writeLine(proxier.natChains, chain)
+	} else {
+		writeLine(proxier.natChains, utiliptables.MakeChainLine(KubeMarkMasqChain))
+	}
+
+	// Install the kubernetes-specific postrouting rules. We use a whole chain for
+	// this so that it is easier to flush and change, for example if the mark
+	// value should ever change.
+	writeLine(proxier.natRules, []string{
+		"-A", string(kubePostroutingChain),
+		"-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`,
+		"-m", "mark", "--mark", proxier.masqueradeMark,
+		"-j", "MASQUERADE",
+	}...)
+
+	// Install the kubernetes-specific masquerade mark rule. We use a whole chain for
+	// this so that it is easier to flush and change, for example if the mark
+	// value should ever change.
+	writeLine(proxier.natRules, []string{
+		"-A", string(KubeMarkMasqChain),
+		"-j", "MARK", "--set-xmark", proxier.masqueradeMark,
+	}...)
+
+	// Accumulate NAT chains to keep.
+	activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set
+
+	// Accumulate the set of local ports that we will be holding open once this update is complete
+	replacementPortsMap := map[localPort]closeable{}
+
+	// We are creating those slices ones here to avoid memory reallocations
+	// in every loop. Note that reuse the memory, instead of doing:
+	//   slice = <some new slice>
+	// you should always do one of the below:
+	//   slice = slice[:0] // and then append to it
+	//   slice = append(slice[:0], ...)
+	endpoints := make([]*endpointsInfo, 0)
+	endpointChains := make([]utiliptables.Chain, 0)
+	// To avoid growing this slice, we arbitrarily set its size to 64,
+	// there is never more than that many arguments for a single line.
+	// Note that even if we go over 64, it will still be correct - it
+	// is just for efficiency, not correctness.
+	args := make([]string, 64)
+
+	// Build rules for each service.
+	var svcNameString string
+	for svcName, svcInfo := range proxier.serviceMap {
+		protocol := strings.ToLower(string(svcInfo.protocol))
+		svcNameString = svcInfo.serviceNameString
+
+		// Create the per-service chain, retaining counters if possible.
+		svcChain := svcInfo.servicePortChainName
+		if chain, ok := existingNATChains[svcChain]; ok {
+			writeLine(proxier.natChains, chain)
+		} else {
+			writeLine(proxier.natChains, utiliptables.MakeChainLine(svcChain))
+		}
+		activeNATChains[svcChain] = true
+
+		svcXlbChain := svcInfo.serviceLBChainName
+		if svcInfo.onlyNodeLocalEndpoints {
+			// Only for services request OnlyLocal traffic
+			// create the per-service LB chain, retaining counters if possible.
+			if lbChain, ok := existingNATChains[svcXlbChain]; ok {
+				writeLine(proxier.natChains, lbChain)
+			} else {
+				writeLine(proxier.natChains, utiliptables.MakeChainLine(svcXlbChain))
+			}
+			activeNATChains[svcXlbChain] = true
+		} else if activeNATChains[svcXlbChain] {
+			// Cleanup the previously created XLB chain for this service
+			delete(activeNATChains, svcXlbChain)
+		}
+
+		// Capture the clusterIP.
+		args = append(args[:0],
+			"-A", string(kubeServicesChain),
+			"-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcNameString),
+			"-m", protocol, "-p", protocol,
+			"-d", fmt.Sprintf("%s/32", svcInfo.clusterIP.String()),
+			"--dport", strconv.Itoa(svcInfo.port),
+		)
+		if proxier.masqueradeAll {
+			writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
+		} else if len(proxier.clusterCIDR) > 0 {
+			// This masquerades off-cluster traffic to a service VIP.  The idea
+			// is that you can establish a static route for your Service range,
+			// routing to any node, and that node will bridge into the Service
+			// for you.  Since that might bounce off-node, we masquerade here.
+			// If/when we support "Local" policy for VIPs, we should update this.
+			writeLine(proxier.natRules, append(args, "! -s", proxier.clusterCIDR, "-j", string(KubeMarkMasqChain))...)
+		}
+		writeLine(proxier.natRules, append(args, "-j", string(svcChain))...)
+
+		// Capture externalIPs.
+		for _, externalIP := range svcInfo.externalIPs {
+			// If the "external" IP happens to be an IP that is local to this
+			// machine, hold the local port open so no other process can open it
+			// (because the socket might open but it would never work).
+			if local, err := isLocalIP(externalIP); err != nil {
+				glog.Errorf("can't determine if IP is local, assuming not: %v", err)
+			} else if local {
+				lp := localPort{
+					desc:     "externalIP for " + svcNameString,
+					ip:       externalIP,
+					port:     svcInfo.port,
+					protocol: protocol,
+				}
+				if proxier.portsMap[lp] != nil {
+					glog.V(4).Infof("Port %s was open before and is still needed", lp.String())
+					replacementPortsMap[lp] = proxier.portsMap[lp]
+				} else {
+					socket, err := proxier.portMapper.OpenLocalPort(&lp)
+					if err != nil {
+						msg := fmt.Sprintf("can't open %s, skipping this externalIP: %v", lp.String(), err)
+
+						proxier.recorder.Eventf(
+							&clientv1.ObjectReference{
+								Kind:      "Node",
+								Name:      proxier.hostname,
+								UID:       types.UID(proxier.hostname),
+								Namespace: "",
+							}, api.EventTypeWarning, err.Error(), msg)
+						glog.Error(msg)
+						continue
+					}
+					replacementPortsMap[lp] = socket
+				}
+			} // We're holding the port, so it's OK to install iptables rules.
+			args = append(args[:0],
+				"-A", string(kubeServicesChain),
+				"-m", "comment", "--comment", fmt.Sprintf(`"%s external IP"`, svcNameString),
+				"-m", protocol, "-p", protocol,
+				"-d", fmt.Sprintf("%s/32", externalIP),
+				"--dport", strconv.Itoa(svcInfo.port),
+			)
+			// We have to SNAT packets to external IPs.
+			writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
+
+			// Allow traffic for external IPs that does not come from a bridge (i.e. not from a container)
+			// nor from a local process to be forwarded to the service.
+			// This rule roughly translates to "all traffic from off-machine".
+			// This is imperfect in the face of network plugins that might not use a bridge, but we can revisit that later.
+			externalTrafficOnlyArgs := append(args,
+				"-m", "physdev", "!", "--physdev-is-in",
+				"-m", "addrtype", "!", "--src-type", "LOCAL")
+			writeLine(proxier.natRules, append(externalTrafficOnlyArgs, "-j", string(svcChain))...)
+			dstLocalOnlyArgs := append(args, "-m", "addrtype", "--dst-type", "LOCAL")
+			// Allow traffic bound for external IPs that happen to be recognized as local IPs to stay local.
+			// This covers cases like GCE load-balancers which get added to the local routing table.
+			writeLine(proxier.natRules, append(dstLocalOnlyArgs, "-j", string(svcChain))...)
+
+			// If the service has no endpoints then reject packets coming via externalIP
+			// Install ICMP Reject rule in filter table for destination=externalIP and dport=svcport
+			if len(proxier.endpointsMap[svcName]) == 0 {
+				writeLine(proxier.filterRules,
+					"-A", string(kubeServicesChain),
+					"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
+					"-m", protocol, "-p", protocol,
+					"-d", fmt.Sprintf("%s/32", externalIP),
+					"--dport", strconv.Itoa(svcInfo.port),
+					"-j", "REJECT",
+				)
+			}
+		}
+
+		// Capture load-balancer ingress.
+		fwChain := svcInfo.serviceFirewallChainName
+		for _, ingress := range svcInfo.loadBalancerStatus.Ingress {
+			if ingress.IP != "" {
+				// create service firewall chain
+				if chain, ok := existingNATChains[fwChain]; ok {
+					writeLine(proxier.natChains, chain)
+				} else {
+					writeLine(proxier.natChains, utiliptables.MakeChainLine(fwChain))
+				}
+				activeNATChains[fwChain] = true
+				// The service firewall rules are created based on ServiceSpec.loadBalancerSourceRanges field.
+				// This currently works for loadbalancers that preserves source ips.
+				// For loadbalancers which direct traffic to service NodePort, the firewall rules will not apply.
+
+				args = append(args[:0],
+					"-A", string(kubeServicesChain),
+					"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
+					"-m", protocol, "-p", protocol,
+					"-d", fmt.Sprintf("%s/32", ingress.IP),
+					"--dport", strconv.Itoa(svcInfo.port),
+				)
+				// jump to service firewall chain
+				writeLine(proxier.natRules, append(args, "-j", string(fwChain))...)
+
+				args = append(args[:0],
+					"-A", string(fwChain),
+					"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
+				)
+
+				// Each source match rule in the FW chain may jump to either the SVC or the XLB chain
+				chosenChain := svcXlbChain
+				// If we are proxying globally, we need to masquerade in case we cross nodes.
+				// If we are proxying only locally, we can retain the source IP.
+				if !svcInfo.onlyNodeLocalEndpoints {
+					writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
+					chosenChain = svcChain
+				}
+
+				if len(svcInfo.loadBalancerSourceRanges) == 0 {
+					// allow all sources, so jump directly to the KUBE-SVC or KUBE-XLB chain
+					writeLine(proxier.natRules, append(args, "-j", string(chosenChain))...)
+				} else {
+					// firewall filter based on each source range
+					allowFromNode := false
+					for _, src := range svcInfo.loadBalancerSourceRanges {
+						writeLine(proxier.natRules, append(args, "-s", src, "-j", string(chosenChain))...)
+						// ignore error because it has been validated
+						_, cidr, _ := net.ParseCIDR(src)
+						if cidr.Contains(proxier.nodeIP) {
+							allowFromNode = true
+						}
+					}
+					// generally, ip route rule was added to intercept request to loadbalancer vip from the
+					// loadbalancer's backend hosts. In this case, request will not hit the loadbalancer but loop back directly.
+					// Need to add the following rule to allow request on host.
+					if allowFromNode {
+						writeLine(proxier.natRules, append(args, "-s", fmt.Sprintf("%s/32", ingress.IP), "-j", string(chosenChain))...)
+					}
+				}
+
+				// If the packet was able to reach the end of firewall chain, then it did not get DNATed.
+				// It means the packet cannot go thru the firewall, then mark it for DROP
+				writeLine(proxier.natRules, append(args, "-j", string(KubeMarkDropChain))...)
+			}
+		}
+
+		// Capture nodeports.  If we had more than 2 rules it might be
+		// worthwhile to make a new per-service chain for nodeport rules, but
+		// with just 2 rules it ends up being a waste and a cognitive burden.
+		if svcInfo.nodePort != 0 {
+			// Hold the local port open so no other process can open it
+			// (because the socket might open but it would never work).
+			lp := localPort{
+				desc:     "nodePort for " + svcNameString,
+				ip:       "",
+				port:     svcInfo.nodePort,
+				protocol: protocol,
+			}
+			if proxier.portsMap[lp] != nil {
+				glog.V(4).Infof("Port %s was open before and is still needed", lp.String())
+				replacementPortsMap[lp] = proxier.portsMap[lp]
+			} else {
+				socket, err := proxier.portMapper.OpenLocalPort(&lp)
+				if err != nil {
+					glog.Errorf("can't open %s, skipping this nodePort: %v", lp.String(), err)
+					continue
+				}
+				if lp.protocol == "udp" {
+					proxier.clearUDPConntrackForPort(lp.port)
+				}
+				replacementPortsMap[lp] = socket
+			} // We're holding the port, so it's OK to install iptables rules.
+
+			args = append(args[:0],
+				"-A", string(kubeNodePortsChain),
+				"-m", "comment", "--comment", svcNameString,
+				"-m", protocol, "-p", protocol,
+				"--dport", strconv.Itoa(svcInfo.nodePort),
+			)
+			if !svcInfo.onlyNodeLocalEndpoints {
+				// Nodeports need SNAT, unless they're local.
+				writeLine(proxier.natRules, append(args, "-j", string(KubeMarkMasqChain))...)
+				// Jump to the service chain.
+				writeLine(proxier.natRules, append(args, "-j", string(svcChain))...)
+			} else {
+				// TODO: Make all nodePorts jump to the firewall chain.
+				// Currently we only create it for loadbalancers (#33586).
+				writeLine(proxier.natRules, append(args, "-j", string(svcXlbChain))...)
+			}
+
+			// If the service has no endpoints then reject packets.  The filter
+			// table doesn't currently have the same per-service structure that
+			// the nat table does, so we just stick this into the kube-services
+			// chain.
+			if len(proxier.endpointsMap[svcName]) == 0 {
+				writeLine(proxier.filterRules,
+					"-A", string(kubeServicesChain),
+					"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
+					"-m", "addrtype", "--dst-type", "LOCAL",
+					"-m", protocol, "-p", protocol,
+					"--dport", strconv.Itoa(svcInfo.nodePort),
+					"-j", "REJECT",
+				)
+			}
+		}
+
+		// If the service has no endpoints then reject packets.
+		if len(proxier.endpointsMap[svcName]) == 0 {
+			writeLine(proxier.filterRules,
+				"-A", string(kubeServicesChain),
+				"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
+				"-m", protocol, "-p", protocol,
+				"-d", fmt.Sprintf("%s/32", svcInfo.clusterIP.String()),
+				"--dport", strconv.Itoa(svcInfo.port),
+				"-j", "REJECT",
+			)
+			continue
+		}
+
+		// From here on, we assume there are active endpoints.
+
+		// Generate the per-endpoint chains.  We do this in multiple passes so we
+		// can group rules together.
+		// These two slices parallel each other - keep in sync
+		endpoints = endpoints[:0]
+		endpointChains = endpointChains[:0]
+		var endpointChain utiliptables.Chain
+		for _, ep := range proxier.endpointsMap[svcName] {
+			endpoints = append(endpoints, ep)
+			endpointChain = ep.endpointChain(svcNameString, protocol)
+			endpointChains = append(endpointChains, endpointChain)
+
+			// Create the endpoint chain, retaining counters if possible.
+			if chain, ok := existingNATChains[utiliptables.Chain(endpointChain)]; ok {
+				writeLine(proxier.natChains, chain)
+			} else {
+				writeLine(proxier.natChains, utiliptables.MakeChainLine(endpointChain))
+			}
+			activeNATChains[endpointChain] = true
+		}
+
+		// First write session affinity rules, if applicable.
+		if svcInfo.sessionAffinityType == api.ServiceAffinityClientIP {
+			for _, endpointChain := range endpointChains {
+				writeLine(proxier.natRules,
+					"-A", string(svcChain),
+					"-m", "comment", "--comment", svcNameString,
+					"-m", "recent", "--name", string(endpointChain),
+					"--rcheck", "--seconds", strconv.Itoa(svcInfo.stickyMaxAgeMinutes*60), "--reap",
+					"-j", string(endpointChain))
+			}
+		}
+
+		// Now write loadbalancing & DNAT rules.
+		n := len(endpointChains)
+		for i, endpointChain := range endpointChains {
+			// Balancing rules in the per-service chain.
+			args = append(args[:0], []string{
+				"-A", string(svcChain),
+				"-m", "comment", "--comment", svcNameString,
+			}...)
+			if i < (n - 1) {
+				// Each rule is a probabilistic match.
+				args = append(args,
+					"-m", "statistic",
+					"--mode", "random",
+					"--probability", proxier.probability(n-i))
+			}
+			// The final (or only if n == 1) rule is a guaranteed match.
+			args = append(args, "-j", string(endpointChain))
+			writeLine(proxier.natRules, args...)
+
+			// Rules in the per-endpoint chain.
+			args = append(args[:0],
+				"-A", string(endpointChain),
+				"-m", "comment", "--comment", svcNameString,
+			)
+			// Handle traffic that loops back to the originator with SNAT.
+			writeLine(proxier.natRules, append(args,
+				"-s", fmt.Sprintf("%s/32", endpoints[i].IPPart()),
+				"-j", string(KubeMarkMasqChain))...)
+			// Update client-affinity lists.
+			if svcInfo.sessionAffinityType == api.ServiceAffinityClientIP {
+				args = append(args, "-m", "recent", "--name", string(endpointChain), "--set")
+			}
+			// DNAT to final destination.
+			args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", endpoints[i].endpoint)
+			writeLine(proxier.natRules, args...)
+		}
+
+		// The logic below this applies only if this service is marked as OnlyLocal
+		if !svcInfo.onlyNodeLocalEndpoints {
+			continue
+		}
+
+		// Now write ingress loadbalancing & DNAT rules only for services that request OnlyLocal traffic.
+		// TODO - This logic may be combinable with the block above that creates the svc balancer chain
+		localEndpoints := make([]*endpointsInfo, 0)
+		localEndpointChains := make([]utiliptables.Chain, 0)
+		for i := range endpointChains {
+			if endpoints[i].isLocal {
+				// These slices parallel each other; must be kept in sync
+				localEndpoints = append(localEndpoints, endpoints[i])
+				localEndpointChains = append(localEndpointChains, endpointChains[i])
+			}
+		}
+		// First rule in the chain redirects all pod -> external VIP traffic to the
+		// Service's ClusterIP instead. This happens whether or not we have local
+		// endpoints; only if clusterCIDR is specified
+		if len(proxier.clusterCIDR) > 0 {
+			args = append(args[:0],
+				"-A", string(svcXlbChain),
+				"-m", "comment", "--comment",
+				`"Redirect pods trying to reach external loadbalancer VIP to clusterIP"`,
+				"-s", proxier.clusterCIDR,
+				"-j", string(svcChain),
+			)
+			writeLine(proxier.natRules, args...)
+		}
+
+		numLocalEndpoints := len(localEndpointChains)
+		if numLocalEndpoints == 0 {
+			// Blackhole all traffic since there are no local endpoints
+			args = append(args[:0],
+				"-A", string(svcXlbChain),
+				"-m", "comment", "--comment",
+				fmt.Sprintf(`"%s has no local endpoints"`, svcNameString),
+				"-j",
+				string(KubeMarkDropChain),
+			)
+			writeLine(proxier.natRules, args...)
+		} else {
+			// Setup probability filter rules only over local endpoints
+			for i, endpointChain := range localEndpointChains {
+				// Balancing rules in the per-service chain.
+				args = append(args[:0],
+					"-A", string(svcXlbChain),
+					"-m", "comment", "--comment",
+					fmt.Sprintf(`"Balancing rule %d for %s"`, i, svcNameString),
+				)
+				if i < (numLocalEndpoints - 1) {
+					// Each rule is a probabilistic match.
+					args = append(args,
+						"-m", "statistic",
+						"--mode", "random",
+						"--probability", proxier.probability(numLocalEndpoints-i))
+				}
+				// The final (or only if n == 1) rule is a guaranteed match.
+				args = append(args, "-j", string(endpointChain))
+				writeLine(proxier.natRules, args...)
+			}
+		}
+	}
+
+	// Delete chains no longer in use.
+	for chain := range existingNATChains {
+		if !activeNATChains[chain] {
+			chainString := string(chain)
+			if !strings.HasPrefix(chainString, "KUBE-SVC-") && !strings.HasPrefix(chainString, "KUBE-SEP-") && !strings.HasPrefix(chainString, "KUBE-FW-") && !strings.HasPrefix(chainString, "KUBE-XLB-") {
+				// Ignore chains that aren't ours.
+				continue
+			}
+			// We must (as per iptables) write a chain-line for it, which has
+			// the nice effect of flushing the chain.  Then we can remove the
+			// chain.
+			writeLine(proxier.natChains, existingNATChains[chain])
+			writeLine(proxier.natRules, "-X", chainString)
+		}
+	}
+
+	// Finally, tail-call to the nodeports chain.  This needs to be after all
+	// other service portal rules.
+	writeLine(proxier.natRules,
+		"-A", string(kubeServicesChain),
+		"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
+		"-m", "addrtype", "--dst-type", "LOCAL",
+		"-j", string(kubeNodePortsChain))
+
+	// Write the end-of-table markers.
+	writeLine(proxier.filterRules, "COMMIT")
+	writeLine(proxier.natRules, "COMMIT")
+
+	// Sync rules.
+	// NOTE: NoFlushTables is used so we don't flush non-kubernetes chains in the table
+	proxier.iptablesData.Reset()
+	proxier.iptablesData.Write(proxier.filterChains.Bytes())
+	proxier.iptablesData.Write(proxier.filterRules.Bytes())
+	proxier.iptablesData.Write(proxier.natChains.Bytes())
+	proxier.iptablesData.Write(proxier.natRules.Bytes())
+
+	glog.V(5).Infof("Restoring iptables rules: %s", proxier.iptablesData.Bytes())
+	err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
+	if err != nil {
+		glog.Errorf("Failed to execute iptables-restore: %v", err)
+		glog.V(2).Infof("Rules:\n%s", proxier.iptablesData.Bytes())
+		// Revert new local ports.
+		revertPorts(replacementPortsMap, proxier.portsMap)
+		return
+	}
+
+	// Close old local ports and save new ones.
+	for k, v := range proxier.portsMap {
+		if replacementPortsMap[k] == nil {
+			v.Close()
+		}
+	}
+	proxier.portsMap = replacementPortsMap
+
+	// Update healthz timestamp.
+	if proxier.healthzServer != nil {
+		proxier.healthzServer.UpdateTimestamp()
+	}
+
+	// Update healthchecks.  The endpoints list might include services that are
+	// not "OnlyLocal", but the services list will not, and the healthChecker
+	// will just drop those endpoints.
+	if err := proxier.healthChecker.SyncServices(serviceUpdateResult.hcServices); err != nil {
+		glog.Errorf("Error syncing healtcheck services: %v", err)
+	}
+	if err := proxier.healthChecker.SyncEndpoints(endpointUpdateResult.hcEndpoints); err != nil {
+		glog.Errorf("Error syncing healthcheck endoints: %v", err)
+	}
+
+	// Finish housekeeping.
+	// TODO: these and clearUDPConntrackForPort() could be made more consistent.
+	utilproxy.DeleteServiceConnections(proxier.exec, staleServices.List())
+	proxier.deleteEndpointConnections(endpointUpdateResult.staleEndpoints)
+}
+
+// Clear UDP conntrack for port or all conntrack entries when port equal zero.
+// 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 issus:
+// https://github.com/docker/docker/issues/8795
+// https://github.com/kubernetes/kubernetes/issues/31983
+func (proxier *Proxier) clearUDPConntrackForPort(port int) {
+	glog.V(2).Infof("Deleting conntrack entries for udp connections")
+	if port > 0 {
+		err := utilproxy.ExecConntrackTool(proxier.exec, "-D", "-p", "udp", "--dport", strconv.Itoa(port))
+		if err != nil && !strings.Contains(err.Error(), noConnectionToDelete) {
+			glog.Errorf("conntrack return with error: %v", err)
+		}
+	} else {
+		glog.Errorf("Wrong port number. The port number must be greater than zero")
+	}
+}
+
+// Join all words with spaces, terminate with newline and write to buf.
+func writeLine(buf *bytes.Buffer, words ...string) {
+	// We avoid strings.Join for performance reasons.
+	for i := range words {
+		buf.WriteString(words[i])
+		if i < len(words)-1 {
+			buf.WriteByte(' ')
+		} else {
+			buf.WriteByte('\n')
+		}
+	}
+}
+
+func isLocalIP(ip string) (bool, error) {
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		return false, err
+	}
+	for i := range addrs {
+		intf, _, err := net.ParseCIDR(addrs[i].String())
+		if err != nil {
+			return false, err
+		}
+		if net.ParseIP(ip).Equal(intf) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func openLocalPort(lp *localPort) (closeable, error) {
+	// For ports on node IPs, open the actual port and hold it, even though we
+	// use iptables to redirect traffic.
+	// This ensures a) that it's safe to use that port and b) that (a) stays
+	// true.  The risk is that some process on the node (e.g. sshd or kubelet)
+	// is using a port and we give that same port out to a Service.  That would
+	// be bad because iptables would silently claim the traffic but the process
+	// would never know.
+	// NOTE: We should not need to have a real listen()ing socket - bind()
+	// should be enough, but I can't figure out a way to e2e test without
+	// it.  Tools like 'ss' and 'netstat' do not show sockets that are
+	// bind()ed but not listen()ed, and at least the default debian netcat
+	// has no way to avoid about 10 seconds of retries.
+	var socket closeable
+	switch lp.protocol {
+	case "tcp":
+		listener, err := net.Listen("tcp", net.JoinHostPort(lp.ip, strconv.Itoa(lp.port)))
+		if err != nil {
+			return nil, err
+		}
+		socket = listener
+	case "udp":
+		addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(lp.ip, strconv.Itoa(lp.port)))
+		if err != nil {
+			return nil, err
+		}
+		conn, err := net.ListenUDP("udp", addr)
+		if err != nil {
+			return nil, err
+		}
+		socket = conn
+	default:
+		return nil, fmt.Errorf("unknown protocol %q", lp.protocol)
+	}
+	glog.V(2).Infof("Opened local port %s", lp.String())
+	return socket, nil
+}
+
+// revertPorts is closing ports in replacementPortsMap but not in originalPortsMap. In other words, it only
+// closes the ports opened in this sync.
+func revertPorts(replacementPortsMap, originalPortsMap map[localPort]closeable) {
+	for k, v := range replacementPortsMap {
+		// Only close newly opened local ports - leave ones that were open before this update
+		if originalPortsMap[k] == nil {
+			glog.V(2).Infof("Closing local port %s after iptables-restore failure", k.String())
+			v.Close()
+		}
+	}
+}
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/types.go b/vendor/k8s.io/kubernetes/pkg/proxy/types.go
new file mode 100644
index 000000000..578baff69
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/types.go
@@ -0,0 +1,44 @@
+/*
+Copyright 2015 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"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// ProxyProvider is the interface provided by proxier implementations.
+type ProxyProvider interface {
+	// Sync immediately synchronizes the ProxyProvider's current state to iptables.
+	Sync()
+	// SyncLoop runs periodic work.
+	// This is expected to run as a goroutine or as the main loop of the app.
+	// It does not return.
+	SyncLoop()
+}
+
+// ServicePortName carries a namespace + name + portname.  This is the unique
+// identfier for a load-balanced service.
+type ServicePortName struct {
+	types.NamespacedName
+	Port string
+}
+
+func (spn ServicePortName) String() string {
+	return fmt.Sprintf("%s:%s", spn.NamespacedName.String(), spn.Port)
+}
diff --git a/vendor/k8s.io/kubernetes/pkg/proxy/util/conntrack.go b/vendor/k8s.io/kubernetes/pkg/proxy/util/conntrack.go
new file mode 100644
index 000000000..436045ecb
--- /dev/null
+++ b/vendor/k8s.io/kubernetes/pkg/proxy/util/conntrack.go
@@ -0,0 +1,58 @@
+/*
+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"
+	"strings"
+
+	"k8s.io/kubernetes/pkg/util/exec"
+
+	"github.com/golang/glog"
+)
+
+// Utilities for dealing with conntrack
+
+const noConnectionToDelete = "0 flow entries have been deleted"
+
+// DeleteServiceConnection uses the conntrack tool to delete the conntrack entries
+// for the UDP connections specified by the given service IPs
+func DeleteServiceConnections(execer exec.Interface, svcIPs []string) {
+	for _, ip := range svcIPs {
+		glog.V(2).Infof("Deleting connection tracking state for service IP %s", ip)
+		err := ExecConntrackTool(execer, "-D", "--orig-dst", ip, "-p", "udp")
+		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.
+			glog.Errorf("conntrack returned error: %v", err)
+		}
+	}
+}
+
+// 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
+}