func (ssc *defaultStatefulSetControl) updateStatefulSet(
	set *nuwav1.StatefulSet,
	currentRevision *apps.ControllerRevision,
	updateRevision *apps.ControllerRevision,
	collisionCount int32,
	pods []*v1.Pod,
	revisions []*apps.ControllerRevision) (*nuwav1.StatefulSetStatus, error) {
	currentSet, err := ApplyRevision(set, currentRevision)
	if err != nil {
		return nil, err
	}
	updateSet, err := ApplyRevision(set, updateRevision)
	if err != nil {
		return nil, err
	}
	status := nuwav1.StatefulSetStatus{}
	status.ObservedGeneration = set.Generation
	status.CurrentRevision = currentRevision.Name
	status.UpdateRevision = updateRevision.Name
	status.CollisionCount = new(int32)
	*status.CollisionCount = collisionCount
	replicaCount := int(*set.Spec.Replicas)
	replicas := make([]*v1.Pod, replicaCount)
	condemned := make([]*v1.Pod, 0, len(pods))
	unhealthy := 0
	firstUnhealthyOrdinal := math.MaxInt32
	var firstUnhealthyPod *v1.Pod
	for i := range pods {
		status.Replicas++
		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 {
			replicas[ord] = pods[i]
		} else if ord >= replicaCount {
			condemned = append(condemned, pods[i])
		}
	}
	for ord := 0; ord < replicaCount; ord++ {
		if replicas[ord] == nil {
			replicas[ord] = newVersionedStatefulSetPod(
				ssc.Client,
				currentSet,
				updateSet,
				currentRevision.Name,
				updateRevision.Name, ord)
		}
	}
	sort.Sort(ascendingOrdinal(condemned))
	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 {
		ssc.Log.Info("StatefulSet has unhealthy Pods starting with",
			"ns", set.Namespace,
			"statefulset_name", set.Name,
			"status", unhealthy,
			"pod_name", firstUnhealthyPod.Name)
	}
	if set.DeletionTimestamp != nil {
		return &status, nil
	}
	monotonic := !allowsBurst(set)
	for i := range replicas {
		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(
				ssc.Client,
				currentSet,
				updateSet,
				currentRevision.Name,
				updateRevision.Name,
				i)
		}
		if !isCreated(replicas[i]) {
			if err := ssc.podControl.CreateStatefulPod(set, replicas[i]); err != nil {
				msg := fmt.Sprintf("StatefulPodControl failed to create Pod error: %s", err)
				condition := NewStatefulsetCondition(nuwav1.FailedCreatePod, v1.ConditionTrue, "", msg)
				SetStatefulsetCondition(&status, condition)
				return &status, err
			}
			status.Replicas++
			if getPodRevision(replicas[i]) == currentRevision.Name {
				status.CurrentReplicas++
			}
			if getPodRevision(replicas[i]) == updateRevision.Name {
				status.UpdatedReplicas++
			}
			if monotonic {
				return &status, nil
			}
			continue
		}
		if isTerminating(replicas[i]) && monotonic {
			ssc.Log.Info(
				"StatefulSet is waiting for Pod to Terminate",
				"ns", set.Namespace,
				"statefulset_name", set.Name,
				"replicas_name", replicas[i].Name)
			return &status, nil
		}
		if shouldUpdateInPlaceReady(replicas[i]) {
			ssc.Log.Info(
				"StatefulSet update Pod InPlaceUpdateReady condition to true",
				"ns", set.Namespace,
				"name", set.Name,
				"podName", replicas[i].Name)
			newCondition := v1.PodCondition{
				Type:   nuwav1.StatefulSetInPlaceUpdateReady,
				Status: v1.ConditionTrue,
			}
			if err := ssc.podControl.UpdateStatefulPodCondition(set, replicas[i], newCondition); err != nil {
				return &status, err
			}
		}
		if !isRunningAndReady(replicas[i]) && monotonic {
			ssc.Log.Info(
				"StatefulSet is waiting for Pod to be Running and Ready",
				"ns", set.Namespace,
				"name", set.Name,
				"podName", replicas[i].Name)
			return &status, nil
		}
		if identityMatches(set, replicas[i]) && storageMatches(set, replicas[i]) {
			continue
		}
		replica := replicas[i].DeepCopy()
		if err := ssc.podControl.UpdateStatefulPod(updateSet, replica); err != nil {
			msg := fmt.Sprintf("StatefulPodControl failed to update Pod error: %s", err)
			condition := NewStatefulsetCondition(nuwav1.FailedUpdatePod, v1.ConditionTrue, "", msg)
			SetStatefulsetCondition(&status, condition)
			return &status, err
		}
	}
	for target := len(condemned) - 1; target >= 0; target-- {
		if isTerminating(condemned[target]) {
			ssc.Log.Info(
				"StatefulSet is waiting for Pod to Terminate prior to scale down",
				"ns", set.Namespace,
				"name", set.Name,
				"podName", condemned[target].Name)
			if monotonic {
				return &status, nil
			}
			continue
		}
		if !isRunningAndReady(condemned[target]) && monotonic && condemned[target] != firstUnhealthyPod {
			ssc.Log.Info(
				"StatefulSet is waiting for Pod to be Running and Ready prior to scale down",
				"ns", set.Namespace,
				"statefulset_name", set.Name,
				"firstUnhealthyPod_name", firstUnhealthyPod.Name)
			return &status, nil
		}
		ssc.Log.Info("StatefulSet terminating Pod for scale down",
			"ns", set.Namespace,
			"statefulset_name", set.Name,
			"condemnedPod_name", condemned[target].Name)
		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--
		}
		if monotonic {
			return &status, nil
		}
	}
	if set.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
		return &status, nil
	}
	if updateSatisfied, updateDirtyPods := updateExpectations.SatisfiedExpectations(getStatefulSetKey(set), updateRevision.Name); !updateSatisfied {
		ssc.Log.Info("Not satisfied update for statefulset, updateDirtyPods", "statefulset", getStatefulSetKey(set), "updateDirtyPods", updateDirtyPods)
		return &status, nil
	}
	updateMin := 0
	maxUnavailable := 1
	if set.Spec.UpdateStrategy.RollingUpdate != nil {
		updateMin = int(*set.Spec.UpdateStrategy.RollingUpdate.Partition)
		maxUnavailable, err = intstrutil.GetValueFromIntOrPercent(intstrutil.ValueOrDefault(set.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable, intstrutil.FromInt(1)), int(replicaCount), true)
		if err != nil {
			return &status, err
		}
		if set.Spec.UpdateStrategy.RollingUpdate.Paused {
			return &status, nil
		}
	}
	var unavailablePods []string
	for target := len(replicas) - 1; target >= updateMin; target-- {
		isInPlaceUpdateCompleted := checkInPlaceUpdateCompleted(replicas[target])
		if getPodRevision(replicas[target]) != updateRevision.Name && !isTerminating(replicas[target]) {
			var skipUpdating bool
			useInPlaceUpdate, inPlaceUpdateSpec := shouldDoInPlaceUpdate(set, updateRevision, getPodRevision(replicas[target]), revisions)
			if useInPlaceUpdate {
				if isInPlaceUpdateCompleted != nil {
					ssc.Log.Info("StatefulSet waiting for Pod last in-place update completed",
						"ns", set.Namespace,
						"statefulset_name", set.Name,
						"replica_name", replicas[target].Name)
					skipUpdating = true
				} else {
					ssc.Log.Info("StatefulSet patching Pod for in-place update",
						"ns", set.Namespace,
						"statefulset_name", set.Name,
						"replica_name", replicas[target].Name)
					updateErr := ssc.inPlaceUpdatePod(set, replicas[target], inPlaceUpdateSpec)
					if updateErr != nil && !errors.IsConflict(updateErr) && !isInPlaceOnly(set) {
						useInPlaceUpdate = false
					}
				}
			}
			if !useInPlaceUpdate {
				ssc.Log.Info("StatefulSet terminating Pod for update",
					"ns", set.Namespace,
					"statefulset_name", set.Name,
					"replicas_name", replicas[target].Name)
				if err := ssc.podControl.DeleteStatefulPod(set, replicas[target]); err != nil {
					return &status, err
				}
			}
			if !skipUpdating && getPodRevision(replicas[target]) == currentRevision.Name {
				status.CurrentReplicas--
			}
		}
		if getPodRevision(replicas[target]) != updateRevision.Name || !isHealthy(replicas[target]) {
			unavailablePods = append(unavailablePods, replicas[target].Name)
		} else if isInPlaceUpdateCompleted != nil {
			ssc.Log.Info("StatefulSet check Pod in-place update not-ready",
				"ns", set.Namespace,
				"statefulset_name", set.Name,
				"replicas_name", replicas[target].Name,
				"status", isInPlaceUpdateCompleted)
			unavailablePods = append(unavailablePods, replicas[target].Name)
		}
		if len(unavailablePods) >= maxUnavailable {
			ssc.Log.Info(" is waiting for unavailable Pods to update",
				"ns", set.Namespace,
				"statefulset_name", set.Name,
				"status", unavailablePods)
			return &status, nil
		}
	}
	return &status, nil
}
updateStatefulSet performs the update function for a StatefulSet. This method creates, updates, and deletes Pods in
the set in order to conform the system to the target state for the set. The target state always contains
set.Spec.Replicas Pods with a Ready Condition. If the UpdateStrategy.Type for the set is
RollingUpdateStatefulSetStrategyType then all Pods in the set must be at set.Status.CurrentRevision.
If the UpdateStrategy.Type for the set is OnDeleteStatefulSetStrategyType, the target state implies nothing about
the revisions of Pods in the set. If the UpdateStrategy.Type for the set is PartitionStatefulSetStrategyType, then
all Pods with ordinal less than UpdateStrategy.Partition.Ordinal must be at Status.CurrentRevision and all other
Pods must be at Status.UpdateRevision. If the returned error is nil, the returned StatefulSetStatus is valid and the
update must be recorded. If the error is not nil, the method should be retried until successful.
package controllers
import (
	"fmt"
	"github.com/go-logr/logr"
	nuwav1 "github.com/yametech/nuwa/api/v1"
	apps "k8s.io/api/apps/v1"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	intstrutil "k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/tools/record"
	"k8s.io/kubernetes/pkg/controller/history"
	"math"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sort"
)
type ControlInterface interface {
	UpdateStatefulSet(set *nuwav1.StatefulSet, pods []*v1.Pod) error
	ListRevisions(set *nuwav1.StatefulSet) ([]*apps.ControllerRevision, error)
	AdoptOrphanRevisions(set *nuwav1.StatefulSet, revisions []*apps.ControllerRevision) error
}
func NewDefaultStatefulSetControl(
	client client.Client,
	podControl StatefulPodControlInterface,
	statusUpdater StatusUpdaterInterface,
	controllerHistory history.Interface,
	recorder record.EventRecorder,
	log logr.Logger,
) ControlInterface {
	return &defaultStatefulSetControl{client, podControl, statusUpdater, controllerHistory, recorder, log}
}
type defaultStatefulSetControl struct {
	client.Client
	podControl        StatefulPodControlInterface
	statusUpdater     StatusUpdaterInterface
	controllerHistory history.Interface
	recorder          record.EventRecorder
	Log               logr.Logger
}
func (ssc *defaultStatefulSetControl) UpdateStatefulSet(set *nuwav1.StatefulSet, pods []*v1.Pod) error {
	revisions, err := ssc.ListRevisions(set)
	if err != nil {
		return err
	}
	history.SortControllerRevisions(revisions)
	currentRevision, updateRevision, collisionCount, err := ssc.getStatefulSetRevisions(set, revisions)
	if err != nil {
		return err
	}
	for _, pod := range pods {
		updateExpectations.ObserveUpdated(getStatefulSetKey(set), updateRevision.Name, pod)
	}
	status, getStatusErr := ssc.updateStatefulSet(set, currentRevision, updateRevision, collisionCount, pods, revisions)
	if getStatusErr != nil {
		return getStatusErr
	}
	updateStateusErr := ssc.updateStatefulSetStatus(set, status)
	if updateStateusErr != nil {
		return updateStateusErr
	}
	ssc.Log.Info("StatefulSe pod status",
		"ns", set.Namespace,
		"name", set.Name,
		"replicas", status.Replicas,
		"readyReplicas", status.ReadyReplicas,
		"currentReplicas", status.CurrentReplicas,
		"updateReplicas", status.UpdatedReplicas)
	ssc.Log.Info("StatefulSet revisions",
		"ns", set.Namespace,
		"name", set.Name,
		"currentRevision", status.CurrentRevision,
		"updateRevision", status.UpdateRevision,
	)
	return ssc.truncateHistory(set, pods, revisions, currentRevision, updateRevision)
}
func (ssc *defaultStatefulSetControl) ListRevisions(set *nuwav1.StatefulSet) ([]*apps.ControllerRevision, error) {
	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
	if err != nil {
		return nil, err
	}
	return ssc.controllerHistory.ListControllerRevisions(set, selector)
}
func (ssc *defaultStatefulSetControl) AdoptOrphanRevisions(
	set *nuwav1.StatefulSet,
	revisions []*apps.ControllerRevision) error {
	for i := range revisions {
		adopted, err := ssc.controllerHistory.AdoptControllerRevision(set, controllerKind, revisions[i])
		if err != nil {
			return err
		}
		revisions[i] = adopted
	}
	return nil
}
func (ssc *defaultStatefulSetControl) truncateHistory(
	set *nuwav1.StatefulSet,
	pods []*v1.Pod,
	revisions []*apps.ControllerRevision,
	current *apps.ControllerRevision,
	update *apps.ControllerRevision) error {
	history := make([]*apps.ControllerRevision, 0, len(revisions))
	live := map[string]bool{current.Name: true, update.Name: true}
	for i := range pods {
		live[getPodRevision(pods[i])] = true
	}
	for i := range revisions {
		if !live[revisions[i].Name] {
			history = append(history, revisions[i])
		}
	}
	historyLen := len(history)
	historyLimit := int(set.Spec.RevisionHistoryLimit)
	if historyLen <= historyLimit {
		return nil
	}
	history = history[:(historyLen - historyLimit)]
	for i := 0; i < len(history); i++ {
		if err := ssc.controllerHistory.DeleteControllerRevision(history[i]); err != nil {
			return err
		}
	}
	return nil
}
func (ssc *defaultStatefulSetControl) getStatefulSetRevisions(
	set *nuwav1.StatefulSet,
	revisions []*apps.ControllerRevision) (*apps.ControllerRevision, *apps.ControllerRevision, int32, error) {
	var currentRevision, updateRevision *apps.ControllerRevision
	revisionCount := len(revisions)
	history.SortControllerRevisions(revisions)
	var collisionCount int32
	if set.Status.CollisionCount != nil {
		collisionCount = *set.Status.CollisionCount
	}
	updateRevision, err := newRevision(set, nextRevision(revisions), &collisionCount)
	if err != nil {
		return nil, nil, collisionCount, err
	}
	equalRevisions := history.FindEqualRevisions(revisions, updateRevision)
	equalCount := len(equalRevisions)
	if equalCount > 0 && history.EqualRevision(revisions[revisionCount-1], equalRevisions[equalCount-1]) {
		updateRevision = revisions[revisionCount-1]
	} else if equalCount > 0 {
		updateRevision, err = ssc.controllerHistory.UpdateControllerRevision(
			equalRevisions[equalCount-1],
			updateRevision.Revision)
		if err != nil {
			return nil, nil, collisionCount, err
		}
	} else {
		updateRevision, err = ssc.controllerHistory.CreateControllerRevision(set, updateRevision, &collisionCount)
		if err != nil {
			return nil, nil, collisionCount, err
		}
	}
	for i := range revisions {
		if revisions[i].Name == set.Status.CurrentRevision {
			currentRevision = revisions[i]
		}
	}
	if currentRevision == nil {
		currentRevision = updateRevision
	}
	return currentRevision, updateRevision, collisionCount, nil
}