Kubernetes StatefulSet源码分析

Author: xidianwangtao@gmail.com，Based on Kubernetes 1.9

摘要：Kubernetes StatefulSet在1.9版本中stable了，相信以后会有越老越多的企业会使用它来部署有状态应用，比如Mysql、Zookeeper、ElasticSearch、Redis等等。本文是对StatefulSet的源码分析，包括其Inner Structure、Sync的核心逻辑、Update的主要流程说明、完整的Code Logic Diagram及一些思考。

Inner Structure

下面是简单的StatefulSet Controller工作的内部结构图。

NewStatefulSetController

同其他Controller一样，StatefulSet Controller也是由ControllerManager初始化时负责启动。

1// NewStatefulSetController creates a new statefulset controller.
2func NewStatefulSetController(
3	podInformer coreinformers.PodInformer,
4	setInformer appsinformers.StatefulSetInformer,
5	pvcInformer coreinformers.PersistentVolumeClaimInformer,
6	revInformer appsinformers.ControllerRevisionInformer,
7	kubeClient clientset.Interface,
8) *StatefulSetController {
9	
10    ...
11
12	ssc := &StatefulSetController{
13		kubeClient: kubeClient,
14		control: NewDefaultStatefulSetControl(
15			NewRealStatefulPodControl(
16				kubeClient,
17				setInformer.Lister(),
18				podInformer.Lister(),
19				pvcInformer.Lister(),
20				recorder),
21			NewRealStatefulSetStatusUpdater(kubeClient, setInformer.Lister()),
22			history.NewHistory(kubeClient, revInformer.Lister()),
23		),
24		pvcListerSynced: pvcInformer.Informer().HasSynced,
25		queue:           workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "statefulset"),
26		podControl:      controller.RealPodControl{KubeClient: kubeClient, Recorder: recorder},
27
28		revListerSynced: revInformer.Informer().HasSynced,
29	}
30
31	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
32		// lookup the statefulset and enqueue
33		AddFunc: ssc.addPod,
34		// lookup current and old statefulset if labels changed
35		UpdateFunc: ssc.updatePod,
36		// lookup statefulset accounting for deletion tombstones
37		DeleteFunc: ssc.deletePod,
38	})
39	ssc.podLister = podInformer.Lister()
40	ssc.podListerSynced = podInformer.Informer().HasSynced
41
42	setInformer.Informer().AddEventHandlerWithResyncPeriod(
43		cache.ResourceEventHandlerFuncs{
44			AddFunc: ssc.enqueueStatefulSet,
45			UpdateFunc: func(old, cur interface{}) {
46				oldPS := old.(*apps.StatefulSet)
47				curPS := cur.(*apps.StatefulSet)
48				if oldPS.Status.Replicas != curPS.Status.Replicas {
49					glog.V(4).Infof("Observed updated replica count for StatefulSet: %v, %d->%d", curPS.Name, oldPS.Status.Replicas, curPS.Status.Replicas)
50				}
51				ssc.enqueueStatefulSet(cur)
52			},
53			DeleteFunc: ssc.enqueueStatefulSet,
54		},
55		statefulSetResyncPeriod,
56	)
57	ssc.setLister = setInformer.Lister()
58	ssc.setListerSynced = setInformer.Informer().HasSynced
59
60	// TODO: Watch volumes
61	return ssc
62}

很熟悉的代码风格，也是创建对应的eventBroadcaster，然后给对应的objectInformer注册对应的eventHandler：

• StatefulSetController主要ListWatch Pod和StatefulSet对象；
• Pod Informer注册了add/update/delete EventHandler，这三个EventHandler都会将Pod对应的StatefulSet加入到StatefulSet Queue中。
• StatefulSet Informer同样注册了add/update/event EventHandler，也都会将StatefulSet加入到StatefulSet Queue中。
• 目前StatefulSetController还未感知PVC Informer的EventHandler，这里继续按照PVC Controller全部处理。在StatefulSet Controller创建和删除Pod时，会调用apiserver创建和删除对应的PVC。
• RevisionController类似，在StatefulSet Controller Reconcile时会创建或者删除对应的Revision。

StatefulSetController sync

接下来，会进入StatefulSetController的worker（只有一个worker，也就是只一个go routine），worker会从StatefulSet Queue中pop out一个StatefulSet对象，然后执行sync进行Reconcile操作。

1// sync syncs the given statefulset.
2func (ssc *StatefulSetController) sync(key string) error {
3	startTime := time.Now()
4	defer func() {
5		glog.V(4).Infof("Finished syncing statefulset %q (%v)", key, time.Now().Sub(startTime))
6	}()
7
8	namespace, name, err := cache.SplitMetaNamespaceKey(key)
9	if err != nil {
10		return err
11	}
12	set, err := ssc.setLister.StatefulSets(namespace).Get(name)
13	if errors.IsNotFound(err) {
14		glog.Infof("StatefulSet has been deleted %v", key)
15		return nil
16	}
17	if err != nil {
18		utilruntime.HandleError(fmt.Errorf("unable to retrieve StatefulSet %v from store: %v", key, err))
19		return err
20	}
21
22	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
23	if err != nil {
24		utilruntime.HandleError(fmt.Errorf("error converting StatefulSet %v selector: %v", key, err))
25		// This is a non-transient error, so don't retry.
26		return nil
27	}
28
29	if err := ssc.adoptOrphanRevisions(set); err != nil {
30		return err
31	}
32
33	pods, err := ssc.getPodsForStatefulSet(set, selector)
34	if err != nil {
35		return err
36	}
37
38	return ssc.syncStatefulSet(set, pods)
39}

• sync中根据setLabel匹配出所有revisions、然后检查这些revisions中是否有OwnerReference为空的，如果有，那说明存在Orphaned的Revisions。

  注意：只要检查到有一个History Revision就会触发给所有的Resivions打上Patch：
  {"metadata":{"ownerReferences":[{"apiVersion":"%s","kind":"%s","name":"%s","uid":"%s","controller":true,"blockOwnerDeletion":true}],"uid":"%s"}}

• 调用getPodsForStatefulSet获取这个StatefulSet应该管理的Pods。

  • 获取该StatefulSet对应Namesapce下所有的Pods；
  • 执行ClaimPods操作：检查set和pod的Label是否匹配上，如果Label不匹配，那么需要release这个Pod，然后检查pod的name和StatefulSet name的格式是否能匹配上。对于都匹配上的，并且ControllerRef UID也相同的，则不需要处理。
  • 如果Selector和ControllerRef都匹配不上，则执行ReleasePod操作，给Pod打Patch: {“metadata":{"ownerReferences":[{"$patch":"delete","uid":"%s"}],"uid":"%s"}}
  • 对于Label和name格式能匹配上的，但是controllerRef为空的Pods,就执行AdoptPod，给Pod打上Patch： {“metadata":{"ownerReferences":[{"apiVersion":"%s","kind":"%s","name":"%s","uid":"%s","controller":true,"blockOwnerDeletion":true}],"uid":"%s"}}

UpdateStatefulSet

syncStatefulSet的实现只是调用UpdateStatefulSet。

1func (ssc *defaultStatefulSetControl) UpdateStatefulSet(set *apps.StatefulSet, pods []*v1.Pod) error {
2
3	// list all revisions and sort them
4	revisions, err := ssc.ListRevisions(set)
5	if err != nil {
6		return err
7	}
8	history.SortControllerRevisions(revisions)
9
10	// get the current, and update revisions
11	currentRevision, updateRevision, collisionCount, err := ssc.getStatefulSetRevisions(set, revisions)
12	if err != nil {
13		return err
14	}
15
16	// perform the main update function and get the status
17	status, err := ssc.updateStatefulSet(set, currentRevision, updateRevision, collisionCount, pods)
18	if err != nil {
19		return err
20	}
21
22	// update the set's status
23	err = ssc.updateStatefulSetStatus(set, status)
24	if err != nil {
25		return err
26	}
27
28	glog.V(4).Infof("StatefulSet %s/%s pod status replicas=%d ready=%d current=%d updated=%d",
29		set.Namespace,
30		set.Name,
31		status.Replicas,
32		status.ReadyReplicas,
33		status.CurrentReplicas,
34		status.UpdatedReplicas)
35
36	glog.V(4).Infof("StatefulSet %s/%s revisions current=%s update=%s",
37		set.Namespace,
38		set.Name,
39		status.CurrentRevision,
40		status.UpdateRevision)
41
42	// maintain the set's revision history limit
43	return ssc.truncateHistory(set, pods, revisions, currentRevision, updateRevision)
44}

UpdateStatefulSet主要流程为：

• ListRevisions获取该StatefulSet的所有Revisions，并按照Revision从小到大进行排序。
• getStatefulSetRevisions获取currentRevison和UpdateRevision。
  • 只有当RollingUpdate策略时Partition不为0时，才会有部分Pods是updateRevision。
  • 其他情况，所有Pods都得维持currentRevision。
• updateStatefulSet是StatefulSet Controller的核心逻辑，负责创建、更新、删除Pods，使得声明式target得以维护：
  • 使得target state始终有Spec.Replicas个Running And Ready的Pods。
  • 如果更新策略是RollingUpdate，并且Partition为0，则保证所有Pods都对应Status.CurrentRevision。
  • 如果更新策略是RollingUpdate，并且Partition不为0，则ordinal小于Partition的Pods保持Status.CurrentRevision，而ordinal大于等于Partition的Pods更新到Status.UpdateRevision。
  • 如果更新策略是OnDelete，则只有删除Pods时才会触发对应Pods的更新，也就是说与Revisions不关联。
• truncateHistory维护History Revision个数不超过.Spec.RevisionHistoryLimit。

updateStatefulSet

updateStatefulSet是整个StatefulSetController的核心。

1func (ssc *defaultStatefulSetControl) updateStatefulSet(
2	set *apps.StatefulSet,
3	currentRevision *apps.ControllerRevision,
4	updateRevision *apps.ControllerRevision,
5	collisionCount int32,
6	pods []*v1.Pod) (*apps.StatefulSetStatus, error) {
7	// get the current and update revisions of the set.
8	currentSet, err := ApplyRevision(set, currentRevision)
9	if err != nil {
10		return nil, err
11	}
12	updateSet, err := ApplyRevision(set, updateRevision)
13	if err != nil {
14		return nil, err
15	}
16
17	// set the generation, and revisions in the returned status
18	status := apps.StatefulSetStatus{}
19	status.ObservedGeneration = new(int64)
20	*status.ObservedGeneration = set.Generation
21	status.CurrentRevision = currentRevision.Name
22	status.UpdateRevision = updateRevision.Name
23	status.CollisionCount = new(int32)
24	*status.CollisionCount = collisionCount
25
26	replicaCount := int(*set.Spec.Replicas)
27	// slice that will contain all Pods such that 0 <= getOrdinal(pod) < set.Spec.Replicas
28	replicas := make([]*v1.Pod, replicaCount)
29	// slice that will contain all Pods such that set.Spec.Replicas <= getOrdinal(pod)
30	condemned := make([]*v1.Pod, 0, len(pods))
31	unhealthy := 0
32	firstUnhealthyOrdinal := math.MaxInt32
33	var firstUnhealthyPod *v1.Pod
34
35	// First we partition pods into two lists valid replicas and condemned Pods
36	for i := range pods {
37		status.Replicas++
38
39		// count the number of running and ready replicas
40		if isRunningAndReady(pods[i]) {
41			status.ReadyReplicas++
42		}
43
44		// count the number of current and update replicas
45		if isCreated(pods[i]) && !isTerminating(pods[i]) {
46			if getPodRevision(pods[i]) == currentRevision.Name {
47				status.CurrentReplicas++
48			} else if getPodRevision(pods[i]) == updateRevision.Name {
49				status.UpdatedReplicas++
50			}
51		}
52
53		if ord := getOrdinal(pods[i]); 0 <= ord && ord < replicaCount {
54			// if the ordinal of the pod is within the range of the current number of replicas,
55			// insert it at the indirection of its ordinal
56			replicas[ord] = pods[i]
57
58		} else if ord >= replicaCount {
59			// if the ordinal is greater than the number of replicas add it to the condemned list
60			condemned = append(condemned, pods[i])
61		}
62		// If the ordinal could not be parsed (ord < 0), ignore the Pod.
63	}
64
65	// for any empty indices in the sequence [0,set.Spec.Replicas) create a new Pod at the correct revision
66	for ord := 0; ord < replicaCount; ord++ {
67		if replicas[ord] == nil {
68			replicas[ord] = newVersionedStatefulSetPod(
69				currentSet,
70				updateSet,
71				currentRevision.Name,
72				updateRevision.Name, ord)
73		}
74	}
75
76	// sort the condemned Pods by their ordinals
77	sort.Sort(ascendingOrdinal(condemned))
78
79	// find the first unhealthy Pod
80	for i := range replicas {
81		if !isHealthy(replicas[i]) {
82			unhealthy++
83			if ord := getOrdinal(replicas[i]); ord < firstUnhealthyOrdinal {
84				firstUnhealthyOrdinal = ord
85				firstUnhealthyPod = replicas[i]
86			}
87		}
88	}
89
90	for i := range condemned {
91		if !isHealthy(condemned[i]) {
92			unhealthy++
93			if ord := getOrdinal(condemned[i]); ord < firstUnhealthyOrdinal {
94				firstUnhealthyOrdinal = ord
95				firstUnhealthyPod = condemned[i]
96			}
97		}
98	}
99
100	if unhealthy > 0 {
101		glog.V(4).Infof("StatefulSet %s/%s has %d unhealthy Pods starting with %s",
102			set.Namespace,
103			set.Name,
104			unhealthy,
105			firstUnhealthyPod.Name)
106	}
107
108	// If the StatefulSet is being deleted, don't do anything other than updating
109	// status.
110	if set.DeletionTimestamp != nil {
111		return &status, nil
112	}
113
114	monotonic := !allowsBurst(set)
115
116	// Examine each replica with respect to its ordinal
117	for i := range replicas {
118		// delete and recreate failed pods
119		if isFailed(replicas[i]) {
120			glog.V(4).Infof("StatefulSet %s/%s is recreating failed Pod %s",
121				set.Namespace,
122				set.Name,
123				replicas[i].Name)
124			if err := ssc.podControl.DeleteStatefulPod(set, replicas[i]); err != nil {
125				return &status, err
126			}
127			if getPodRevision(replicas[i]) == currentRevision.Name {
128				status.CurrentReplicas--
129			} else if getPodRevision(replicas[i]) == updateRevision.Name {
130				status.UpdatedReplicas--
131			}
132			status.Replicas--
133			replicas[i] = newVersionedStatefulSetPod(
134				currentSet,
135				updateSet,
136				currentRevision.Name,
137				updateRevision.Name,
138				i)
139		}
140		// If we find a Pod that has not been created we create the Pod
141		if !isCreated(replicas[i]) {
142			if err := ssc.podControl.CreateStatefulPod(set, replicas[i]); err != nil {
143				return &status, err
144			}
145			status.Replicas++
146			if getPodRevision(replicas[i]) == currentRevision.Name {
147				status.CurrentReplicas++
148			} else if getPodRevision(replicas[i]) == updateRevision.Name {
149				status.UpdatedReplicas++
150			}
151
152			// if the set does not allow bursting, return immediately
153			if monotonic {
154				return &status, nil
155			}
156			// pod created, no more work possible for this round
157			continue
158		}
159		// If we find a Pod that is currently terminating, we must wait until graceful deletion
160		// completes before we continue to make progress.
161		if isTerminating(replicas[i]) && monotonic {
162			glog.V(4).Infof(
163				"StatefulSet %s/%s is waiting for Pod %s to Terminate",
164				set.Namespace,
165				set.Name,
166				replicas[i].Name)
167			return &status, nil
168		}
169		// If we have a Pod that has been created but is not running and ready we can not make progress.
170		// We must ensure that all for each Pod, when we create it, all of its predecessors, with respect to its
171		// ordinal, are Running and Ready.
172		if !isRunningAndReady(replicas[i]) && monotonic {
173			glog.V(4).Infof(
174				"StatefulSet %s/%s is waiting for Pod %s to be Running and Ready",
175				set.Namespace,
176				set.Name,
177				replicas[i].Name)
178			return &status, nil
179		}
180		// Enforce the StatefulSet invariants
181		if identityMatches(set, replicas[i]) && storageMatches(set, replicas[i]) {
182			continue
183		}
184		// Make a deep copy so we don't mutate the shared cache
185		replica := replicas[i].DeepCopy()
186		if err := ssc.podControl.UpdateStatefulPod(updateSet, replica); err != nil {
187			return &status, err
188		}
189	}
190
191	// At this point, all of the current Replicas are Running and Ready, we can consider termination.
192	// We will wait for all predecessors to be Running and Ready prior to attempting a deletion.
193	// We will terminate Pods in a monotonically decreasing order over [len(pods),set.Spec.Replicas).
194	// Note that we do not resurrect Pods in this interval. Also not that scaling will take precedence over
195	// updates.
196	for target := len(condemned) - 1; target >= 0; target-- {
197		// wait for terminating pods to expire
198		if isTerminating(condemned[target]) {
199			glog.V(4).Infof(
200				"StatefulSet %s/%s is waiting for Pod %s to Terminate prior to scale down",
201				set.Namespace,
202				set.Name,
203				condemned[target].Name)
204			// block if we are in monotonic mode
205			if monotonic {
206				return &status, nil
207			}
208			continue
209		}
210		// if we are in monotonic mode and the condemned target is not the first unhealthy Pod block
211		if !isRunningAndReady(condemned[target]) && monotonic && condemned[target] != firstUnhealthyPod {
212			glog.V(4).Infof(
213				"StatefulSet %s/%s is waiting for Pod %s to be Running and Ready prior to scale down",
214				set.Namespace,
215				set.Name,
216				firstUnhealthyPod.Name)
217			return &status, nil
218		}
219		glog.V(4).Infof("StatefulSet %s/%s terminating Pod %s for scale dowm",
220			set.Namespace,
221			set.Name,
222			condemned[target].Name)
223
224		if err := ssc.podControl.DeleteStatefulPod(set, condemned[target]); err != nil {
225			return &status, err
226		}
227		if getPodRevision(condemned[target]) == currentRevision.Name {
228			status.CurrentReplicas--
229		} else if getPodRevision(condemned[target]) == updateRevision.Name {
230			status.UpdatedReplicas--
231		}
232		if monotonic {
233			return &status, nil
234		}
235	}
236
237	// for the OnDelete strategy we short circuit. Pods will be updated when they are manually deleted.
238	if set.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
239		return &status, nil
240	}
241
242	// we compute the minimum ordinal of the target sequence for a destructive update based on the strategy.
243	updateMin := 0
244	if set.Spec.UpdateStrategy.RollingUpdate != nil {
245		updateMin = int(*set.Spec.UpdateStrategy.RollingUpdate.Partition)
246	}
247	// we terminate the Pod with the largest ordinal that does not match the update revision.
248	for target := len(replicas) - 1; target >= updateMin; target-- {
249
250		// delete the Pod if it is not already terminating and does not match the update revision.
251		if getPodRevision(replicas[target]) != updateRevision.Name && !isTerminating(replicas[target]) {
252			glog.V(4).Infof("StatefulSet %s/%s terminating Pod %s for update",
253				set.Namespace,
254				set.Name,
255				replicas[target].Name)
256			err := ssc.podControl.DeleteStatefulPod(set, replicas[target])
257			status.CurrentReplicas--
258			return &status, err
259		}
260
261		// wait for unhealthy Pods on update
262		if !isHealthy(replicas[target]) {
263			glog.V(4).Infof(
264				"StatefulSet %s/%s is waiting for Pod %s to update",
265				set.Namespace,
266				set.Name,
267				replicas[target].Name)
268			return &status, nil
269		}
270
271	}
272	return &status, nil
273}

主要流程：

• 获取currentRevision和updateRevision对应的StatefulSet Object，并设置generation，currentRevision, updateRevision等信息到StatefulSet status。

• 将前面getPodsForStatefulSet获取到的pods分成两个slice：

  • valid replicas slice: : 0 <= getOrdinal(pod) < set.Spec.Replicas
  • condemned pods slice: set.Spec.Replicas <= getOrdinal(pod)

• 如果valid replicas中存在某些ordinal没有对应的Pod，则创建对应Revision的Pods Object，后面会检测到该Pod没有真实创建就会去创建对应的Pod实例：

  • 如果更新策略是RollingUpdate且Partition为0或者ordinal < Partition，则使用currentRevision创建该Pod Object。
  • 如果更新策略时RollingUpdate且Partition不为0且ordinal >= Partition，则使用updateRevision创建该Pod Object。

• 从valid repilcas和condemned pods两个slices中找出第一个unhealthy的Pod。（ordinal最小的unhealth pod）

  healthy pods means：pods is running and ready, and not terminating.

• 对于正在删除(DeletionTimestamp非空)的StatefulSet，不做任何操作，直接返回当前status。

• 遍历valid replicas中pods，保证valid replicas中index在[0，spec.replicas）的pod都是Running And Ready的：

  • 如果检测到某个pod Failed (pod.Status.Phase = Failed), 则删除这个Pod，并重新new这个pod object（注意revisions匹配）
  • 如果这个pod还没有recreate,则Create it。
  • 如果ParallelPodManagement = "OrderedReady”，则直接返回当前status。否则ParallelPodManagement = "Parallel”,则循环检测下一个。
  • 如果pod正在删除并且ParallelPodManagement = "OrderedReady”，则返回status结束。
  • 如果pod不是RunningAndReady状态，并且ParallelPodManagement = "OrderedReady”，则返回status结束。
  • 检测该pod与statefulset的identity和storage是否匹配，如果有一个不匹配，则调用apiserver Update Stateful Pod进行updateIdentity和updateStorage（并创建对应的PVC），返回status，结束。

  Pod is Running and Ready means:
  pod.Status.Phase = Runnin,
  pod.Status.Condition = Ready

• 遍历condemned replicas中pods，index由大到小的顺序，确保这些pods最终都被删除：

  • 如果这个Pod正在删除(DeletionTimestamp)，并且Pod Management是OrderedReady，则进行Block住，返回status，流程结束。
  • 如果是OrderedReady策略，Pod不是处于Running and Ready状态，且该pod不是first unhealthy pod，则返回status，流程结束。
  • 其他情况，则删除该statefulset pod。
  • 根据该pod的controller-revision-hash Label获取Revision，如果等于currentRevision，则更新status.CurrentReplicas；如果等于updateRevision，则更新status.UpdatedReplicas；
  • 如果是OrderedReady策略，则返回status，流程结束。

• OnDelete更新策略：删除Pod才会触发更新这个ordinal的更新 如果UpdateStrategy Type是OnDelete, 意味着只有当对应的Pods被手动删除后，才会触发Recreate，因此直接返回status，流程结束。

• RollingUpdate更新策略：（Partition不设置就相当于0，意味着全部pods进行滚动更新） 如果UpdateStrategy Type是RollingUpdate, 根据RollingUpdate中Partition配置得到updateMin作为update replicas index区间最小值，遍历valid replicas，index从最大值到updateMin递减的顺序：

  • 如果pod revision不是updateRevision，并且不是正在删除的，则删除这个pod，并更新status.CurrentReplicas，然后返回status，流程结束。
  • 如果pod不是healthy的，那么将等待它变成healthy，因此这里就直接返回status，流程结束。

Identity Match

updateStatefulSet Reconcile中，会检查identity match的情况，具体包含哪些？

1StatefulSetPodNameLabel        = "statefulset.kubernetes.io/pod-name"
2
3
4// identityMatches returns true if pod has a valid identity and network identity for a member of set.
5func identityMatches(set *apps.StatefulSet, pod *v1.Pod) bool {
6	parent, ordinal := getParentNameAndOrdinal(pod)
7	return ordinal >= 0 &&
8		set.Name == parent &&
9		pod.Name == getPodName(set, ordinal) &&
10		pod.Namespace == set.Namespace &&
11		pod.Labels[apps.StatefulSetPodNameLabel] == pod.Name
12}

• pod name和statefulset name内容匹配。
• namespace匹配。
• Pod的Label：statefulset.kubernetes.io/pod-name与Pod name真实匹配。

Storage Match

updateStatefulSet Reconcile中，会检查Storage match的情况，具体怎么匹配的呢？

1// storageMatches returns true if pod's Volumes cover the set of PersistentVolumeClaims
2func storageMatches(set *apps.StatefulSet, pod *v1.Pod) bool {
3	ordinal := getOrdinal(pod)
4	if ordinal < 0 {
5		return false
6	}
7	volumes := make(map[string]v1.Volume, len(pod.Spec.Volumes))
8	for _, volume := range pod.Spec.Volumes {
9		volumes[volume.Name] = volume
10	}
11	for _, claim := range set.Spec.VolumeClaimTemplates {
12		volume, found := volumes[claim.Name]
13		if !found ||
14			volume.VolumeSource.PersistentVolumeClaim == nil ||
15			volume.VolumeSource.PersistentVolumeClaim.ClaimName !=
16				getPersistentVolumeClaimName(set, &claim, ordinal) {
17			return false
18		}
19	}
20	return true
21}

Code Logic Diagram

基于上述分析，下面是一个相对完整的StatefulSetController的代码逻辑图。 (不支持大于2MB的图片，所以不太清晰，不过基本在前面内容都提到了。)

思考

滚动更新过程中出现异常

在上一篇博文浅析Kubernetes StatefulSet中遗留了一个问题：StatefulSet滚动更新时，如果某个Pod更新失败，会怎么办呢？

通过上面源码分析中滚动更新部分的分析，我们知道：

• 如果UpdateStrategy Type是RollingUpdate, 根据RollingUpdate中Partition（Partition不设置就相当于0，意味着全部pods进行滚动更新）配置得到updateMin作为update replicas index区间最小值，遍历valid replicas，index从最大值到updateMin递减的顺序：
  • 如果pod revision不是updateRevision，并且不是正在删除的，则删除这个pod，并更新status.CurrentReplicas，然后返回status，流程结束。
  • 如果pod不是healthy的，那么将等待它变成healthy，因此这里就直接返回status，流程结束。

知道这一点后，就能回答这个问题了，答案很简单：

• 如果更新策略是RollingUpdate，则逐个滚动更新过程中，如果在更新某个ordinal replica时这个Pod一直无法达到Running and Ready状态，那么整个滚动更新流程将Block在这里。还没有更新的replicas将不会触发更新，已经更新成功的replicas就保持更新后的版本，并不存在什么自动回滚的机制。在下一次sync时，检测到这个Pod isFailed（pod.Status.Phase = Failed）,会delete and recreate这个failed pod。

podManagementPolicy设置为Parallel时，体现在哪？

问题：podManagementPolicy: "Parallel"体现在什么时候呢？Scale的时候？RollingUpdate的时候？

• 在前面代码分析中updateStatefulSet中那段-"遍历valid replicas中pods，保证valid replicas中index在[0，spec.replicas）的pod都是Running And Ready的"：如果发现某个ordinal replica应该创建但是还没被创建，则会触发create。如果podManagementPolicy设置为Parallel，则会继续delete then create其他应该创建的replicas，而不会等待前面创建的replicas成为Running and Ready。
• 在前面代码分析中updateStatefulSet中那段-”遍历condemned replicas中pods，index由大到小的顺序，确保这些pods最终都被删除"：podManagementPolicy设置为Parallel，如果发现某个ordinal replica正在删除，则继续删除其他应该删除的replicas，而不会等待之前删除的replica重建并成为Running and Ready状态。

因此Parallel体现在以下场景：

• 初始化部署StatefulSet时，并行create pods。
• 级联删除StatefulSet时，并行delete pods。
• Scale up时，并行create pods。
• Scale down时，并行delete pods。

而在滚动更新时，是不会受podManagementPolicy的配置影响的，都是按照逐个地、ordinal从大到小的的顺序，保证前者Running and Ready的原则，进行RollingUpdate。

如果更新策略是OnDelete呢？那情况就不同于RollingUpdate了，因为update的流程就体现在前面提到的两个阶段了，因此Parallel是会启作用的。