Feature: upgrade the cluster (#2340)

* Feature: upgrade the cluster

delete 'TTLAfterFinished=true',the TTLAtferFinished wouldn't be recognized after kubeadm:v1.25

upgrade k8s implementation

Signed-off-by: mond77 <1208119827@qq.com>

retry uncordon node every five seconds

add license

delele unused args

* same result if upgrade multiple times or retry after failure

modify license

* improve upgrade process and support upgrade to v1.25

Signed-off-by: mond77 <1208119827@qq.com>

* add the upgrade limit of minor version

* add func pingApiServer

* add func tryUncordonNode

improve uncordon cmd

* docs: how to upgrade cluster

docs/4.0/docs/examples/how-to-upgrade-cluster.md

@@ -0,0 +1,38 @@
+# how to upgrade cluster
+
+Just run the command '**sealos run labring/kubernetes:<new-version>**' . **Make sure that you have built the cluster before. **
+
+## For examples
+
+1. You have run the command :
+
+```sh
+sealos run labring/kubernetes:v1.24.0 labring/calico:v3.22.1 --masters 192.168.64.8 --nodes 192.168.64.7
+```
+
+2. To upgrade cluster to v1.25.0 :
+
+```sh
+sealos run labring/kubernetes:v1.25.0 
+```
+
+​	And when it comes to 'kubeadm upgrade v1.25.0', you will see :
+
+```txt
+[upgrade/version] You have chosen to change the cluster version to "v1.25.0"
+[upgrade/versions] Cluster version: v1.24.0
+[upgrade/versions] kubeadm version: v1.25.0
+[upgrade] Are you sure you want to proceed? [y/N]: 
+```
+
+​	Input 'y' so that cluster upgrade goes on. 
+
+​	If **error happen**s, you can run the command 'sealos run labring/kubernetes:v1.25.0' again. It assure the same result after failure. 
+
+## Usage
+
+1. **Upgrade cannot skip minor version**. Like 'v1.23.0 to v1.25.0' that is not allowed. If really need to upgrade from v1.23.0 to v1.25.0, you can split to two steps like 'v1.23.0 to v1.24.0' and 'v1.24.0 to v1.25.0'.
+
+2. Upgrade once succeed the cluster mount image of old version was replaced. Add masters or nodes will apply the new version.
+
+​	

pkg/apply/processor/install.go

@@ -28,8 +28,10 @@ import (
 	"github.com/labring/sealos/pkg/config"
 	"github.com/labring/sealos/pkg/filesystem"
 	"github.com/labring/sealos/pkg/guest"
+	runtime "github.com/labring/sealos/pkg/runtime"
 	v2 "github.com/labring/sealos/pkg/types/v1beta1"
 	"github.com/labring/sealos/pkg/utils/confirm"
+	"github.com/labring/sealos/pkg/utils/images"
 	"github.com/labring/sealos/pkg/utils/logger"
 	"github.com/labring/sealos/pkg/utils/rand"
 )
@@ -39,6 +41,7 @@ var ForceOverride bool
 type InstallProcessor struct {
 	ClusterFile      clusterfile.Interface
 	Buildah          buildah.Interface
+	Runtime          runtime.Interface
 	Guest            guest.Interface
 	NewMounts        []v2.MountImage
 	NewImages        []string
@@ -87,6 +90,7 @@ func (c *InstallProcessor) GetPipeLine() ([]func(cluster *v2.Cluster) error, err
 		c.RunConfig,
 		c.MountRootfs,
 		c.MirrorRegistry,
+		c.UpgradeIfNeed,
 		// i.GetPhasePluginFunc(plugin.PhasePreGuest),
 		c.RunGuest,
 		c.PostProcess,
@@ -160,6 +164,30 @@ func (c *InstallProcessor) PreProcess(cluster *v2.Cluster) error {
 			c.NewMounts = append(c.NewMounts, *mount)
 		}
 	}
+	runtime, err := runtime.NewDefaultRuntime(cluster, c.ClusterFile.GetKubeadmConfig())
+	if err != nil {
+		return fmt.Errorf("failed to init runtime, %v", err)
+	}
+	c.Runtime = runtime
+	return nil
+}
+
+func (c *InstallProcessor) UpgradeIfNeed(cluster *v2.Cluster) error {
+	logger.Info("Executing UpgradeIfNeed Pipeline in InstallProcessor")
+	for _, img := range c.NewMounts {
+		version := images.GetKubeVersionFromImage(img)
+		if version == "" {
+			continue
+		}
+		logger.Debug("try Upgrade Cluster to %s", version)
+		err := c.Runtime.UpgradeCluster(version)
+		if err != nil {
+			logger.Info("upgrade cluster failed")
+			return err
+		}
+		//upgrade success; replace the old cluster mount
+		cluster.ReplaceRootfsImage()
+	}
 	return nil
 }
 
@@ -199,6 +227,7 @@ func (c *InstallProcessor) RunConfig(cluster *v2.Cluster) error {
 }
 
 func (c *InstallProcessor) MountRootfs(cluster *v2.Cluster) error {
+	logger.Info("Executing pipeline MountRootfs in InstallProcessor.")
 	if len(c.NewMounts) == 0 {
 		return nil
 	}

pkg/apply/run.go

@@ -161,7 +161,7 @@ func (r *ClusterArgs) SetClusterRunArgs(imageList []string, args *RunArgs) error
 		}
 	}
 
-	r.cluster.Spec.Image = append(r.cluster.Spec.Image, imageList...)
+	r.cluster.SetNewImages(imageList)
 
 	// set host when cluster is not yet initialized
 	if !r.cluster.CreationTimestamp.IsZero() {

pkg/constants/data.go

@@ -48,6 +48,7 @@ const (
 	EtcDirName                       = "etc"
 	ChartsDirName                    = "charts"
 	ManifestsDirName                 = "manifests"
+	BinDirName                       = "bin"
 	RegistryDirName                  = "registry"
 	ImagesDirName                    = "images"
 	ImageShimDirName                 = "shim"
@@ -96,6 +97,7 @@ type Data interface {
 
 	RootFSCharsPath() string
 	RootFSManifestsPath() string
+	RootFSBinPath() string
 	RootFSSealctlPath() string
 }
 
@@ -126,6 +128,10 @@ func (d *data) RootFSManifestsPath() string {
 	return filepath.Join(d.RootFSPath(), ManifestsDirName)
 }
 
+func (d *data) RootFSBinPath() string {
+	return filepath.Join(d.RootFSPath(), BinDirName)
+}
+
 func (d *data) EtcPath() string {
 	return filepath.Join(ClusterDir(d.clusterName), EtcDirName)
 }

pkg/runtime/runtime.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 
 	"github.com/labring/sealos/pkg/utils/logger"
+	"github.com/labring/sealos/pkg/utils/versionutil"
 
 	v2 "github.com/labring/sealos/pkg/types/v1beta1"
 )
@@ -61,6 +62,7 @@ type Interface interface {
 	DeleteMasters(mastersIPList []string) error
 	SyncNodeIPVS(mastersIPList, nodeIPList []string) error
 	UpdateCert(certs []string) error
+	UpgradeCluster(version string) error
 }
 
 func (k *KubeadmRuntime) Reset() error {
@@ -142,3 +144,21 @@ func (k *KubeadmRuntime) Validate() error {
 	}
 	return nil
 }
+
+func (k *KubeadmRuntime) UpgradeCluster(version string) error {
+	curversion := k.getKubeVersionFromImage()
+	if curversion == version {
+		logger.Info("The cluster version has not changed")
+		return nil
+	} else if versionutil.Compare(version, curversion) {
+		if err := versionutil.UpgradeVersionLimit(curversion, version); err != nil {
+			return err
+		}
+		logger.Info("cluster vesion: %s will be upgraded into %s.", curversion, version)
+		return k.upgradeCluster(version)
+	} else if versionutil.Compare(curversion, version) {
+		logger.Info("new cluster version %s behind the current version %s", version, curversion)
+		return nil
+	}
+	return fmt.Errorf("verion format error")
+}

pkg/runtime/runtime_getter.go

@@ -46,13 +46,13 @@ func (k *KubeadmRuntime) getKubeVersion() string {
 	return k.ClusterConfiguration.KubernetesVersion
 }
 
+// old implementation doesn't consider multiple rootfs images; here get the first rootfs image
 func (k *KubeadmRuntime) getKubeVersionFromImage() string {
-	labels := k.getImageLabels()
-	image := labels[v1beta1.ImageKubeVersionKey]
-	if image == "" {
+	img := k.Cluster.GetRootfsImage("")
+	if img.Labels == nil {
 		return ""
 	}
-	return image
+	return img.Labels[v1beta1.ImageKubeVersionKey]
 }
 
 func (k *KubeadmRuntime) getMaster0IP() string {

pkg/runtime/upgrade.go

@@ -0,0 +1,201 @@
+// Copyright © 2022 sealos.
+//
+// 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 runtime
+
+import (
+	"context"
+	"fmt"
+	str "strings"
+	"time"
+
+	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/labring/sealos/pkg/client-go/kubernetes"
+	"github.com/labring/sealos/pkg/utils/logger"
+	"github.com/labring/sealos/pkg/utils/versionutil"
+)
+
+const (
+	upgradeApplyCmd = "kubeadm upgrade apply %s"
+	upradeNodeCmd   = "kubeadm upgrade node"
+	//drainNodeCmd    = "kubectl drain %s --ignore-daemonsets"
+	cordonNodeCmd   = "kubectl cordon %s"
+	uncordonNodeCmd = "kubectl uncordon %s"
+	daemonReload    = "systemctl daemon-reload"
+	restartKubelet  = "systemctl restart kubelet"
+
+	installKubeadmCmd = "cp -rf %s/kubeadm /usr/bin"
+	installKubeletCmd = "cp -rf %s/kubelet /usr/bin"
+	installKubectlCmd = "cp -rf %s/kubectl /usr/bin"
+)
+
+func (k *KubeadmRuntime) upgradeCluster(version string) error {
+	//v1.25.0 some flag unsupported
+	if versionutil.Compare(version, V1250) {
+		logger.Info("start change ClusterConfiguration up to v1.25")
+		if err := k.ChangeConfigToV125(); err != nil {
+			return err
+		}
+	}
+	//upgrade master0
+	logger.Info("start to upgrade master0")
+	err := k.upgradeMaster0(version)
+	if err != nil {
+		return err
+	}
+	//upgrade other control-planes and worker nodes
+	upgradeNodes := []string{}
+	for _, ip := range append(k.getMasterIPList(), k.getNodeIPList()...) {
+		if ip == k.getMaster0IP() {
+			continue
+		}
+		upgradeNodes = append(upgradeNodes, ip)
+	}
+	logger.Info("start to upgrade other control-planes and worker nodes")
+	return k.upgradeOtherNodes(upgradeNodes)
+}
+
+func (k *KubeadmRuntime) upgradeMaster0(version string) error {
+	master0ip := k.getMaster0IP()
+	master0Name, err := k.getRemoteInterface().Hostname(master0ip)
+	if err != nil {
+		return err
+	}
+	kubeBinaryPath := k.getContentData().RootFSBinPath()
+	//assure the connection to api-server succeed before executing upgrade cmds
+	if err = k.pingAPIServer(); err != nil {
+		return err
+	}
+	err = k.sshCmdAsync(master0ip,
+		//install kubeadm:{version} at master0
+		fmt.Sprintf(installKubeadmCmd, kubeBinaryPath),
+		//execute  kubeadm upgrade apply {version} at master0
+		fmt.Sprintf(upgradeApplyCmd, version),
+		//kubectl cordon <node-to-cordon>
+		fmt.Sprintf(cordonNodeCmd, master0Name),
+		//install kubelet:{version},kubectl{version} at master0
+		fmt.Sprintf(installKubectlCmd, kubeBinaryPath),
+		fmt.Sprintf(installKubeletCmd, kubeBinaryPath),
+		//reload kubelet daemon
+		daemonReload,
+		restartKubelet,
+	)
+	if err != nil {
+		return err
+	}
+	return k.tryUncordonNode(master0ip)
+}
+
+func (k *KubeadmRuntime) upgradeOtherNodes(ips []string) error {
+	for _, ip := range ips {
+		nodename, err := k.getRemoteInterface().Hostname(ip)
+		if err != nil {
+			return err
+		}
+		kubeBinaryPath := k.getContentData().RootFSBinPath()
+		//assure the connection to api-server succeed before executing upgrade cmds
+		if err = k.pingAPIServer(); err != nil {
+			return err
+		}
+		logger.Info("upgrade node %s", nodename)
+		err = k.sshCmdAsync(ip,
+			//install kubeadm:{version} at the node
+			fmt.Sprintf(installKubeadmCmd, kubeBinaryPath),
+			//upgrade other control-plane and nodes
+			upradeNodeCmd,
+			//kubectl cordon <node-to-cordon>
+			fmt.Sprintf(cordonNodeCmd, nodename),
+			//install kubelet:{version},kubectl{version} at the node
+			fmt.Sprintf(installKubectlCmd, kubeBinaryPath),
+			fmt.Sprintf(installKubeletCmd, kubeBinaryPath),
+			//reload kubelet daemon
+			daemonReload,
+			restartKubelet,
+		)
+		if err != nil {
+			return err
+		}
+		if err = k.tryUncordonNode(ip); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (k *KubeadmRuntime) ChangeConfigToV125() error {
+	cli, err := kubernetes.NewKubernetesClient(k.getContentData().AdminFile(), k.getMaster0IPAPIServer())
+	if err != nil {
+		logger.Info("get k8s-client failure : %s", err)
+		return err
+	}
+	KubeadmConfig, err := cli.Kubernetes().CoreV1().ConfigMaps("kube-system").Get(context.TODO(), "kubeadm-config", metaV1.GetOptions{})
+	if err != nil {
+		logger.Info("get kubeadmConfig with k8s-client failure : %s", err)
+		return err
+	}
+	ccf := KubeadmConfig.Data[ClusterConfiguration]
+	logger.Debug("get configmap data:\n%s", ccf)
+
+	ccf = str.ReplaceAll(ccf, "TTLAfterFinished=true,", "")
+	ccf = str.ReplaceAll(ccf, "\n    experimental-cluster-signing-duration: 876000h", "")
+
+	logger.Debug("update config:\n%s", ccf)
+	KubeadmConfig.Data[ClusterConfiguration] = ccf
+	_, err = cli.Kubernetes().CoreV1().ConfigMaps("kube-system").Update(context.TODO(), KubeadmConfig, metaV1.UpdateOptions{})
+	if err != nil {
+		logger.Info("update kubeadmConfig with k8s-client failure : %s", err)
+		return err
+	}
+	return nil
+}
+
+func (k *KubeadmRuntime) pingAPIServer() error {
+	cli, err := kubernetes.NewKubernetesClient(k.getContentData().AdminFile(), k.getMaster0IPAPIServer())
+	if err != nil {
+		return err
+	}
+	timeout := time.Now().Add(1 * time.Minute)
+	for {
+		_, err = cli.Kubernetes().CoreV1().Nodes().List(context.TODO(), metaV1.ListOptions{})
+		if err == nil {
+			break
+		}
+		if time.Now().After(timeout) {
+			return fmt.Errorf("restart api-server timeout within one minute")
+		}
+		time.Sleep(5 * time.Second)
+	}
+	return nil
+}
+
+func (k *KubeadmRuntime) tryUncordonNode(ip string) error {
+	nodename, err := k.getRemoteInterface().Hostname(ip)
+	if err != nil {
+		return err
+	}
+	err = k.sshCmdAsync(ip, fmt.Sprintf(uncordonNodeCmd, nodename))
+	timeout := time.Now().Add(1 * time.Minute)
+	for err != nil {
+		time.Sleep(5 * time.Second)
+		err = k.sshCmdAsync(ip, fmt.Sprintf(uncordonNodeCmd, nodename))
+		if err == nil {
+			break
+		}
+		if time.Now().After(timeout) {
+			return fmt.Errorf("try uncordon node %s timeout one minute", nodename)
+		}
+	}
+	return nil
+}

pkg/types/v1beta1/cluster_args.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/labring/sealos/pkg/utils/iputils"
 	"github.com/labring/sealos/pkg/utils/maps"
+	"github.com/labring/sealos/pkg/utils/versionutil"
 )
 
 func (c *Cluster) GetSSH() SSH {
@@ -178,6 +179,44 @@ func (c *Cluster) SetMountImage(targetMount *MountImage) {
 	}
 }
 
+func (c *Cluster) ReplaceRootfsImage() {
+	i1, i2 := -1, -1
+	var v1, v2 string
+	for i := range c.Status.Mounts {
+		img := c.Status.Mounts[i]
+		if img.Type == RootfsImage {
+			if v1 == "" {
+				v1, i1 = img.Labels[ImageKubeVersionKey], i
+			} else {
+				v2, i2 = img.Labels[ImageKubeVersionKey], i
+			}
+		}
+	}
+	//if no two rootfsImages, never replace
+	if v1 == "" || v2 == "" {
+		return
+	}
+	//if version format error, never replace
+	if versionutil.Compare(v2, v1) {
+		c.Status.Mounts[i1], c.Status.Mounts[i2] = c.Status.Mounts[i2], c.Status.Mounts[i1]
+		c.Status.Mounts = append(c.Status.Mounts[:i2], c.Status.Mounts[i2+1:]...)
+	} else if versionutil.Compare(v1, v2) {
+		c.Status.Mounts[i2], c.Status.Mounts[i1] = c.Status.Mounts[i1], c.Status.Mounts[i2]
+		c.Status.Mounts = append(c.Status.Mounts[:i1], c.Status.Mounts[i1+1:]...)
+	}
+}
+
+func (c *Cluster) SetNewImages(images []string) {
+	imageSets := map[string]struct{}{}
+	for _, img := range c.Spec.Image {
+		imageSets[img] = struct{}{}
+	}
+	for _, img := range images {
+		if _, ok := imageSets[img]; !ok {
+			c.Spec.Image = append(c.Spec.Image, img)
+		}
+	}
+}
 func (c *Cluster) GetImageLabels() map[string]string {
 	var imageLabelMap map[string]string
 	for _, img := range c.Status.Mounts {

pkg/utils/images/images.go

@@ -23,6 +23,7 @@ import (
 	"github.com/containers/image/v5/docker/reference"
 	"github.com/pkg/errors"
 
+	"github.com/labring/sealos/pkg/types/v1beta1"
 	"github.com/labring/sealos/pkg/utils/file"
 	"github.com/labring/sealos/pkg/utils/logger"
 	str "github.com/labring/sealos/pkg/utils/strings"
@@ -129,3 +130,13 @@ func normalizeTaggedDigestedNamed(named reference.Named) (reference.Named, error
 func toLocalImageName(name string) string {
 	return "localhost/" + strings.TrimLeft(name, "/")
 }
+
+func GetKubeVersionFromImage(img v1beta1.MountImage) string {
+	if img.Type != v1beta1.RootfsImage {
+		return ""
+	}
+	if img.Labels == nil {
+		return ""
+	}
+	return img.Labels[v1beta1.ImageKubeVersionKey]
+}

pkg/utils/versionutil/version.go

@@ -17,6 +17,8 @@ limitations under the License.
 package versionutil
 
 import (
+	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/labring/sealos/pkg/utils/logger"
@@ -46,8 +48,34 @@ func Compare(v1, v2 string) bool {
 	} else if v1List[1] < v2List[1] {
 		return false
 	}
-	if v1List[2] > v2List[2] {
+	if v1List[2] >= v2List[2] {
 		return true
 	}
-	return true
+	return false
+}
+
+// assure version format right and new >=
+// The upgrade of minor version number cannot be skipped
+func UpgradeVersionLimit(old, new string) error {
+	new = strings.Replace(new, "v", "", -1)
+	old = strings.Replace(old, "v", "", -1)
+	new = strings.Split(new, "-")[0]
+	old = strings.Split(old, "-")[0]
+	newList := strings.Split(new, ".")
+	oldList := strings.Split(old, ".")
+
+	minorNewV, err := strconv.Atoi(newList[1])
+	if err != nil {
+		return err
+	}
+	minorOldV, err := strconv.Atoi(oldList[1])
+	if err != nil {
+		return err
+	}
+	if newList[0] > oldList[0] {
+		return fmt.Errorf("upgrade of senior version cannot be executed")
+	} else if minorNewV > minorOldV+1 {
+		return fmt.Errorf("upgrade of minor version number cannot be skipped")
+	}
+	return nil
 }