deployment | 南风未起｜云原生/Go技术分享博客

# 1.删除

# 1.1.syncStatusOnly

deployment.DeletionTimestamp不为空，执行syncStatusOnly()同步正在删除的deployment状态，基于dm和rs获取新旧副本集对象。

// syncStatusOnly only updates Deployments Status and doesn't take any mutating actions.
func (dc *DeploymentController) syncStatusOnly(ctx context.Context, d *apps.Deployment, rsList...) error {
	// 获取新的RS和所有旧的RS及同步RV
    newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, false)
	...

	allRSs := append(oldRSs, newRS)
    
    // 同步状态
	return dc.syncDeploymentStatus(ctx, allRSs, newRS, d)
}

// getAllReplicaSetsAndSyncRevision returns all the replica sets for the provided deployment (new and all old).
func (dc *DeploymentController) getAllReplicaSetsAndSyncRevision(...) (...) {
    // 获取所有旧的RS
	_, allOldRSs := deploymentutil.FindOldReplicaSets(d, rsList)

	// 获取新的RS
	newRS, err := dc.getNewReplicaSet(ctx, d, rsList, allOldRSs, createIfNotExisted)
	...

	return newRS, allOldRSs, nil
}

// FindOldReplicaSets returns the old replica sets targeted by the given Deployment, with the given slice。
func FindOldReplicaSets(deployment *apps.Deployment, rsList []*apps.ReplicaSet) (...) {
	...
    
    // 找到podTemplate匹配的最近RS
	newRS := FindNewReplicaSet(deployment, rsList)
    // 收集其它RS和副本数不为0的RS
	for _, rs := range rsList {
		// Filter out new replica set
		if newRS != nil && rs.UID == newRS.UID {
			continue
		}
		allRSs = append(allRSs, rs)
		if *(rs.Spec.Replicas) != 0 {
			requiredRSs = append(requiredRSs, rs)
		}
	}
	return requiredRSs, allRSs
}

// Returns a replica set that matches the intent of the given deployment.
func (dc *DeploymentController) getNewReplicaSet(...) (*apps.ReplicaSet, error) {
	// 找到podTemplate匹配的创建最久RS
	existingNewRS := deploymentutil.FindNewReplicaSet(d, rsList)

	// 所有旧版本的最大RV
	maxOldRevision := deploymentutil.MaxRevision(oldRSs)
	// newRV = max+1
	newRevision := strconv.FormatInt(maxOldRevision+1, 10)

	// 更新已存在的新RS
	if existingNewRS != nil {
		rsCopy := existingNewRS.DeepCopy()

		// 1.deploy部分annotation同步到newRS
        // 2.更新newRS的RVAnnotation
        // 3.更新newRS的RVHistoryAnnotation,长度溢出(2000)则剔除起始索引RV
		annotationsUpdated := deploymentutil.SetNewReplicaSetAnnotations(ctx, d, rsCopy, newRevision, true, maxRevHistoryLengthInChars)
        // 更新minReadySeconds,Patch变化到RS
		minReadySecondsNeedsUpdate := rsCopy.Spec.MinReadySeconds != d.Spec.MinReadySeconds
		if annotationsUpdated || minReadySecondsNeedsUpdate {
			rsCopy.Spec.MinReadySeconds = d.Spec.MinReadySeconds
			return dc.client.AppsV1().ReplicaSets(rsCopy.ObjectMeta.Namespace).Update(ctx, rsCopy, metav1.UpdateOptions{})
		}

		// 基于newRS的RVAnnotation更新deploy RVAnnotation
		needsUpdate := deploymentutil.SetDeploymentRevision(d, rsCopy.Annotations[deploymentutil.RevisionAnnotation])
		
        // 获取deploy进度condition
		cond := deploymentutil.GetDeploymentCondition(d.Status, apps.DeploymentProgressing)
        
        // deploy设置进度监听,未处理过进度事件
		if deploymentutil.HasProgressDeadline(d) && cond == nil {
			// 更新进度condition
			condition := deploymentutil.NewDeploymentCondition(apps.DeploymentProgressing, v1.ConditionTrue, deploymentutil.FoundNewRSReason, msg)
			deploymentutil.SetDeploymentCondition(&d.Status, *condition)
			needsUpdate = true
		}

        // 更新deploy
		if needsUpdate {
			dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(ctx, d, metav1.UpdateOptions{})
            ...
		}
        
        // 返回newRS
		return rsCopy, nil
	}
    
    if !createIfNotExisted {
		return nil, nil
	}
	...
    
    return createdRS, err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

注意

1.newRS获取会更新相应的Annotation/minReadySeconds及Patch到集群存储，若相关数据无变化会更新deploy的RV/进度Condition

2.getNewReplicaSet()其实包括已有的newRS更新及未找到时创建newRS，由于这里限制未找到不创建，因此只截取newRS更新部分

# 1.2.syncDeployStatus

syncDeploymentStatus()基于newRS和all计算deployment当前的status，基于计算结果与deployment实际status对比更新。

// syncDeploymentStatus checks if the status is up-to-date and sync it if necessary
func (dc *DeploymentController) syncDeploymentStatus(...) error {
    // 计算status
	newStatus := calculateStatus(allRSs, newRS, d)

	if reflect.DeepEqual(d.Status, newStatus) {
		return nil
	}

    // 存在差异,更新deploymeny status
	newDeployment := d
	newDeployment.Status = newStatus
	_, err := dc.client.AppsV1().Deployments(newDeployment.Namespace).UpdateStatus(ctx, newDeployment, metav1.UpdateOptions{})
	return err
}

// calculateStatus calculates the latest status for the provided deployment by looking into the provided rs.
func calculateStatus(...) apps.DeploymentStatus {
    // 所有rs可用副本数
	availableReplicas := deploymentutil.GetAvailableReplicaCountForReplicaSets(allRSs)
    // 所有rs总副本数
	totalReplicas := deploymentutil.GetReplicaCountForReplicaSets(allRSs)
    // 计算不可用副本数
	unavailableReplicas := totalReplicas - availableReplicas
	...

	status := apps.DeploymentStatus{
		ObservedGeneration:  deployment.Generation,
		Replicas:            deploymentutil.GetActualReplicaCountForReplicaSets(allRSs),	 // 实际总副本数
		UpdatedReplicas:     GetActualReplicaCountForReplicaSets([]*apps.ReplicaSet{newRS}), // 更新副本数(newRS)
		ReadyReplicas:       deploymentutil.GetReadyReplicaCountForReplicaSets(allRSs), 	 // 就绪副本数
		AvailableReplicas:   availableReplicas,												 // 可用副本数
		UnavailableReplicas: unavailableReplicas,											 // 不可用副本数
		CollisionCount:      deployment.Status.CollisionCount,								 // hash盐,冲突会递增
	}

	// 继承conditions
	conditions := deployment.Status.Conditions
	for i := range conditions {
		status.Conditions = append(status.Conditions, conditions[i])
	}

    // 基于可用副本数要求更新deploy condition
	if availableReplicas >= *(deployment.Spec.Replicas)-deploymentutil.MaxUnavailable(*deployment) {
		minAvailability := deploymentutil.NewDeploymentCondition(apps.DeploymentAvailable, v1.ConditionTrue, deploymentutil.MinimumReplicasAvailable, "Deployment has minimum availability.")
		deploymentutil.SetDeploymentCondition(&status, *minAvailability)
	} else {
		noMinAvailability := deploymentutil.NewDeploymentCondition(apps.DeploymentAvailable, v1.ConditionFalse, deploymentutil.MinimumReplicasUnavailable, "Deployment does not have minimum availability.")
		deploymentutil.SetDeploymentCondition(&status, *noMinAvailability)
	}

	return status
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

注意

1.status计算本质上就是聚合RS的相关副本数，这里计算可用副本数、就绪副本数参考了所有RS是因为oldRS副本数均为0

2.deploy真正删除由GC控制器触发，其它控制里的删除仅设置DeletionTimestamp字段，标记orphan、background或foreground

# 2.扩容

# 2.1.checkPaused

dc.checkPausedConditions()会维护deployment 暂停或回复的进度条件，以防止误报超时或缺乏进展，避免controller错误认为卡住。

// checkPausedConditions checks if the given deployment is paused or not and adds an appropriate condition.
func (dc *DeploymentController) checkPausedConditions(ctx context.Context, d *apps.Deployment) error {
    // 没有progressDeadline,提前退出
	if !deploymentutil.HasProgressDeadline(d) {
		return nil
	}
    
    // 获取progressing condition
	cond := deploymentutil.GetDeploymentCondition(d.Status, apps.DeploymentProgressing)
    
    // progress timeout,不允许用paused/resumed覆盖
	if cond != nil && cond.Reason == deploymentutil.TimedOutReason {
		// If we have reported lack of progress, do not overwrite it with a paused condition.
		return nil
	}
    
    // 检查condition paused条件
	pausedCondExists := cond != nil && cond.Reason == deploymentutil.PausedDeployReason

	needsUpdate := false
    
    // deployment处于paused,condition未标记paused
	if d.Spec.Paused && !pausedCondExists {
        // 更新Progressing paused condition
		condition := deploymentutil.NewDeploymentCondition(apps.DeploymentProgressing, v1.ConditionUnknown, deploymentutil.PausedDeployReason, "Deployment is paused")
		deploymentutil.SetDeploymentCondition(&d.Status, *condition)
		needsUpdate = true
    // deployment非paused,condition标记paused
	} else if !d.Spec.Paused && pausedCondExists {
        // 更新Progressing resumed condition
		condition := deploymentutil.NewDeploymentCondition(apps.DeploymentProgressing, v1.ConditionUnknown, deploymentutil.ResumedDeployReason, "Deployment is resumed")
		deploymentutil.SetDeploymentCondition(&d.Status, *condition)
		needsUpdate = true
	}

    // 无需更新
	if !needsUpdate {
		return nil
	}
	...
    
	// 更新deployment status
	_, err = dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(ctx, d, metav1.UpdateOptions{})
	return err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

注意

checkPausedConditions()仅更新设置processDeadline deployment的paused/resumed condition，后续还会进入sync处理

# 2.2.scalingEvent

dc.isScalingEvent()会获取所有RS，过滤rs.spec.replicas>0的activeRS，检查rs.desired与deploy.spec.replicas一致性。

// isScalingEvent checks whether the provided deployment has been updated with a scaling event.
func (dc *DeploymentController) isScalingEvent(...) (bool, error) {
    // 获取newRS和所有oldRS
	newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, false)
	...
	allRSs := append(oldRSs, newRS)
	
    // 获取activeRS遍历
	for _, rs := range controller.FilterActiveReplicaSets(allRSs) {
        // 获取期望值
		desired, ok := deploymentutil.GetDesiredReplicasAnnotation(logger, rs)
		if !ok {
			continue
		}
        // rs期望值与deploy不一致,说明是扩缩容事件
		if desired != *(d.Spec.Replicas) {
			return true, nil
		}
	}
	return false, nil
}

// FilterActiveReplicaSets returns replica sets that have (or at least ought to have) pods.
func FilterActiveReplicaSets(replicaSets []*apps.ReplicaSet) []*apps.ReplicaSet {
	activeFilter := func(rs *apps.ReplicaSet) bool {
		return rs != nil && *(rs.Spec.Replicas) > 0
	}
    
    // rs.spec.replicas > 0就是activeRS
	return FilterReplicaSets(replicaSets, activeFilter)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

注意

dc.isScalingEvent()相对简单，获取activeRS(其实就是最新的)，检查desiredAnnotation的期望值与deploy是否一致

# 2.3.sync

dc.sync()会处理paused或scaling阶段的deloyment，基于新旧副本集调用scale执行扩缩容方法，将最新的状态同步到deployment。

// sync is responsible for reconciling deployments on scaling events or when they  are paused.
func (dc *DeploymentController) sync(ctx context.Context, d *apps.Deployment, rsList []*apps.ReplicaSet) error {
	// 获取newRS和所有oldRS
    newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, false)
	...
    
    // 执行扩缩容(paused仅限制rollouting)
	dc.scale(ctx, d, newRS, oldRSs)
    ...

	// paused&非rollbacking
	if d.Spec.Paused && getRollbackTo(d) == nil {
        // 清理旧的RS,保留最新revisionHistoryLimit个
		dc.cleanupDeployment(ctx, oldRSs, d)
        ... 
	}

    // 同步deployment status状态
	allRSs := append(oldRSs, newRS)
	return dc.syncDeploymentStatus(ctx, allRSs, newRS, d)
}

// cleanupDeployment is responsible for cleaning up a deployment ie. retains all but the latest N old RS.
func (dc *DeploymentController) cleanupDeployment(...) error {
	// 未设置RVHistoryLimit
	if !deploymentutil.HasRevisionHistoryLimit(deployment) {
		return nil
	}

    // RS存活检测(避免回收正在删除的RS)
	aliveFilter := func(rs *apps.ReplicaSet) bool {
		return rs != nil && rs.ObjectMeta.DeletionTimestamp == nil
	}
    
    // 筛选存活的RS作为可清理项
	cleanableRSes := controller.FilterReplicaSets(oldRSs, aliveFilter)

    // 计算超出的RS
	diff := int32(len(cleanableRSes)) - *deployment.Spec.RevisionHistoryLimit
	if diff <= 0 {
		return nil
	}

    // 根据创建时间排序RS
	sort.Sort(deploymentutil.ReplicaSetsByRevision(cleanableRSes))
	
	// 依次删除创建最久的N个
	for i := int32(0); i < diff; i++ {
		rs := cleanableRSes[i]
		// 副本数不为0/Generation未同步/正在删除的RS跳过
		if rs.Status.Replicas != 0 || *(rs.Spec.Replicas) != 0 || rs.Generation > rs.Status.ObservedGeneration || rs.DeletionTimestamp != nil {
			continue
		}
		
        // 删除RS
		dc.client.AppsV1().ReplicaSets(rs.Namespace).Delete(ctx, rs.Name, metav1.DeleteOptions{})
        ...
	}

	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

注意

dc.sync()会优先处理扩缩容，尝试删除旧的RS以维护最大历史副本数，最后同步deployment最新的status

# 2.4.scale

dc.scale()会计算需要扩容的pod数量，根据策略排序RS，进行newRS和oldRS交错扩缩容，实现pod逐步切换的滚动效果。

// scale scales proportionally in order to mitigate risk.
func (dc *DeploymentController) scale(...) error {
	// 尝试获取唯一的activeRS执行扩缩容
	if activeOrLatest := deploymentutil.FindActiveOrLatest(newRS, oldRSs); activeOrLatest != nil {
        // 副本数一致
		if *(activeOrLatest.Spec.Replicas) == *(deployment.Spec.Replicas) {
			return nil
		}
        // 执行扩缩容
		err := dc.scaleReplicaSetAndRecordEvent(ctx, activeOrLatest, *(deployment.Spec.Replicas), deployment)
		return err
	}

	// newRS已经饱和,缩容oldRS.spec.replicas=0
	if deploymentutil.IsSaturated(deployment, newRS) {
        // 释放oldRS副本
		for _, old := range controller.FilterActiveReplicaSets(oldRSs) {
			dc.scaleReplicaSetAndRecordEvent(ctx, old, 0, deployment)
            ...
		}
		return nil
	}

    // 滚动更新(template变化)
	if deploymentutil.IsRollingUpdate(deployment) {
        // activeRS的pod
		allRSs := controller.FilterActiveReplicaSets(append(oldRSs, newRS))
		allRSsReplicas := deploymentutil.GetReplicaCountForReplicaSets(allRSs)

        // 最大可创建的pod
		allowedSize := int32(0)
		if *(deployment.Spec.Replicas) > 0 {
			allowedSize = *(deployment.Spec.Replicas) + deploymentutil.MaxSurge(*deployment)
		}

		// 需要扩容的pod
		deploymentReplicasToAdd := allowedSize - allRSsReplicas

		...
		switch {
         //扩容,newRS放在前面
		case deploymentReplicasToAdd > 0:
			sort.Sort(controller.ReplicaSetsBySizeNewer(allRSs))
			scalingOperation = "up"
		// 缩容,newRS放在后边
		case deploymentReplicasToAdd < 0:
			sort.Sort(controller.ReplicaSetsBySizeOlder(allRSs))
			scalingOperation = "down"
		}
		...
        
        // 遍历activeRS,计算rs扩缩容的期望副本数
		for i := range allRSs {
			rs := allRSs[i]

			if deploymentReplicasToAdd != 0 {
                // 估算rs期望扩缩容副本数(newRS和oldRS方向一致,根据比例均摊)
				proportion := deploymentutil.GetProportion(logger, rs, *deployment, deploymentReplicasToAdd, deploymentReplicasAdded)
				// 累加rs扩缩容的副本数
				nameToSize[rs.Name] = *(rs.Spec.Replicas) + proportion
				deploymentReplicasAdded += proportion
			} else {
				nameToSize[rs.Name] = *(rs.Spec.Replicas)
			}
		}

		// Update all replica sets
		for i := range allRSs {
			rs := allRSs[i]
			// 剩余扩容副本优先给newRS,剩余缩容副本优先给oldRS
			if i == 0 && deploymentReplicasToAdd != 0 {
				leftover := deploymentReplicasToAdd - deploymentReplicasAdded
				nameToSize[rs.Name] = nameToSize[rs.Name] + leftover
				if nameToSize[rs.Name] < 0 {
					nameToSize[rs.Name] = 0
				}
			}

			// TODO: Use transactions when we have them.
			dc.scaleReplicaSet(ctx, rs, nameToSize[rs.Name], deployment, scalingOperation)
            ...
		}
	}
	return nil
}

func (dc *DeploymentController) scaleReplicaSetAndRecordEvent(...) (bool, *apps.ReplicaSet, error) {
	// 扩缩容完成
	if *(rs.Spec.Replicas) == newScale {
		return false, rs, nil
	}
	...
    
    // 计算行为(UP/Down)
	if *(rs.Spec.Replicas) < newScale {
		scalingOperation = "up"
	} else {
		scalingOperation = "down"
	}
    
    // 更新RS副本数
	scaled, newRS, err := dc.scaleReplicaSet(ctx, rs, newScale, deployment, scalingOperation)
	return scaled, newRS, err
}

func (dc *DeploymentController) scaleReplicaSet(...) (bool, *apps.ReplicaSet, error) {

	sizeNeedsUpdate := *(rs.Spec.Replicas) != newScale

    // 检查rs的desiredAnno和maxReplicaAnno是否需要更新
	annotationsNeedUpdate := deploymentutil.ReplicasAnnotationsNeedUpdate(rs, *(deployment.Spec.Replicas), *(deployment.Spec.Replicas)+deploymentutil.MaxSurge(*deployment))
	...
	
    // spec.replicas或annotation需更新
	if sizeNeedsUpdate || annotationsNeedUpdate {
		...
		rsCopy := rs.DeepCopy()
		// 更新spec.replicas
		*(rsCopy.Spec.Replicas) = newScale
        // 更新desiredAnno和maxReplicaAnno
		deploymentutil.SetReplicasAnnotations(rsCopy, *(deployment.Spec.Replicas), *(deployment.Spec.Replicas)+deploymentutil.MaxSurge(*deployment))
        // 更新到rs
		rs, err = dc.client.AppsV1().ReplicaSets(rsCopy.Namespace).Update(ctx, rsCopy, metav1.UpdateOptions{})
		...
	}
	return scaled, rs, err
}

注意

dc.scale()会将activeRS根据scaleUP/scaleDown向同一个方向分摊扩缩一定数量，至于active oldRS则会在滚动处理阶段缩容

# 3.回退

# 3.1.rollback

dc.rollback()负责执行回退，根据deployment注解标记的reversion选中历史版本RS，基于历史版本RS对deployment进行回滚及调整。

func getRollbackTo(d *apps.Deployment) *extensions.RollbackConfig {
	// 获取annotation标注的预期历史版本
	revision := d.Annotations[apps.DeprecatedRollbackTo]
	if revision == "" {
		return nil
	}
	revision64, err := strconv.ParseInt(revision, 10, 64)
	...
    
	return &extensions.RollbackConfig{
		Revision: revision64,
	}
}

// rollback the deployment to the specified revision. In any case cleanup the rollback spec.
func (dc *DeploymentController) rollback(...) error {
	// 获取newRS和所有oldRS
	newRS, allOldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, true)
	...

	allRSs := append(allOldRSs, newRS)
    
    // 获取回退的RS
	rollbackTo := getRollbackTo(d)
    
	// reversion为0,标识回退到上一个版本
	if rollbackTo.Revision == 0 {
        // 获取上一个版本(未找到则放弃回滚)
		if rollbackTo.Revision = deploymentutil.LastRevision(allRSs); rollbackTo.Revision == 0 {
			// 清理rollbackTo注解及更新deployment
			return dc.updateDeploymentAndClearRollbackTo(ctx, d)
		}
	}
    
    // 遍历RS
	for _, rs := range allRSs {
        // 获取当前rs的reversion
		v, err := deploymentutil.Revision(rs)
		...
        
        // 当前rs是回退目标
		if v == rollbackTo.Revision {
			...
            
            // 回退
			performedRollback, err := dc.rollbackToTemplate(ctx, d, rs)
			...
            
			return err
		}
	}
	
	// 未找到匹配RS,清理rollbackTo注解及更新deployment
	return dc.updateDeploymentAndClearRollbackTo(ctx, d)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

注意

dc.rollback()匹配desired rs，基于template更新deployment，具体的副本管理其实还是会走到rolloutUpdate分支

# 3.2.rollbackTemp

dc.rollbackToTemplate()会检查deployment.spec.template和rs.spec.template差异，不一致会基于rs更新deployment定义。

// rollbackToTemplate compares the templates of the provided deployment and replica and update deployment.
func (dc *DeploymentController) rollbackToTemplate(...) (bool, error) {
	...
    
    // deployment和rs的spec.template不一致
	if !deploymentutil.EqualIgnoreHash(&d.Spec.Template, &rs.Spec.Template) {
		// 更新deployment.spec.template
		deploymentutil.SetFromReplicaSetTemplate(d, rs.Spec.Template)
		// 同步deployment.annotations
		deploymentutil.SetDeploymentAnnotationsTo(d, rs)
		performedRollback = true
	}
    ...
	
	return performedRollback, dc.updateDeploymentAndClearRollbackTo(ctx, d)
}

// updateDeploymentAndClearRollbackTo sets .spec.rollbackTo to nil and update the input deployment.
func (dc *DeploymentController) updateDeploymentAndClearRollbackTo(...) error {
	// 清理回退annotation
	setRollbackTo(d, nil)
    // 更新
	_, err := dc.client.AppsV1().Deployments(d.Namespace).Update(ctx, d, metav1.UpdateOptions{})
	return err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

注意

rollback本质上是基于历史replicaset更新deployment，基于注解声明期望的历史版本，这种注解回退模式即将废弃

# 3.3.rsAndSyncRV

dc.getAllRSAndSyncRV()在删除阶段提到过会更新exist newRS，其实它会根据传入的条件更新或创建newRS，这里可以看一下创建流程。

// getAllReplicaSetsAndSyncRevision returns all the replica sets for the provided deployment.
func (dc *DeploymentController) getAllReplicaSetsAndSyncRevision(...) (...) {
    // 获取所有oldRS
	_, allOldRSs := deploymentutil.FindOldReplicaSets(d, rsList)

	// 获取newRS,涉及创建或更新
	newRS, err := dc.getNewReplicaSet(ctx, d, rsList, allOldRSs, createIfNotExisted)
	...

	return newRS, allOldRSs, nil
}

// Returns a replica set that matches the intent of the given deployment.
func (dc *DeploymentController) getNewReplicaSet(...) (*apps.ReplicaSet, error) {
	
	existingNewRS := deploymentutil.FindNewReplicaSet(d, rsList)

	// Calculate the max revision number among all old RSes
	maxOldRevision := deploymentutil.MaxRevision(oldRSs)
	// Calculate revision number for this new replica set
	newRevision := strconv.FormatInt(maxOldRevision+1, 10)
	...

    // rollback/rollout会置为创建
	if !createIfNotExisted {
		return nil, nil
	}
	...
    
	// 构造rs定义
	newRS := apps.ReplicaSet{...}
    
	allRSs := append(oldRSs, &newRS)
    // 设置newRS的初始副本数
	newReplicasCount, err := deploymentutil.NewRSNewReplicas(d, allRSs, &newRS)
	...
	*(newRS.Spec.Replicas) = newReplicasCount
    
	// 设置annotation
	deploymentutil.SetNewReplicaSetAnnotations(ctx, d, &newRS, newRevision, false, maxRevHistoryLengthInChars)
	...
    
    // 创建
	createdRS, err := dc.client.AppsV1().ReplicaSets(d.Namespace).Create(ctx, &newRS, metav1.CreateOptions{})
	
    switch {
	/// 已存在
	case errors.IsAlreadyExists(err):
		alreadyExists = true

		// 由informer缓存获取
		rs, rsErr := dc.rsLister.ReplicaSets(newRS.Namespace).Get(newRS.Name)
		...

		// 确认owner是当前deployment
		controllerRef := metav1.GetControllerOf(rs)
		if controllerRef != nil && controllerRef.UID == d.UID && dt.EqualIgnoreHash(&d.Spec.Template, &rs.Spec.Template) {
			createdRS = rs
            ...
			break
		}
		...
        
		// hash冲突,将collisionCount盐自增
		preCollisionCount := *d.Status.CollisionCount
		*d.Status.CollisionCount++
		// 更新到deployment,触发下一次协调重新生成hash
		_, dErr := dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(ctx, d, metav1.UpdateOptions{})
		...
        
		return nil, err
    // namespace正在删除
	case errors.HasStatusCause(err, v1.NamespaceTerminatingCause):
		// if the namespace is terminating, all subsequent creates will fail and we can safely do nothing
		return nil, err
    // 其它错误
	case err != nil:
		...
        // 更新deployment进度condition
		if deploymentutil.HasProgressDeadline(d) {
			cond := deploymentutil.NewDeploymentCondition(apps.DeploymentProgressing, v1.ConditionFalse, deploymentutil.FailedRSCreateReason, msg)
			deploymentutil.SetDeploymentCondition(&d.Status, *cond)
		}
		...
		return nil, err
	}
    ...

    // 正常创建,更新deployment关联RV
	needsUpdate := deploymentutil.SetDeploymentRevision(d, newRevision)
    // 更新deplyment进度condition
	if !alreadyExists && deploymentutil.HasProgressDeadline(d) {
		msg := fmt.Sprintf("Created new replica set %q", createdRS.Name)
		condition := deploymentutil.NewDeploymentCondition(apps.DeploymentProgressing, v1.ConditionTrue, deploymentutil.NewReplicaSetReason, msg)
		deploymentutil.SetDeploymentCondition(&d.Status, *condition)
		needsUpdate = true
	}
    
    // 更新deployment
	if needsUpdate {
		_, err = dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(ctx, d, metav1.UpdateOptions{})
	}
	return createdRS, err
}

补充

dc.getAllReplicaSetsAndSyncRevision()的newRS创建部分相对简单，主要基于deploy.spec.template匹配或创建满足的rs

# 4.rollout

# 4.1.rolloutRecreate

dc.rolloutRecreate()设计上简单粗暴，不像滚动更新逐步替换，直接将oldRS缩容到0，pod全部notRunning触发newRS创建。

// rolloutRecreate implements the logic for recreating a replica set.
func (dc *DeploymentController) rolloutRecreate(...) error {
    // 获取newRS和所有oldRS(未找到不会创建newRS)
	newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, false)
	...
    
	allRSs := append(oldRSs, newRS)
    // 获取activeRS
	activeOldRSs := controller.FilterActiveReplicaSets(oldRSs)

	// oldRS缩容为0
	scaledDown, err := dc.scaleDownOldReplicaSetsForRecreate(ctx, activeOldRSs, d)
	...
    
    // 缩容完成,同步状态
	if scaledDown {
		// Update DeploymentStatus.
		return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
	}

	// oldPod还在运行,仅同步状态
	if oldPodsRunning(newRS, oldRSs, podMap) {
		return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
	}

	// 创建newRS
	if newRS == nil {
		newRS, oldRSs, err = dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, true)
		...
		allRSs = append(oldRSs, newRS)
	}

    // newRS扩容(deploy.spec.replicas)
	dc.scaleUpNewReplicaSetForRecreate(ctx, newRS, d)
    ...

    // 清理过期的oldRS
	if util.DeploymentComplete(d, &d.Status) {
		dc.cleanupDeployment(ctx, oldRSs, d)
        ...
	}

	// 同步状态
	return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

注意

dc.rolloutRecreate()相对直接，先将oldRS.spec.replicas均设置0，没有pod处于Running后直接创建newRS及设置期望副本数

# 4.2.rolloutRolling

dc.rolloutRolling()基于滚动更新，尝试先对newRS扩容，再对oldRS缩容，通过循环反复执行扩缩容进行一边增一边减，达到逐步替换效果。

// rolloutRolling implements the logic for rolling a new replica set.
func (dc *DeploymentController) rolloutRolling(...) error {
    // 获取newRS及oldRS(涉及创建)
	newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, true)
	...
    
	allRSs := append(oldRSs, newRS)

	// 先尝试newRS扩容
	scaledUp, err := dc.reconcileNewReplicaSet(ctx, allRSs, newRS, d)
	...
    
    // 扩容完成,同步状态
	if scaledUp {
		// Update DeploymentStatus
		return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
	}

	// 再尝试oldRS缩容
	scaledDown, err := dc.reconcileOldReplicaSets(ctx, allRSs, FilterActiveReplicaSets(oldRSs), newRS, d)
	...
    
    // 缩容完成,同步状态
	if scaledDown {
		// Update DeploymentStatus
		return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
	}

    // scale完成,清理多出的oldRS
	if deploymentutil.DeploymentComplete(d, &d.Status) {
		dc.cleanupDeployment(ctx, oldRSs, d)
        ...
	}

	// Sync deployment status
	return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
}

func (dc *DeploymentController) reconcileNewReplicaSet(...) (bool, error) {
    // 期望副本数一致,无需操作
	if *(newRS.Spec.Replicas) == *(deployment.Spec.Replicas) {
		// Scaling not required.
		return false, nil
	}
    
    // newRS副本数超出,重新设置为deploy.spec.replicas
	if *(newRS.Spec.Replicas) > *(deployment.Spec.Replicas) {
		// Scale down.
		scaled, _, err := dc.scaleReplicaSetAndRecordEvent(ctx, newRS, *(deployment.Spec.Replicas), deployment)
		return scaled, err
	}
    
    // 计算预期副本数(replicas = maxSugre - curPod + replicas)
	newReplicasCount, err := deploymentutil.NewRSNewReplicas(deployment, allRSs, newRS)
	...
    
    // scale up
	scaled, _, err := dc.scaleReplicaSetAndRecordEvent(ctx, newRS, newReplicasCount, deployment)
	return scaled, err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

注意

滚动更新就是根据maxSugrePod-curPod差异先扩容newRS，再根据avalibalePod-minAvaliable缩容oldRS，直到newRS达到预期

# 4.3.reconcileOldRS

dc.reconcileOldRS()会计算oldRS可缩容数量，优先从unhealthyRS缩容replicas，其次再从healthyRS缩容一定数量replicas。

func (dc *DeploymentController) reconcileOldReplicaSets(...) (bool, error) {
	// 计算oldPod数量
	oldPodsCount := deploymentutil.GetReplicaCountForReplicaSets(oldRSs)
	if oldPodsCount == 0 {
		// Can't scale down further
		return false, nil
	}
    // 计算Pod数量
	allPodsCount := deploymentutil.GetReplicaCountForReplicaSets(allRSs)
	...
    
    // 计算maxScaleDown(maxScaledDown = total - 最小可用数 - newRS不可用Pod数)
	maxUnavailable := deploymentutil.MaxUnavailable(*deployment)
	minAvailable := *(deployment.Spec.Replicas) - maxUnavailable
	newRSUnavailablePodCount := *(newRS.Spec.Replicas) - newRS.Status.AvailableReplicas
	maxScaledDown := allPodsCount - minAvailable - newRSUnavailablePodCount
	if maxScaledDown <= 0 {
		return false, nil
	}

	// 优先缩容异常的oldRS
	oldRSs, cleanupCount, err := dc.cleanupUnhealthyReplicas(ctx, oldRSs, deployment, maxScaledDown)
	...

	allRSs = append(oldRSs, newRS)
    // 缩容oldRS
	scaledDownCount, err := dc.scaleDownOldReplicaSetsForRollingUpdate(ctx, allRSs, oldRSs, deployment)
	...

	totalScaledDown := cleanupCount + scaledDownCount
	return totalScaledDown > 0, nil
}

// cleanupUnhealthyReplicas will scale down old replica sets with unhealthy replica.
func (dc *DeploymentController) cleanupUnhealthyReplicas(...) ([]*apps.ReplicaSet, int32, error) {
	// 根据创建时间排序
	sort.Sort(controller.ReplicaSetsByCreationTimestamp(oldRSs))
	
	totalScaledDown := int32(0)
	for i, targetRS := range oldRSs {
        // 已达到最大清理数量
		if totalScaledDown >= maxCleanupCount {
			break
		}
        
        // 缩容到0的RS
		if *(targetRS.Spec.Replicas) == 0 {
			// cannot scale down this replica set.
			continue
		}
		
        // 副本均健康,无需清理
		if *(targetRS.Spec.Replicas) == targetRS.Status.AvailableReplicas {
			// no unhealthy replicas found, no scaling required.
			continue
		}

        // 计算剩余可清理名额
		scaledDownCount := int32(integer.IntMin(int(maxCleanupCount-totalScaledDown), int(*(targetRS.Spec.Replicas)-targetRS.Status.AvailableReplicas)))
        
        // 设置缩容期望副本数
		newReplicasCount := *(targetRS.Spec.Replicas) - scaledDownCount
		...
        
        // 调整rs副本数
		_, updatedOldRS, err := dc.scaleReplicaSetAndRecordEvent(ctx, targetRS, newReplicasCount, deployment)
		...
        
        // 记录缩容信息
		totalScaledDown += scaledDownCount
		oldRSs[i] = updatedOldRS
	}
	return oldRSs, totalScaledDown, nil
}

// scaleDownOldReplicaSetsForRollingUpdate scales down old replica sets.
func (dc *DeploymentController) scaleDownOldReplicaSetsForRollingUpdate(...) (int32, error) {
	...
    
    // 最小可用副本检查
	if availablePodCount <= minAvailable {
		// Cannot scale down.
		return 0, nil
	}
	
	// 根据创建时间排序oldRS
	sort.Sort(controller.ReplicaSetsByCreationTimestamp(oldRSs))

	totalScaledDown := int32(0)
    // 最大可缩容数量
	totalScaleDownCount := availablePodCount - minAvailable
	for _, targetRS := range oldRSs {
        // 已达到缩容目标
		if totalScaledDown >= totalScaleDownCount {
			// No further scaling required.
			break
		}
        
        // oldRS已缩容到0
		if *(targetRS.Spec.Replicas) == 0 {
			// cannot scale down this ReplicaSet.
			continue
		}
        
        // 计算本次缩容数量
		scaleDownCount := int32(integer.IntMin(int(*(targetRS.Spec.Replicas)), int(totalScaleDownCount-totalScaledDown)))
		newReplicasCount := *(targetRS.Spec.Replicas) - scaleDownCount
		...
        
        // 调整oldRS副本数
		_, _, err := dc.scaleReplicaSetAndRecordEvent(ctx, targetRS, newReplicasCount, deployment)
		...

		totalScaledDown += scaleDownCount
	}

	return totalScaledDown, nil
}

补充

dc.reconcileOldRS()会减少oldRS持有副本数，会优先减少unhealthy oldRS不健康副本，再尝试由healthy oldRS缩一定数量

# 5.总结

deployment根据不同阶段分发任务，依次处理delete——>pause——>rollback——>scale——>rollout任务，确保deployment期望副本一致。