func (tcc *Controller) deleteStatefulSet(obj interface{}) {
	set, ok := obj.(*apps.StatefulSet)
	ns := set.GetNamespace()
	setName := set.GetName()
	if !ok {
		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
		if !ok {
			utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %+v", obj))
			return
		}
		set, ok = tombstone.Obj.(*apps.StatefulSet)
		if !ok {
			utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a statefuset %+v", obj))
			return
		}
	}
	tc := tcc.resolveTikvClusterFromSet(ns, set)
	if tc == nil {
		return
	}
	klog.V(4).Infof("StatefulSet %s/%s deleted through %v.", ns, setName, utilruntime.GetCaller())
	tcc.enqueueTikvCluster(tc)
}
deleteStatefulSet enqueues the tikvcluster for the statefulset accounting for deletion tombstones.
package tikvcluster
import (
	"fmt"
	"time"
	perrors "github.com/pingcap/errors"
	"github.com/tikv/tikv-operator/pkg/apis/tikv/v1alpha1"
	"github.com/tikv/tikv-operator/pkg/client/clientset/versioned"
	informers "github.com/tikv/tikv-operator/pkg/client/informers/externalversions"
	listers "github.com/tikv/tikv-operator/pkg/client/listers/tikv/v1alpha1"
	"github.com/tikv/tikv-operator/pkg/controller"
	mm "github.com/tikv/tikv-operator/pkg/manager/member"
	"github.com/tikv/tikv-operator/pkg/manager/meta"
	"github.com/tikv/tikv-operator/pkg/pdapi"
	apps "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	eventv1 "k8s.io/client-go/kubernetes/typed/core/v1"
	appslisters "k8s.io/client-go/listers/apps/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog"
	"sigs.k8s.io/controller-runtime/pkg/client"
)
type Controller struct {
	kubeClient kubernetes.Interface
	cli versioned.Interface
	control ControlInterface
	tcLister listers.TikvClusterLister
	tcListerSynced cache.InformerSynced
	setLister appslisters.StatefulSetLister
	setListerSynced cache.InformerSynced
	queue workqueue.RateLimitingInterface
}
func NewController(
	kubeCli kubernetes.Interface,
	cli versioned.Interface,
	genericCli client.Client,
	informerFactory informers.SharedInformerFactory,
	kubeInformerFactory kubeinformers.SharedInformerFactory,
	autoFailover bool,
	pdFailoverPeriod time.Duration,
	tikvFailoverPeriod time.Duration,
) *Controller {
	eventBroadcaster := record.NewBroadcasterWithCorrelatorOptions(record.CorrelatorOptions{QPS: 1})
	eventBroadcaster.StartLogging(klog.V(2).Infof)
	eventBroadcaster.StartRecordingToSink(&eventv1.EventSinkImpl{
		Interface: eventv1.New(kubeCli.CoreV1().RESTClient()).Events("")})
	recorder := eventBroadcaster.NewRecorder(v1alpha1.Scheme, corev1.EventSource{Component: "tikv-controller-manager"})
	tcInformer := informerFactory.Tikv().V1alpha1().TikvClusters()
	setInformer := kubeInformerFactory.Apps().V1().StatefulSets()
	svcInformer := kubeInformerFactory.Core().V1().Services()
	epsInformer := kubeInformerFactory.Core().V1().Endpoints()
	pvcInformer := kubeInformerFactory.Core().V1().PersistentVolumeClaims()
	pvInformer := kubeInformerFactory.Core().V1().PersistentVolumes()
	podInformer := kubeInformerFactory.Core().V1().Pods()
	nodeInformer := kubeInformerFactory.Core().V1().Nodes()
	tcControl := controller.NewRealTikvClusterControl(cli, tcInformer.Lister(), recorder)
	pdControl := pdapi.NewDefaultPDControl(kubeCli)
	setControl := controller.NewRealStatefuSetControl(kubeCli, setInformer.Lister(), recorder)
	svcControl := controller.NewRealServiceControl(kubeCli, svcInformer.Lister(), recorder)
	pvControl := controller.NewRealPVControl(kubeCli, pvcInformer.Lister(), pvInformer.Lister(), recorder)
	pvcControl := controller.NewRealPVCControl(kubeCli, recorder, pvcInformer.Lister())
	podControl := controller.NewRealPodControl(kubeCli, pdControl, podInformer.Lister(), recorder)
	typedControl := controller.NewTypedControl(controller.NewRealGenericControl(genericCli, recorder))
	pdScaler := mm.NewPDScaler(pdControl, pvcInformer.Lister(), pvcControl)
	tikvScaler := mm.NewTiKVScaler(pdControl, pvcInformer.Lister(), pvcControl, podInformer.Lister())
	pdFailover := mm.NewPDFailover(cli, pdControl, pdFailoverPeriod, podInformer.Lister(), podControl, pvcInformer.Lister(), pvcControl, pvInformer.Lister(), recorder)
	tikvFailover := mm.NewTiKVFailover(tikvFailoverPeriod, recorder)
	pdUpgrader := mm.NewPDUpgrader(pdControl, podControl, podInformer.Lister())
	tikvUpgrader := mm.NewTiKVUpgrader(pdControl, podControl, podInformer.Lister())
	tcc := &Controller{
		kubeClient: kubeCli,
		cli:        cli,
		control: NewDefaultTikvClusterControl(
			tcControl,
			mm.NewPDMemberManager(
				pdControl,
				setControl,
				svcControl,
				podControl,
				typedControl,
				setInformer.Lister(),
				svcInformer.Lister(),
				podInformer.Lister(),
				epsInformer.Lister(),
				pvcInformer.Lister(),
				pdScaler,
				pdUpgrader,
				autoFailover,
				pdFailover,
			),
			mm.NewTiKVMemberManager(
				pdControl,
				setControl,
				svcControl,
				typedControl,
				setInformer.Lister(),
				svcInformer.Lister(),
				podInformer.Lister(),
				nodeInformer.Lister(),
				autoFailover,
				tikvFailover,
				tikvScaler,
				tikvUpgrader,
			),
			meta.NewMetaManager(
				pvcInformer.Lister(),
				pvcControl,
				pvInformer.Lister(),
				pvControl,
				podInformer.Lister(),
				podControl,
			),
			mm.NewOrphanPodsCleaner(
				podInformer.Lister(),
				podControl,
				pvcInformer.Lister(),
				kubeCli,
			),
			mm.NewPDDiscoveryManager(typedControl),
			&tikvClusterConditionUpdater{},
			recorder,
		),
		queue: workqueue.NewNamedRateLimitingQueue(
			workqueue.DefaultControllerRateLimiter(),
			"tikvcluster",
		),
	}
	tcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: tcc.enqueueTikvCluster,
		UpdateFunc: func(old, cur interface{}) {
			tcc.enqueueTikvCluster(cur)
		},
		DeleteFunc: tcc.enqueueTikvCluster,
	})
	tcc.tcLister = tcInformer.Lister()
	tcc.tcListerSynced = tcInformer.Informer().HasSynced
	setInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: tcc.addStatefulSet,
		UpdateFunc: func(old, cur interface{}) {
			tcc.updateStatefuSet(old, cur)
		},
		DeleteFunc: tcc.deleteStatefulSet,
	})
	tcc.setLister = setInformer.Lister()
	tcc.setListerSynced = setInformer.Informer().HasSynced
	return tcc
}
func (tcc *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tcc.queue.ShutDown()
	klog.Info("Starting tikvcluster controller")
	defer klog.Info("Shutting down tikvcluster controller")
	for i := 0; i < workers; i++ {
		go wait.Until(tcc.worker, time.Second, stopCh)
	}
	<-stopCh
}
func (tcc *Controller) worker() {
	for tcc.processNextWorkItem() {
	}
}
func (tcc *Controller) processNextWorkItem() bool {
	key, quit := tcc.queue.Get()
	if quit {
		return false
	}
	defer tcc.queue.Done(key)
	if err := tcc.sync(key.(string)); err != nil {
		if perrors.Find(err, controller.IsRequeueError) != nil {
			klog.Infof("TikvCluster: %v, still need sync: %v, requeuing", key.(string), err)
		} else {
			utilruntime.HandleError(fmt.Errorf("TikvCluster: %v, sync failed %v, requeuing", key.(string), err))
		}
		tcc.queue.AddRateLimited(key)
	} else {
		tcc.queue.Forget(key)
	}
	return true
}
func (tcc *Controller) sync(key string) error {
	startTime := time.Now()
	defer func() {
		klog.V(4).Infof("Finished syncing TikvCluster %q (%v)", key, time.Since(startTime))
	}()
	ns, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}
	tc, err := tcc.tcLister.TikvClusters(ns).Get(name)
	if errors.IsNotFound(err) {
		klog.Infof("TikvCluster has been deleted %v", key)
		return nil
	}
	if err != nil {
		return err
	}
	return tcc.syncTikvCluster(tc.DeepCopy())
}
func (tcc *Controller) syncTikvCluster(tc *v1alpha1.TikvCluster) error {
	return tcc.control.UpdateTikvCluster(tc)
}
func (tcc *Controller) enqueueTikvCluster(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("Cound't get key for object %+v: %v", obj, err))
		return
	}
	tcc.queue.Add(key)
}
func (tcc *Controller) addStatefulSet(obj interface{}) {
	set := obj.(*apps.StatefulSet)
	ns := set.GetNamespace()
	setName := set.GetName()
	if set.DeletionTimestamp != nil {
		tcc.deleteStatefulSet(set)
		return
	}
	tc := tcc.resolveTikvClusterFromSet(ns, set)
	if tc == nil {
		return
	}
	klog.V(4).Infof("StatefuSet %s/%s created, TikvCluster: %s/%s", ns, setName, ns, tc.Name)
	tcc.enqueueTikvCluster(tc)
}
func (tcc *Controller) updateStatefuSet(old, cur interface{}) {
	curSet := cur.(*apps.StatefulSet)
	oldSet := old.(*apps.StatefulSet)
	ns := curSet.GetNamespace()
	setName := curSet.GetName()
	if curSet.ResourceVersion == oldSet.ResourceVersion {
		return
	}
	tc := tcc.resolveTikvClusterFromSet(ns, curSet)
	if tc == nil {
		return
	}
	klog.V(4).Infof("StatefulSet %s/%s updated, %+v -> %+v.", ns, setName, oldSet.Spec, curSet.Spec)
	tcc.enqueueTikvCluster(tc)
}