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

# 1.调谐

# 1.1.updateState

ssc.updateStatefulSet()会基于sts.spec.replicas划分Pod为合法存在或非法删除部分，更新过程保持Pod数量固定，有序递增或递减。

// updateStatefulSet performs the update function for a StatefulSet. This method creates/updates/deletes Pods.
func (ssc *defaultStatefulSetControl) updateStatefulSet(...) (*apps.StatefulSetStatus, error) {
	...
    
	// 生成curSet
	curSet, err := ApplyRevision(set, currentRevision)
	...
    
    // 生成updateSet
	updateSet, err := ApplyRevision(set, updateRevision)
	...

	// 重新计算status
	status := apps.StatefulSetStatus{}
	status.ObservedGeneration = set.Generation
	status.CurrentRevision = currentRevision.Name
	status.UpdateRevision = updateRevision.Name
	...
	*status.CollisionCount = collisionCount

    // 更新status的replicas/readyReplicas/AvaReplicas/curReplicas/updateReplicas
	updateStatus(&status, set.Spec.MinReadySeconds, currentRevision, updateRevision, pods)
	...

	// 合法性检查
	for _, pod := range pods {
        // Pod索引位于[start,start+replicas-1]
		if podInOrdinalRange(pod, set) {
			// 记录合法Pod
			replicas[getOrdinal(pod)-getStartOrdinal(set)] = pod
        // Pod索引范围超出
		} else if getOrdinal(pod) >= 0 {
			// 记录非法Pod
			condemned = append(condemned, pod)
		}
	}

	// 遍历合法索引[start,start+replicas-1]
	for ord := getStartOrdinal(set); ord <= getEndOrdinal(set); ord++ {
        // podIndex = missIndex - start
		replicaIdx := ord - getStartOrdinal(set)
        // 补充缺失Pod
		if replicas[replicaIdx] == nil {
            // 滚动更新+{未设置partition&podIndex合法 || 设置partition&podIndex<partition}则使用curReversion生成Pod
            // 否则使用updateReversion生成Pod
			replicas[replicaIdx] = newVersionedStatefulSetPod(
				currentSet,
				updateSet,
				currentRevision.Name,
				updateRevision.Name, ord)
		}
	}

	// 非法Pod基于索引排序
	sort.Sort(descendingOrdinal(condemned))
	...

	// sts正在删除,仅返回状态不进行操作
	if set.DeletionTimestamp != nil {
		return &status, nil
	}

    // 确认调度模式(OrderedReady模式 || Parallel模式)
	monotonic := !allowsBurst(set)

	// 处理合法副本的闭包函数
	processReplicaFn := func(i int) (bool, error) {
		return ssc.processReplica(ctx,set,curRevision,updateRevision,curSet,updateSet,monotonic,replicas,i)
	}
    
    // 基于调度模式执行processReplicaFn处理replicas副本
	if shouldExit, err := runForAll(replicas, processReplicaFn, monotonic); shouldExit || err != nil {
        // 更新status
		updateStatus(&status, set.Spec.MinReadySeconds, currentRevision, updateRevision, replicas, condemned)
		return &status, err
	}

	// 激活自动PVC删除
	if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetAutoDeletePVC) {
		fixPodClaim := func(i int) (bool, error) {
            // 保留策略与ownerRef匹配
			matchPolicy, err := ssc.podControl.ClaimsMatchRetentionPolicy(ctx, updateSet, condemned[i])
            ...
            
            // 未匹配
			if !matchPolicy {
                // 重置ownerRef及更新PVC
				ssc.podControl.UpdatePodClaimForRetentionPolicy(ctx, updateSet, condemned[i])
                ...
			}
            
            // 不退出循环
			return false, nil
		}
        
        // 执行PVC修复
		if shouldExit, err := runForAll(condemned, fixPodClaim, monotonic); shouldExit || err != nil {
            // 出错更新status
			updateStatus(&status, set.Spec.MinReadySeconds, curRevision, updateRevision, replicas, condemned)
			return &status, err
		}
	}

	// 处理非法副本的闭包函数
	processCondemnedFn := func(i int) (bool, error) {
		return ssc.processCondemned(ctx, set, firstUnhealthyPod, monotonic, condemned, i)
	}
    
    // 执行非法副本清理
	if shouldExit, err := runForAll(condemned, processCondemnedFn, monotonic); shouldExit || err != nil {
        // 出错仅更新status
		updateStatus(&status, set.Spec.MinReadySeconds, currentRevision, updateRevision, replicas, condemned)
		return &status, err
	}

    // 更新status
	updateStatus(&status, set.Spec.MinReadySeconds, currentRevision, updateRevision, replicas, condemned)

	// OnDelete策略不进行Pod更新
	if set.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
		return &status, nil
	}

    // 允许更新器件最大不可用副本的同步分支
	if utilfeature.DefaultFeatureGate.Enabled(features.MaxUnavailableStatefulSet) {
		return updateStatefulSetAfterInvariantEstablished(ctx,
			ssc,
			set,
			replicas,
			updateRevision,
			status,
		)
	}

	// 更新的起始位置
	updateMin := 0
	if set.Spec.UpdateStrategy.RollingUpdate != nil {
		updateMin = int(*set.Spec.UpdateStrategy.RollingUpdate.Partition)
	}
    
	// 仅更新partition分区后的Pod
	for target := len(replicas) - 1; target >= updateMin; target-- {

		// reversion不匹配
		if getPodRevision(replicas[target]) != updateRevision.Name && !isTerminating(replicas[target]) {
			// 清理不匹配的Pod
			ssc.podControl.DeleteStatefulPod(set, replicas[target])
            ...
            
            // 更新status.CurrentReplicas,触发删除Pod基于updateReversion重建
			status.CurrentReplicas--
			return &status, err
		}

		// 等待前一个newPod健康
		if !isHealthy(replicas[target]) {
			return &status, nil
		}

	}
	return &status, 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
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

注意

ssc.updateStatefulSet()优先补全Pod，其次修复PVC OwnerRef及清理非法Pod，最后才会处理滚动更新

# 1.2.fixPodClaim

fixPodClaim()是内部命名函数，检查保留策略与PVC OwnerRef是否冲突，出现冲突会基于保留策略及sts/Pod情况重置ownerRef并更新。

fixPodClaim := func(i int) (bool, error) {
    // 保留策略与ownerRef匹配
    matchPolicy, err := ssc.podControl.ClaimsMatchRetentionPolicy(ctx, updateSet, condemned[i])
    ...

    // 未匹配
    if !matchPolicy {
        // 重置ownerRef及更新PVC
        ssc.podControl.UpdatePodClaimForRetentionPolicy(ctx, updateSet, condemned[i])
        ...
    }

    // 不退出循环
    return false, nil
}

// returns false if the PVCs for pod are not consistent with set's PVC deletion policy.
func (spc *StatefulPodControl) ClaimsMatchRetentionPolicy(...) (bool, error) {
	...
    
    // 遍历sts VolumeClaimTemplates定义
	for i := range templates {
        // 获取PVC
		claimName := getPersistentVolumeClaimName(set, &templates[i], ordinal)
		claim, err := spc.objectMgr.GetClaim(set.Namespace, claimName)
		switch {
		...
		default:
            // 检查ownerRef
			if !claimOwnerMatchesSetAndPod(claim, set, pod) {
				return false, nil
			}
		}
	}
	return true, nil
}

// returns false if ownerRefs of the claim are not set consistently with the PVC deletion policy for the sts.
func claimOwnerMatchesSetAndPod(claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
	// 获取PVC保留策略
    policy := getPersistentVolumeClaimRetentionPolicy(set)
	...
    
	switch {
	...
    // 扩缩容保留,删除sts保留
	case policy.WhenScaled == retain && policy.WhenDeleted == retain:
		// PVC不能有sts或Pod的ownerRef
        if hasOwnerRef(claim, set) ||
			hasOwnerRef(claim, pod) {
			return false
		}
    // 扩缩容保留,删除sts清理
	case policy.WhenScaled == retain && policy.WhenDeleted == delete:
		// PVC只能有sts ownerRef
        if !hasOwnerRef(claim, set) ||
			hasOwnerRef(claim, pod) {
			return false
		}
    // 扩缩容清理,删除sts保留
	case policy.WhenScaled == delete && policy.WhenDeleted == retain:
        // PVC不应该有sts ownerRef
		if hasOwnerRef(claim, set) {
			return false
		}
        // PodIndex检查(缩容)
		podScaledDown := !podInOrdinalRange(pod, set)
        // 缩容的Pod应该是PVC的ownerRef
		if podScaledDown != hasOwnerRef(claim, pod) {
			return false
		}
    // 扩缩容清理,删除sts均清理
	case policy.WhenScaled == delete && policy.WhenDeleted == delete:
        // PodIndex检查
		podScaledDown := !podInOrdinalRange(pod, set)
		
        // Pod未缩容,PVC属于sts
		if podScaledDown == hasOwnerRef(claim, set) {
			return false
		}
        
        // Pod缩容,PVC属于Pod
		if podScaledDown != hasOwnerRef(claim, pod) {
			return false
		}
	}
	return true
}

注意

fixPodClaim()修复PVC执行于非法Pod回收前，主要确保PVC自动清晰或保留

# 2.同步

# 2.1.processRFn

runForAll+processReplicaFn负责执行replicas同步，涉及已存在Pod维护及补位Pod 的创建，确保statefulset副本数固定。

func runForAll(pods []*v1.Pod, fn func(i int) (bool, error), monotonic bool) (bool, error) {
	// 串行顺序模式
    if monotonic {
        // 依次执行processReplica
		for i := range pods {
            // 串行模式返回true或报错提前结束
			if shouldExit, err := fn(i); shouldExit || err != nil {
				return true, err
			}
		}
	} else {
        // 并行执行(step=1,2,4,8,...,podSize)
		if _, err := slowStartBatch(1, len(pods), fn); err != nil {
			return true, err
		}
	}
	return false, nil
}

func (ssc *defaultStatefulSetControl) processReplica(...) (bool, error) {
	// Pod走到终态进行重建
	if isFailed(replicas[i]) || isSucceeded(replicas[i]) {
		...
        // 删除Pod
		ssc.podControl.DeleteStatefulPod(set, replicas[i])
        ...
		
        // 生成curReversion Pod草稿
		replicas[i] = newVersionedStatefulSetPod(currentSet, updateSet, currentRevision.Name,
			updateRevision.Name, replicaOrd)
	}
    
	// 未创建过
	if !isCreated(replicas[i]) {
        // PVC自动清理开启
		if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetAutoDeletePVC) {
            // PVC陈旧检测(非保留、Pod关联的PVC存在ownerRef UID对不上)
			isStale, _ := ssc.podControl.PodClaimIsStale(set, replicas[i])
            ...
            
            // PVC陈旧不允许创建Pod
            if isStale {
				return true, err
			}
		}
        
        // 创建Pod
		ssc.podControl.CreateStatefulPod(ctx, set, replicas[i])
        ...
        
        // 串行排序模式,提前结束
		if monotonic {
			// if the set does not allow bursting, return immediately
			return true, nil
		}
	}

	// Pod Pending状态
	if isPending(replicas[i]) {
		...
        // 补充缺失的PVC,基于保留策略尝试修复PVC的ownerRef
		ssc.podControl.createMissingPersistentVolumeClaims(ctx, set, replicas[i])
        ...
	}

	// Pod终止状态且串行模式,提前结束
	if isTerminating(replicas[i]) && monotonic {
		return true, nil
	}

	// 非ready且串行模式,提前结束
	if !isRunningAndReady(replicas[i]) && monotonic {
		return true, nil
	}

	// 非可用且串行模式
	if !isRunningAndAvailable(replicas[i], set.Spec.MinReadySeconds) && monotonic {
		return true, nil
	}
	...
    
    // 开启PVC自动清理
	if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetAutoDeletePVC) {
		...
        // 保留策略与ownerRef匹配
		retentionMatch, err = ssc.podControl.ClaimsMatchRetentionPolicy(ctx, updateSet, replicas[i])
		
        // 出错或匹配均为true
		if err != nil {
			retentionMatch = true
		}
	}

    // 身份匹配&PVC匹配(Pod存在声明的Volume+Volume必须是PVC+PVC名称符合规范)&保留策略匹配ownerRef
	if identityMatches(set, replicas[i]) && storageMatches(set, replicas[i]) && retentionMatch {
		return false, nil
	}
	...
    
    // 更新Pod相关信息
	ssc.podControl.UpdateStatefulPod(ctx, updateSet, replicas[i])
    ...

	return false, nil
}

注意

runForAll+processReplica会创建Pod及修复PVC

# 2.2.processCFn

runForAll+processCondemned负责清理Index非法的Pod,PVC已基于fixPodClaim修复ownerRef，Pod删除后会基于GC清理。

func runForAll(pods []*v1.Pod, fn func(i int) (bool, error), monotonic bool) (bool, error) {
    // 串行模式
	if monotonic {
		for i := range pods {
			if shouldExit, err := fn(i); shouldExit || err != nil {
				return true, err
			}
		}
    // 并行模式
	} else {
		if _, err := slowStartBatch(1, len(pods), fn); err != nil {
			return true, err
		}
	}
	return false, nil
}

func (ssc *defaultStatefulSetControl) processCondemned(...) (bool, error) {
	// Pod正在终止
	if isTerminating(condemned[i]) {
		// 串行模式阻塞等待上一个Pod清理完成
		if monotonic {
			return true, nil
		}
        
		return false, nil
	}
    
	// Podf非ready&串行模式&非首个不健康Pod,需等待
	if !isRunningAndReady(condemned[i]) && monotonic && condemned[i] != firstUnhealthyPod {
		return true, nil
	}
    
	// Podf非可用&串行模式&非首个不健康Pod,需等待
	if !isRunningAndAvailable(condemned[i], set.Spec.MinReadySeconds) && monotonic && condemned[i] != firstUnhealthyPod {
		return true, nil
	}

	// 否则执行Pod删除
	return true, ssc.podControl.DeleteStatefulPod(set, condemned[i])
}

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

补充

Pod执行清理区分串行模式和并行模式，串行模式会阻塞至上一个Pod或首个UnhealthyPod清理完成

# 2.3.updateLimit

ssc.updateStatefulSetAfterInvariantEstablished()用于滚动更新过程中限制不可用Pod数量，有序由高到低重建需更新Pod。

// GetScaledValueFromIntOrPercent is meant to replace GetValueFromIntOrPercent.
func GetScaledValueFromIntOrPercent(intOrPercent *IntOrString, total int, roundUp bool) (int, error) {
	...
    
    // 解析值或百分比
	value, isPercent, err := getIntOrPercentValueSafely(intOrPercent)
	...
    
    // 百分比类型
	if isPercent {
        // 扩容
		if roundUp {
            // value = ceil(total*percent)
			value = int(math.Ceil(float64(value) * (float64(total)) / 100))
        // 缩容
		} else {
            // value = floor(total*percent)
			value = int(math.Floor(float64(value) * (float64(total)) / 100))
		}
	}
	return value, nil
}

// calculates the real maxUnavailable number according to replica count and maxUnavailable from rollingUpdate.
func getStatefulSetMaxUnavailable(maxUnavailable *intstr.IntOrString, replicaCount int) (int, error) {
	// 获取最大不可用数量
    maxUnavailableNum, err := intstr.GetScaledValueFromIntOrPercent(intstr.ValueOrDefault(maxUnavailable, intstr.FromInt(1)), replicaCount, false)
	...

	if maxUnavailableNum < 1 {
		maxUnavailableNum = 1
	}
    
	return maxUnavailableNum, nil
}

func updateStatefulSetAfterInvariantEstablished(...) (*apps.StatefulSetStatus, error) {
	...
    
    // 初始化updateMin及maxUnavailable
	if set.Spec.UpdateStrategy.RollingUpdate != nil {
		updateMin = int(*set.Spec.UpdateStrategy.RollingUpdate.Partition)
		...
        
		maxUnavailable, _ = getStatefulSetMaxUnavailable(set.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable, replicaCount)
		...
	}
	...
    
    // 统计不可用Pod
	for target := len(replicas) - 1; target >= 0; target-- {
		if !isHealthy(replicas[target]) {
			unavailablePods++
		}
	}

    // 实际不可用超出最大限制,提前结束
	if unavailablePods >= maxUnavailable {
		return &status, nil
	}

	// 剩余可删除Pod
	podsToDelete := maxUnavailable - unavailablePods
	...
    
    // 只更新[updateMin,replicaSize]的Pod
	for target := len(replicas) - 1; target >= updateMin && deletedPods < podsToDelete; target-- {

		// Pod未终止且未匹配updateReversion
		if getPodRevision(replicas[target]) != updateRevision.Name && !isTerminating(replicas[target]) {
            // 删除Pod
			ssc.podControl.DeleteStatefulPod(set, replicas[target])
            ...
            
			deletedPods++
			status.CurrentReplicas--
		}
	}
	return &status, nil
}

注意

ssc.updateStatefulSetAfterInvariantEstablished()会删除多个Pod，直至达到maxUnavailable

# 3.更新

# 3.1.updateStatus

ssc.updateStatefulSetStatus()基于副本调整计算的status和statefulset对比，更新status部分数据及Patch到APIServer。

// completes a rolling update when all of set's replica Pods have been updated to the updateRevision.
func completeRollingUpdate(set *apps.StatefulSet, status *apps.StatefulSetStatus) {
	if set.Spec.UpdateStrategy.Type == apps.RollingUpdateStatefulSetStrategyType &&
		status.UpdatedReplicas == *set.Spec.Replicas &&
		status.ReadyReplicas == *set.Spec.Replicas &&
		status.Replicas == *set.Spec.Replicas {
		status.CurrentReplicas = status.UpdatedReplicas
		status.CurrentRevision = status.UpdateRevision
	}
}

// updates set's Status to be equal to status. If status indicates a complete update.
func (ssc *defaultStatefulSetControl) updateStatefulSetStatus(...) error {
	// 滚动更新完成标记status数据
	completeRollingUpdate(set, status)

	// 状态一致
	if !inconsistentStatus(set, status) {
		return nil
	}

	// copy set and update its status
	set = set.DeepCopy()
	if err := ssc.statusUpdater.UpdateStatefulSetStatus(ctx, set, status); err != nil {
		return err
	}

	return nil
}

func (ssu *realStatefulSetStatusUpdater) UpdateStatefulSetStatus(...) error {
	// don't wait due to limited number of clients, but backoff after the default number of steps
	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
		set.Status = *status
        
		// 更新sts status
		_, updateErr := ssu.client.AppsV1().StatefulSets(set.Namespace).UpdateStatus(context.TODO(), set, ...)
		...
        
        // 重新获取底层的sts
        updated, _ := ssu.setLister.StatefulSets(set.Namespace).Get(set.Name)
        ...
        
        // make a copy so we don't mutate the shared cache
        set = updated.DeepCopy()
		...

		return updateErr
	})
}

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

补充

基于sts完成情况，会进一步计算status状态，对比sts与status差异，内容不同会Patch到APIServer