Merge pull request #33543 from taimir/kubeadm

Automatic merge from submit-queue

Decouple master bootstrap from CSR in kubeadm

<!--  Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, read our contributor guidelines https://github.com/kubernetes/kubernetes/blob/master/CONTRIBUTING.md and developer guide https://github.com/kubernetes/kubernetes/blob/master/docs/devel/development.md
2. If you want *faster* PR reviews, read how: https://github.com/kubernetes/kubernetes/blob/master/docs/devel/faster_reviews.md
3. Follow the instructions for writing a release note: https://github.com/kubernetes/kubernetes/blob/master/docs/devel/pull-requests.md#release-notes
-->

**What this PR does / why we need it**: enhance `kubeadm` to allow for parallel provisioning of API endpoints and slave nodes

**Fixes**: https://github.com/kubernetes/kubernetes/issues/33542

**Special notes for your reviewer**:
This is work in progress, trying to 

 * Introduce a concurrent retry mechanism for bootstrapping
   with a single API endpoint
 * Refactor API client creation, decouple from CSR

cmd/kubeadm/app/cmd/join.go

@@ -76,7 +76,17 @@ func RunJoin(out io.Writer, cmd *cobra.Command, args []string, s *kubeadmapi.Nod
 		return fmt.Errorf("Must specify --token (see --help)\n")
 	}
 
-	kubeconfig, err := kubenode.RetrieveTrustedClusterInfo(s)
+	clusterInfo, err := kubenode.RetrieveTrustedClusterInfo(s)
+	if err != nil {
+		return err
+	}
+
+	connectionDetails, err := kubenode.EstablishMasterConnection(s, clusterInfo)
+	if err != nil {
+		return err
+	}
+
+	kubeconfig, err := kubenode.PerformTLSBootstrap(connectionDetails)
 	if err != nil {
 		return err
 	}

cmd/kubeadm/app/node/bootstrap.go

@@ -0,0 +1,103 @@
+/*
+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 node
+
+import (
+	"fmt"
+	"os"
+
+	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
+	kubeadmutil "k8s.io/kubernetes/cmd/kubeadm/app/util"
+	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	certclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/certificates/unversioned"
+	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
+	"k8s.io/kubernetes/pkg/types"
+)
+
+// ConnectionDetails represents a master API endpoint connection
+type ConnectionDetails struct {
+	CertClient *certclient.CertificatesClient
+	Endpoint   string
+	CACert     []byte
+	NodeName   types.NodeName
+}
+
+// EstablishMasterConnection establishes a connection with exactly one of the provided API endpoints or errors.
+func EstablishMasterConnection(s *kubeadmapi.NodeConfiguration, clusterInfo *kubeadmapi.ClusterInfo) (*ConnectionDetails, error) {
+	hostName, err := os.Hostname()
+	if err != nil {
+		return nil, fmt.Errorf("<node/bootstrap> failed to get node hostname [%v]", err)
+	}
+	// TODO(phase1+) https://github.com/kubernetes/kubernetes/issues/33641
+	nodeName := types.NodeName(hostName)
+
+	endpoints := clusterInfo.Endpoints
+	caCert := []byte(clusterInfo.CertificateAuthorities[0])
+
+	var establishedConnection *ConnectionDetails
+	// TODO: add a wait mechanism for the API endpoints (retrying to connect to at least one)
+	for _, endpoint := range endpoints {
+		clientSet, err := createClients(caCert, endpoint, s.Secrets.BearerToken, nodeName)
+		if err != nil {
+			fmt.Printf("<node/bootstrap> warning: %s. Skipping endpoint %s\n", err, endpoint)
+			continue
+		}
+		fmt.Printf("<node/bootstrap> trying to connect to endpoint %s\n", endpoint)
+
+		// TODO: add a simple GET /version request to fail early if needed before attempting
+		// to connect with a discovery client.
+		if err := checkCertsAPI(clientSet.DiscoveryClient); err != nil {
+			fmt.Printf("<node/bootstrap> warning: failed to connect to %s: %v\n", endpoint, err)
+			continue
+		}
+
+		fmt.Printf("<node/bootstrap> successfully established connection with endpoint %s\n", endpoint)
+		// connection established
+		establishedConnection = &ConnectionDetails{
+			CertClient: clientSet.CertificatesClient,
+			Endpoint:   endpoint,
+			CACert:     caCert,
+			NodeName:   nodeName,
+		}
+		break
+	}
+
+	if establishedConnection == nil {
+		return nil, fmt.Errorf("<node/bootstrap> failed to create bootstrap clients " +
+			"for any of the provided API endpoints")
+	}
+	return establishedConnection, nil
+}
+
+// Creates a set of clients for this endpoint
+func createClients(caCert []byte, endpoint, token string, nodeName types.NodeName) (*clientset.Clientset, error) {
+	bareClientConfig := kubeadmutil.CreateBasicClientConfig("kubernetes", endpoint, caCert)
+	bootstrapClientConfig, err := clientcmd.NewDefaultClientConfig(
+		*kubeadmutil.MakeClientConfigWithToken(
+			bareClientConfig, "kubernetes", fmt.Sprintf("kubelet-%s", nodeName), token,
+		),
+		&clientcmd.ConfigOverrides{},
+	).ClientConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create API client configuration [%v]", err)
+	}
+	clientSet, err := clientset.NewForConfig(bootstrapClientConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create clients for the API endpoint %s [%v]", endpoint, err)
+	}
+	return clientSet, nil
+}

cmd/kubeadm/app/node/csr.go

@@ -18,55 +18,19 @@ package node
 
 import (
 	"fmt"
-	"os"
 
-	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
 	kubeadmutil "k8s.io/kubernetes/cmd/kubeadm/app/util"
 	"k8s.io/kubernetes/pkg/apis/certificates"
-	unversionedcertificates "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/certificates/unversioned"
-	"k8s.io/kubernetes/pkg/client/restclient"
 	"k8s.io/kubernetes/pkg/client/typed/discovery"
-	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
 	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
 	"k8s.io/kubernetes/pkg/kubelet/util/csr"
-	"k8s.io/kubernetes/pkg/types"
 	certutil "k8s.io/kubernetes/pkg/util/cert"
 )
 
-// PerformTLSBootstrap creates a RESTful client in order to execute certificate signing request.
+// PerformTLSBootstrap executes a certificate signing request with the
-func PerformTLSBootstrap(s *kubeadmapi.NodeConfiguration, apiEndpoint string, caCert []byte) (*clientcmdapi.Config, error) {
+// provided connection details.
-	// TODO(phase1+) try all the api servers until we find one that works
+func PerformTLSBootstrap(connection *ConnectionDetails) (*clientcmdapi.Config, error) {
-	bareClientConfig := kubeadmutil.CreateBasicClientConfig("kubernetes", apiEndpoint, caCert)
+	csrClient := connection.CertClient.CertificateSigningRequests()
-
-	hostName, err := os.Hostname()
-	if err != nil {
-		return nil, fmt.Errorf("<node/csr> failed to get node hostname [%v]", err)
-	}
-
-	// TODO(phase1+) https://github.com/kubernetes/kubernetes/issues/33641
-	nodeName := types.NodeName(hostName)
-
-	bootstrapClientConfig, err := clientcmd.NewDefaultClientConfig(
-		*kubeadmutil.MakeClientConfigWithToken(
-			bareClientConfig, "kubernetes", fmt.Sprintf("kubelet-%s", nodeName), s.Secrets.BearerToken,
-		),
-		&clientcmd.ConfigOverrides{},
-	).ClientConfig()
-	if err != nil {
-		return nil, fmt.Errorf("<node/csr> failed to create API client configuration [%v]", err)
-	}
-
-	client, err := unversionedcertificates.NewForConfig(bootstrapClientConfig)
-	if err != nil {
-		return nil, fmt.Errorf("<node/csr> failed to create API client [%v]", err)
-	}
-	csrClient := client.CertificateSigningRequests()
-
-	// TODO(phase1+) https://github.com/kubernetes/kubernetes/issues/33643
-
-	if err := checkCertsAPI(bootstrapClientConfig); err != nil {
-		return nil, fmt.Errorf("<node/csr> failed to proceed due to API compatibility issue - %v", err)
-	}
 
 	fmt.Println("<node/csr> created API client to obtain unique certificate for this node, generating keys and certificate signing request")
 
@@ -74,7 +38,7 @@ func PerformTLSBootstrap(s *kubeadmapi.NodeConfiguration, apiEndpoint string, ca
 	if err != nil {
 		return nil, fmt.Errorf("<node/csr> failed to generating private key [%v]", err)
 	}
-	cert, err := csr.RequestNodeCertificate(csrClient, key, nodeName)
+	cert, err := csr.RequestNodeCertificate(csrClient, key, connection.NodeName)
 	if err != nil {
 		return nil, fmt.Errorf("<node/csr> failed to request signed certificate from the API server [%v]", err)
 	}
@@ -84,21 +48,18 @@ func PerformTLSBootstrap(s *kubeadmapi.NodeConfiguration, apiEndpoint string, ca
 	}
 	fmt.Printf("<node/csr> received signed certificate from the API server:\n%s\n", fmtCert)
 	fmt.Println("<node/csr> generating kubelet configuration")
+
+	bareClientConfig := kubeadmutil.CreateBasicClientConfig("kubernetes", connection.Endpoint, connection.CACert)
 	finalConfig := kubeadmutil.MakeClientConfigWithCerts(
-		bareClientConfig, "kubernetes", fmt.Sprintf("kubelet-%s", nodeName),
+		bareClientConfig, "kubernetes", fmt.Sprintf("kubelet-%s", connection.NodeName),
 		key, cert,
 	)
 
 	return finalConfig, nil
 }
 
-func checkCertsAPI(config *restclient.Config) error {
+// Checks if the certificates API for this endpoint is functional
-	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
+func checkCertsAPI(discoveryClient *discovery.DiscoveryClient) error {
-
-	if err != nil {
-		return fmt.Errorf("failed to create API discovery client [%v]", err)
-	}
-
 	serverGroups, err := discoveryClient.ServerGroups()
 
 	if err != nil {

cmd/kubeadm/app/node/discovery.go

@@ -25,10 +25,9 @@ import (
 
 	jose "github.com/square/go-jose"
 	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
-	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
 )
 
-func RetrieveTrustedClusterInfo(s *kubeadmapi.NodeConfiguration) (*clientcmdapi.Config, error) {
+func RetrieveTrustedClusterInfo(s *kubeadmapi.NodeConfiguration) (*kubeadmapi.ClusterInfo, error) {
 	host, port := s.MasterAddresses[0], 9898
 	requestURL := fmt.Sprintf("http://%s:%d/cluster-info/v1/?token-id=%s", host, port, s.Secrets.TokenID)
 	req, err := http.NewRequest("GET", requestURL, nil)
@@ -71,9 +70,5 @@ func RetrieveTrustedClusterInfo(s *kubeadmapi.NodeConfiguration) (*clientcmdapi.
 	// TODO(phase1+) print summary info about the CA certificate, along with the the checksum signature
 	// we also need an ability for the user to configure the client to validate recieved CA cert agains a checksum
 	fmt.Printf("<node/discovery> cluster info signature and contents are valid, will use API endpoints %v\n", clusterInfo.Endpoints)
-
+	return &clusterInfo, nil
-	apiServer := clusterInfo.Endpoints[0]
-	caCert := []byte(clusterInfo.CertificateAuthorities[0])
-
-	return PerformTLSBootstrap(s, apiServer, caCert)
 }