Merge pull request #55010 from sbezverk/kubeadm_etcd_upgrade_apply

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

@@ -46,6 +46,7 @@ type applyFlags struct {
 	nonInteractiveMode bool
 	force              bool
 	dryRun             bool
+	etcdUpgrade        bool
 	newK8sVersionStr   string
 	newK8sVersion      *version.Version
 	imagePullTimeout   time.Duration
@@ -62,6 +63,7 @@ func NewCmdApply(parentFlags *cmdUpgradeFlags) *cobra.Command {
 	flags := &applyFlags{
 		parent:           parentFlags,
 		imagePullTimeout: 15 * time.Minute,
+		etcdUpgrade:      false,
 	}
 
 	cmd := &cobra.Command{
@@ -91,6 +93,7 @@ func NewCmdApply(parentFlags *cmdUpgradeFlags) *cobra.Command {
 	cmd.Flags().BoolVarP(&flags.nonInteractiveMode, "yes", "y", flags.nonInteractiveMode, "Perform the upgrade and do not prompt for confirmation (non-interactive mode).")
 	cmd.Flags().BoolVarP(&flags.force, "force", "f", flags.force, "Force upgrading although some requirements might not be met. This also implies non-interactive mode.")
 	cmd.Flags().BoolVar(&flags.dryRun, "dry-run", flags.dryRun, "Do not change any state, just output what actions would be performed.")
+	cmd.Flags().BoolVar(&flags.etcdUpgrade, "etcd-upgrade", flags.etcdUpgrade, "Perform the upgrade of etcd.")
 	cmd.Flags().DurationVar(&flags.imagePullTimeout, "image-pull-timeout", flags.imagePullTimeout, "The maximum amount of time to wait for the control plane pods to be downloaded.")
 
 	return cmd
@@ -231,17 +234,18 @@ func PerformControlPlaneUpgrade(flags *applyFlags, client clientset.Interface, w
 	if flags.dryRun {
 		return DryRunStaticPodUpgrade(internalcfg)
 	}
-	return PerformStaticPodUpgrade(client, waiter, internalcfg)
+
+	return PerformStaticPodUpgrade(client, waiter, internalcfg, flags.etcdUpgrade)
 }
 
 // PerformStaticPodUpgrade performs the upgrade of the control plane components for a static pod hosted cluster
-func PerformStaticPodUpgrade(client clientset.Interface, waiter apiclient.Waiter, internalcfg *kubeadmapi.MasterConfiguration) error {
+func PerformStaticPodUpgrade(client clientset.Interface, waiter apiclient.Waiter, internalcfg *kubeadmapi.MasterConfiguration, etcdUpgrade bool) error {
 	pathManager, err := upgrade.NewKubeStaticPodPathManagerUsingTempDirs(constants.GetStaticPodDirectory())
 	if err != nil {
 		return err
 	}
 
-	return upgrade.StaticPodControlPlane(waiter, pathManager, internalcfg)
+	return upgrade.StaticPodControlPlane(waiter, pathManager, internalcfg, etcdUpgrade)
 }
 
 // DryRunStaticPodUpgrade fakes an upgrade of the control plane

cmd/kubeadm/app/constants/BUILD

@@ -35,4 +35,5 @@ go_test(
     srcs = ["constants_test.go"],
     importpath = "k8s.io/kubernetes/cmd/kubeadm/app/constants",
     library = ":go_default_library",
+    deps = ["//pkg/util/version:go_default_library"],
 )

cmd/kubeadm/app/constants/constants.go

@@ -223,8 +223,32 @@ var (
 
 	// MinimumKubeletVersion specifies the minimum version of kubelet which kubeadm supports
 	MinimumKubeletVersion = version.MustParseSemantic("v1.8.0")
+
+	// SupportedEtcdVersion lists officially supported etcd versions with corresponding kubernetes releases
+	SupportedEtcdVersion = map[uint8]string{
+		8: "3.0.17",
+		9: "3.1.10",
+	}
 )
 
+// EtcdSupportedVersion returns officially supported version of etcd for a specific kubernetes release
+// if passed version is not listed, the function returns nil and an error
+func EtcdSupportedVersion(versionString string) (*version.Version, error) {
+	kubernetesVersion, err := version.ParseSemantic(versionString)
+	if err != nil {
+		return nil, err
+	}
+
+	if etcdStringVersion, ok := SupportedEtcdVersion[uint8(kubernetesVersion.Minor())]; ok {
+		etcdVersion, err := version.ParseSemantic(etcdStringVersion)
+		if err != nil {
+			return nil, err
+		}
+		return etcdVersion, nil
+	}
+	return nil, fmt.Errorf("Unsupported or unknown kubernetes version")
+}
+
 // GetStaticPodDirectory returns the location on the disk where the Static Pod should be present
 func GetStaticPodDirectory() string {
 	return filepath.Join(KubernetesDir, ManifestsSubDirName)

cmd/kubeadm/app/constants/constants_test.go

@@ -17,6 +17,9 @@ limitations under the License.
 package constants
 
 import (
+	"fmt"
+	"k8s.io/kubernetes/pkg/util/version"
+	"strings"
 	"testing"
 )
 
@@ -110,3 +113,58 @@ func TestAddSelfHostedPrefix(t *testing.T) {
 		}
 	}
 }
+
+func TestEtcdSupportedVersion(t *testing.T) {
+	var tests = []struct {
+		kubernetesVersion string
+		expectedVersion   *version.Version
+		expectedError     error
+	}{
+		{
+			kubernetesVersion: "1.8.0",
+			expectedVersion:   version.MustParseSemantic("3.0.17"),
+			expectedError:     nil,
+		},
+		{
+			kubernetesVersion: "1.80.0",
+			expectedVersion:   nil,
+			expectedError:     fmt.Errorf("Unsupported or unknown kubernetes version"),
+		},
+		{
+			kubernetesVersion: "1.9.0",
+			expectedVersion:   version.MustParseSemantic("3.1.10"),
+			expectedError:     nil,
+		},
+		{
+			kubernetesVersion: "1.10.0",
+			expectedVersion:   nil,
+			expectedError:     fmt.Errorf("Unsupported or unknown kubernetes version"),
+		},
+		{
+			kubernetesVersion: "1.8.6",
+			expectedVersion:   version.MustParseSemantic("3.0.17"),
+			expectedError:     nil,
+		},
+	}
+	for _, rt := range tests {
+		actualVersion, actualError := EtcdSupportedVersion(rt.kubernetesVersion)
+		if actualError != nil {
+			if actualError.Error() != rt.expectedError.Error() {
+				t.Errorf(
+					"failed EtcdSupportedVersion:\n\texpected error: %v\n\t  actual error: %v",
+					rt.expectedError,
+					actualError,
+				)
+			}
+
+		} else {
+			if strings.Compare(actualVersion.String(), rt.expectedVersion.String()) != 0 {
+				t.Errorf(
+					"failed EtcdSupportedVersion:\n\texpected version: %s\n\t  actual version: %s",
+					rt.expectedVersion.String(),
+					actualVersion.String(),
+				)
+			}
+		}
+	}
+}

cmd/kubeadm/app/images/images.go

@@ -30,8 +30,13 @@ func GetCoreImage(image, repoPrefix, k8sVersion, overrideImage string) string {
 		return overrideImage
 	}
 	kubernetesImageTag := kubeadmutil.KubernetesVersionToImageTag(k8sVersion)
+	etcdImageTag := constants.DefaultEtcdVersion
+	etcdImageVersion, err := constants.EtcdSupportedVersion(k8sVersion)
+	if err == nil {
+		etcdImageTag = etcdImageVersion.String()
+	}
 	return map[string]string{
-		constants.Etcd:                  fmt.Sprintf("%s/%s-%s:%s", repoPrefix, "etcd", runtime.GOARCH, constants.DefaultEtcdVersion),
+		constants.Etcd:                  fmt.Sprintf("%s/%s-%s:%s", repoPrefix, "etcd", runtime.GOARCH, etcdImageTag),
 		constants.KubeAPIServer:         fmt.Sprintf("%s/%s-%s:%s", repoPrefix, "kube-apiserver", runtime.GOARCH, kubernetesImageTag),
 		constants.KubeControllerManager: fmt.Sprintf("%s/%s-%s:%s", repoPrefix, "kube-controller-manager", runtime.GOARCH, kubernetesImageTag),
 		constants.KubeScheduler:         fmt.Sprintf("%s/%s-%s:%s", repoPrefix, "kube-scheduler", runtime.GOARCH, kubernetesImageTag),

cmd/kubeadm/app/phases/etcd/local.go

@@ -36,7 +36,6 @@ func CreateLocalEtcdStaticPodManifestFile(manifestDir string, cfg *kubeadmapi.Ma
 
 	// gets etcd StaticPodSpec, actualized for the current MasterConfiguration
 	spec := GetEtcdPodSpec(cfg)
-
 	// writes etcd StaticPod to disk
 	if err := staticpodutil.WriteStaticPodToDisk(kubeadmconstants.Etcd, manifestDir, spec); err != nil {
 		return err
@@ -56,7 +55,7 @@ func GetEtcdPodSpec(cfg *kubeadmapi.MasterConfiguration) v1.Pod {
 	return staticpodutil.ComponentPod(v1.Container{
 		Name:    kubeadmconstants.Etcd,
 		Command: getEtcdCommand(cfg),
-		Image:   images.GetCoreImage(kubeadmconstants.Etcd, cfg.ImageRepository, "", cfg.Etcd.Image),
+		Image:   images.GetCoreImage(kubeadmconstants.Etcd, cfg.ImageRepository, cfg.KubernetesVersion, cfg.Etcd.Image),
 		// Mount the etcd datadir path read-write so etcd can store data in a more persistent manner
 		VolumeMounts:  []v1.VolumeMount{staticpodutil.NewVolumeMount(etcdVolumeName, cfg.Etcd.DataDir, false)},
 		LivenessProbe: staticpodutil.ComponentProbe(cfg, kubeadmconstants.Etcd, 2379, "/health", v1.URISchemeHTTP),

cmd/kubeadm/app/phases/upgrade/BUILD

@@ -26,6 +26,7 @@ go_library(
         "//cmd/kubeadm/app/phases/bootstraptoken/clusterinfo:go_default_library",
         "//cmd/kubeadm/app/phases/bootstraptoken/node:go_default_library",
         "//cmd/kubeadm/app/phases/controlplane:go_default_library",
+        "//cmd/kubeadm/app/phases/etcd:go_default_library",
         "//cmd/kubeadm/app/phases/selfhosting:go_default_library",
         "//cmd/kubeadm/app/phases/uploadconfig:go_default_library",
         "//cmd/kubeadm/app/util:go_default_library",
@@ -73,6 +74,7 @@ go_test(
         "//cmd/kubeadm/app/apis/kubeadm/v1alpha1:go_default_library",
         "//cmd/kubeadm/app/constants:go_default_library",
         "//cmd/kubeadm/app/phases/controlplane:go_default_library",
+        "//cmd/kubeadm/app/phases/etcd:go_default_library",
         "//cmd/kubeadm/app/util/apiclient:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/util/version:go_default_library",

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

@@ -30,6 +30,7 @@ import (
 	kubeadmapiext "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1alpha1"
 	"k8s.io/kubernetes/cmd/kubeadm/app/constants"
 	"k8s.io/kubernetes/cmd/kubeadm/app/phases/controlplane"
+	etcdphase "k8s.io/kubernetes/cmd/kubeadm/app/phases/etcd"
 	"k8s.io/kubernetes/cmd/kubeadm/app/util/apiclient"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 )
@@ -108,6 +109,11 @@ func (w *fakeWaiter) WaitForStaticPodControlPlaneHashes(_ string) (map[string]st
 	return map[string]string{}, w.errsToReturn[waitForHashes]
 }
 
+// WaitForStaticPodSingleHash returns an error if set from errsToReturn
+func (w *fakeWaiter) WaitForStaticPodSingleHash(_ string, _ string) (string, error) {
+	return "", w.errsToReturn[waitForHashes]
+}
+
 // WaitForStaticPodControlPlaneHashChange returns an error if set from errsToReturn
 func (w *fakeWaiter) WaitForStaticPodControlPlaneHashChange(_, _, _ string) error {
 	return w.errsToReturn[waitForHashChange]
@@ -122,6 +128,7 @@ type fakeStaticPodPathManager struct {
 	realManifestDir   string
 	tempManifestDir   string
 	backupManifestDir string
+	backupEtcdDir     string
 	MoveFileFunc      func(string, string) error
 }
 
@@ -140,11 +147,16 @@ func NewFakeStaticPodPathManager(moveFileFunc func(string, string) error) (Stati
 	if err != nil {
 		return nil, fmt.Errorf("couldn't create a temporary directory for the upgrade: %v", err)
 	}
+	backupEtcdDir, err := ioutil.TempDir("", "kubeadm-backup-etcd")
+	if err != nil {
+		return nil, err
+	}
 
 	return &fakeStaticPodPathManager{
 		realManifestDir:   realManifestsDir,
 		tempManifestDir:   upgradedManifestsDir,
 		backupManifestDir: backupManifestsDir,
+		backupEtcdDir:     backupEtcdDir,
 		MoveFileFunc:      moveFileFunc,
 	}, nil
 }
@@ -174,6 +186,10 @@ func (spm *fakeStaticPodPathManager) BackupManifestDir() string {
 	return spm.backupManifestDir
 }
 
+func (spm *fakeStaticPodPathManager) BackupEtcdDir() string {
+	return spm.backupEtcdDir
+}
+
 func TestStaticPodControlPlane(t *testing.T) {
 	tests := []struct {
 		waitErrsToReturn     map[string]error
@@ -280,7 +296,6 @@ func TestStaticPodControlPlane(t *testing.T) {
 	}
 
 	for _, rt := range tests {
-
 		waiter := NewFakeStaticPodWaiter(rt.waitErrsToReturn)
 		pathMgr, err := NewFakeStaticPodPathManager(rt.moveFileFunc)
 		if err != nil {
@@ -299,6 +314,10 @@ func TestStaticPodControlPlane(t *testing.T) {
 		if err != nil {
 			t.Fatalf("couldn't run CreateInitStaticPodManifestFiles: %v", err)
 		}
+		err = etcdphase.CreateLocalEtcdStaticPodManifestFile(pathMgr.RealManifestDir(), oldcfg)
+		if err != nil {
+			t.Fatalf("couldn't run CreateLocalEtcdStaticPodManifestFile: %v", err)
+		}
 		// Get a hash of the v1.7 API server manifest to compare later (was the file re-written)
 		oldHash, err := getAPIServerHash(pathMgr.RealManifestDir())
 		if err != nil {
@@ -310,7 +329,7 @@ func TestStaticPodControlPlane(t *testing.T) {
 			t.Fatalf("couldn't create config: %v", err)
 		}
 
-		actualErr := StaticPodControlPlane(waiter, pathMgr, newcfg)
+		actualErr := StaticPodControlPlane(waiter, pathMgr, newcfg, false)
 		if (actualErr != nil) != rt.expectedErr {
 			t.Errorf(
 				"failed UpgradeStaticPodControlPlane\n\texpected error: %t\n\tgot: %t",

cmd/kubeadm/app/util/BUILD

@@ -10,8 +10,10 @@ go_library(
     name = "go_default_library",
     srcs = [
         "arguments.go",
+        "copy.go",
         "endpoint.go",
         "error.go",
+        "etcd.go",
         "marshal.go",
         "template.go",
         "version.go",
@@ -20,6 +22,7 @@ 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",

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

@@ -40,6 +40,8 @@ type Waiter interface {
 	WaitForPodsWithLabel(kvLabel string) error
 	// WaitForPodToDisappear waits for the given Pod in the kube-system namespace to be deleted
 	WaitForPodToDisappear(staticPodName string) error
+	// WaitForStaticPodSingleHash fetches sha256 hash for the control plane static pod
+	WaitForStaticPodSingleHash(nodeName string, component string) (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.
@@ -154,17 +156,40 @@ func (w *KubeWaiter) SetTimeout(timeout time.Duration) {
 // WaitForStaticPodControlPlaneHashes blocks until it timeouts or gets a hash map for all components and their Static Pods
 func (w *KubeWaiter) WaitForStaticPodControlPlaneHashes(nodeName string) (map[string]string, error) {
 
-	var mirrorPodHashes map[string]string
-	err := wait.PollImmediate(constants.APICallRetryInterval, w.timeout, func() (bool, error) {
+	componentHash := ""
+	var err error
+	mirrorPodHashes := map[string]string{}
+	for _, component := range constants.MasterComponents {
+		err = wait.PollImmediate(constants.APICallRetryInterval, w.timeout, func() (bool, error) {
+			componentHash, err = getStaticPodSingleHash(w.client, nodeName, component)
+			if err != nil {
+				return false, nil
+			}
+			return true, nil
+		})
+		if err != nil {
+			return nil, err
+		}
+		mirrorPodHashes[component] = componentHash
+	}
 
-		hashes, err := getStaticPodControlPlaneHashes(w.client, nodeName)
+	return mirrorPodHashes, nil
+}
+
+// WaitForStaticPodSingleHash blocks until it timeouts or gets a hash for a single component and its Static Pod
+func (w *KubeWaiter) WaitForStaticPodSingleHash(nodeName string, component string) (string, error) {
+
+	componentPodHash := ""
+	var err error
+	err = wait.PollImmediate(constants.APICallRetryInterval, w.timeout, func() (bool, error) {
+		componentPodHash, err = getStaticPodSingleHash(w.client, nodeName, component)
 		if err != nil {
 			return false, nil
 		}
-		mirrorPodHashes = hashes
 		return true, nil
 	})
-	return mirrorPodHashes, err
+
+	return componentPodHash, err
 }
 
 // WaitForStaticPodControlPlaneHashChange blocks until it timeouts or notices that the Mirror Pod (for the Static Pod, respectively) has changed
@@ -190,20 +215,30 @@ func getStaticPodControlPlaneHashes(client clientset.Interface, nodeName string)
 
 	mirrorPodHashes := map[string]string{}
 	for _, component := range constants.MasterComponents {
-		staticPodName := fmt.Sprintf("%s-%s", component, nodeName)
-		staticPod, err := client.CoreV1().Pods(metav1.NamespaceSystem).Get(staticPodName, metav1.GetOptions{})
+		hash, err := getStaticPodSingleHash(client, nodeName, component)
 		if err != nil {
 			return nil, err
 		}
+		mirrorPodHashes[component] = hash
+	}
+	return mirrorPodHashes, nil
+}
 
-		podBytes, err := json.Marshal(staticPod)
-		if err != nil {
-			return nil, err
-		}
+// getStaticSinglePodHash computes hashes for a single Static Pod resource
+func getStaticPodSingleHash(client clientset.Interface, nodeName string, component string) (string, error) {
 
-		mirrorPodHashes[component] = fmt.Sprintf("%x", sha256.Sum256(podBytes))
+	staticPodName := fmt.Sprintf("%s-%s", component, nodeName)
+	staticPod, err := client.CoreV1().Pods(metav1.NamespaceSystem).Get(staticPodName, metav1.GetOptions{})
+	if err != nil {
+		return "", err
 	}
-	return mirrorPodHashes, nil
+
+	podBytes, err := json.Marshal(staticPod)
+	if err != nil {
+		return "", err
+	}
+
+	return fmt.Sprintf("%x", sha256.Sum256(podBytes)), 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/copy.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 (
+	"os/exec"
+)
+
+// CopyDir copies the content of a folder
+func CopyDir(src string, dst string) error {
+	cmd := exec.Command("cp", "-r", src, dst)
+	err := cmd.Run()
+	if err != nil {
+		return err
+	}
+	return nil
+}

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

@@ -116,6 +116,12 @@ func (w *Waiter) WaitForStaticPodControlPlaneHashes(_ string) (map[string]string
 	}, nil
 }
 
+// WaitForStaticPodSingleHash returns an empty hash
+// but the empty strings there are needed
+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 {
 	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"
+)
+
+// GetEtcdClusterStatus returns nil for status Up or error for status Down
+func 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
+}