kubernetes的rolling update機制解析

commit: d577db99873cbf04b8e17b78f17ec8f3a27eca30 Date: Fri Apr 10 23:45:36 2015 -0700

##0.命令行和依賴的基礎知識

Synopsis

Perform a rolling update of the given ReplicationController.

Replaces the specified controller with new controller, updating one pod at a time to use the
new PodTemplate. The new-controller.json must specify the same namespace as the
existing controller and overwrite at least one (common) label in its replicaSelector.


kubectl rolling-update OLD_CONTROLLER_NAME -f NEW_CONTROLLER_SPEC

Examples

// Update pods of frontend-v1 using new controller data in frontend-v2.json.
$ kubectl rolling-update frontend-v1 -f frontend-v2.json

// Update pods of frontend-v1 using JSON data passed into stdin.
$ cat frontend-v2.json | kubectl rolling-update frontend-v1 -f -

ReplicationController，簡稱rc，是kubernet體系中某一種類型pod的集合，rc有一個關鍵參數叫作replicas，也是就是pod的數量。

那麼rc有什麼用呢？這是爲了解決在集羣上一堆pod中有些若是掛了，那麼就在別的宿主機上把容器啓動起來，並讓業務流量導入到正確啓動的pod上。也就是說，rc保證了集羣服務的可用性，當你有不少個服務啓動在一個集羣中，你須要用程序去監控這些服務的運行情況，並動態保證服務可用。

rc和pod的對應關係是怎麼樣的？rc經過selector來選擇一些pod做爲他的控制範圍。只要pod的標籤（label）符合seletor，則屬於這個rc，下面是pod和rc的示例。

xx-controller.json

"spec":{
      "replicas":1,
      "selector":{
         "name":"redis",
         "role":"master"
      },

xx-pod.json

"labels": {
    "name": "redis"
  },

kubernetes被咱們簡稱爲k8s，若是對其中的基礎概念有興趣能夠看這篇

##1.kubctl入口

/cmd/kubectl/kubctl.go

func main() {
	runtime.GOMAXPROCS(runtime.NumCPU())
	cmd := cmd.NewKubectlCommand(cmdutil.NewFactory(nil), os.Stdin, os.Stdout, os.Stderr)
	if err := cmd.Execute(); err != nil {
		os.Exit(1)
	}
}

##2.實際調用

源代碼在pkg包內，/pkg/kubectl/cmd/cmd.go，每一個子命令都實現統一的接口，rollingupdate這行是：

cmds.AddCommand(NewCmdRollingUpdate(f, out))

這個函數的實如今：/pkg/kubectl/cmd/rollingupdate.go

func NewCmdRollingUpdate(f *cmdutil.Factory, out io.Writer) *cobra.Command {
	cmd := &cobra.Command{
		Use: "rolling-update OLD_CONTROLLER_NAME -f NEW_CONTROLLER_SPEC",
		// rollingupdate is deprecated.
		Aliases: []string{"rollingupdate"},
		Short:   "Perform a rolling update of the given ReplicationController.",
		Long:    rollingUpdate_long,
		Example: rollingUpdate_example,
		Run: func(cmd *cobra.Command, args []string) {
			err := RunRollingUpdate(f, out, cmd, args)
			cmdutil.CheckErr(err)
		},
	}
}

能夠看到實際調用時的執行函數是RunRollingUpdate，算是進入正題了

func RunRollingUpdate(f *cmdutil.Factory, out io.Writer, cmd *cobra.Command, args []string) error {
...
	mapper, typer := f.Object()
	// TODO: use resource.Builder instead
	obj, err := resource.NewBuilder(mapper, typer, f.ClientMapperForCommand()).
		NamespaceParam(cmdNamespace).RequireNamespace().
		FilenameParam(filename).
		Do().
		Object()
	if err != nil {
		return err
	}
	newRc, ok := obj.(*api.ReplicationController)
	if !ok {
		return cmdutil.UsageError(cmd, "%s does not specify a valid ReplicationController", filename)
	}

這是創建一個新的rc的代碼，其中resource是kubneter全部資源(pod,service,rc)的基類。能夠看到新的rc從json參數文件中獲取全部信息，而後轉義爲ReplicationController這個類。

if oldName == newName {
		return cmdutil.UsageError(cmd, "%s cannot have the same name as the existing ReplicationController %s",
			filename, oldName)
	}

	var hasLabel bool
	for key, oldValue := range oldRc.Spec.Selector {
		if newValue, ok := newRc.Spec.Selector[key]; ok && newValue != oldValue {
			hasLabel = true
			break
		}
	}
	if !hasLabel {
		return cmdutil.UsageError(cmd, "%s must specify a matching key with non-equal value in Selector for %s",
			filename, oldName)
	}

這裏能夠看到，對於新的rc和舊的rc，有2項限制，一個是新舊名字須要不一樣，另外一個是rc的selector中須要至少有一項的值不同。

updater := kubectl.NewRollingUpdater(newRc.Namespace, client)

	// fetch rc
	oldRc, err := client.ReplicationControllers(newRc.Namespace).Get(oldName)
	if err != nil {
		return err
	}
...
	err = updater.Update(out, oldRc, newRc, period, interval, timeout)
	if err != nil {
		return err
	}

在作rolling update的時候，有兩個條件限制，一個是新的rc的名字須要和舊的不同，第二是至少有個一個標籤的值不同。其中namespace是k8s用來作多租戶資源隔離的，能夠先忽略不計。

##3. 數據結構和實現

這段代碼出現了NewRollingUpdater，是在上一層的/pkg/kubectl/rollingupdate.go這個文件中，更加接近主體了

// RollingUpdater provides methods for updating replicated pods in a predictable,
// fault-tolerant way.
type RollingUpdater struct {
	// Client interface for creating and updating controllers
	c client.Interface
	// Namespace for resources
	ns string
}

能夠看到這裏的RollingUpdater裏面是一個k8s的client的結構來向api server發送命令

func (r *RollingUpdater) Update(out io.Writer, oldRc, newRc *api.ReplicationController, updatePeriod, interval, timeout time.Duration) error {
	oldName := oldRc.ObjectMeta.Name
	newName := newRc.ObjectMeta.Name
	retry := &RetryParams{interval, timeout}
	waitForReplicas := &RetryParams{interval, timeout}
	if newRc.Spec.Replicas <= 0 {
		return fmt.Errorf("Invalid controller spec for %s; required: > 0 replicas, actual: %s\n", newName, newRc.Spec)
	}
	desired := newRc.Spec.Replicas
	sourceId := fmt.Sprintf("%s:%s", oldName, oldRc.ObjectMeta.UID)

	// look for existing newRc, incase this update was previously started but interrupted
	rc, existing, err := r.getExistingNewRc(sourceId, newName)
	if existing {
		fmt.Fprintf(out, "Continuing update with existing controller %s.\n", newName)
		if err != nil {
			return err
		}
		replicas := rc.ObjectMeta.Annotations[desiredReplicasAnnotation]
		desired, err = strconv.Atoi(replicas)
		if err != nil {
			return fmt.Errorf("Unable to parse annotation for %s: %s=%s",
				newName, desiredReplicasAnnotation, replicas)
		}
		newRc = rc
	} else {
		fmt.Fprintf(out, "Creating %s\n", newName)
		if newRc.ObjectMeta.Annotations == nil {
			newRc.ObjectMeta.Annotations = map[string]string{}
		}
		newRc.ObjectMeta.Annotations[desiredReplicasAnnotation] = fmt.Sprintf("%d", desired)
		newRc.ObjectMeta.Annotations[sourceIdAnnotation] = sourceId
		newRc.Spec.Replicas = 0
		newRc, err = r.c.ReplicationControllers(r.ns).Create(newRc)
		if err != nil {
			return err
		}
	}

	// +1, -1 on oldRc, newRc until newRc has desired number of replicas or oldRc has 0 replicas
	for newRc.Spec.Replicas < desired && oldRc.Spec.Replicas != 0 {
		newRc.Spec.Replicas += 1
		oldRc.Spec.Replicas -= 1
		fmt.Printf("At beginning of loop: %s replicas: %d, %s replicas: %d\n",
			oldName, oldRc.Spec.Replicas,
			newName, newRc.Spec.Replicas)
		fmt.Fprintf(out, "Updating %s replicas: %d, %s replicas: %d\n",
			oldName, oldRc.Spec.Replicas,
			newName, newRc.Spec.Replicas)

		newRc, err = r.resizeAndWait(newRc, retry, waitForReplicas)
		if err != nil {
			return err
		}
		time.Sleep(updatePeriod)
		oldRc, err = r.resizeAndWait(oldRc, retry, waitForReplicas)
		if err != nil {
			return err
		}
		fmt.Printf("At end of loop: %s replicas: %d, %s replicas: %d\n",
			oldName, oldRc.Spec.Replicas,
			newName, newRc.Spec.Replicas)
	}
	// delete remaining replicas on oldRc
	if oldRc.Spec.Replicas != 0 {
		fmt.Fprintf(out, "Stopping %s replicas: %d -> %d\n",
			oldName, oldRc.Spec.Replicas, 0)
		oldRc.Spec.Replicas = 0
		oldRc, err = r.resizeAndWait(oldRc, retry, waitForReplicas)
		// oldRc, err = r.resizeAndWait(oldRc, interval, timeout)
		if err != nil {
			return err
		}
	}
	// add remaining replicas on newRc
	if newRc.Spec.Replicas != desired {
		fmt.Fprintf(out, "Resizing %s replicas: %d -> %d\n",
			newName, newRc.Spec.Replicas, desired)
		newRc.Spec.Replicas = desired
		newRc, err = r.resizeAndWait(newRc, retry, waitForReplicas)
		if err != nil {
			return err
		}
	}
	// Clean up annotations
	if newRc, err = r.c.ReplicationControllers(r.ns).Get(newName); err != nil {
		return err
	}
	delete(newRc.ObjectMeta.Annotations, sourceIdAnnotation)
	delete(newRc.ObjectMeta.Annotations, desiredReplicasAnnotation)
	newRc, err = r.updateAndWait(newRc, interval, timeout)
	if err != nil {
		return err
	}
	// delete old rc
	fmt.Fprintf(out, "Update succeeded. Deleting %s\n", oldName)
	return r.c.ReplicationControllers(r.ns).Delete(oldName)
}

這段代碼很長，但作的事情很簡單：

1. 若是新的rc沒有被建立，就先創一下，若是已經建立了（在上次的rolling_update中建立了但超時了）
2. 用幾個循環，把新的rc的replicas增長上去，舊的rc的replicas下降下來，主要調用的函數是resizeAndWait和updateAndWait

##4. 底層調用

接上一節的resizeAndWait，代碼在/pkg/kubectl/resize.go，這裏的具體代碼就不貼了 其他的全部調用都發生/pkg/client這個目錄下，這是一個http/json的client，主要功能就是向api-server發送請求 總體來講，上面的wait的實現都是比較土的，就是發一個update請求過去，後面輪詢的調用get來檢測狀態是否符合最終須要的狀態。

##5. 總結

先說一下這三個時間參數的做用：

update-period：新rc增長一個pod後，等待這個period，而後從舊rc縮減一個pod poll-interval：這個函數名來源於linux上的poll調用，就是每過一個poll-interval，向服務端發起請求，直到這個請求成功或者報失敗 timeout：總操做的超時時間

rolling update主要是客戶端這邊實現的，分析完了，但仍是有一些未知的問題，例如：

1. api-server, cadvisor, kubelet, proxy, etcd這些服務端組件是怎麼交互的？怎麼保證在服務一直可用的狀況下增減pod？
2. 是否有可能在pod增減的時候插入本身的一些代碼或者過程？由於咱們目前的架構中沒有使用k8s的proxy，須要本身去調用負載均衡的系統給這些pod導流量
3. 對於具體的pod，咱們怎麼去作內部程序的健康檢查？在業務不可用的狀況下向k8s系統發送消息，幹掉這個pod，在別的機器上建立新的來替代。