Merge pull request #44053 from thockin/proxy-healthchecks

pkg/proxy/healthcheck/BUILD

@@ -11,20 +11,17 @@ load(
 go_library(
     name = "go_default_library",
     srcs = [
-        "api.go",
         "doc.go",
         "healthcheck.go",
-        "http.go",
-        "listener.go",
-        "worker.go",
     ],
     tags = ["automanaged"],
     deps = [
         "//vendor:github.com/golang/glog",
+        "//vendor:github.com/renstrom/dedent",
         "//vendor:k8s.io/apimachinery/pkg/types",
-        "//vendor:k8s.io/apimachinery/pkg/util/sets",
-        "//vendor:k8s.io/apimachinery/pkg/util/wait",
-        "//vendor:k8s.io/client-go/tools/cache",
+        "//vendor:k8s.io/client-go/pkg/api",
+        "//vendor:k8s.io/client-go/pkg/api/v1",
+        "//vendor:k8s.io/client-go/tools/record",
     ],
 )
 
@@ -34,6 +31,7 @@ go_test(
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
+        "//vendor:github.com/davecgh/go-spew/spew",
         "//vendor:k8s.io/apimachinery/pkg/types",
         "//vendor:k8s.io/apimachinery/pkg/util/sets",
     ],

pkg/proxy/healthcheck/api.go

-/*
-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 (
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/util/wait"
-)
-
-// All public API Methods for this package
-
-// UpdateEndpoints Update the set of local endpoints for a service
-func UpdateEndpoints(serviceName types.NamespacedName, endpointUids sets.String) {
-	req := &proxyMutationRequest{
-		serviceName:  serviceName,
-		endpointUids: &endpointUids,
-	}
-	healthchecker.mutationRequestChannel <- req
-}
-
-func updateServiceListener(serviceName types.NamespacedName, listenPort int, add bool) bool {
-	responseChannel := make(chan bool)
-	req := &proxyListenerRequest{
-		serviceName:     serviceName,
-		listenPort:      uint16(listenPort),
-		add:             add,
-		responseChannel: responseChannel,
-	}
-	healthchecker.listenerRequestChannel <- req
-	return <-responseChannel
-}
-
-// AddServiceListener Request addition of a listener for a service's health check
-func AddServiceListener(serviceName types.NamespacedName, listenPort int) bool {
-	return updateServiceListener(serviceName, listenPort, true)
-}
-
-// DeleteServiceListener Request deletion of a listener for a service's health check
-func DeleteServiceListener(serviceName types.NamespacedName, listenPort int) bool {
-	return updateServiceListener(serviceName, listenPort, false)
-}
-
-// Run Start the healthchecker main loop
-func Run() {
-	healthchecker = proxyHealthCheckFactory()
-	// Wrap with a wait.Forever to handle panics.
-	go wait.Forever(func() {
-		healthchecker.handlerLoop()
-	}, 0)
-}

pkg/proxy/healthcheck/doc.go

@@ -14,5 +14,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package healthcheck LoadBalancer Healthcheck responder library for kubernetes network proxies
+// Package healthcheck provides tools for serving kube-proxy healthchecks.
 package healthcheck // import "k8s.io/kubernetes/pkg/proxy/healthcheck"

pkg/proxy/healthcheck/http.go

-/*
-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/http"
-
-	"github.com/golang/glog"
-)
-
-// A healthCheckHandler serves http requests on /healthz on the service health check node port,
-// and responds to every request with either:
-// 200 OK and the count of endpoints for the given service that are local to this node.
-// or
-// 503 Service Unavailable If the count is zero or the service does not exist
-type healthCheckHandler struct {
-	svcNsName string
-}
-
-// HTTP Utility function to send the required statusCode and error text to a http.ResponseWriter object
-func sendHealthCheckResponse(rw http.ResponseWriter, statusCode int, error string) {
-	rw.Header().Set("Content-Type", "text/plain")
-	rw.WriteHeader(statusCode)
-	fmt.Fprint(rw, error)
-}
-
-// ServeHTTP: Interface callback method for net.Listener Handlers
-func (h healthCheckHandler) ServeHTTP(response http.ResponseWriter, req *http.Request) {
-	glog.V(4).Infof("Received HC Request Service %s from Cloud Load Balancer", h.svcNsName)
-	healthchecker.handleHealthCheckRequest(response, h.svcNsName)
-}

pkg/proxy/healthcheck/listener.go

-/*
-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
-
-// Create/Delete dynamic listeners on the required nodePorts
-
-import (
-	"fmt"
-	"net"
-	"net/http"
-
-	"github.com/golang/glog"
-)
-
-// handleServiceListenerRequest: receive requests to add/remove service health check listening ports
-func (h *proxyHC) handleServiceListenerRequest(req *proxyListenerRequest) bool {
-	sr, serviceFound := h.serviceResponderMap[req.serviceName]
-	if !req.add {
-		if !serviceFound {
-			return false
-		}
-		glog.Infof("Deleting HealthCheckListenPort for service %s port %d",
-			req.serviceName, req.listenPort)
-		delete(h.serviceResponderMap, req.serviceName)
-		(*sr.listener).Close()
-		return true
-	} else if serviceFound {
-		if req.listenPort == sr.listenPort {
-			// Addition requested but responder for service already exists and port is unchanged
-			return true
-		}
-		// Addition requested but responder for service already exists but the listen port has changed
-		glog.Infof("HealthCheckListenPort for service %s changed from %d to %d - closing old listening port",
-			req.serviceName, sr.listenPort, req.listenPort)
-		delete(h.serviceResponderMap, req.serviceName)
-		(*sr.listener).Close()
-	}
-	// Create a service responder object and start listening and serving on the provided port
-	glog.V(2).Infof("Adding health check listener for service %s on nodePort %d", req.serviceName, req.listenPort)
-	server := http.Server{
-		Addr:    fmt.Sprintf(":%d", req.listenPort),
-		Handler: healthCheckHandler{svcNsName: req.serviceName.String()},
-	}
-	listener, err := net.Listen("tcp", server.Addr)
-	if err != nil {
-		glog.Warningf("FAILED to listen on address %s (%s)\n", server.Addr, err)
-		return false
-	}
-	h.serviceResponderMap[req.serviceName] = serviceResponder{serviceName: req.serviceName,
-		listenPort: req.listenPort,
-		listener:   &listener,
-		server:     &server}
-	go func() {
-		// Anonymous goroutine to block on Serve for this listen port - Serve will exit when the listener is closed
-		glog.V(3).Infof("Goroutine blocking on serving health checks for %s on port %d", req.serviceName, req.listenPort)
-		if err := server.Serve(listener); err != nil {
-			glog.V(3).Infof("Proxy HealthCheck listen socket %d for service %s closed with error %s\n", req.listenPort, req.serviceName, err)
-			return
-		}
-		glog.V(3).Infof("Proxy HealthCheck listen socket %d for service %s closed\n", req.listenPort, req.serviceName)
-	}()
-	return true
-}

pkg/proxy/healthcheck/worker.go

-/*
-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 LoadBalancer Healthcheck responder library for kubernetes network proxies
-package healthcheck // import "k8s.io/kubernetes/pkg/proxy/healthcheck"
-
-import (
-	"time"
-
-	"github.com/golang/glog"
-)
-
-var healthchecker *proxyHC
-
-// handlerLoop Serializes all requests to prevent concurrent access to the maps
-func (h *proxyHC) handlerLoop() {
-	ticker := time.NewTicker(1 * time.Minute)
-	defer ticker.Stop()
-	for {
-		select {
-		case req := <-h.mutationRequestChannel:
-			h.handleMutationRequest(req)
-		case req := <-h.listenerRequestChannel:
-			req.responseChannel <- h.handleServiceListenerRequest(req)
-		case <-ticker.C:
-			go h.sync()
-		}
-	}
-}
-
-func (h *proxyHC) sync() {
-	glog.V(4).Infof("%d Health Check Listeners", len(h.serviceResponderMap))
-	glog.V(4).Infof("%d Services registered for health checking", len(h.serviceEndpointsMap.List()))
-	for _, svc := range h.serviceEndpointsMap.ListKeys() {
-		if e, ok := h.serviceEndpointsMap.Get(svc); ok {
-			endpointList := e.(*serviceEndpointsList)
-			glog.V(4).Infof("Service %s has %d local endpoints", svc, endpointList.endpoints.Len())
-		}
-	}
-}

pkg/proxy/iptables/BUILD

@@ -50,7 +50,6 @@ go_test(
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/types",
         "//vendor:k8s.io/apimachinery/pkg/util/intstr",
-        "//vendor:k8s.io/apimachinery/pkg/util/sets",
     ],
 )
 

pkg/proxy/iptables/proxier.go

@@ -219,7 +219,7 @@ type Proxier struct {
 	nodeIP         net.IP
 	portMapper     portOpener
 	recorder       record.EventRecorder
-	healthChecker  healthChecker
+	healthChecker  healthcheck.Server
 }
 
 type localPort struct {
@@ -251,17 +251,6 @@ func (l *listenPortOpener) OpenLocalPort(lp *localPort) (closeable, error) {
 	return openLocalPort(lp)
 }
 
-type healthChecker interface {
-	UpdateEndpoints(serviceName types.NamespacedName, endpointUIDs sets.String)
-}
-
-// TODO: the healthcheck pkg should offer a type
-type globalHealthChecker struct{}
-
-func (globalHealthChecker) UpdateEndpoints(serviceName types.NamespacedName, endpointUIDs sets.String) {
-	healthcheck.UpdateEndpoints(serviceName, endpointUIDs)
-}
-
 // Proxier implements ProxyProvider
 var _ proxy.ProxyProvider = &Proxier{}
 
@@ -315,8 +304,7 @@ func NewProxier(ipt utiliptables.Interface,
 		glog.Warningf("clusterCIDR not specified, unable to distinguish between internal and external traffic")
 	}
 
-	healthChecker := globalHealthChecker{}
-	go healthcheck.Run()
+	healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) // use default implementations of deps
 
 	var throttle flowcontrol.RateLimiter
 	// Defaulting back to not limit sync rate when minSyncPeriod is 0.
@@ -451,18 +439,12 @@ func (proxier *Proxier) SyncLoop() {
 	}
 }
 
-type healthCheckPort struct {
-	namespace types.NamespacedName
-	nodeport  int
-}
-
 // Accepts a list of Services and the existing service map.  Returns the new
-// service map, a list of healthcheck ports to add to or remove from the health
-// checking listener service, and a set of stale UDP services.
-func buildNewServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMap) (proxyServiceMap, []healthCheckPort, []healthCheckPort, sets.String) {
+// service map, a map of healthcheck ports, and a set of stale UDP
+// services.
+func buildNewServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMap) (proxyServiceMap, map[types.NamespacedName]uint16, sets.String) {
 	newServiceMap := make(proxyServiceMap)
-	healthCheckAdd := make([]healthCheckPort, 0)
-	healthCheckDel := make([]healthCheckPort, 0)
+	hcPorts := make(map[types.NamespacedName]uint16)
 
 	for _, service := range allServices {
 		svcName := types.NamespacedName{
@@ -498,12 +480,8 @@ func buildNewServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMa
 				glog.V(1).Infof("Updating existing service %q at %s:%d/%s", serviceName, info.clusterIP, servicePort.Port, servicePort.Protocol)
 			}
 
-			if !exists || !equal {
-				if info.onlyNodeLocalEndpoints && info.healthCheckNodePort > 0 {
-					healthCheckAdd = append(healthCheckAdd, healthCheckPort{serviceName.NamespacedName, info.healthCheckNodePort})
-				} else {
-					healthCheckDel = append(healthCheckDel, healthCheckPort{serviceName.NamespacedName, 0})
-				}
+			if info.onlyNodeLocalEndpoints {
+				hcPorts[svcName] = uint16(info.healthCheckNodePort)
 			}
 
 			newServiceMap[serviceName] = info
@@ -511,6 +489,13 @@ func buildNewServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMa
 		}
 	}
 
+	for nsn, port := range hcPorts {
+		if port == 0 {
+			glog.Errorf("Service %q has no healthcheck nodeport", nsn)
+			delete(hcPorts, nsn)
+		}
+	}
+
 	staleUDPServices := sets.NewString()
 	// Remove serviceports missing from the update.
 	for name, info := range oldServiceMap {
@@ -519,13 +504,10 @@ func buildNewServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMa
 			if info.protocol == api.ProtocolUDP {
 				staleUDPServices.Insert(info.clusterIP.String())
 			}
-			if info.onlyNodeLocalEndpoints && info.healthCheckNodePort > 0 {
-				healthCheckDel = append(healthCheckDel, healthCheckPort{name.NamespacedName, info.healthCheckNodePort})
-			}
 		}
 	}
 
-	return newServiceMap, healthCheckAdd, healthCheckDel, staleUDPServices
+	return newServiceMap, hcPorts, staleUDPServices
 }
 
 // OnServiceUpdate tracks the active set of service proxies.
@@ -538,19 +520,11 @@ func (proxier *Proxier) OnServiceUpdate(allServices []*api.Service) {
 	}
 	proxier.allServices = allServices
 
-	newServiceMap, hcAdd, hcDel, staleUDPServices := buildNewServiceMap(allServices, proxier.serviceMap)
-	for _, hc := range hcAdd {
-		glog.V(4).Infof("Adding health check for %+v, port %v", hc.namespace, hc.nodeport)
-		// Turn on healthcheck responder to listen on the health check nodePort
-		// FIXME: handle failures from adding the service
-		healthcheck.AddServiceListener(hc.namespace, hc.nodeport)
-	}
-	for _, hc := range hcDel {
-		// Remove ServiceListener health check nodePorts from the health checker
-		// TODO - Stats
-		glog.V(4).Infof("Deleting health check for %+v, port %v", hc.namespace, hc.nodeport)
-		// FIXME: handle failures from deleting the service
-		healthcheck.DeleteServiceListener(hc.namespace, hc.nodeport)
+	newServiceMap, hcPorts, staleUDPServices := buildNewServiceMap(allServices, proxier.serviceMap)
+
+	// update healthcheck ports
+	if err := proxier.healthChecker.SyncServices(hcPorts); err != nil {
+		glog.Errorf("Error syncing healtcheck ports: %v", err)
 	}
 
 	if len(newServiceMap) != len(proxier.serviceMap) || !reflect.DeepEqual(newServiceMap, proxier.serviceMap) {
@@ -573,7 +547,13 @@ func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
 	proxier.allEndpoints = allEndpoints
 
 	// TODO: once service has made this same transform, move this into proxier.syncProxyRules()
-	newMap, staleConnections := buildNewEndpointsMap(proxier.allEndpoints, proxier.endpointsMap, proxier.hostname, proxier.healthChecker)
+	newMap, hcEndpoints, staleConnections := buildNewEndpointsMap(proxier.allEndpoints, proxier.endpointsMap, proxier.hostname)
+
+	// update healthcheck endpoints
+	if err := proxier.healthChecker.SyncEndpoints(hcEndpoints); err != nil {
+		glog.Errorf("Error syncing healthcheck endoints: %v", err)
+	}
+
 	if len(newMap) != len(proxier.endpointsMap) || !reflect.DeepEqual(newMap, proxier.endpointsMap) {
 		proxier.endpointsMap = newMap
 		proxier.syncProxyRules(syncReasonEndpoints)
@@ -585,11 +565,11 @@ func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
 }
 
 // Convert a slice of api.Endpoints objects into a map of service-port -> endpoints.
-func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap, hostname string,
-	healthChecker healthChecker) (newMap proxyEndpointMap, staleSet map[endpointServicePair]bool) {
+func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap, hostname string) (newMap proxyEndpointMap, hcEndpoints map[types.NamespacedName]int, staleSet map[endpointServicePair]bool) {
 
 	// return values
 	newMap = make(proxyEndpointMap)
+	hcEndpoints = make(map[types.NamespacedName]int)
 	staleSet = make(map[endpointServicePair]bool)
 
 	// Update endpoints for services.
@@ -615,19 +595,30 @@ func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap
 		}
 	}
 
-	// Update service health check
-	allSvcPorts := make(map[proxy.ServicePortName]bool)
-	for svcPort := range curMap {
-		allSvcPorts[svcPort] = true
+	if !utilfeature.DefaultFeatureGate.Enabled(features.ExternalTrafficLocalOnly) {
+		return
 	}
+
+	// accumulate local IPs per service, ignoring ports
+	localIPs := map[types.NamespacedName]sets.String{}
 	for svcPort := range newMap {
-		allSvcPorts[svcPort] = true
+		for _, ep := range newMap[svcPort] {
+			if ep.isLocal {
+				nsn := svcPort.NamespacedName
+				if localIPs[nsn] == nil {
+					localIPs[nsn] = sets.NewString()
+				}
+				ip := strings.Split(ep.endpoint, ":")[0] // just the IP part
+				localIPs[nsn].Insert(ip)
+			}
+		}
 	}
-	for svcPort := range allSvcPorts {
-		updateHealthCheckEntries(svcPort.NamespacedName, newMap[svcPort], healthChecker)
+	// produce a count per service
+	for nsn, ips := range localIPs {
+		hcEndpoints[nsn] = len(ips)
 	}
 
-	return newMap, staleSet
+	return newMap, hcEndpoints, staleSet
 }
 
 // Gather information about all the endpoint state for a given api.Endpoints.
@@ -675,23 +666,6 @@ func accumulateEndpointsMap(endpoints *api.Endpoints, hostname string,
 	}
 }
 
-// updateHealthCheckEntries - send the new set of local endpoints to the health checker
-func updateHealthCheckEntries(name types.NamespacedName, endpoints []*endpointsInfo, healthChecker healthChecker) {
-	if !utilfeature.DefaultFeatureGate.Enabled(features.ExternalTrafficLocalOnly) {
-		return
-	}
-
-	// Use a set instead of a slice to provide deduplication
-	epSet := sets.NewString()
-	for _, portInfo := range endpoints {
-		if portInfo.isLocal {
-			// kube-proxy health check only needs local endpoints
-			epSet.Insert(fmt.Sprintf("%s/%s", name.Namespace, name.Name))
-		}
-	}
-	healthChecker.UpdateEndpoints(name, epSet)
-}
-
 // 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