apiVersion: v1
kind: Service
metadata:
  name: nginx
  labels:
    app: nginx
spec:
  ports:
  - port: 80
    name: web
  clusterIP: None # headless service
  selector:
    app: nginx
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: web
spec:
  serviceName: "nginx"
  podManagementPolicy: "OrderedReady" # default is OrderedReady
  replicas: 2
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: k8s.gcr.io/nginx-slim:0.8
        ports:
        - containerPort: 80
          name: web
        volumeMounts:
        - name: www
          mountPath: /usr/share/nginx/html
  volumeClaimTemplates:
  - metadata:
      name: www
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 1Gi

func startStatefulSetController(ctx ControllerContext) (http.Handler, bool, error) {
	if !ctx.AvailableResources[schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "statefulsets"}] {
		return nil, false, nil
	}
	go statefulset.NewStatefulSetController(
		ctx.InformerFactory.Core().V1().Pods(),
		ctx.InformerFactory.Apps().V1().StatefulSets(),
		ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
		ctx.InformerFactory.Apps().V1().ControllerRevisions(),
		ctx.ClientBuilder.ClientOrDie("statefulset-controller"),
	).Run(int(ctx.ComponentConfig.StatefulSetController.ConcurrentStatefulSetSyncs), ctx.Stop)
	return nil, true, nil
}

func (ssc *StatefulSetController) sync(key string) error {
	...
    // 1. 解析 namespace 和 name
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	...
    
    // 2. 根据 ns 和 name，获取 sts 对象
	set, err := ssc.setLister.StatefulSets(namespace).Get(name)
	...

    // 3. 获取 selector 对象，用于筛选 Pod
	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
	...

    // ssc.adoptOrphanRevisions() 
    // -> ssc.control.AdoptOrphanRevisions() 
    // -> ssc.controllerHistory.AdoptControllerRevision() 
    // -> Patch()
    // 4、筛选 sts 的孤儿 controllerrevisions，并尝试与 sts 重新关联（添加 ControllerRef）
	if err := ssc.adoptOrphanRevisions(set); err != nil {
		return err
	}

    // 5. 获取 sts 所有关联的 pod
    // ssc.getPodsForStatefulSet() 
    // -> cm.ClaimPods() 
    // -> m.ClaimObject() 
    // -> release()/adopt()
	pods, err := ssc.getPodsForStatefulSet(set, selector)
	...

    // 6. 真正执行 sync 操作
	return ssc.syncStatefulSet(set, pods)
}

func (ssc *StatefulSetController) syncStatefulSet(set *apps.StatefulSet, pods []*v1.Pod) error {
	...
    // ssc.control.UpdateStatefulSet() 
    // -> ssc.performUpdate()
    // -> ssc.updateStatefulSet() + ssc.updateStatefulSetStatus()
	if err := ssc.control.UpdateStatefulSet(set.DeepCopy(), pods); err != nil {
		return err
	}
	...
	return nil
}

func (ssc *defaultStatefulSetControl) UpdateStatefulSet(set *apps.StatefulSet, pods []*v1.Pod) error {

	// 1. 获取历史 revisions，并排序
	revisions, err := ssc.ListRevisions(set)
	...
	history.SortControllerRevisions(revisions)
    
    // 2. 计算 currentRevision 和 updateRevision
	currentRevision, updateRevision, err := ssc.performUpdate(set, pods, revisions)
	if err != nil {
		return utilerrors.NewAggregate([]error{err, ssc.truncateHistory(set, pods, revisions, currentRevision, updateRevision)})
	}

	// 3. 清理过期的历史版本
	return ssc.truncateHistory(set, pods, revisions, currentRevision, updateRevision)
}

func (ssc *defaultStatefulSetControl) performUpdate(
	set *apps.StatefulSet, pods []*v1.Pod, revisions []*apps.ControllerRevision) (*apps.ControllerRevision, *apps.ControllerRevision, error) {

	// 2.1. 计算 currentRevision 和 updateRevision
	currentRevision, updateRevision, collisionCount, err := ssc.getStatefulSetRevisions(set, revisions)
	if err != nil {
		return currentRevision, updateRevision, err
	}

	// 2.2. 执行实际的 sync 操作
	status, err := ssc.updateStatefulSet(set, currentRevision, updateRevision, collisionCount, pods)
	if err != nil {
		return currentRevision, updateRevision, err
	}

	// 2.3. 更新 sts 状态
	err = ssc.updateStatefulSetStatus(set, status)
	if err != nil {
		return currentRevision, updateRevision, err
	}

	...
	return currentRevision, updateRevision, nil
}

func (ssc *defaultStatefulSetControl) updateStatefulSet(...) (*apps.StatefulSetStatus, error) {
	// 1. 分别获取 currentRevision 和 updateRevision 对应的的 statefulset object
	currentSet, err := ApplyRevision(set, currentRevision)
	...
	updateSet, err := ApplyRevision(set, updateRevision)
	...

	// 设置 status 的 generation 和 revisions
	status := apps.StatefulSetStatus{}
	status.ObservedGeneration = set.Generation
	status.CurrentRevision = currentRevision.Name
	status.UpdateRevision = updateRevision.Name
	status.CollisionCount = new(int32)
	*status.CollisionCount = collisionCount

    // 3. 将 statefulset 的 pods 按序分到 replicas 和 condemned 两个切片
	replicaCount := int(*set.Spec.Replicas)
    // 用于保存序号在 [0,replicas) 范围内的 pod
	replicas := make([]*v1.Pod, replicaCount)
	// 用于序号大于 replicas 的 Pod
	condemned := make([]*v1.Pod, 0, len(pods))
	unhealthy := 0
	firstUnhealthyOrdinal := math.MaxInt32
	var firstUnhealthyPod *v1.Pod

    // 4. 计算 status 字段中的值，将 pod 分配到 replicas 和 condemned 两个数组中
	for i := range pods {
		status.Replicas++

		// 计算 Ready 的副本数
		if isRunningAndReady(pods[i]) {
			status.ReadyReplicas++
		}

		// 计算当前的副本数和已经更新的副本数
		if isCreated(pods[i]) && !isTerminating(pods[i]) {
			if getPodRevision(pods[i]) == currentRevision.Name {
				status.CurrentReplicas++
			}
			if getPodRevision(pods[i]) == updateRevision.Name {
				status.UpdatedReplicas++
			}
		}

		if ord := getOrdinal(pods[i]); 0 <= ord && ord < replicaCount {
			// 保存序号在 [0,replicas) 范围内的 Pod
			replicas[ord] = pods[i]

		} else if ord >= replicaCount {
			// 序号大于 replicas 的 Pod，则保存在 condemned
			condemned = append(condemned, pods[i])
		}
		// 其余的 Pod 忽略
	}

    // 5. 检查 replicas 数组中 [0,set.Spec.Replicas) 下标是否有缺失的 pod，若有缺失的则创建对应的 Pod 
	// 在 newVersionedStatefulSetPod 中会判断是使用 currentSet 还是 updateSet 来创建
	for ord := 0; ord < replicaCount; ord++ {
		if replicas[ord] == nil {
			replicas[ord] = newVersionedStatefulSetPod(
				currentSet,
				updateSet,
				currentRevision.Name,
				updateRevision.Name, ord)
		}
	}

	// 6. 对 condemned 数组进行排序
	sort.Sort(ascendingOrdinal(condemned))

    // 7、根据 ordinal 在 replicas 和 condemned 数组中找出第一个处于 NotReady 或 Terminating 的 Pod
	for i := range replicas {
		if !isHealthy(replicas[i]) {
			unhealthy++
			if ord := getOrdinal(replicas[i]); ord < firstUnhealthyOrdinal {
				firstUnhealthyOrdinal = ord
				firstUnhealthyPod = replicas[i]
			}
		}
	}

	for i := range condemned {
		if !isHealthy(condemned[i]) {
			unhealthy++
			if ord := getOrdinal(condemned[i]); ord < firstUnhealthyOrdinal {
				firstUnhealthyOrdinal = ord
				firstUnhealthyPod = condemned[i]
			}
		}
	}

	if unhealthy > 0 {
		klog.V(4).Infof("StatefulSet %s/%s has %d unhealthy Pods starting with %s",
			set.Namespace,
			set.Name,
			unhealthy,
			firstUnhealthyPod.Name)
	}

	// 8. 如果 sts 正在删除中，只要更新 status 即可
	if set.DeletionTimestamp != nil {
		return &status, nil
	}

    // 9. 默认设置为非 Parallel
	monotonic := !allowsBurst(set)

    // 10. 确保 replicas 数组中所有的 pod 是 running 的
	for i := range replicas {
		// 11. 对于 failed 的 pod 删除并重建
		if isFailed(replicas[i]) {
			ssc.recorder.Eventf(set, v1.EventTypeWarning, "RecreatingFailedPod",
				"StatefulSet %s/%s is recreating failed Pod %s",
				set.Namespace,
				set.Name,
				replicas[i].Name)
           // 删除
			if err := ssc.podControl.DeleteStatefulPod(set, replicas[i]); err != nil {
				return &status, err
			}
			if getPodRevision(replicas[i]) == currentRevision.Name {
				status.CurrentReplicas--
			}
			if getPodRevision(replicas[i]) == updateRevision.Name {
				status.UpdatedReplicas--
			}
			status.Replicas--
           // 新建
			replicas[i] = newVersionedStatefulSetPod(
				currentSet,
				updateSet,
				currentRevision.Name,
				updateRevision.Name,
				i)
		}
		// 12、如果 pod.Status.Phase 不为空，说明该 pod 未创建，则直接重新创建该 pod
		if !isCreated(replicas[i]) {
			if err := ssc.podControl.CreateStatefulPod(set, replicas[i]); err != nil {
				return &status, err
			}
			status.Replicas++
			if getPodRevision(replicas[i]) == currentRevision.Name {
				status.CurrentReplicas++
			}
			if getPodRevision(replicas[i]) == updateRevision.Name {
				status.UpdatedReplicas++
			}

			// 13. 如果为 Parallel，直接 return status 结束；如果为 OrderedReady，循环处理下一个 pod。
			if monotonic {
				return &status, nil
			}
			// pod 创建完成，本轮循环结束
			continue
		}
        // 14、如果 pod 正在删除中，且 Spec.PodManagementPolicy 不为 Parallel，
        // 直接 return status 结束，结束后会在下一个 syncLoop 继续进行处理，pod 状态的改变会触发下一次 syncLoop
		if isTerminating(replicas[i]) && monotonic {
			klog.V(4).Infof(
				"StatefulSet %s/%s is waiting for Pod %s to Terminate",
				set.Namespace,
				set.Name,
				replicas[i].Name)
			return &status, nil
		}
        // 15. 如果 pod 状态不是 Running & Ready，且 Spec.PodManagementPolicy 不为 Parallel
        // 则直接 return status 结束
		if !isRunningAndReady(replicas[i]) && monotonic {
			klog.V(4).Infof(
				"StatefulSet %s/%s is waiting for Pod %s to be Running and Ready",
				set.Namespace,
				set.Name,
				replicas[i].Name)
			return &status, nil
		}
		// 16. 检查 pod 的信息是否与 statefulset 的匹配，若不匹配则更新 pod 的状态
		if identityMatches(set, replicas[i]) && storageMatches(set, replicas[i]) {
			continue
		}
		// 深拷贝对象，避免修改缓存
		replica := replicas[i].DeepCopy()
		if err := ssc.podControl.UpdateStatefulPod(updateSet, replica); err != nil {
			return &status, err
		}
	}

    // 17. 逆序处理 condemned 中的 pod
	for target := len(condemned) - 1; target >= 0; target-- {
        // 18. 如果 pod 正在删除，检查 Spec.PodManagementPolicy 的值
        // 如果为 Parallel，循环处理下一个 pod 否则直接退出
		if isTerminating(condemned[target]) {
			klog.V(4).Infof(
				"StatefulSet %s/%s is waiting for Pod %s to Terminate prior to scale down",
				set.Namespace,
				set.Name,
				condemned[target].Name)
			// 如果不为 Parallel，直接 return
			if monotonic {
				return &status, nil
			}
			continue
		}
		// 19. 不满足以下条件说明该 pod 是更新前创建的，正处于创建中
		if !isRunningAndReady(condemned[target]) && monotonic && condemned[target] != firstUnhealthyPod {
			klog.V(4).Infof(
				"StatefulSet %s/%s is waiting for Pod %s to be Running and Ready prior to scale down",
				set.Namespace,
				set.Name,
				firstUnhealthyPod.Name)
			return &status, nil
		}
		klog.V(2).Infof("StatefulSet %s/%s terminating Pod %s for scale down",
			set.Namespace,
			set.Name,
			condemned[target].Name)
        // 20、否则直接删除该 pod
		if err := ssc.podControl.DeleteStatefulPod(set, condemned[target]); err != nil {
			return &status, err
		}
		if getPodRevision(condemned[target]) == currentRevision.Name {
			status.CurrentReplicas--
		}
		if getPodRevision(condemned[target]) == updateRevision.Name {
			status.UpdatedReplicas--
		}
        // 21. 如果为 OrderedReady 方式则返回否则继续处理下一个 pod
		if monotonic {
			return &status, nil
		}
	}

	// 22. 对于 OnDelete 策略直接返回
	if set.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
		return &status, nil
	}

	// 23. 若为 RollingUpdate 策略，则倒序处理 replicas 数组中下标大于等于
    // 	   Spec.UpdateStrategy.RollingUpdate.Partition 的 pod
	updateMin := 0
	if set.Spec.UpdateStrategy.RollingUpdate != nil {
		updateMin = int(*set.Spec.UpdateStrategy.RollingUpdate.Partition)
	}
	// 用与更新版本不匹配的最大序号终止 Pod
	for target := len(replicas) - 1; target >= updateMin; target-- {

		// 24. 如果 Pod 的 Revision 不等于 updateRevision，且 pod 没有处于删除状态，则直接删除 pod
		if getPodRevision(replicas[target]) != updateRevision.Name && !isTerminating(replicas[target]) {
			klog.V(2).Infof("StatefulSet %s/%s terminating Pod %s for update",
				set.Namespace,
				set.Name,
				replicas[target].Name)
			err := ssc.podControl.DeleteStatefulPod(set, replicas[target])
			status.CurrentReplicas--
			return &status, err
		}

		// 25. 如果 pod 非 healthy 状态直接返回
		if !isHealthy(replicas[target]) {
			klog.V(4).Infof(
				"StatefulSet %s/%s is waiting for Pod %s to update",
				set.Namespace,
				set.Name,
				replicas[target].Name)
			return &status, nil
		}

	}
	return &status, nil
}