Merge pull request #63110 from stealthybox/automated-cherry-pick-of-#62655-upstream-release-1.10

cmd/kubeadm/app/cmd/upgrade/BUILD

@@ -24,6 +24,7 @@ go_library(
         "//cmd/kubeadm/app/util/apiclient:go_default_library",
         "//cmd/kubeadm/app/util/config:go_default_library",
         "//cmd/kubeadm/app/util/dryrun:go_default_library",
+        "//cmd/kubeadm/app/util/etcd:go_default_library",
         "//cmd/kubeadm/app/util/kubeconfig:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/util/version:go_default_library",

cmd/kubeadm/app/cmd/upgrade/apply.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/kubernetes/cmd/kubeadm/app/util/apiclient"
 	configutil "k8s.io/kubernetes/cmd/kubeadm/app/util/config"
 	dryrunutil "k8s.io/kubernetes/cmd/kubeadm/app/util/dryrun"
+	etcdutil "k8s.io/kubernetes/cmd/kubeadm/app/util/etcd"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/util/version"
 )
@@ -268,7 +269,9 @@ func PerformStaticPodUpgrade(client clientset.Interface, waiter apiclient.Waiter
 		return err
 	}
 
-	return upgrade.StaticPodControlPlane(waiter, pathManager, internalcfg, etcdUpgrade)
+	// These are uninitialized because passing in the clients allow for mocking the client during testing
+	var oldEtcdClient, newEtdClient etcdutil.Client
+	return upgrade.StaticPodControlPlane(waiter, pathManager, internalcfg, etcdUpgrade, oldEtcdClient, newEtdClient)
 }
 
 // DryRunStaticPodUpgrade fakes an upgrade of the control plane

cmd/kubeadm/app/cmd/upgrade/plan.go

@@ -26,9 +26,11 @@ import (
 	"github.com/spf13/cobra"
 
 	"k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/validation"
+	"k8s.io/kubernetes/cmd/kubeadm/app/constants"
 	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	"k8s.io/kubernetes/cmd/kubeadm/app/phases/upgrade"
 	kubeadmutil "k8s.io/kubernetes/cmd/kubeadm/app/util"
+	etcdutil "k8s.io/kubernetes/cmd/kubeadm/app/util/etcd"
 )
 
 // NewCmdPlan returns the cobra command for `kubeadm upgrade plan`
@@ -61,10 +63,18 @@ func RunPlan(parentFlags *cmdUpgradeFlags) error {
 	}
 
 	// Define Local Etcd cluster to be able to retrieve information
-	etcdCluster := kubeadmutil.LocalEtcdCluster{}
+	etcdClient, err := etcdutil.NewStaticPodClient(
+		[]string{"localhost:2379"},
+		constants.GetStaticPodDirectory(),
+		upgradeVars.cfg.CertificatesDir,
+	)
+	if err != nil {
+		return err
+	}
 
 	// Compute which upgrade possibilities there are
-	availUpgrades, err := upgrade.GetAvailableUpgrades(upgradeVars.versionGetter, parentFlags.allowExperimentalUpgrades, parentFlags.allowRCUpgrades, etcdCluster, upgradeVars.cfg.FeatureGates)
+	fmt.Println("[upgrade/plan] computing upgrade possibilities")
+	availUpgrades, err := upgrade.GetAvailableUpgrades(upgradeVars.versionGetter, parentFlags.allowExperimentalUpgrades, parentFlags.allowRCUpgrades, etcdClient, upgradeVars.cfg.FeatureGates)
 	if err != nil {
 		return fmt.Errorf("[upgrade/versions] FATAL: %v", err)
 	}

cmd/kubeadm/app/constants/constants.go

@@ -317,7 +317,7 @@ func EtcdSupportedVersion(versionString string) (*version.Version, error) {
 		}
 		return etcdVersion, nil
 	}
-	return nil, fmt.Errorf("Unsupported or unknown kubernetes version")
+	return nil, fmt.Errorf("Unsupported or unknown kubernetes version(%v)", kubernetesVersion)
 }
 
 // GetStaticPodDirectory returns the location on the disk where the Static Pod should be present

cmd/kubeadm/app/constants/constants_test.go

@@ -123,7 +123,7 @@ func TestEtcdSupportedVersion(t *testing.T) {
 		{
 			kubernetesVersion: "1.99.0",
 			expectedVersion:   nil,
-			expectedError:     fmt.Errorf("Unsupported or unknown kubernetes version"),
+			expectedError:     fmt.Errorf("Unsupported or unknown kubernetes version(1.99.0)"),
 		},
 		{
 			kubernetesVersion: "1.9.0",

cmd/kubeadm/app/phases/upgrade/BUILD

@@ -36,6 +36,7 @@ go_library(
         "//cmd/kubeadm/app/util/apiclient:go_default_library",
         "//cmd/kubeadm/app/util/config:go_default_library",
         "//cmd/kubeadm/app/util/dryrun:go_default_library",
+        "//cmd/kubeadm/app/util/etcd:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/util/version:go_default_library",
         "//pkg/version:go_default_library",
@@ -84,10 +85,12 @@ go_test(
         "//cmd/kubeadm/app/phases/controlplane:go_default_library",
         "//cmd/kubeadm/app/phases/etcd:go_default_library",
         "//cmd/kubeadm/app/util/apiclient:go_default_library",
+        "//cmd/kubeadm/app/util/etcd:go_default_library",
         "//cmd/kubeadm/test:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/util/version:go_default_library",
         "//vendor/github.com/coreos/etcd/clientv3:go_default_library",
+        "//vendor/github.com/coreos/etcd/pkg/transport:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
     ],
 )

cmd/kubeadm/app/phases/upgrade/compute.go

@@ -23,7 +23,7 @@ import (
 	kubeadmconstants "k8s.io/kubernetes/cmd/kubeadm/app/constants"
 	"k8s.io/kubernetes/cmd/kubeadm/app/features"
 	"k8s.io/kubernetes/cmd/kubeadm/app/phases/addons/dns"
-	"k8s.io/kubernetes/cmd/kubeadm/app/util"
+	etcdutil "k8s.io/kubernetes/cmd/kubeadm/app/util/etcd"
 	"k8s.io/kubernetes/pkg/util/version"
 )
 
@@ -74,7 +74,7 @@ type ClusterState struct {
 
 // GetAvailableUpgrades fetches all versions from the specified VersionGetter and computes which
 // kinds of upgrades can be performed
-func GetAvailableUpgrades(versionGetterImpl VersionGetter, experimentalUpgradesAllowed, rcUpgradesAllowed bool, cluster util.EtcdCluster, featureGates map[string]bool) ([]Upgrade, error) {
+func GetAvailableUpgrades(versionGetterImpl VersionGetter, experimentalUpgradesAllowed, rcUpgradesAllowed bool, etcdClient etcdutil.Client, featureGates map[string]bool) ([]Upgrade, error) {
 	fmt.Println("[upgrade] Fetching available versions to upgrade to")
 
 	// Collect the upgrades kubeadm can do in this list
@@ -107,7 +107,7 @@ func GetAvailableUpgrades(versionGetterImpl VersionGetter, experimentalUpgradesA
 	}
 
 	// Get current etcd version
-	etcdStatus, err := cluster.GetEtcdClusterStatus()
+	etcdStatus, err := etcdClient.GetStatus()
 	if err != nil {
 		return nil, err
 	}

cmd/kubeadm/app/phases/upgrade/compute_test.go

@@ -17,10 +17,12 @@ limitations under the License.
 package upgrade
 
 import (
-	"github.com/coreos/etcd/clientv3"
-	versionutil "k8s.io/kubernetes/pkg/util/version"
 	"reflect"
 	"testing"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	versionutil "k8s.io/kubernetes/pkg/util/version"
 )
 
 type fakeVersionGetter struct {
@@ -60,14 +62,20 @@ func (f *fakeVersionGetter) KubeletVersions() (map[string]uint16, error) {
 	}, nil
 }
 
-type fakeEtcdCluster struct{}
+type fakeEtcdCluster struct{ TLS bool }
+
+func (f fakeEtcdCluster) HasTLS() bool { return f.TLS }
 
-func (f fakeEtcdCluster) GetEtcdClusterStatus() (*clientv3.StatusResponse, error) {
+func (f fakeEtcdCluster) GetStatus() (*clientv3.StatusResponse, error) {
 	client := &clientv3.StatusResponse{}
 	client.Version = "3.1.12"
 	return client, nil
 }
 
+func (f fakeEtcdCluster) WaitForStatus(delay time.Duration, retries int, retryInterval time.Duration) (*clientv3.StatusResponse, error) {
+	return f.GetStatus()
+}
+
 func TestGetAvailableUpgrades(t *testing.T) {
 	featureGates := make(map[string]bool)
 	tests := []struct {

cmd/kubeadm/app/util/BUILD

@@ -13,7 +13,6 @@ go_library(
         "copy.go",
         "endpoint.go",
         "error.go",
-        "etcd.go",
         "marshal.go",
         "template.go",
         "version.go",
@@ -22,7 +21,6 @@ go_library(
     deps = [
         "//cmd/kubeadm/app/apis/kubeadm:go_default_library",
         "//cmd/kubeadm/app/preflight:go_default_library",
-        "//vendor/github.com/coreos/etcd/clientv3:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
@@ -64,6 +62,7 @@ filegroup(
         "//cmd/kubeadm/app/util/audit:all-srcs",
         "//cmd/kubeadm/app/util/config:all-srcs",
         "//cmd/kubeadm/app/util/dryrun:all-srcs",
+        "//cmd/kubeadm/app/util/etcd:all-srcs",
         "//cmd/kubeadm/app/util/kubeconfig:all-srcs",
         "//cmd/kubeadm/app/util/pubkeypin:all-srcs",
         "//cmd/kubeadm/app/util/staticpod:all-srcs",

cmd/kubeadm/app/util/apiclient/BUILD

@@ -19,6 +19,7 @@ go_library(
     deps = [
         "//cmd/kubeadm/app/constants:go_default_library",
         "//cmd/kubeadm/app/util:go_default_library",
+        "//pkg/kubelet/types:go_default_library",
         "//pkg/registry/core/service/ipallocator:go_default_library",
         "//vendor/k8s.io/api/apps/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",

cmd/kubeadm/app/util/apiclient/wait.go

@@ -17,8 +17,6 @@ limitations under the License.
 package apiclient
 
 import (
-	"crypto/sha256"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -31,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/cmd/kubeadm/app/constants"
+	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 )
 
 // Waiter is an interface for waiting for criteria in Kubernetes to happen
@@ -43,11 +42,11 @@ type Waiter interface {
 	WaitForPodToDisappear(staticPodName string) error
 	// WaitForStaticPodSingleHash fetches sha256 hash for the control plane static pod
 	WaitForStaticPodSingleHash(nodeName string, component string) (string, error)
+	// WaitForStaticPodHashChange waits for the given static pod component's static pod hash to get updated.
+	// By doing that we can be sure that the kubelet has restarted the given Static Pod
+	WaitForStaticPodHashChange(nodeName, component, previousHash string) error
 	// WaitForStaticPodControlPlaneHashes fetches sha256 hashes for the control plane static pods
 	WaitForStaticPodControlPlaneHashes(nodeName string) (map[string]string, error)
-	// WaitForStaticPodControlPlaneHashChange waits for the given static pod component's static pod hash to get updated.
-	// By doing that we can be sure that the kubelet has restarted the given Static Pod
-	WaitForStaticPodControlPlaneHashChange(nodeName, component, previousHash string) error
 	// WaitForHealthyKubelet blocks until the kubelet /healthz endpoint returns 'ok'
 	WaitForHealthyKubelet(initalTimeout time.Duration, healthzEndpoint string) error
 	// SetTimeout adjusts the timeout to the specified duration
@@ -194,17 +193,17 @@ func (w *KubeWaiter) WaitForStaticPodSingleHash(nodeName string, component strin
 	return componentPodHash, err
 }
 
-// WaitForStaticPodControlPlaneHashChange blocks until it timeouts or notices that the Mirror Pod (for the Static Pod, respectively) has changed
+// WaitForStaticPodHashChange blocks until it timeouts or notices that the Mirror Pod (for the Static Pod, respectively) has changed
 // This implicitly means this function blocks until the kubelet has restarted the Static Pod in question
-func (w *KubeWaiter) WaitForStaticPodControlPlaneHashChange(nodeName, component, previousHash string) error {
+func (w *KubeWaiter) WaitForStaticPodHashChange(nodeName, component, previousHash string) error {
 	return wait.PollImmediate(constants.APICallRetryInterval, w.timeout, func() (bool, error) {
 
-		hashes, err := getStaticPodControlPlaneHashes(w.client, nodeName)
+		hash, err := getStaticPodSingleHash(w.client, nodeName, component)
 		if err != nil {
 			return false, nil
 		}
 		// We should continue polling until the UID changes
-		if hashes[component] == previousHash {
+		if hash == previousHash {
 			return false, nil
 		}
 
@@ -235,12 +234,9 @@ func getStaticPodSingleHash(client clientset.Interface, nodeName string, compone
 		return "", err
 	}
 
-	podBytes, err := json.Marshal(staticPod)
-	if err != nil {
-		return "", err
-	}
-
-	return fmt.Sprintf("%x", sha256.Sum256(podBytes)), nil
+	staticPodHash := staticPod.Annotations[kubetypes.ConfigHashAnnotationKey]
+	fmt.Printf("Static pod: %s hash: %s\n", staticPodName, staticPodHash)
+	return staticPodHash, nil
 }
 
 // TryRunCommand runs a function a maximum of failureThreshold times, and retries on error. If failureThreshold is hit; the last error is returned

cmd/kubeadm/app/util/dryrun/dryrun.go

@@ -106,8 +106,7 @@ func (w *Waiter) WaitForHealthyKubelet(_ time.Duration, healthzEndpoint string)
 // SetTimeout is a no-op; we don't wait in this implementation
 func (w *Waiter) SetTimeout(_ time.Duration) {}
 
-// WaitForStaticPodControlPlaneHashes returns an empty hash for all control plane images; WaitForStaticPodControlPlaneHashChange won't block in any case
-// but the empty strings there are needed
+// WaitForStaticPodControlPlaneHashes returns an empty hash for all control plane images;
 func (w *Waiter) WaitForStaticPodControlPlaneHashes(_ string) (map[string]string, error) {
 	return map[string]string{
 		constants.KubeAPIServer:         "",
@@ -122,7 +121,7 @@ func (w *Waiter) WaitForStaticPodSingleHash(_ string, _ string) (string, error)
 	return "", nil
 }
 
-// WaitForStaticPodControlPlaneHashChange returns a dummy nil error in order for the flow to just continue as we're dryrunning
-func (w *Waiter) WaitForStaticPodControlPlaneHashChange(_, _, _ string) error {
+// WaitForStaticPodHashChange returns a dummy nil error in order for the flow to just continue as we're dryrunning
+func (w *Waiter) WaitForStaticPodHashChange(_, _, _ string) error {
 	return nil
 }

cmd/kubeadm/app/util/etcd.go

-/*
-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 util
-
-import (
-	"context"
-	"github.com/coreos/etcd/clientv3"
-	"time"
-)
-
-// EtcdCluster is an interface to get etcd cluster related information
-type EtcdCluster interface {
-	GetEtcdClusterStatus() (*clientv3.StatusResponse, error)
-}
-
-// LocalEtcdCluster represents an instance of a local etcd cluster
-type LocalEtcdCluster struct{}
-
-// GetEtcdClusterStatus returns nil for status Up or error for status Down
-func (cluster LocalEtcdCluster) GetEtcdClusterStatus() (*clientv3.StatusResponse, error) {
-	ep := []string{"localhost:2379"}
-	cli, err := clientv3.New(clientv3.Config{
-		Endpoints:   ep,
-		DialTimeout: 5 * time.Second,
-	})
-	if err != nil {
-		return nil, err
-	}
-	defer cli.Close()
-
-	resp, err := cli.Status(context.Background(), ep[0])
-	if err != nil {
-		return nil, err
-	}
-
-	return resp, nil
-}

cmd/kubeadm/app/util/etcd/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["etcd.go"],
+    importpath = "k8s.io/kubernetes/cmd/kubeadm/app/util/etcd",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//cmd/kubeadm/app/constants:go_default_library",
+        "//cmd/kubeadm/app/util/staticpod:go_default_library",
+        "//vendor/github.com/coreos/etcd/clientv3:go_default_library",
+        "//vendor/github.com/coreos/etcd/pkg/transport:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["etcd_test.go"],
+    embed = [":go_default_library"],
+    deps = ["//cmd/kubeadm/test:go_default_library"],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

cmd/kubeadm/app/util/etcd/etcd.go

+/*
+Copyright 2018 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 etcd
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/pkg/transport"
+	"k8s.io/kubernetes/cmd/kubeadm/app/constants"
+	"k8s.io/kubernetes/cmd/kubeadm/app/util/staticpod"
+)
+
+// Client is an interface to get etcd cluster related information
+type Client interface {
+	GetStatus() (*clientv3.StatusResponse, error)
+	WaitForStatus(delay time.Duration, retries int, retryInterval time.Duration) (*clientv3.StatusResponse, error)
+	HasTLS() bool
+}
+
+// GenericClient is a common etcd client for supported etcd servers
+type GenericClient struct {
+	Endpoints []string
+	TLSConfig *tls.Config
+}
+
+// HasTLS returns true if etcd is configured for TLS
+func (c GenericClient) HasTLS() bool {
+	return c.TLSConfig != nil
+}
+
+// PodManifestsHaveTLS reads the etcd staticpod manifest from disk and returns false if the TLS flags
+// are missing from the command list. If all the flags are present it returns true.
+func PodManifestsHaveTLS(ManifestDir string) (bool, error) {
+	etcdPodPath := constants.GetStaticPodFilepath(constants.Etcd, ManifestDir)
+	etcdPod, err := staticpod.ReadStaticPodFromDisk(etcdPodPath)
+	if err != nil {
+		return false, fmt.Errorf("failed to check if etcd pod implements TLS: %v", err)
+	}
+
+	tlsFlags := []string{
+		"--cert-file=",
+		"--key-file=",
+		"--trusted-ca-file=",
+		"--client-cert-auth=",
+		"--peer-cert-file=",
+		"--peer-key-file=",
+		"--peer-trusted-ca-file=",
+		"--peer-client-cert-auth=",
+	}
+FlagLoop:
+	for _, flag := range tlsFlags {
+		for _, container := range etcdPod.Spec.Containers {
+			for _, arg := range container.Command {
+				if strings.Contains(arg, flag) {
+					continue FlagLoop
+				}
+			}
+		}
+		// flag not found in any container
+		return false, nil
+	}
+	// all flags were found in container args; pod fully implements TLS
+	return true, nil
+}
+
+// GetStatus gets server status
+func (c GenericClient) GetStatus() (*clientv3.StatusResponse, error) {
+	const dialTimeout = 5 * time.Second
+	cli, err := clientv3.New(clientv3.Config{
+		Endpoints:   c.Endpoints,
+		DialTimeout: dialTimeout,
+		TLS:         c.TLSConfig,
+	})
+	if err != nil {
+		return nil, err
+	}
+	defer cli.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	resp, err := cli.Status(ctx, c.Endpoints[0])
+	cancel()
+	if err != nil {
+		return nil, err
+	}
+
+	return resp, nil
+}
+
+// WaitForStatus returns a StatusResponse after an initial delay and retry attempts
+func (c GenericClient) WaitForStatus(delay time.Duration, retries int, retryInterval time.Duration) (*clientv3.StatusResponse, error) {
+	fmt.Printf("[util/etcd] Waiting %v for initial delay\n", delay)
+	time.Sleep(delay)
+	for i := 0; i < retries; i++ {
+		if i > 0 {
+			fmt.Printf("[util/etcd] Waiting %v until next retry\n", retryInterval)
+			time.Sleep(retryInterval)
+		}
+		fmt.Printf("[util/etcd] Attempting to get etcd status %d/%d\n", i+1, retries)
+		resp, err := c.GetStatus()
+		if err != nil {
+			switch err {
+			case context.DeadlineExceeded:
+				fmt.Println("[util/etcd] Attempt timed out")
+			default:
+				fmt.Printf("[util/etcd] Attempt failed with error: %v\n", err)
+			}
+			continue
+		}
+		return resp, nil
+	}
+	return nil, fmt.Errorf("timeout waiting for etcd cluster status")
+}
+
+// NewClient creates a new EtcdCluster client
+func NewClient(endpoints []string, caFile string, certFile string, keyFile string) (*GenericClient, error) {
+	client := GenericClient{Endpoints: endpoints}
+
+	if caFile != "" || certFile != "" || keyFile != "" {
+		tlsInfo := transport.TLSInfo{
+			CertFile:      certFile,
+			KeyFile:       keyFile,
+			TrustedCAFile: caFile,
+		}
+		tlsConfig, err := tlsInfo.ClientConfig()
+		if err != nil {
+			return nil, err
+		}
+		client.TLSConfig = tlsConfig
+	}
+
+	return &client, nil
+}
+
+// NewStaticPodClient creates a GenericClient from the given endpoints, manifestDir, and certificatesDir
+func NewStaticPodClient(endpoints []string, manifestDir string, certificatesDir string) (*GenericClient, error) {
+	hasTLS, err := PodManifestsHaveTLS(manifestDir)
+	if err != nil {
+		return nil, fmt.Errorf("could not read manifests from: %s, error: %v", manifestDir, err)
+	}
+	if hasTLS {
+		return NewClient(
+			endpoints,
+			filepath.Join(certificatesDir, constants.EtcdCACertName),
+			filepath.Join(certificatesDir, constants.EtcdHealthcheckClientCertName),
+			filepath.Join(certificatesDir, constants.EtcdHealthcheckClientKeyName),
+		)
+	}
+	return NewClient(endpoints, "", "", "")
+}

cmd/kubeadm/app/util/etcd/etcd_test.go

+/*
+Copyright 2018 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 etcd
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	testutil "k8s.io/kubernetes/cmd/kubeadm/test"
+)
+
+const (
+	secureEtcdPod = `# generated by kubeadm v1.10.0
+apiVersion: v1
+kind: Pod
+metadata:
+  annotations:
+    scheduler.alpha.kubernetes.io/critical-pod: ""
+  creationTimestamp: null
+  labels:
+    component: etcd
+    tier: control-plane
+  name: etcd
+  namespace: kube-system
+spec:
+  containers:
+  - command:
+    - etcd
+    - --advertise-client-urls=https://127.0.0.1:2379
+    - --data-dir=/var/lib/etcd
+    - --peer-key-file=/etc/kubernetes/pki/etcd/peer.key
+    - --peer-trusted-ca-file=/etc/kubernetes/pki/etcd/ca.crt
+    - --listen-client-urls=https://127.0.0.1:2379
+    - --peer-client-cert-auth=true
+    - --cert-file=/etc/kubernetes/pki/etcd/server.crt
+    - --key-file=/etc/kubernetes/pki/etcd/server.key
+    - --trusted-ca-file=/etc/kubernetes/pki/etcd/ca.crt
+    - --peer-cert-file=/etc/kubernetes/pki/etcd/peer.crt
+    - --client-cert-auth=true
+    image: k8s.gcr.io/etcd-amd64:3.1.12
+    livenessProbe:
+      exec:
+        command:
+        - /bin/sh
+        - -ec
+        - ETCDCTL_API=3 etcdctl --endpoints=127.0.0.1:2379 --cacert=/etc/kubernetes/pki/etcd/ca.crt
+          --cert=/etc/kubernetes/pki/etcd/healthcheck-client.crt --key=/etc/kubernetes/pki/etcd/healthcheck-client.key
+          get foo
+      failureThreshold: 8
+      initialDelaySeconds: 15
+      timeoutSeconds: 15
+    name: etcd
+    resources: {}
+    volumeMounts:
+    - mountPath: /var/lib/etcd
+      name: etcd-data
+    - mountPath: /etc/kubernetes/pki/etcd
+      name: etcd-certs
+  hostNetwork: true
+  volumes:
+  - hostPath:
+      path: /var/lib/etcd
+      type: DirectoryOrCreate
+    name: etcd-data
+  - hostPath:
+      path: /etc/kubernetes/pki/etcd
+      type: DirectoryOrCreate
+    name: etcd-certs
+status: {}
+`
+	insecureEtcdPod = `# generated by kubeadm v1.9.6
+apiVersion: v1
+kind: Pod
+metadata:
+  annotations:
+    scheduler.alpha.kubernetes.io/critical-pod: ""
+  creationTimestamp: null
+  labels:
+    component: etcd
+    tier: control-plane
+  name: etcd
+  namespace: kube-system
+spec:
+  containers:
+  - command:
+    - etcd
+    - --listen-client-urls=http://127.0.0.1:2379
+    - --advertise-client-urls=http://127.0.0.1:2379
+    - --data-dir=/var/lib/etcd
+    image: gcr.io/google_containers/etcd-amd64:3.1.11
+    livenessProbe:
+      failureThreshold: 8
+      httpGet:
+        host: 127.0.0.1
+        path: /health
+        port: 2379
+        scheme: HTTP
+      initialDelaySeconds: 15
+      timeoutSeconds: 15
+    name: etcd
+    resources: {}
+    volumeMounts:
+    - mountPath: /var/lib/etcd
+      name: etcd
+  hostNetwork: true
+  volumes:
+  - hostPath:
+      path: /var/lib/etcd
+      type: DirectoryOrCreate
+    name: etcd
+status: {}
+`
+	invalidPod = `---{ broken yaml @@@`
+)
+
+func TestPodManifestHasTLS(t *testing.T) {
+	tests := []struct {
+		description   string
+		podYaml       string
+		hasTLS        bool
+		expectErr     bool
+		writeManifest bool
+	}{
+		{
+			description:   "secure etcd returns true",
+			podYaml:       secureEtcdPod,
+			hasTLS:        true,
+			writeManifest: true,
+			expectErr:     false,
+		},
+		{
+			description:   "insecure etcd returns false",
+			podYaml:       insecureEtcdPod,
+			hasTLS:        false,
+			writeManifest: true,
+			expectErr:     false,
+		},
+		{
+			description:   "invalid pod fails to unmarshal",
+			podYaml:       invalidPod,
+			hasTLS:        false,
+			writeManifest: true,
+			expectErr:     true,
+		},
+		{
+			description:   "non-existent file returns error",
+			podYaml:       ``,
+			hasTLS:        false,
+			writeManifest: false,
+			expectErr:     true,
+		},
+	}
+
+	for _, rt := range tests {
+		tmpdir := testutil.SetupTempDir(t)
+		defer os.RemoveAll(tmpdir)
+
+		manifestPath := filepath.Join(tmpdir, "etcd.yaml")
+		if rt.writeManifest {
+			err := ioutil.WriteFile(manifestPath, []byte(rt.podYaml), 0644)
+			if err != nil {
+				t.Fatalf("Failed to write pod manifest\n%s\n\tfatal error: %v", rt.description, err)
+			}
+		}
+
+		hasTLS, actualErr := PodManifestsHaveTLS(tmpdir)
+		if (actualErr != nil) != rt.expectErr {
+			t.Errorf(
+				"PodManifestHasTLS failed\n%s\n\texpected error: %t\n\tgot: %t\n\tactual error: %v",
+				rt.description,
+				rt.expectErr,
+				(actualErr != nil),
+				actualErr,
+			)
+		}
+
+		if hasTLS != rt.hasTLS {
+			t.Errorf("PodManifestHasTLS failed\n%s\n\texpected hasTLS: %t\n\tgot: %t", rt.description, rt.hasTLS, hasTLS)
+		}
+	}
+}

cmd/kubeadm/app/util/marshal.go

@@ -41,3 +41,20 @@ func MarshalToYamlForCodecs(obj runtime.Object, gv schema.GroupVersion, codecs s
 	encoder := codecs.EncoderForVersion(info.Serializer, gv)
 	return runtime.Encode(encoder, obj)
 }
+
+// UnmarshalFromYaml unmarshals yaml into an object.
+func UnmarshalFromYaml(buffer []byte, gv schema.GroupVersion) (runtime.Object, error) {
+	return UnmarshalFromYamlForCodecs(buffer, gv, clientsetscheme.Codecs)
+}
+
+// UnmarshalFromYamlForCodecs unmarshals yaml into an object using the specified codec
+func UnmarshalFromYamlForCodecs(buffer []byte, gv schema.GroupVersion, codecs serializer.CodecFactory) (runtime.Object, error) {
+	mediaType := "application/yaml"
+	info, ok := runtime.SerializerInfoForMediaType(codecs.SupportedMediaTypes(), mediaType)
+	if !ok {
+		return nil, fmt.Errorf("unsupported media type %q", mediaType)
+	}
+
+	decoder := codecs.DecoderToVersion(info.Serializer, gv)
+	return runtime.Decode(decoder, buffer)
+}

cmd/kubeadm/app/util/staticpod/BUILD

@@ -14,6 +14,7 @@ go_test(
         "//cmd/kubeadm/app/apis/kubeadm:go_default_library",
         "//cmd/kubeadm/app/constants:go_default_library",
         "//cmd/kubeadm/app/features:go_default_library",
+        "//cmd/kubeadm/test:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/intstr:go_default_library",

cmd/kubeadm/app/util/staticpod/utils.go

@@ -198,6 +198,23 @@ func WriteStaticPodToDisk(componentName, manifestDir string, pod v1.Pod) error {
 	return nil
 }
 
+// ReadStaticPodFromDisk reads a static pod file from disk
+func ReadStaticPodFromDisk(manifestPath string) (*v1.Pod, error) {
+	buf, err := ioutil.ReadFile(manifestPath)
+	if err != nil {
+		return &v1.Pod{}, fmt.Errorf("failed to read manifest for %q: %v", manifestPath, err)
+	}
+
+	obj, err := util.UnmarshalFromYaml(buf, v1.SchemeGroupVersion)
+	if err != nil {
+		return &v1.Pod{}, fmt.Errorf("failed to unmarshal manifest for %q from YAML: %v", manifestPath, err)
+	}
+
+	pod := obj.(*v1.Pod)
+
+	return pod, nil
+}
+
 // GetProbeAddress returns an IP address or 127.0.0.1 to use for liveness probes
 // in static pod manifests.
 func GetProbeAddress(cfg *kubeadmapi.MasterConfiguration, componentName string) string {

cmd/kubeadm/app/util/staticpod/utils_test.go

@@ -17,6 +17,9 @@ limitations under the License.
 package staticpod
 
 import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"reflect"
 	"sort"
 	"testing"
@@ -28,6 +31,7 @@ import (
 
 	kubeadmapi "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
 	kubeadmconstants "k8s.io/kubernetes/cmd/kubeadm/app/constants"
+	testutil "k8s.io/kubernetes/cmd/kubeadm/test"
 )
 
 func TestComponentResources(t *testing.T) {
@@ -454,3 +458,73 @@ func TestGetExtraParameters(t *testing.T) {
 		}
 	}
 }
+
+const (
+	validPod = `
+apiVersion: v1
+kind: Pod
+metadata:
+  labels:
+    component: etcd
+    tier: control-plane
+  name: etcd
+  namespace: kube-system
+spec:
+  containers:
+  - image: gcr.io/google_containers/etcd-amd64:3.1.11
+status: {}
+`
+	invalidPod = `---{ broken yaml @@@`
+)
+
+func TestReadStaticPodFromDisk(t *testing.T) {
+	tests := []struct {
+		description   string
+		podYaml       string
+		expectErr     bool
+		writeManifest bool
+	}{
+		{
+			description:   "valid pod is marshaled",
+			podYaml:       validPod,
+			writeManifest: true,
+			expectErr:     false,
+		},
+		{
+			description:   "invalid pod fails to unmarshal",
+			podYaml:       invalidPod,
+			writeManifest: true,
+			expectErr:     true,
+		},
+		{
+			description:   "non-existent file returns error",
+			podYaml:       ``,
+			writeManifest: false,
+			expectErr:     true,
+		},
+	}
+
+	for _, rt := range tests {
+		tmpdir := testutil.SetupTempDir(t)
+		defer os.RemoveAll(tmpdir)
+
+		manifestPath := filepath.Join(tmpdir, "pod.yaml")
+		if rt.writeManifest {
+			err := ioutil.WriteFile(manifestPath, []byte(rt.podYaml), 0644)
+			if err != nil {
+				t.Fatalf("Failed to write pod manifest\n%s\n\tfatal error: %v", rt.description, err)
+			}
+		}
+
+		_, actualErr := ReadStaticPodFromDisk(manifestPath)
+		if (actualErr != nil) != rt.expectErr {
+			t.Errorf(
+				"ReadStaticPodFromDisk failed\n%s\n\texpected error: %t\n\tgot: %t\n\tactual error: %v",
+				rt.description,
+				rt.expectErr,
+				(actualErr != nil),
+				actualErr,
+			)
+		}
+	}
+}