Merge pull request kubernetes#89654 from thockin/proxy-cleanup

Proxy cleanup

Author: k8s-ci-robot

cmd/kube-proxy/app/server.go

@@ -576,6 +576,57 @@ func createClients(config componentbaseconfig.ClientConnectionConfiguration, mas
 	return client, eventClient.CoreV1(), nil
 }
 
+func serveHealthz(hz healthcheck.ProxierHealthUpdater) {
+	if hz == nil {
+		return
+	}
+
+	fn := func() {
+		err := hz.Run()
+		if err != nil {
+			// For historical reasons we do not abort on errors here.  We may
+			// change that in the future.
+			klog.Errorf("healthz server failed: %v", err)
+		} else {
+			klog.Errorf("healthz server returned without error")
+		}
+	}
+	go wait.Until(fn, 5*time.Second, wait.NeverStop)
+}
+
+func serveMetrics(bindAddress string, proxyMode string, enableProfiling bool) {
+	if len(bindAddress) == 0 {
+		return
+	}
+
+	proxyMux := mux.NewPathRecorderMux("kube-proxy")
+	healthz.InstallHandler(proxyMux)
+	proxyMux.HandleFunc("/proxyMode", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		w.Header().Set("X-Content-Type-Options", "nosniff")
+		fmt.Fprintf(w, "%s", proxyMode)
+	})
+
+	//lint:ignore SA1019 See the Metrics Stability Migration KEP
+	proxyMux.Handle("/metrics", legacyregistry.Handler())
+
+	if enableProfiling {
+		routes.Profiling{}.Install(proxyMux)
+	}
+
+	configz.InstallHandler(proxyMux)
+
+	fn := func() {
+		err := http.ListenAndServe(bindAddress, proxyMux)
+		if err != nil {
+			// For historical reasons we do not abort on errors here.  We may
+			// change that in the future.
+			utilruntime.HandleError(fmt.Errorf("starting metrics server failed: %v", err))
+		}
+	}
+	go wait.Until(fn, 5*time.Second, wait.NeverStop)
+}
+
 // Run runs the specified ProxyServer.  This should never exit (unless CleanupAndExit is set).
 // TODO: At the moment, Run() cannot return a nil error, otherwise it's caller will never exit. Update callers of Run to handle nil errors.
 func (s *ProxyServer) Run() error {
@@ -595,33 +646,13 @@ func (s *ProxyServer) Run() error {
 		s.Broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: s.EventClient.Events("")})
 	}
 
+	// TODO(thockin): make it possible for healthz and metrics to be on the same port.
+
 	// Start up a healthz server if requested
-	if s.HealthzServer != nil {
-		s.HealthzServer.Run()
-	}
+	serveHealthz(s.HealthzServer)
 
 	// Start up a metrics server if requested
-	if len(s.MetricsBindAddress) > 0 {
-		proxyMux := mux.NewPathRecorderMux("kube-proxy")
-		healthz.InstallHandler(proxyMux)
-		proxyMux.HandleFunc("/proxyMode", func(w http.ResponseWriter, r *http.Request) {
-			w.Header().Set("Content-Type", "text/plain; charset=utf-8")
-			w.Header().Set("X-Content-Type-Options", "nosniff")
-			fmt.Fprintf(w, "%s", s.ProxyMode)
-		})
-		//lint:ignore SA1019 See the Metrics Stability Migration KEP
-		proxyMux.Handle("/metrics", legacyregistry.Handler())
-		if s.EnableProfiling {
-			routes.Profiling{}.Install(proxyMux)
-		}
-		configz.InstallHandler(proxyMux)
-		go wait.Until(func() {
-			err := http.ListenAndServe(s.MetricsBindAddress, proxyMux)
-			if err != nil {
-				utilruntime.HandleError(fmt.Errorf("starting metrics server failed: %v", err))
-			}
-		}, 5*time.Second, wait.NeverStop)
-	}
+	serveMetrics(s.MetricsBindAddress, s.ProxyMode, s.EnableProfiling)
 
 	// Tune conntrack, if requested
 	// Conntracker is always nil for windows

pkg/proxy/healthcheck/BUILD

@@ -20,7 +20,6 @@ go_library(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/client-go/tools/record:go_default_library",
         "//vendor/github.com/lithammer/dedent:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",

pkg/proxy/healthcheck/proxier_health.go

@@ -22,17 +22,13 @@ import (
 	"sync/atomic"
 	"time"
 
-	"k8s.io/klog"
-
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/clock"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog"
 	api "k8s.io/kubernetes/pkg/apis/core"
 )
 
-var proxierHealthzRetryInterval = 60 * time.Second
-
 // ProxierHealthUpdater allows callers to update healthz timestamp only.
 type ProxierHealthUpdater interface {
 	// QueuedUpdate should be called when the proxier receives a Service or Endpoints
@@ -43,8 +39,8 @@ type ProxierHealthUpdater interface {
 	// rules to reflect the current state.
 	Updated()
 
-	// Run starts the healthz http server and returns.
-	Run()
+	// Run starts the healthz HTTP server and blocks until it exits.
+	Run() error
 }
 
 var _ ProxierHealthUpdater = &proxierHealthServer{}
@@ -92,31 +88,28 @@ func (hs *proxierHealthServer) QueuedUpdate() {
 	hs.lastQueued.Store(hs.clock.Now())
 }
 
-// Run starts the healthz http server and returns.
-func (hs *proxierHealthServer) Run() {
+// Run starts the healthz HTTP server and blocks until it exits.
+func (hs *proxierHealthServer) Run() error {
 	serveMux := http.NewServeMux()
 	serveMux.Handle("/healthz", healthzHandler{hs: hs})
 	server := hs.httpFactory.New(hs.addr, serveMux)
 
-	go wait.Until(func() {
-		klog.V(3).Infof("Starting goroutine for proxier healthz on %s", hs.addr)
-
-		listener, err := hs.listener.Listen(hs.addr)
-		if err != nil {
-			msg := fmt.Sprintf("Failed to start proxier healthz on %s: %v", hs.addr, err)
-			if hs.recorder != nil {
-				hs.recorder.Eventf(hs.nodeRef, api.EventTypeWarning, "FailedToStartProxierHealthcheck", msg)
-			}
-			klog.Error(msg)
-			return
+	listener, err := hs.listener.Listen(hs.addr)
+	if err != nil {
+		msg := fmt.Sprintf("failed to start proxier healthz on %s: %v", hs.addr, err)
+		// TODO(thockin): move eventing back to caller
+		if hs.recorder != nil {
+			hs.recorder.Eventf(hs.nodeRef, api.EventTypeWarning, "FailedToStartProxierHealthcheck", msg)
 		}
+		return fmt.Errorf("%v", msg)
+	}
 
-		if err := server.Serve(listener); err != nil {
-			klog.Errorf("Proxier healthz closed with error: %v", err)
-			return
-		}
-		klog.Error("Unexpected proxier healthz closed.")
-	}, proxierHealthzRetryInterval, wait.NeverStop)
+	klog.V(3).Infof("starting healthz on %s", hs.addr)
+
+	if err := server.Serve(listener); err != nil {
+		return fmt.Errorf("proxier healthz closed with error: %v", err)
+	}
+	return nil
 }
 
 type healthzHandler struct {