Merge pull request kubernetes#39230 from irfanurrehman/fed-init-5

Automatic merge from submit-queue (batch tested with PRs 39230, 39718)

[Federation] Kubefed init verifies if control plane pods are up before returning success

This PR updates the functionality as needed in issue kubernetes#37841.

cc @kubernetes/sig-cluster-federation @nikhiljindal @madhusudancs @shashidharatd

Author: Kubernetes Submit Queue

federation/pkg/kubefed/init/init.go

@@ -63,6 +63,7 @@ const (
 	HostClusterLocalDNSZoneName = "cluster.local."
 
 	lbAddrRetryInterval = 5 * time.Second
+	podWaitInterval     = 2 * time.Second
 )
 
 var (
@@ -230,6 +231,15 @@ func initFederation(cmdOut io.Writer, config util.AdminConfig, cmd *cobra.Comman
 	}
 
 	if !dryRun {
+		fedPods := []string{serverName, cmName}
+		err = waitForPods(hostClientset, fedPods, initFlags.FederationSystemNamespace)
+		if err != nil {
+			return err
+		}
+		err = waitSrvHealthy(config, initFlags.Name, initFlags.Kubeconfig)
+		if err != nil {
+			return err
+		}
 		return printSuccess(cmdOut, ips, hostnames)
 	}
 	_, err = fmt.Fprintf(cmdOut, "Federation control plane runs (dry run)\n")
@@ -576,6 +586,48 @@ func createControllerManager(clientset *client.Clientset, namespace, name, svcNa
 	return clientset.Extensions().Deployments(namespace).Create(dep)
 }
 
+func waitForPods(clientset *client.Clientset, fedPods []string, namespace string) error {
+	err := wait.PollInfinite(podWaitInterval, func() (bool, error) {
+		podCheck := len(fedPods)
+		podList, err := clientset.Core().Pods(namespace).List(api.ListOptions{})
+		if err != nil {
+			return false, nil
+		}
+		for _, pod := range podList.Items {
+			for _, fedPod := range fedPods {
+				if strings.HasPrefix(pod.Name, fedPod) && pod.Status.Phase == "Running" {
+					podCheck -= 1
+				}
+			}
+			//ensure that all pods are in running state or keep waiting
+			if podCheck == 0 {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+	return err
+}
+
+func waitSrvHealthy(config util.AdminConfig, context, kubeconfig string) error {
+	fedClientSet, err := config.FederationClientset(context, kubeconfig)
+	if err != nil {
+		return err
+	}
+	fedDiscoveryClient := fedClientSet.Discovery()
+	err = wait.PollInfinite(podWaitInterval, func() (bool, error) {
+		body, err := fedDiscoveryClient.RESTClient().Get().AbsPath("/healthz").Do().Raw()
+		if err != nil {
+			return false, nil
+		}
+		if strings.EqualFold(string(body), "ok") {
+			return true, nil
+		}
+		return false, nil
+	})
+	return err
+}
+
 func printSuccess(cmdOut io.Writer, ips, hostnames []string) error {
 	svcEndpoints := append(ips, hostnames...)
 	_, err := fmt.Fprintf(cmdOut, "Federation API server is running at: %s\n", strings.Join(svcEndpoints, ", "))

federation/pkg/kubefed/init/init_test.go

@@ -693,6 +693,38 @@ func fakeInitHostFactory(federationName, namespaceName, ip, dnsZoneName, image,
 		},
 	}
 
+	podList := v1.PodList{}
+	apiServerPod := v1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: testapi.Extensions.GroupVersion().String(),
+		},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      svcName,
+			Namespace: namespaceName,
+		},
+		Status: v1.PodStatus{
+			Phase: "Running",
+		},
+	}
+
+	cmPod := v1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: testapi.Extensions.GroupVersion().String(),
+		},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      cmName,
+			Namespace: namespaceName,
+		},
+		Status: v1.PodStatus{
+			Phase: "Running",
+		},
+	}
+
+	podList.Items = append(podList.Items, apiServerPod)
+	podList.Items = append(podList.Items, cmPod)
+
 	f, tf, codec, _ := cmdtesting.NewAPIFactory()
 	extCodec := testapi.Extensions.Codec()
 	ns := dynamic.ContentConfig().NegotiatedSerializer
@@ -701,6 +733,8 @@ func fakeInitHostFactory(federationName, namespaceName, ip, dnsZoneName, image,
 		NegotiatedSerializer: ns,
 		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
 			switch p, m := req.URL.Path, req.Method; {
+			case p == "/healthz":
+				return &http.Response{StatusCode: http.StatusOK, Header: kubefedtesting.DefaultHeader(), Body: ioutil.NopCloser(bytes.NewReader([]byte("ok")))}, nil
 			case p == "/api/v1/namespaces" && m == http.MethodPost:
 				body, err := ioutil.ReadAll(req.Body)
 				if err != nil {
@@ -794,6 +828,8 @@ func fakeInitHostFactory(federationName, namespaceName, ip, dnsZoneName, image,
 					return nil, fmt.Errorf("Unexpected deployment object\n\tDiff: %s", diff.ObjectGoPrintDiff(got, want))
 				}
 				return &http.Response{StatusCode: http.StatusCreated, Header: kubefedtesting.DefaultHeader(), Body: kubefedtesting.ObjBody(extCodec, &want)}, nil
+			case p == "/api/v1/namespaces/federation-system/pods" && m == http.MethodGet:
+				return &http.Response{StatusCode: http.StatusOK, Header: kubefedtesting.DefaultHeader(), Body: kubefedtesting.ObjBody(codec, &podList)}, nil
 			default:
 				return nil, fmt.Errorf("unexpected request: %#v\n%#v", req.URL, req)
 			}

federation/pkg/kubefed/testing/BUILD

@@ -12,6 +12,7 @@ go_library(
     srcs = ["testing.go"],
     tags = ["automanaged"],
     deps = [
+        "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/pkg/kubefed/util:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/apimachinery/registered:go_default_library",

federation/pkg/kubefed/testing/testing.go

@@ -23,6 +23,7 @@ import (
 	"net/http"
 	"os"
 
+	fedclient "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	"k8s.io/kubernetes/federation/pkg/kubefed/util"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/apimachinery/registered"
@@ -53,6 +54,18 @@ func (f *fakeAdminConfig) PathOptions() *clientcmd.PathOptions {
 	return f.pathOptions
 }
 
+func (f *fakeAdminConfig) FederationClientset(context, kubeconfigPath string) (*fedclient.Clientset, error) {
+	fakeRestClient, err := f.hostFactory.RESTClient()
+	if err != nil {
+		return nil, err
+	}
+
+	// we ignore the function params and use the client from
+	// the same fakefactory to create a federation clientset
+	// our fake factory exposes only the healthz api for this client
+	return fedclient.New(fakeRestClient), nil
+}
+
 func (f *fakeAdminConfig) HostFactory(host, kubeconfigPath string) cmdutil.Factory {
 	return f.hostFactory
 }

federation/pkg/kubefed/util/BUILD

@@ -12,6 +12,7 @@ go_library(
     srcs = ["util.go"],
     tags = ["automanaged"],
     deps = [
+        "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/unversioned/clientcmd:go_default_library",

federation/pkg/kubefed/util/util.go

@@ -17,6 +17,7 @@ limitations under the License.
 package util
 
 import (
+	fedclient "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
@@ -39,13 +40,16 @@ const (
 
 // AdminConfig provides a filesystem based kubeconfig (via
 // `PathOptions()`) and a mechanism to talk to the federation
-// host cluster.
+// host cluster and the federation control plane api server.
 type AdminConfig interface {
 	// PathOptions provides filesystem based kubeconfig access.
 	PathOptions() *clientcmd.PathOptions
+	// FedClientSet provides a federation API compliant clientset
+	// to communicate with the federation control plane api server
+	FederationClientset(context, kubeconfigPath string) (*fedclient.Clientset, error)
 	// HostFactory provides a mechanism to communicate with the
 	// cluster where federation control plane is hosted.
-	HostFactory(host, kubeconfigPath string) cmdutil.Factory
+	HostFactory(hostcontext, kubeconfigPath string) cmdutil.Factory
 }
 
 // adminConfig implements the AdminConfig interface.
@@ -64,17 +68,30 @@ func (a *adminConfig) PathOptions() *clientcmd.PathOptions {
 	return a.pathOptions
 }
 
-func (a *adminConfig) HostFactory(host, kubeconfigPath string) cmdutil.Factory {
+func (a *adminConfig) FederationClientset(context, kubeconfigPath string) (*fedclient.Clientset, error) {
+	fedConfig := a.getClientConfig(context, kubeconfigPath)
+	fedClientConfig, err := fedConfig.ClientConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	return fedclient.NewForConfigOrDie(fedClientConfig), nil
+}
+
+func (a *adminConfig) HostFactory(hostcontext, kubeconfigPath string) cmdutil.Factory {
+	hostClientConfig := a.getClientConfig(hostcontext, kubeconfigPath)
+	return cmdutil.NewFactory(hostClientConfig)
+}
+
+func (a *adminConfig) getClientConfig(context, kubeconfigPath string) clientcmd.ClientConfig {
 	loadingRules := *a.pathOptions.LoadingRules
 	loadingRules.Precedence = a.pathOptions.GetLoadingPrecedence()
 	loadingRules.ExplicitPath = kubeconfigPath
 	overrides := &clientcmd.ConfigOverrides{
-		CurrentContext: host,
+		CurrentContext: context,
 	}
 
-	hostClientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&loadingRules, overrides)
-
-	return cmdutil.NewFactory(hostClientConfig)
+	return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&loadingRules, overrides)
 }
 
 // SubcommandFlags holds the flags required by the subcommands of