func (s *StatefulSetSyncer) updatePVC(ctx context.Context) error {
	pvcs := corev1.PersistentVolumeClaimList{}
	if err := s.cli.List(ctx,
		&pvcs,
		&client.ListOptions{
			Namespace:     s.sfs.Namespace,
			LabelSelector: s.GetLabels().AsSelector(),
		},
	); err != nil {
		return err
	}
	for _, item := range pvcs.Items {
		if item.DeletionTimestamp != nil {
			log.Info("pvc is being deleted", "pvc", item.Name, "key", s.Unwrap())
			continue
		}
		ordinal, err := utils.GetOrdinal(item.Name)
		if err != nil {
			log.Error(err, "pvc deletion error", "key", s.Unwrap())
			continue
		}
		if ordinal >= int(*s.Spec.Replicas) {
			log.Info("cleaning up pvc", "pvc", item.Name, "key", s.Unwrap())
			if err := s.cli.Delete(ctx, &item); err != nil {
				return err
			}
		}
	}
	return nil
}
updatePVC used to update the pvc, check and remove the extra pvc.
package syncer
import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"io"
	"net/http"
	"time"
	"github.com/iancoleman/strcase"
	"github.com/imdario/mergo"
	"github.com/presslabs/controller-util/mergo/transformers"
	"github.com/presslabs/controller-util/syncer"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	apiv1alpha1 "github.com/radondb/radondb-mysql-kubernetes/api/v1alpha1"
	"github.com/radondb/radondb-mysql-kubernetes/internal"
	"github.com/radondb/radondb-mysql-kubernetes/mysqlcluster"
	"github.com/radondb/radondb-mysql-kubernetes/mysqlcluster/container"
	"github.com/radondb/radondb-mysql-kubernetes/utils"
)
const waitLimit = 2 * 60 * 60
type StatefulSetSyncer struct {
	*mysqlcluster.MysqlCluster
	cli client.Client
	sfs *appsv1.StatefulSet
	cmRev string
	sctRev string
	internal.SQLRunnerFactory
}
func NewStatefulSetSyncer(cli client.Client, c *mysqlcluster.MysqlCluster, cmRev, sctRev string, sqlRunnerFactory internal.SQLRunnerFactory) *StatefulSetSyncer {
	return &StatefulSetSyncer{
		MysqlCluster: c,
		cli:          cli,
		sfs: &appsv1.StatefulSet{
			TypeMeta: metav1.TypeMeta{
				APIVersion: "v1",
				Kind:       "StatefulSet",
			},
			ObjectMeta: metav1.ObjectMeta{
				Name:      c.GetNameForResource(utils.StatefulSet),
				Namespace: c.Namespace,
			},
		},
		cmRev:            cmRev,
		sctRev:           sctRev,
		SQLRunnerFactory: sqlRunnerFactory,
	}
}
func (s *StatefulSetSyncer) Object() interface{} { return s.sfs }
func (s *StatefulSetSyncer) GetObject() interface{} { return s.sfs }
func (s *StatefulSetSyncer) ObjectOwner() runtime.Object { return s.Unwrap() }
func (s *StatefulSetSyncer) GetOwner() runtime.Object { return s.Unwrap() }
func (s *StatefulSetSyncer) Sync(ctx context.Context) (syncer.SyncResult, error) {
	var err error
	var kind string
	result := syncer.SyncResult{}
	result.Operation, err = s.createOrUpdate(ctx)
	key := client.ObjectKeyFromObject(s.sfs)
	gvk, gvkErr := apiutil.GVKForObject(s.sfs, s.cli.Scheme())
	if gvkErr != nil {
		kind = fmt.Sprintf("%T", s.sfs)
	} else {
		kind = gvk.String()
	}
	switch {
	case errors.Is(err, syncer.ErrOwnerDeleted):
		log.Info(string(result.Operation), "key", key, "kind", kind, "error", err)
		err = nil
	case errors.Is(err, syncer.ErrIgnore):
		log.Info("syncer skipped", "key", key, "kind", kind, "error", err)
		err = nil
	case err != nil:
		if k8serrors.IsInvalid(err) && s.canExpandPVC(ctx) {
			result.Operation, err = s.expandPVCs(ctx)
		} else {
			result.SetEventData("Warning", basicEventReason(s.Name, err),
				fmt.Sprintf("%s %s failed syncing: %s", kind, key, err))
			log.Error(err, string(result.Operation), "key", key, "kind", kind)
		}
	default:
		result.SetEventData("Normal", basicEventReason(s.Name, err),
			fmt.Sprintf("%s %s %s successfully", kind, key, result.Operation))
		log.Info(string(result.Operation), "key", key, "kind", kind)
	}
	return result, err
}
func (s *StatefulSetSyncer) canExpandPVC(ctx context.Context) bool {
	if err := s.cli.Get(ctx, client.ObjectKeyFromObject(s.sfs), s.sfs); err != nil {
		if k8serrors.IsNotFound(err) {
			return false
		}
	}
	oldRequest := s.sfs.Spec.VolumeClaimTemplates[0].Spec.Resources.Requests.DeepCopy()
	if err := s.mutate(); err != nil {
		return false
	}
	newStorage := s.sfs.Spec.VolumeClaimTemplates[0].Spec.Resources.Requests.Storage()
	if newStorage.Cmp(*oldRequest.Storage()) != 1 {
		log.Info("canExpandPVC", "result", "can not expand", "reason", "new pvc is not larger than old pvc")
		return false
	}
	return true
}
func (s *StatefulSetSyncer) expandPVCs(ctx context.Context) (controllerutil.OperationResult, error) {
	if err := s.cli.Delete(ctx, s.sfs); err != nil {
		return controllerutil.OperationResultNone, err
	}
	if err := s.doExpandPVCs(ctx); err != nil {
		log.Error(err, "expandPVCs")
		return controllerutil.OperationResultNone, err
	}
	if err := s.cli.Create(ctx, s.sfs); err != nil {
		return controllerutil.OperationResultNone, err
	} else {
		return controllerutil.OperationResultCreated, nil
	}
}
func (s *StatefulSetSyncer) doExpandPVCs(ctx context.Context) error {
	pvcs := corev1.PersistentVolumeClaimList{}
	if err := s.cli.List(ctx,
		&pvcs,
		&client.ListOptions{
			Namespace:     s.sfs.Namespace,
			LabelSelector: s.GetLabels().AsSelector(),
		},
	); err != nil {
		return err
	}
	for _, item := range pvcs.Items {
		item.Spec.Resources.Requests = s.sfs.Spec.VolumeClaimTemplates[0].Spec.Resources.Requests
		if err := s.cli.Update(ctx, &item); err != nil {
			return err
		}
		if err := retry(time.Second*2, time.Duration(waitLimit)*time.Second, func() (bool, error) {
			var currentPVC corev1.PersistentVolumeClaim
			if err2 := s.cli.Get(ctx, client.ObjectKeyFromObject(&item), &currentPVC); err2 != nil {
				return true, err2
			}
			var conditons = currentPVC.Status.Conditions
			if conditons == nil {
				if equality.Semantic.DeepEqual(currentPVC.Status.Capacity, item.Spec.Resources.Requests) {
					return true, nil
				}
				return false, nil
			}
			status := conditons[0].Type
			if status == "FileSystemResizePending" {
				return true, nil
			}
			return false, nil
		}); err != nil {
			return err
		}
	}
	return nil
}
func (s *StatefulSetSyncer) createOrUpdate(ctx context.Context) (controllerutil.OperationResult, error) {
	var err error
	if err = s.cli.Get(ctx, client.ObjectKeyFromObject(s.sfs), s.sfs); err != nil {
		if !k8serrors.IsNotFound(err) {
			return controllerutil.OperationResultNone, err
		}
		if err = s.mutate(); err != nil {
			return controllerutil.OperationResultNone, err
		}
		if err = s.cli.Create(ctx, s.sfs); err != nil {
			return controllerutil.OperationResultNone, err
		} else {
			return controllerutil.OperationResultCreated, nil
		}
	}
	existing := s.sfs.DeepCopyObject()
	if err = s.mutate(); err != nil {
		return controllerutil.OperationResultNone, err
	}
	if equality.Semantic.DeepEqual(existing, s.sfs) {
		return controllerutil.OperationResultNone, nil
	}
	if err := s.cli.Update(ctx, s.sfs); err != nil {
		return controllerutil.OperationResultNone, err
	}
	if err := s.updatePod(ctx); err != nil {
		return controllerutil.OperationResultNone, err
	}
	if err := s.updatePVC(ctx); err != nil {
		return controllerutil.OperationResultNone, err
	}
	return controllerutil.OperationResultUpdated, nil
}
func (s *StatefulSetSyncer) updatePod(ctx context.Context) error {
	if s.sfs.Status.UpdatedReplicas >= s.sfs.Status.Replicas {
		return nil
	}
	log.Info("statefulSet was changed, run update")
	if s.sfs.Status.ReadyReplicas < s.sfs.Status.Replicas {
		log.Info("can't start/continue 'update': waiting for all replicas are ready")
		return nil
	}
	if backuping, _ := s.backupIsRunning(ctx); backuping {
		return fmt.Errorf("can't start/continue 'update': waiting for all backup completed")
	}
	pods := corev1.PodList{}
	if err := s.cli.List(ctx,
		&pods,
		&client.ListOptions{
			Namespace:     s.sfs.Namespace,
			LabelSelector: s.GetLabels().AsSelector(),
		},
	); err != nil {
		return err
	}
	var leaderPod corev1.Pod
	var followerPods []corev1.Pod
	for _, pod := range pods.Items {
		if pod.ObjectMeta.Labels["healthy"] != "yes" {
			return fmt.Errorf("can't start/continue 'update': pod[%s] is unhealthy", pod.Name)
		}
		if pod.ObjectMeta.Labels["role"] == "leader" && leaderPod.Name == "" {
			leaderPod = pod
			continue
		}
		followerPods = append(followerPods, pod)
		if err := s.applyNWait(ctx, &pod); err != nil {
			return err
		}
	}
	if leaderPod.Name != "" {
		if *s.sfs.Spec.Replicas == 2 {
			if err := s.preUpdate(ctx, leaderPod.Name, followerPods[0].Name); err != nil {
				return err
			}
		}
		if err := s.applyNWait(ctx, &leaderPod); err != nil {
			return err
		}
	}
	return nil
}
func (s *StatefulSetSyncer) preUpdate(ctx context.Context, leader, follower string) error {
	leaderRunner, closeConn, err := s.SQLRunnerFactory(internal.NewConfigFromClusterKey(
		s.cli, s.MysqlCluster.GetClusterKey(), utils.OperatorUser, utils.LeaderHost))
	if err != nil {
		return err
	}
	defer closeConn()
	if *s.sfs.Spec.Replicas != 2 {
		return nil
	}
	utils.TouchUpdateFile()
	defer utils.RemoveUpdateFile()
	sctName := s.GetNameForResource(utils.Secret)
	svcName := s.GetNameForResource(utils.HeadlessSVC)
	nameSpace := s.Namespace
	secret := &corev1.Secret{}
	if err := s.cli.Get(context.TODO(),
		types.NamespacedName{
			Namespace: nameSpace,
			Name:      sctName,
		},
		secret,
	); err != nil {
		return fmt.Errorf("failed to get the secret: %s", sctName)
	}
	rootPasswd, ok := secret.Data["root-password"]
	if !ok {
		return fmt.Errorf("failed to get the root password: %s", rootPasswd)
	}
	if err = retry(time.Second*2, time.Duration(waitLimit)*time.Second, func() (bool, error) {
		if err = leaderRunner.QueryExec(internal.NewQuery("SET GLOBAL super_read_only=on;")); err != nil {
			log.Error(err, "failed to set leader read only", "node", leader)
			return false, err
		}
		success, err := internal.CheckProcesslist(leaderRunner)
		if err != nil {
			return false, err
		}
		if success {
			return true, nil
		}
		return false, nil
	}); err != nil {
		return err
	}
	followerHost := fmt.Sprintf("%s.%s.%s", follower, svcName, nameSpace)
	if err = retry(time.Second*5, time.Second*60, func() (bool, error) {
		status, err := checkRole(followerHost, rootPasswd)
		if err != nil {
			log.Error(err, "failed to check role", "pod", follower)
			return false, nil
		}
		if status == corev1.ConditionTrue {
			return true, nil
		}
		xenonHttpRequest(followerHost, "POST", "/v1/raft/trytoleader", rootPasswd, nil)
		return false, nil
	}); err != nil {
		return err
	}
	return nil
}
func (s *StatefulSetSyncer) mutate() error {
	s.sfs.Spec.ServiceName = s.GetNameForResource(utils.StatefulSet)
	s.sfs.Spec.Replicas = s.Spec.Replicas
	s.sfs.Spec.Selector = metav1.SetAsLabelSelector(s.GetSelectorLabels())
	s.sfs.Spec.UpdateStrategy = appsv1.StatefulSetUpdateStrategy{
		Type: appsv1.OnDeleteStatefulSetStrategyType,
	}
	s.sfs.Spec.Template.ObjectMeta.Labels = s.GetLabels()
	for k, v := range s.Spec.PodPolicy.Labels {
		s.sfs.Spec.Template.ObjectMeta.Labels[k] = v
	}
	s.sfs.Spec.Template.ObjectMeta.Labels["role"] = "candidate"
	s.sfs.Spec.Template.ObjectMeta.Labels["healthy"] = "no"
	s.sfs.Spec.Template.Annotations = s.Spec.PodPolicy.Annotations
	if len(s.sfs.Spec.Template.ObjectMeta.Annotations) == 0 {
		s.sfs.Spec.Template.ObjectMeta.Annotations = make(map[string]string)
	}
	if s.Spec.MetricsOpts.Enabled {
		s.sfs.Spec.Template.ObjectMeta.Annotations["prometheus.io/scrape"] = "true"
		s.sfs.Spec.Template.ObjectMeta.Annotations["prometheus.io/port"] = fmt.Sprintf("%d", utils.MetricsPort)
	}
	s.sfs.Spec.Template.ObjectMeta.Annotations["config_rev"] = s.cmRev
	s.sfs.Spec.Template.ObjectMeta.Annotations["secret_rev"] = s.sctRev
	err := mergo.Merge(&s.sfs.Spec.Template.Spec, s.ensurePodSpec(), mergo.WithTransformers(transformers.PodSpec))
	if err != nil {
		return err
	}
	s.sfs.Spec.Template.Spec.Tolerations = s.Spec.PodPolicy.Tolerations
	if s.Spec.Persistence.Enabled {
		if s.sfs.Spec.VolumeClaimTemplates, err = s.EnsureVolumeClaimTemplates(s.cli.Scheme()); err != nil {
			return err
		}
	}
	if s.Unwrap().GetDeletionTimestamp().IsZero() {
		if err := controllerutil.SetControllerReference(s.Unwrap(), s.sfs, s.cli.Scheme()); err != nil {
			return err
		}
	} else if ctime := s.Unwrap().GetCreationTimestamp(); ctime.IsZero() {
		return fmt.Errorf("owner is deleted")
	}
	return nil
}
func (s *StatefulSetSyncer) ensurePodSpec() corev1.PodSpec {
	initSidecar := container.EnsureContainer(utils.ContainerInitSidecarName, s.MysqlCluster)
	initMysql := container.EnsureContainer(utils.ContainerInitMysqlName, s.MysqlCluster)
	initContainers := []corev1.Container{initSidecar, initMysql}
	mysql := container.EnsureContainer(utils.ContainerMysqlName, s.MysqlCluster)
	xenon := container.EnsureContainer(utils.ContainerXenonName, s.MysqlCluster)
	backup := container.EnsureContainer(utils.ContainerBackupName, s.MysqlCluster)
	containers := []corev1.Container{mysql, xenon, backup}
	if s.Spec.MetricsOpts.Enabled {
		containers = append(containers, container.EnsureContainer(utils.ContainerMetricsName, s.MysqlCluster))
	}
	if s.Spec.PodPolicy.SlowLogTail {
		containers = append(containers, container.EnsureContainer(utils.ContainerSlowLogName, s.MysqlCluster))
	}
	if s.Spec.PodPolicy.AuditLogTail {
		containers = append(containers, container.EnsureContainer(utils.ContainerAuditLogName, s.MysqlCluster))
	}
	return corev1.PodSpec{
		InitContainers:     initContainers,
		Containers:         containers,
		Volumes:            s.EnsureVolumes(),
		SchedulerName:      s.Spec.PodPolicy.SchedulerName,
		ServiceAccountName: s.GetNameForResource(utils.ServiceAccount),
		Affinity:           s.Spec.PodPolicy.Affinity,
		PriorityClassName:  s.Spec.PodPolicy.PriorityClassName,
		Tolerations:        s.Spec.PodPolicy.Tolerations,
	}
}