This adds an etcd health check endpoint to kube-apiserver

addressing kubernetes#48215.

cmd/kube-apiserver/app/BUILD

@@ -18,7 +18,6 @@ go_library(
     tags = ["automanaged"],
     deps = [
         "//cmd/kube-apiserver/app/options:go_default_library",
-        "//cmd/kube-apiserver/app/preflight:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/apis/apps:go_default_library",
         "//pkg/apis/batch:go_default_library",
@@ -94,6 +93,7 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/server/options:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/server/storage:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/storage/etcd3/preflight:go_default_library",
         "//vendor/k8s.io/client-go/informers:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration:go_default_library",
@@ -116,7 +116,6 @@ filegroup(
     srcs = [
         ":package-srcs",
         "//cmd/kube-apiserver/app/options:all-srcs",
-        "//cmd/kube-apiserver/app/preflight:all-srcs",
         "//cmd/kube-apiserver/app/testing:all-srcs",
     ],
     tags = ["automanaged"],

cmd/kube-apiserver/app/server.go

@@ -54,11 +54,11 @@ import (
 	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
 	//aggregatorinformers "k8s.io/kube-aggregator/pkg/client/informers/internalversion"
 
+	"k8s.io/apiserver/pkg/storage/etcd3/preflight"
 	clientgoinformers "k8s.io/client-go/informers"
 	clientgoclientset "k8s.io/client-go/kubernetes"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
-	"k8s.io/kubernetes/cmd/kube-apiserver/app/preflight"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/apis/apps"
 	"k8s.io/kubernetes/pkg/apis/batch"

staging/src/k8s.io/apiserver/pkg/server/options/BUILD

@@ -62,7 +62,9 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/registry/generic:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/server:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/server/healthz:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/server/storage:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/storage/etcd3/preflight:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/flag:go_default_library",

staging/src/k8s.io/apiserver/pkg/server/options/etcd.go

@@ -18,6 +18,7 @@ package options
 
 import (
 	"fmt"
+	"net/http"
 
 	"github.com/spf13/pflag"
 
@@ -26,7 +27,9 @@ import (
 	"k8s.io/apiserver/pkg/registry/generic"
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/server/healthz"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
+	"k8s.io/apiserver/pkg/storage/etcd3/preflight"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 )
 
@@ -127,15 +130,30 @@ func (s *EtcdOptions) AddFlags(fs *pflag.FlagSet) {
 }
 
 func (s *EtcdOptions) ApplyTo(c *server.Config) error {
+	s.addEtcdHealthEndpoint(c)
 	c.RESTOptionsGetter = &SimpleRestOptionsFactory{Options: *s}
 	return nil
 }
 
 func (s *EtcdOptions) ApplyWithStorageFactoryTo(factory serverstorage.StorageFactory, c *server.Config) error {
+	s.addEtcdHealthEndpoint(c)
 	c.RESTOptionsGetter = &storageFactoryRestOptionsFactory{Options: *s, StorageFactory: factory}
 	return nil
 }
 
+func (s *EtcdOptions) addEtcdHealthEndpoint(c *server.Config) {
+	c.HealthzChecks = append(c.HealthzChecks, healthz.NamedCheck("etcd", func(r *http.Request) error {
+		done, err := preflight.EtcdConnection{ServerList: s.StorageConfig.ServerList}.CheckEtcdServers()
+		if !done {
+			return fmt.Errorf("etcd failed")
+		}
+		if err != nil {
+			return err
+		}
+		return nil
+	}))
+}
+
 type SimpleRestOptionsFactory struct {
 	Options EtcdOptions
 }

cmd/kube-apiserver/app/preflight/BUILD → staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight/BUILD

@@ -21,16 +21,3 @@ go_test(
     tags = ["automanaged"],
     deps = ["//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library"],
 )
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-)

cmd/kube-apiserver/app/preflight/checks.go → staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight/checks.go

@@ -36,20 +36,24 @@ type EtcdConnection struct {
 	ServerList []string
 }
 
-func (EtcdConnection) serverReachable(address string) bool {
-	if conn, err := net.DialTimeout("tcp", address, connectionTimeout); err == nil {
+func (EtcdConnection) serverReachable(connURL *url.URL) bool {
+	scheme := connURL.Scheme
+	if scheme == "http" || scheme == "https" || scheme == "tcp" {
+		scheme = "tcp"
+	}
+	if conn, err := net.DialTimeout(scheme, connURL.Host, connectionTimeout); err == nil {
 		defer conn.Close()
 		return true
 	}
 	return false
 }
 
-func parseServerURI(serverURI string) (string, error) {
+func parseServerURI(serverURI string) (*url.URL, error) {
 	connURL, err := url.Parse(serverURI)
 	if err != nil {
-		return "", fmt.Errorf("unable to parse etcd url: %v", err)
+		return &url.URL{}, fmt.Errorf("unable to parse etcd url: %v", err)
 	}
-	return connURL.Host, nil
+	return connURL, nil
 }
 
 // CheckEtcdServers will attempt to reach all etcd servers once. If any

cmd/kube-apiserver/app/preflight/checks_test.go → staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight/checks_test.go

@@ -17,21 +17,34 @@ limitations under the License.
 package preflight
 
 import (
+	"net/url"
 	"testing"
 	"time"
 
 	utilwait "k8s.io/apimachinery/pkg/util/wait"
 )
 
 func TestParseServerURIGood(t *testing.T) {
-	host, err := parseServerURI("https://127.0.0.1:2379")
+	connURL, err := parseServerURI("https://127.0.0.1:2379")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 
 	reference := "127.0.0.1:2379"
-	if host != reference {
-		t.Fatalf("server uri was not parsed correctly, host %s was invalid", host)
+	if connURL.Host != reference {
+		t.Fatalf("server uri was not parsed correctly, host %s was invalid", connURL.Host)
+	}
+}
+
+func TestParseServerURIGoodUnix(t *testing.T) {
+	connURL, err := parseServerURI("unix://127.0.0.1:21002112605")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	reference := "127.0.0.1:21002112605"
+	if connURL.Host != reference {
+		t.Fatalf("server uri was not parsed correctly, host %s was invalid", connURL.Host)
 	}
 }
 
@@ -45,7 +58,7 @@ func TestParseServerURIBad(t *testing.T) {
 func TestEtcdConnection(t *testing.T) {
 	etcd := new(EtcdConnection)
 
-	result := etcd.serverReachable("-not a real network address-")
+	result := etcd.serverReachable(&url.URL{Host: "-not a real network address-", Scheme: "tcp"})
 	if result {
 		t.Fatal("checkConnection should not have succeeded")
 	}