add healthz to genericapiserver

Author: deads2k

Diff for: cmd/kube-apiserver/app/server.go

@@ -305,6 +305,7 @@ func Run(s *options.ServerRunOptions) error {
 	genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
 	genericConfig.OpenAPIConfig.Definitions = generatedopenapi.OpenAPIDefinitions
 	genericConfig.EnableOpenAPISupport = true
+	genericConfig.EnableMetrics = true
 	genericConfig.OpenAPIConfig.SecurityDefinitions = securityDefinitions
 
 	config := &master.Config{

Diff for: pkg/genericapiserver/config.go

@@ -449,6 +449,8 @@ func (c completedConfig) New() (*GenericAPIServer, error) {
 
 		enableOpenAPISupport: c.EnableOpenAPISupport,
 		openAPIConfig:        c.OpenAPIConfig,
+
+		postStartHooks: map[string]postStartHookEntry{},
 	}
 
 	s.HandlerContainer = mux.NewAPIContainer(http.NewServeMux(), c.Serializer)

Diff for: pkg/genericapiserver/genericapiserver.go

@@ -42,6 +42,7 @@ import (
 	genericmux "k8s.io/kubernetes/pkg/genericapiserver/mux"
 	"k8s.io/kubernetes/pkg/genericapiserver/openapi/common"
 	"k8s.io/kubernetes/pkg/genericapiserver/routes"
+	"k8s.io/kubernetes/pkg/healthz"
 	"k8s.io/kubernetes/pkg/runtime"
 	utilnet "k8s.io/kubernetes/pkg/util/net"
 	"k8s.io/kubernetes/pkg/util/sets"
@@ -143,10 +144,15 @@ type GenericAPIServer struct {
 	// PostStartHooks are each called after the server has started listening, in a separate go func for each
 	// with no guaranteee of ordering between them.  The map key is a name used for error reporting.
 	// It may kill the process with a panic if it wishes to by returning an error
-	postStartHooks       map[string]PostStartHookFunc
 	postStartHookLock    sync.Mutex
+	postStartHooks       map[string]postStartHookEntry
 	postStartHooksCalled bool
 
+	// healthz checks
+	healthzLock    sync.Mutex
+	healthzChecks  []healthz.HealthzChecker
+	healthzCreated bool
+
 	// See Config.$name for documentation of these flags:
 
 	MasterCount               int
@@ -188,6 +194,9 @@ func (s *GenericAPIServer) PrepareRun() preparedGenericAPIServer {
 			Config: s.openAPIConfig,
 		}.Install(s.HandlerContainer)
 	}
+
+	s.installHealthz()
+
 	return preparedGenericAPIServer{s}
 }
 

Diff for: pkg/genericapiserver/healthz.go

@@ -0,0 +1,45 @@
+/*
+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 genericapiserver
+
+import (
+	"fmt"
+
+	"k8s.io/kubernetes/pkg/healthz"
+)
+
+// AddHealthzCheck allows you to add a HealthzCheck.
+func (s *GenericAPIServer) AddHealthzChecks(checks ...healthz.HealthzChecker) error {
+	s.healthzLock.Lock()
+	defer s.healthzLock.Unlock()
+
+	if s.healthzCreated {
+		return fmt.Errorf("unable to add because the healthz endpoint has already been created")
+	}
+
+	s.healthzChecks = append(s.healthzChecks, checks...)
+	return nil
+}
+
+// installHealthz creates the healthz endpoint for this server
+func (s *GenericAPIServer) installHealthz() {
+	s.healthzLock.Lock()
+	defer s.healthzLock.Unlock()
+	s.healthzCreated = true
+
+	healthz.InstallHandler(&s.HandlerContainer.NonSwaggerRoutes, s.healthzChecks...)
+}

Diff for: pkg/genericapiserver/hooks.go

@@ -17,11 +17,14 @@ limitations under the License.
 package genericapiserver
 
 import (
+	"errors"
 	"fmt"
+	"net/http"
 
 	"github.com/golang/glog"
 
 	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/healthz"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 )
 
@@ -47,6 +50,13 @@ type PostStartHookProvider interface {
 	PostStartHook() (string, PostStartHookFunc, error)
 }
 
+type postStartHookEntry struct {
+	hook PostStartHookFunc
+
+	// done will be closed when the postHook is finished
+	done chan struct{}
+}
+
 // AddPostStartHook allows you to add a PostStartHook.
 func (s *GenericAPIServer) AddPostStartHook(name string, hook PostStartHookFunc) error {
 	if len(name) == 0 {
@@ -62,14 +72,15 @@ func (s *GenericAPIServer) AddPostStartHook(name string, hook PostStartHookFunc)
 	if s.postStartHooksCalled {
 		return fmt.Errorf("unable to add %q because PostStartHooks have already been called", name)
 	}
-	if s.postStartHooks == nil {
-		s.postStartHooks = map[string]PostStartHookFunc{}
-	}
 	if _, exists := s.postStartHooks[name]; exists {
 		return fmt.Errorf("unable to add %q because it is already registered", name)
 	}
 
-	s.postStartHooks[name] = hook
+	// done is closed when the poststarthook is finished.  This is used by the health check to be able to indicate
+	// that the poststarthook is finished
+	done := make(chan struct{})
+	s.AddHealthzChecks(postStartHookHealthz{name: "poststarthook/" + name, done: done})
+	s.postStartHooks[name] = postStartHookEntry{hook: hook, done: done}
 
 	return nil
 }
@@ -82,20 +93,48 @@ func (s *GenericAPIServer) RunPostStartHooks() {
 
 	context := PostStartHookContext{LoopbackClientConfig: s.LoopbackClientConfig}
 
-	for hookName, hook := range s.postStartHooks {
-		go runPostStartHook(hookName, hook, context)
+	for hookName, hookEntry := range s.postStartHooks {
+		go runPostStartHook(hookName, hookEntry, context)
 	}
 }
 
-func runPostStartHook(name string, hook PostStartHookFunc, context PostStartHookContext) {
+func runPostStartHook(name string, entry postStartHookEntry, context PostStartHookContext) {
 	var err error
 	func() {
 		// don't let the hook *accidentally* panic and kill the server
 		defer utilruntime.HandleCrash()
-		err = hook(context)
+		err = entry.hook(context)
 	}()
 	// if the hook intentionally wants to kill server, let it.
 	if err != nil {
 		glog.Fatalf("PostStartHook %q failed: %v", name, err)
 	}
+
+	close(entry.done)
+}
+
+// postStartHookHealthz implements a healthz check for poststarthooks.  It will return a "hookNotFinished"
+// error until the poststarthook is finished.
+type postStartHookHealthz struct {
+	name string
+
+	// done will be closed when the postStartHook is finished
+	done chan struct{}
+}
+
+var _ healthz.HealthzChecker = postStartHookHealthz{}
+
+func (h postStartHookHealthz) Name() string {
+	return h.name
+}
+
+var hookNotFinished = errors.New("not finished")
+
+func (h postStartHookHealthz) Check(req *http.Request) error {
+	select {
+	case <-h.done:
+		return nil
+	default:
+		return hookNotFinished
+	}
 }

Diff for: pkg/master/master.go

@@ -192,6 +192,9 @@ func (c completedConfig) New() (*Master, error) {
 	}
 	m.InstallAPIs(c.Config.GenericConfig.APIResourceConfigSource, restOptionsFactory.NewFor, restStorageProviders...)
 
+	if c.Tunneler != nil {
+		m.installTunneler(c.Tunneler, coreclient.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig).Nodes())
+	}
 	m.InstallGeneralEndpoints(c.Config)
 
 	return m, nil
@@ -215,29 +218,23 @@ func (m *Master) InstallLegacyAPI(c *Config, restOptionsGetter genericapiserver.
 	}
 }
 
+func (m *Master) installTunneler(tunneler genericapiserver.Tunneler, nodeClient coreclient.NodeInterface) {
+	tunneler.Run(nodeAddressProvider{nodeClient}.externalAddresses)
+	m.GenericAPIServer.AddHealthzChecks(healthz.NamedCheck("SSH Tunnel Check", genericapiserver.TunnelSyncHealthChecker(tunneler)))
+	prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Name: "apiserver_proxy_tunnel_sync_latency_secs",
+		Help: "The time since the last successful synchronization of the SSH tunnels for proxy requests.",
+	}, func() float64 { return float64(tunneler.SecondsSinceSync()) })
+}
+
 // TODO this needs to be refactored so we have a way to add general health checks to genericapiserver
 // TODO profiling should be generic
 func (m *Master) InstallGeneralEndpoints(c *Config) {
-	// Run the tunneler.
-	healthzChecks := []healthz.HealthzChecker{}
-	if c.Tunneler != nil {
-		nodeClient := coreclient.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig).Nodes()
-		c.Tunneler.Run(nodeAddressProvider{nodeClient}.externalAddresses)
-
-		healthzChecks = append(healthzChecks, healthz.NamedCheck("SSH Tunnel Check", genericapiserver.TunnelSyncHealthChecker(c.Tunneler)))
-		prometheus.NewGaugeFunc(prometheus.GaugeOpts{
-			Name: "apiserver_proxy_tunnel_sync_latency_secs",
-			Help: "The time since the last successful synchronization of the SSH tunnels for proxy requests.",
-		}, func() float64 { return float64(c.Tunneler.SecondsSinceSync()) })
-	}
-	healthz.InstallHandler(&m.GenericAPIServer.HandlerContainer.NonSwaggerRoutes, healthzChecks...)
-
 	if c.GenericConfig.EnableProfiling {
 		routes.MetricsWithReset{}.Install(m.GenericAPIServer.HandlerContainer)
 	} else {
 		routes.DefaultMetrics{}.Install(m.GenericAPIServer.HandlerContainer)
 	}
-
 }
 
 // InstallAPIs will install the APIs for the restStorageProviders if they are enabled.

Diff for: pkg/master/master_test.go

@@ -90,6 +90,7 @@ func setUp(t *testing.T) (*Master, *etcdtesting.EtcdTestServer, Config, *assert.
 	config.GenericConfig.APIResourceConfigSource = DefaultAPIResourceConfigSource()
 	config.GenericConfig.RequestContextMapper = api.NewRequestContextMapper()
 	config.GenericConfig.LoopbackClientConfig = &restclient.Config{APIPath: "/api", ContentConfig: restclient.ContentConfig{NegotiatedSerializer: api.Codecs}}
+	config.GenericConfig.EnableMetrics = true
 	config.EnableCoreControllers = false
 	config.KubeletClientConfig = kubeletclient.KubeletClientConfig{Port: 10250}
 	config.ProxyTransport = utilnet.SetTransportDefaults(&http.Transport{

Diff for: test/integration/auth/rbac_test.go

@@ -475,4 +475,10 @@ func TestBootstrapping(t *testing.T) {
 	}
 
 	t.Errorf("missing cluster-admin: %v", clusterRoles)
+
+	healthBytes, err := clientset.Discovery().RESTClient().Get().AbsPath("/healthz/poststarthooks/rbac/bootstrap-roles").DoRaw()
+	if err != nil {
+		t.Error(err)
+	}
+	t.Errorf("expected %v, got %v", "asdf", string(healthBytes))
 }

Diff for: test/integration/framework/master_utils.go

@@ -235,6 +235,12 @@ func startMasterOrDie(masterConfig *master.Config, incomingServer *httptest.Serv
 		masterReceiver.SetMaster(m)
 	}
 
+	// TODO have this start method actually use the normal start sequence for the API server
+	// this method never actually calls the `Run` method for the API server
+	// fire the post hooks ourselves
+	m.GenericAPIServer.PrepareRun()
+	m.GenericAPIServer.RunPostStartHooks()
+
 	cfg := *masterConfig.GenericConfig.LoopbackClientConfig
 	cfg.ContentConfig.GroupVersion = &unversioned.GroupVersion{}
 	privilegedClient, err := restclient.RESTClientFor(&cfg)
@@ -254,12 +260,6 @@ func startMasterOrDie(masterConfig *master.Config, incomingServer *httptest.Serv
 		glog.Fatal(err)
 	}
 
-	// TODO have this start method actually use the normal start sequence for the API server
-	// this method never actually calls the `Run` method for the API server
-	// fire the post hooks ourselves
-	m.GenericAPIServer.PrepareRun()
-	m.GenericAPIServer.RunPostStartHooks()
-
 	// wait for services to be ready
 	if masterConfig.EnableCoreControllers {
 		// TODO Once /healthz is updated for posthooks, we'll wait for good health
@@ -350,6 +350,7 @@ func NewMasterConfig() *master.Config {
 	genericConfig.APIResourceConfigSource = master.DefaultAPIResourceConfigSource()
 	genericConfig.Authorizer = authorizer.NewAlwaysAllowAuthorizer()
 	genericConfig.AdmissionControl = admit.NewAlwaysAdmit()
+	genericConfig.EnableMetrics = true
 
 	return &master.Config{
 		GenericConfig:         genericConfig,

Diff for: test/integration/openshift/openshift_test.go

@@ -29,6 +29,7 @@ func TestMasterExportsSymbols(t *testing.T) {
 	_ = &master.Config{
 		GenericConfig: &genericapiserver.Config{
 			EnableSwaggerSupport: false,
+			EnableMetrics:        true,
 		},
 		EnableCoreControllers: false,
 		EnableUISupport:       false,