Kubernetes源码解析之controller-manager deployment同步流程

apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: nginx
  
spec:
  replicas: 3
    template:
      metadata:
        labels:
          app: nginx
      spec:
        containers:
        - name: nginx
         image: nginx:1.9.1

//直接使用 kubectl set 更新对象
$ kubectl set image deployment/nginx  nginx=nginx:1.9.1
//直接更新nginx yaml文件
$ kubectl edit deployment/nginx

$ kubectl rollout pause deployment/nginx  //暂停
$ kubectl rollout resume  deployment/nginx  //恢复

func (dc *DeploymentController) syncDeployment(key string) error {
	//由key值获取Dp的namespaces和name
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	
	//根据ns、name从系统中获取deployment当前信息(此时有可能已被delete，在处理同步中会不断检查删除状态)
	deployment, err := dc.dLister.Deployments(namespace).Get(name)
	
	//deepcopy信息，更新状态时更新拷贝信息然后将副本更新到server
	d := deployment.DeepCopy()

	//行为：获取属于当前Dp的所有Rs，同时进行一些adopt和release操作
	rsList, err := dc.getReplicaSetsForDeployment(d)
	
	//获取rs列表下的所有pod，返回Map(key为Rs.UID  value为PodList)
	podMap, err := dc.getPodMapForDeployment(d, rsList)
	
	//如果Pod已经被delete，调用getAllReplicaSetsAndSyncRevision更新版本信息，并同步状态信息。不明白这里，为什么已删除还要同步状态
	if d.DeletionTimestamp != nil {
		return dc.syncStatusOnly(d, rsList)
	}

	//检查是否为暂停或恢复事件。
	//暂停时将condition中Progressing中 status=Unknown reason=DpPaused，此时不对其进行处理超时等检查
	//检查为恢复请求并且当前为暂停时，更新Progressing为 status=Unknown reason=DpResume
	if err = dc.checkPausedConditions(d)

	//暂停态时，执行sync同步状态(本节会单独分析函数)
	if d.Spec.Paused {
		return dc.sync(d, rsList)
	}

	//检查有回滚事件时，回滚版本(下节会分析此函数)
	if getRollbackTo(d) != nil {
		return dc.rollback(d, rsList)
	}

	//发现desire与dp.replicas不符时，确定为正在进行扩缩容事件，调用sync同步
	scalingEvent, err := dc.isScalingEvent(d, rsList)
	if scalingEvent {
		return dc.sync(d, rsList)
	}

	//根据两种发布策略检查并更新deployment到最新状态(下节会分析处理函数)
	switch d.Spec.Strategy.Type {
	case apps.RecreateDeploymentStrategyType:
		return dc.rolloutRecreate(d, rsList, podMap)
	case apps.RollingUpdateDeploymentStrategyType:
		return dc.rolloutRolling(d, rsList)
	}
}

status := apps.DeploymentStatus{
		Replicas:    //Dp下所有Rs的期望实例数
		UpdatedReplicas:    //新版本Rs期望实例数(不论running与否)
		ReadyReplicas:      //Dp下所有Rs拥有的ReadyPod数 
		AvailableReplicas:   ///Dp下所有Rs拥有的AvaliblePod数
		UnavailableReplicas: //Dp下所有Rs拥有的unavailablePod数
}
//更新condition中type=AvaliblePod状态,下面是avaliblePod个数是否达标时设置的conditon
deploymentutil.NewDeploymentCondition("Available", "True", "MinimumReplicasAvailable", "Deployment has minimum availability.")
deploymentutil.NewDeploymentCondition("Available", "False", "MinimumReplicasAvailable", "Deployment does not have minimum availability")

replicas: 5 				#deployment期望实例数
strategy:			 		#升级策略提示符 位于yaml中 .spec下
	rollingUpdate:   
		maxSurge: 1        	#更新中允许存在的最大pod数
		maxUnavailable: 1  	#更新中允许存在的最大不可用pod数, dp.replicas-maxUnava为最小可用数
	type: RollingUpdate  	#升级策略

更新过程：
	对5b69版本scaleUp 1个pod  
    对85bb版本scaleDown 1个pod
    对5b69版本scaleUp 1个pod
    ！对85bb版本scaleDown 1个pod  此时因为maxUnavaluble限制，此版本不能再缩容，又因为maxSurge限制，新版本pod不能再发起，由此stuck在上图状态

//升级过程中发生的scale up/down，实际上是操作Rs.replicas
func (dc *DeploymentController) rolloutRolling(d *apps.Deployment, rsList []*apps.ReplicaSet) error {
	//从rslist中获取newRs、oldRs,并更新revision信息。参数为true时，如果不存在newRs就创建它，检查存在newRs与否的标准是：rs.template = dp.spec.template 
	newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(d, rsList, true)
	allRSs := append(oldRSs, newRS)

	//调整newRs-yaml信息，通常为扩容事件。如果操作了rs.replicas, scaledUp=true.
	scaledUp, err := dc.reconcileNewReplicaSet(allRSs, newRS, d)
	if scaledUp {
		return dc.syncRolloutStatus(allRSs, newRS, d)
	}
	//调整oldRs-yaml信息，通常为缩容事件。如果操作了rs.replicas,scaledDown=false
	scaledDown, err := dc.reconcileOldReplicaSets(allRSs, controller.FilterActiveReplicaSets(oldRSs), newRS, d)
	if scaledDown {
		return dc.syncRolloutStatus(allRSs, newRS, d)
	}
	//清理deployment下Rs版本信息
	if deploymentutil.DeploymentComplete(d, &d.Status) {
		if err := dc.cleanupDeployment(oldRSs, d); err != nil {
			return err
		}
	}
	//同步rs状态
	return dc.syncRolloutStatus(allRSs, newRS, d)
}

//最大可清理旧版本pod数量，缩容时，RS下列表pod是经过排序的，保证优先清理Unhealthy Pod
maxCleanCount = allPodsCount - minAvailable - newRSUnavailablePodCount  
//遍历每个ActiveoldRs
totalScaledDown=0    //记录已清理副本数，不能超过maxCleanCount值
for _, targetRs := range oldRsList {
	scaledDownCount := Min(maxCleanCount - totalScaledDown, targetRS.Spec.Replicas-targetRS.Status.AvailableReplicas)   //当前Rs缩容数
	//向server发送缩容请求
	totalScaledDown+=scaledDownCount 
}

totalScaleDownCount := availablePodCount - minAvailable
totalScaledDown := int32(0)  //记录本次总scaleDown数目，不能超过totalScaleDownCount值
for _, targetRs := range oldRsList {
	scaleDownCount := Min(targetRS.Spec.Replicas, totalScaleDownCount-totalScaledDown)
	totalScaledDown += scaleDownCount
}

func (dc *DeploymentController) rolloutRecreate(d *apps.Deployment, rsList []*apps.ReplicaSet, podMap map[types.UID]*v1.PodList) error {
	//getRs并更新版本信息，false参数表示如果没有新版本则不创建
	newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(d, rsList, false)
	if err != nil {
		return err
	}
	allRSs := append(oldRSs, newRS)
	activeOldRSs := controller.FilterActiveReplicaSets(oldRSs)

	//缩容所有active的旧实例(其下有pod即为active)，如果有缩容操作，则更新状态
	scaledDown, err := dc.scaleDownOldReplicaSetsForRecreate(activeOldRSs, d)
	if scaledDown {
		return dc.syncRolloutStatus(allRSs, newRS, d)
	}

	//确认Rs下已经没有 running/pending/unknown 状态的pod
	if oldPodsRunning(newRS, oldRSs, podMap) {
		return dc.syncRolloutStatus(allRSs, newRS, d)
	}

	//创建newRs并扩容
	if newRS == nil {
		newRS, oldRSs, err = dc.getAllReplicaSetsAndSyncRevision(d, rsList, true)
		allRSs = append(oldRSs, newRS)
	}
	dc.scaleUpNewReplicaSetForRecreate(newRS, d)

	if util.DeploymentComplete(d, &d.Status) {
		if err := dc.cleanupDeployment(oldRSs, d); err != nil {
			return err
		}
	}
	return dc.syncRolloutStatus(allRSs, newRS, d)

func (dc *DeploymentController) rollback(d *apps.Deployment, rsList []*apps.ReplicaSet) error {
	//getRs并更新版本号
	newRS, allOldRSs, err := dc.getAllReplicaSetsAndSyncRevision(d, rsList, true)
	allRSs := append(allOldRSs, newRS)

	rollbackTo := getRollbackTo(d)
	if rollbackTo.Revision == 0 {
		//如果未指定Revision,遍历rs找到第二max的revision
		rollbackTo.Revision = deploymentutil.LastRevision(allRSs)}
	//根据revision遍历RsList找到对应rs,将rs.spec.template复制给dp.spec.template，并更新此版本为最新版，deployment-controller将在下次调用getAllReplicaSetsAndSyncRevision时创建newRs	
	for _, rs := range allRSs {
		v, err := deploymentutil.Revision(rs)
		if v == rollbackTo.Revision {
			performedRollback, err := dc.rollbackToTemplate(d, rs)
			return err
		}
	}
	//清理一些anntition相关的信息
	return dc.updateDeploymentAndClearRollbackTo(d)
}

func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
	startTime := time.Now()
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	
	rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
	//rs是否需要同步
	rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
	
	//获取Rs所在ns下所有Pod
	//过滤非success和已删除pod
	//获取ns下所有pod,获取过程参考Dp获取Rs
	selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
	allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
	filteredPods := controller.FilterActivePods(allPods)
	filteredPods, err = rsc.claimPods(rs, selector, filteredPods)
	
	var manageReplicasErr error
	//管理此rs下pod, 增删pod(具体实现下面讲?)
	if rsNeedsSync && rs.DeletionTimestamp == nil {
		manageReplicasErr = rsc.manageReplicas(filteredPods, rs)
	}
	
	//返回rs下关于pod的running/avalidble等状态数量汇总和status相关的东西
	rs = rs.DeepCopy()
	newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)
	updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)

	//状态还需要同步时，加入队列
	if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
		updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
		updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
		rsc.enqueueReplicaSetAfter(updatedRS, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
	}
	return manageReplicasErr
}

func (rsc *ReplicaSetController) manageReplicas(filteredPods []*v1.Pod, rs *apps.ReplicaSet) error {
	diff := len(filteredPods) - int(*(rs.Spec.Replicas))
	rsKey, err := controller.KeyFunc(rs)
	
	//avaliblePod数未达到期望值，需要扩容。 burstReplicas为单次创建Pod限制
	if diff < 0 {
		diff *= -1
		if diff > rsc.burstReplicas {
			diff = rsc.burstReplicas
		}
		//批量创建Pod,批量数字从1开始double增加，这样可以防止出现相同错误的pod大量失败的情况
		//例如，一个尝试创建大量Pod的低quota的任务将在第一个Pod创建失败时被停止，返回成功创建数量
		//successfulCreations表示成功调用创建pod函数的次数
		successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {
			err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
		
		//重新调起pod数量未达到diff值，这里应该是一个记录此次调用失败，提醒informer下次继续调用的过程
		if skippedPods := diff - successfulCreations; skippedPods > 0 {
			for i := 0; i < skippedPods; i++ {
				rsc.expectations.CreationObserved(rsKey)
			}
		}
	} else if diff > 0 {
		//存在pod超过期望值，getPodsToDelete中对pod按照状态进行排序，根据数量返回需要删除pod数组，优先删除unhealthy等unAvailible态
		if diff > rsc.burstReplicas {
			diff = rsc.burstReplicas
		}
		podsToDelete := getPodsToDelete(filteredPods, diff)

		for _, pod := range podsToDelete {
			go func(targetPod *v1.Pod) {
				defer wg.Done()
				rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs)
			}(pod)
		}
		wg.Wait()
	}
}