Merge pull request #46865 from sttts/sttts-kube-apiserver-run-test

Automatic merge from submit-queue

kube-apiserver: tests for aggregation and CRDs via delegation

In our integration tests we do not use the real kube-apiserver setup code, but mock our own. Here I use the actual `cmd/kube-apiserver/app.Run()` func with an testing etcd server. This can test the whole delegation chain of aggregator, apiextensions and kube-apiserver.

cmd/kube-apiserver/app/BUILD

@@ -118,6 +118,7 @@ filegroup(
         ":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

@@ -110,30 +110,39 @@ func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
 	// To help debugging, immediately log version
 	glog.Infof("Version: %+v", version.Get())
 
-	nodeTunneler, proxyTransport, err := CreateNodeDialer(runOptions)
+	server, err := CreateServerChain(runOptions, stopCh)
 	if err != nil {
 		return err
 	}
 
+	return server.PrepareRun().Run(stopCh)
+}
+
+// CreateServerChain creates the apiservers connected via delegation.
+func CreateServerChain(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) (*genericapiserver.GenericAPIServer, error) {
+	nodeTunneler, proxyTransport, err := CreateNodeDialer(runOptions)
+	if err != nil {
+		return nil, err
+	}
 	kubeAPIServerConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, err := CreateKubeAPIServerConfig(runOptions, nodeTunneler, proxyTransport)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// TPRs are enabled and not yet beta, since this these are the successor, they fall under the same enablement rule
 	// If additional API servers are added, they should be gated.
 	apiExtensionsConfig, err := createAPIExtensionsConfig(*kubeAPIServerConfig.GenericConfig, runOptions)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.EmptyDelegate)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer, sharedInformers)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// if we're starting up a hacked up version of this API server for a weird test case,
@@ -142,11 +151,11 @@ func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
 		if insecureServingOptions != nil {
 			insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(kubeAPIServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
 			if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, stopCh); err != nil {
-				return err
+				return nil, err
 			}
 		}
 
-		return kubeAPIServer.GenericAPIServer.PrepareRun().Run(stopCh)
+		return kubeAPIServer.GenericAPIServer, nil
 	}
 
 	// otherwise go down the normal path of standing the aggregator up in front of the API server
@@ -156,24 +165,24 @@ func Run(runOptions *options.ServerRunOptions, stopCh <-chan struct{}) error {
 	// aggregator comes last in the chain
 	aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, runOptions, versionedInformers, serviceResolver, proxyTransport)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	aggregatorConfig.ProxyTransport = proxyTransport
 	aggregatorConfig.ServiceResolver = serviceResolver
 	aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
 	if err != nil {
 		// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
-		return err
+		return nil, err
 	}
 
 	if insecureServingOptions != nil {
 		insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(aggregatorServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
 		if err := kubeserver.NonBlockingRun(insecureServingOptions, insecureHandlerChain, stopCh); err != nil {
-			return err
+			return nil, err
 		}
 	}
 
-	return aggregatorServer.GenericAPIServer.PrepareRun().Run(stopCh)
+	return aggregatorServer.GenericAPIServer, nil
 }
 
 // CreateKubeAPIServer creates and wires a workable kube-apiserver
@@ -255,9 +264,11 @@ func CreateKubeAPIServerConfig(s *options.ServerRunOptions, nodeTunneler tunnele
 		return nil, nil, nil, nil, nil, err
 	}
 
+	if _, port, err := net.SplitHostPort(s.Etcd.StorageConfig.ServerList[0]); err == nil && port != "0" && len(port) != 0 {
 		if err := utilwait.PollImmediate(etcdRetryInterval, etcdRetryLimit*etcdRetryInterval, preflight.EtcdConnection{ServerList: s.Etcd.StorageConfig.ServerList}.CheckEtcdServers); err != nil {
 			return nil, nil, nil, nil, nil, fmt.Errorf("error waiting for etcd connection: %v", err)
 		}
+	}
 
 	capabilities.Initialize(capabilities.Capabilities{
 		AllowPrivileged: s.AllowPrivileged,

cmd/kube-apiserver/app/testing/BUILD

@@ -0,0 +1,56 @@
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["server_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//vendor/k8s.io/api/apps/v1beta1:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/api/networking/v1:go_default_library",
+        "//vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1:go_default_library",
+        "//vendor/k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/client-go/dynamic:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes:go_default_library",
+    ],
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["testserver.go"],
+    tags = ["automanaged"],
+    deps = [
+        "//cmd/kube-apiserver/app:go_default_library",
+        "//cmd/kube-apiserver/app/options:go_default_library",
+        "//pkg/api:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/storage/etcd/testing:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes:go_default_library",
+        "//vendor/k8s.io/client-go/rest: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/testing/server_test.go

@@ -0,0 +1,231 @@
+/*
+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 testing
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	appsv1beta1 "k8s.io/api/apps/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
+	apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+)
+
+func TestRun(t *testing.T) {
+	config, tearDown := StartTestServerOrDie(t)
+	defer tearDown()
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// test whether the server is really healthy after /healthz told us so
+	t.Logf("Creating Deployment directly after being healthy")
+	var replicas int32 = 1
+	_, err = client.AppsV1beta1().Deployments("default").Create(&appsv1beta1.Deployment{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Deployment",
+			APIVersion: "apps/v1beta1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "test",
+		},
+		Spec: appsv1beta1.DeploymentSpec{
+			Replicas: &replicas,
+			Strategy: appsv1beta1.DeploymentStrategy{
+				Type: appsv1beta1.RollingUpdateDeploymentStrategyType,
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{"foo": "bar"},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "foo",
+							Image: "foo",
+						},
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Failed to create deployment: %v", err)
+	}
+}
+
+func TestCRDShadowGroup(t *testing.T) {
+	config, tearDown := StartTestServerOrDie(t)
+	defer tearDown()
+
+	kubeclient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	apiextensionsclient, err := apiextensionsclientset.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	t.Logf("Creating a NetworkPolicy")
+	nwPolicy, err := kubeclient.NetworkingV1().NetworkPolicies("default").Create(&networkingv1.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{Name: "abc", Namespace: metav1.NamespaceDefault},
+		Spec: networkingv1.NetworkPolicySpec{
+			PodSelector: metav1.LabelSelector{MatchLabels: map[string]string{"foo": "bar"}},
+			Ingress:     []networkingv1.NetworkPolicyIngressRule{},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Failed to create NetworkPolicy: %v", err)
+	}
+
+	t.Logf("Trying to shadow networking group")
+	crd := &apiextensionsv1beta1.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foos." + networkingv1.GroupName,
+		},
+		Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{
+			Group:   networkingv1.GroupName,
+			Version: networkingv1.SchemeGroupVersion.Version,
+			Scope:   apiextensionsv1beta1.ClusterScoped,
+			Names: apiextensionsv1beta1.CustomResourceDefinitionNames{
+				Plural: "foos",
+				Kind:   "Foo",
+			},
+		},
+	}
+	if _, err = apiextensionsclient.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd); err != nil {
+		t.Fatalf("Failed to create networking group CRD: %v", err)
+	}
+	if err := waitForEstablishedCRD(apiextensionsclient, crd.Name); err != nil {
+		t.Fatalf("Failed to establish networking group CRD: %v", err)
+	}
+	// wait to give aggregator time to update
+	time.Sleep(2 * time.Second)
+
+	t.Logf("Checking that we still see the NetworkPolicy")
+	_, err = kubeclient.NetworkingV1().NetworkPolicies(nwPolicy.Namespace).Get(nwPolicy.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Errorf("Failed to get NetworkPolocy: %v", err)
+	}
+
+	t.Logf("Checking that crd resource does not show up in networking group")
+	found, err := crdExistsInDiscovery(apiextensionsclient, crd)
+	if err != nil {
+		t.Fatalf("unexpected discovery error: %v", err)
+	}
+	if found {
+		t.Errorf("CRD resource shows up in discovery, but shouldn't.")
+	}
+}
+
+func TestCRD(t *testing.T) {
+	config, tearDown := StartTestServerOrDie(t)
+	defer tearDown()
+
+	apiextensionsclient, err := apiextensionsclientset.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	t.Logf("Trying to create a custom resource without conflict")
+	crd := &apiextensionsv1beta1.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foos.cr.bar.com",
+		},
+		Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{
+			Group:   "cr.bar.com",
+			Version: "v1",
+			Scope:   apiextensionsv1beta1.NamespaceScoped,
+			Names: apiextensionsv1beta1.CustomResourceDefinitionNames{
+				Plural: "foos",
+				Kind:   "Foo",
+			},
+		},
+	}
+	if _, err = apiextensionsclient.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd); err != nil {
+		t.Fatalf("Failed to create foos.cr.bar.com CRD; %v", err)
+	}
+	if err := waitForEstablishedCRD(apiextensionsclient, crd.Name); err != nil {
+		t.Fatalf("Failed to establish foos.cr.bar.com CRD: %v", err)
+	}
+	if err := wait.PollImmediate(500*time.Millisecond, 30*time.Second, func() (bool, error) {
+		return crdExistsInDiscovery(apiextensionsclient, crd)
+	}); err != nil {
+		t.Fatalf("Failed to see foos.cr.bar.com in discovery: %v", err)
+	}
+
+	t.Logf("Trying to access foos.cr.bar.com with dynamic client")
+	barComConfig := *config
+	barComConfig.GroupVersion = &schema.GroupVersion{Group: "cr.bar.com", Version: "v1"}
+	barComConfig.APIPath = "/apis"
+	barComClient, err := dynamic.NewClient(&barComConfig)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	_, err = barComClient.Resource(&metav1.APIResource{Name: "foos", Namespaced: true}, "default").List(metav1.ListOptions{})
+	if err != nil {
+		t.Errorf("Failed to list foos.cr.bar.com instances: %v", err)
+	}
+}
+
+func waitForEstablishedCRD(client apiextensionsclientset.Interface, name string) error {
+	return wait.PollImmediate(500*time.Millisecond, wait.ForeverTestTimeout, func() (bool, error) {
+		crd, err := client.ApiextensionsV1beta1().CustomResourceDefinitions().Get(name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		for _, cond := range crd.Status.Conditions {
+			switch cond.Type {
+			case apiextensionsv1beta1.Established:
+				if cond.Status == apiextensionsv1beta1.ConditionTrue {
+					return true, err
+				}
+			case apiextensionsv1beta1.NamesAccepted:
+				if cond.Status == apiextensionsv1beta1.ConditionFalse {
+					fmt.Printf("Name conflict: %v\n", cond.Reason)
+				}
+			}
+		}
+		return false, nil
+	})
+}
+
+func crdExistsInDiscovery(client apiextensionsclientset.Interface, crd *apiextensionsv1beta1.CustomResourceDefinition) (bool, error) {
+	resourceList, err := client.Discovery().ServerResourcesForGroupVersion(crd.Spec.Group + "/" + crd.Spec.Version)
+	if err != nil {
+		return false, nil
+	}
+	for _, resource := range resourceList.APIResources {
+		if resource.Name == crd.Spec.Names.Plural {
+			return true, nil
+		}
+	}
+	return false, nil
+}

cmd/kube-apiserver/app/testing/testserver.go

@@ -0,0 +1,157 @@
+/*
+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 testing
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	etcdtesting "k8s.io/apiserver/pkg/storage/etcd/testing"
+	"k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/kubernetes/cmd/kube-apiserver/app"
+	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
+	"k8s.io/kubernetes/pkg/api"
+)
+
+// TearDownFunc is to be called to tear down a test server.
+type TearDownFunc func()
+
+// StartTestServer starts a etcd server and kube-apiserver. A rest client config and a tear-down func
+// are returned.
+//
+// Note: we return a tear-down func instead of a stop channel because the later will leak temporariy
+// 		 files that becaues Golang testing's call to os.Exit will not give a stop channel go routine
+// 		 enough time to remove temporariy files.
+func StartTestServer(t *testing.T) (result *restclient.Config, tearDownForCaller TearDownFunc, err error) {
+	var tmpDir string
+	var etcdServer *etcdtesting.EtcdTestServer
+	stopCh := make(chan struct{})
+	tearDown := func() {
+		close(stopCh)
+		if etcdServer != nil {
+			etcdServer.Terminate(t)
+		}
+		if len(tmpDir) != 0 {
+			os.RemoveAll(tmpDir)
+		}
+	}
+	defer func() {
+		if tearDownForCaller == nil {
+			tearDown()
+		}
+	}()
+
+	t.Logf("Starting etcd...")
+	etcdServer, storageConfig := etcdtesting.NewUnsecuredEtcd3TestClientServer(t, api.Scheme)
+
+	tmpDir, err = ioutil.TempDir("", "kubernetes-kube-apiserver")
+	if err != nil {
+		return nil, nil, fmt.Errorf("Failed to create temp dir: %v", err)
+	}
+
+	s := options.NewServerRunOptions()
+	s.InsecureServing.BindPort = 0
+	s.SecureServing.BindPort = freePort()
+	s.SecureServing.ServerCert.CertDirectory = tmpDir
+	s.ServiceClusterIPRange.IP = net.IPv4(10, 0, 0, 0)
+	s.ServiceClusterIPRange.Mask = net.CIDRMask(16, 32)
+	s.Etcd.StorageConfig = *storageConfig
+	s.Etcd.DefaultStorageMediaType = "application/json"
+	s.Admission.PluginNames = strings.Split("Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,PersistentVolumeLabel,DefaultStorageClass,ResourceQuota,DefaultTolerationSeconds", ",")
+
+	t.Logf("Starting kube-apiserver...")
+	runErrCh := make(chan error, 1)
+	server, err := app.CreateServerChain(s, stopCh)
+	if err != nil {
+		return nil, nil, fmt.Errorf("Failed to create server chain: %v", err)
+	}
+	go func(stopCh <-chan struct{}) {
+		if err := server.PrepareRun().Run(stopCh); err != nil {
+			t.Logf("kube-apiserver exited uncleanly: %v", err)
+			runErrCh <- err
+		}
+	}(stopCh)
+
+	t.Logf("Waiting for /healthz to be ok...")
+	client, err := kubernetes.NewForConfig(server.LoopbackClientConfig)
+	if err != nil {
+		return nil, nil, fmt.Errorf("Failed to create a client: %v", err)
+	}
+	err = wait.Poll(100*time.Millisecond, 30*time.Second, func() (bool, error) {
+		select {
+		case err := <-runErrCh:
+			return false, err
+		default:
+		}
+
+		result := client.CoreV1Client.RESTClient().Get().AbsPath("/healthz").Do()
+		status := 0
+		result.StatusCode(&status)
+		if status == 200 {
+			return true, nil
+		}
+		return false, nil
+	})
+	if err != nil {
+		return nil, nil, fmt.Errorf("Failed to wait for /healthz to return ok: %v", err)
+	}
+
+	// from here the caller must call tearDown
+	return server.LoopbackClientConfig, tearDown, nil
+}
+
+// StartTestServerOrDie calls StartTestServer with up to 5 retries on bind error and dies with
+// t.Fatal if it does not succeed.
+func StartTestServerOrDie(t *testing.T) (*restclient.Config, TearDownFunc) {
+	// retry test because the bind might fail due to a race with another process
+	// binding to the port. We cannot listen to :0 (then the kernel would give us
+	// a port which is free for sure), so we need this workaround.
+	for retry := 0; retry < 5 && !t.Failed(); retry++ {
+		config, td, err := StartTestServer(t)
+		if err == nil {
+			return config, td
+		}
+		if err != nil && !strings.Contains(err.Error(), "bind") {
+			break
+		}
+		t.Logf("Bind error, retrying...")
+	}
+
+	t.Fatalf("Failed to launch server")
+	return nil, nil
+}
+
+func freePort() int {
+	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
+	if err != nil {
+		panic(err)
+	}
+
+	l, err := net.ListenTCP("tcp", addr)
+	if err != nil {
+		panic(err)
+	}
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port
+}

hack/.linted_packages

@@ -14,6 +14,7 @@ cmd/kube-apiserver
 cmd/kube-apiserver/app
 cmd/kube-apiserver/app/options
 cmd/kube-apiserver/app/preflight
+cmd/kube-apiserver/app/testing
 cmd/kube-controller-manager
 cmd/kube-controller-manager/app/options
 cmd/kube-proxy

test/integration/framework/BUILD

@@ -21,7 +21,6 @@ go_library(
         "//pkg/apis/batch:go_default_library",
         "//pkg/apis/policy/v1alpha1:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
-        "//pkg/client/clientset_generated/clientset/typed/core/v1:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/externalversions:go_default_library",
         "//pkg/controller:go_default_library",
@@ -51,7 +50,6 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/versioning:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authentication/authenticator:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authentication/authenticatorfactory:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authentication/request/union:go_default_library",

test/integration/framework/master_utils.go

@@ -42,7 +42,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apimachinery/pkg/watch"
 	authauthenticator "k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/apiserver/pkg/authentication/authenticatorfactory"
 	authenticatorunion "k8s.io/apiserver/pkg/authentication/request/union"
@@ -63,7 +62,6 @@ import (
 	"k8s.io/kubernetes/pkg/apis/batch"
 	policy "k8s.io/kubernetes/pkg/apis/policy/v1alpha1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
-	coreclient "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/typed/core/v1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/externalversions"
 	"k8s.io/kubernetes/pkg/controller"
@@ -289,30 +287,6 @@ func startMasterOrDie(masterConfig *master.Config, incomingServer *httptest.Serv
 		glog.Fatal(err)
 	}
 
-	// wait for services to be ready
-	if masterConfig.EnableCoreControllers {
-		// TODO Once /healthz is updated for posthooks, we'll wait for good health
-		coreClient := coreclient.NewForConfigOrDie(&cfg)
-		svcWatch, err := coreClient.Services(metav1.NamespaceDefault).Watch(metav1.ListOptions{})
-		if err != nil {
-			closeFn()
-			glog.Fatal(err)
-		}
-		_, err = watch.Until(30*time.Second, svcWatch, func(event watch.Event) (bool, error) {
-			if event.Type != watch.Added {
-				return false, nil
-			}
-			if event.Object.(*v1.Service).Name == "kubernetes" {
-				return true, nil
-			}
-			return false, nil
-		})
-		if err != nil {
-			closeFn()
-			glog.Fatal(err)
-		}
-	}
-
 	return m, s, closeFn
 }
 

test/integration/master/master_test.go

@@ -94,6 +94,16 @@ func TestExtensionsPrefix(t *testing.T) {
 	testPrefix(t, "/apis/extensions/")
 }
 
+func TestKubernetesService(t *testing.T) {
+	config := framework.NewMasterConfig()
+	_, _, closeFn := framework.RunAMaster(config)
+	defer closeFn()
+	coreClient := clientset.NewForConfigOrDie(config.GenericConfig.LoopbackClientConfig)
+	if _, err := coreClient.Services(metav1.NamespaceDefault).Get("kubernetes", metav1.GetOptions{}); err != nil {
+		t.Fatalf("Expected kubernetes service to exists, got: %v", err)
+	}
+}
+
 func TestEmptyList(t *testing.T) {
 	_, s, closeFn := framework.RunAMaster(nil)
 	defer closeFn()