func WatchForController(informer cache.SharedIndexInformer, q workqueue.Interface, fn GetControllerFn, m map[string]string) {
	enqueueFn := func(obj interface{}) {
		meta, ok := obj.(metav1.Object)
		if !ok {
			utilruntime.HandleError(fmt.Errorf("%+v is not a runtime.Object, cannot get controller from it", obj))
			return
		}
		if m != nil {
			l := meta.GetLabels()
			if !util.IsSubMapOf(m, l) {
				return
			}
		}
		ref := metav1.GetControllerOf(meta)
		if ref == nil {
			return
		}
		refGV, err := schema.ParseGroupVersion(ref.APIVersion)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("cannot parse group version for the controller %v of %s/%s",
				ref, meta.GetNamespace(), meta.GetName()))
			return
		}
		controllerObj, err := fn(meta.GetNamespace(), ref.Name)
		if err != nil {
			if errors.IsNotFound(err) {
				klog.V(4).Infof("controller %s/%s of %s/%s not found, ignore",
					meta.GetNamespace(), ref.Name, meta.GetNamespace(), meta.GetName())
			} else {
				utilruntime.HandleError(fmt.Errorf("cannot get controller %s/%s of %s/%s",
					meta.GetNamespace(), ref.Name, meta.GetNamespace(), meta.GetName()))
			}
			return
		}
		if ref.Kind == controllerObj.GetObjectKind().GroupVersionKind().Kind &&
			refGV.Group == controllerObj.GetObjectKind().GroupVersionKind().Group {
			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(controllerObj)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("Cound't get key for object %+v: %v", controllerObj, err))
				return
			}
			q.Add(key)
		}
	}
	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: enqueueFn,
		UpdateFunc: func(_, cur interface{}) {
			enqueueFn(cur)
		},
		DeleteFunc: enqueueFn,
	})
}
WatchForController watch the object change from informer and add it's controller to workqueue
package controller
import (
	"context"
	"fmt"
	"time"
	"github.com/dustin/go-humanize"
	"github.com/tikv/tikv-operator/pkg/apis/tikv/v1alpha1"
	"github.com/tikv/tikv-operator/pkg/scheme"
	"github.com/tikv/tikv-operator/pkg/util"
	corev1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/retry"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog"
	"sigs.k8s.io/controller-runtime/pkg/client"
)
var (
	ControllerKind = v1alpha1.SchemeGroupVersion.WithKind("TikvCluster")
	ClusterScoped bool
	TestMode bool
	ResyncDuration time.Duration
	PDDiscoveryImage string
)
const (
	defaultTiDBLogTailerImage = "busybox:1.26.2"
)
type RequeueError struct {
	s string
}
func (re *RequeueError) Error() string {
	return re.s
}
func RequeueErrorf(format string, a ...interface{}) error {
	return &RequeueError{fmt.Sprintf(format, a...)}
}
func IsRequeueError(err error) bool {
	_, ok := err.(*RequeueError)
	return ok
}
type IgnoreError struct {
	s string
}
func (re *IgnoreError) Error() string {
	return re.s
}
func IgnoreErrorf(format string, a ...interface{}) error {
	return &IgnoreError{fmt.Sprintf(format, a...)}
}
func IsIgnoreError(err error) bool {
	_, ok := err.(*IgnoreError)
	return ok
}
func GetOwnerRef(tc *v1alpha1.TikvCluster) metav1.OwnerReference {
	controller := true
	blockOwnerDeletion := true
	return metav1.OwnerReference{
		APIVersion:         ControllerKind.GroupVersion().String(),
		Kind:               ControllerKind.Kind,
		Name:               tc.GetName(),
		UID:                tc.GetUID(),
		Controller:         &controller,
		BlockOwnerDeletion: &blockOwnerDeletion,
	}
}
func TiKVCapacity(limits corev1.ResourceList) string {
	defaultArgs := "0"
	if limits == nil {
		return defaultArgs
	}
	q, ok := limits[corev1.ResourceStorage]
	if !ok {
		return defaultArgs
	}
	i, b := q.AsInt64()
	if !b {
		klog.Errorf("quantity %s can't be converted to int64", q.String())
		return defaultArgs
	}
	if i%humanize.GiByte == 0 {
		return fmt.Sprintf("%dGB", i/humanize.GiByte)
	}
	return fmt.Sprintf("%dMB", i/humanize.MiByte)
}
func PDMemberName(clusterName string) string {
	return fmt.Sprintf("%s-pd", clusterName)
}
func PDPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-pd-peer", clusterName)
}
func TiKVMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tikv", clusterName)
}
func TiKVPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tikv-peer", clusterName)
}
func TiFlashMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tiflash", clusterName)
}
func TiCDCMemberName(clusterName string) string {
	return fmt.Sprintf("%s-ticdc", clusterName)
}
func TiFlashPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tiflash-peer", clusterName)
}
func TiCDCPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-ticdc-peer", clusterName)
}
func TiDBMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tikv", clusterName)
}
func TiDBPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tikv-peer", clusterName)
}
func PumpMemberName(clusterName string) string {
	return fmt.Sprintf("%s-pump", clusterName)
}
func TiDBInitializerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-tikv-initializer", clusterName)
}
func PumpPeerMemberName(clusterName string) string {
	return fmt.Sprintf("%s-pump", clusterName)
}
func DiscoveryMemberName(clusterName string) string {
	return fmt.Sprintf("%s-discovery", clusterName)
}
func AnnProm(port int32) map[string]string {
	return map[string]string{
		"prometheus.io/scrape": "true",
		"prometheus.io/path":   "/metrics",
		"prometheus.io/port":   fmt.Sprintf("%d", port),
	}
}
func AnnAdditionalProm(name string, port int32) map[string]string {
	return map[string]string{
		fmt.Sprintf("%s.prometheus.io/port", name): fmt.Sprintf("%d", port),
	}
}
func ParseStorageRequest(req corev1.ResourceList) (corev1.ResourceRequirements, error) {
	if req == nil {
		return corev1.ResourceRequirements{}, nil
	}
	q, ok := req[corev1.ResourceStorage]
	if !ok {
		return corev1.ResourceRequirements{}, fmt.Errorf("storage request is not set")
	}
	return corev1.ResourceRequirements{
		Requests: corev1.ResourceList{
			corev1.ResourceStorage: q,
		},
	}, nil
}
func ContainerResource(req corev1.ResourceRequirements) corev1.ResourceRequirements {
	trimmed := req.DeepCopy()
	if trimmed.Limits != nil {
		delete(trimmed.Limits, corev1.ResourceStorage)
	}
	if trimmed.Requests != nil {
		delete(trimmed.Requests, corev1.ResourceStorage)
	}
	return *trimmed
}
func MemberConfigMapName(tc *v1alpha1.TikvCluster, member v1alpha1.MemberType) string {
	nameKey := fmt.Sprintf("%s-%s", tc.Name, member)
	return nameKey + getConfigMapSuffix(tc, member.String(), nameKey)
}
func getConfigMapSuffix(tc *v1alpha1.TikvCluster, component string, name string) string {
	if tc.Annotations == nil {
		return ""
	}
	sha := tc.Annotations[fmt.Sprintf("tikv.org/%s.%s.sha", component, name)]
	if len(sha) == 0 {
		return ""
	}
	return "-" + sha
}
func setIfNotEmpty(container map[string]string, key, value string) {
	if value != "" {
		container[key] = value
	}
}
func Int32Ptr(i int32) *int32 {
	return &i
}
type RequestTracker struct {
	requests int
	err      error
	after    int
}
func (rt *RequestTracker) ErrorReady() bool {
	return rt.err != nil && rt.requests >= rt.after
}
func (rt *RequestTracker) Inc() {
	rt.requests++
}
func (rt *RequestTracker) Reset() {
	rt.err = nil
	rt.after = 0
}
func (rt *RequestTracker) SetError(err error) *RequestTracker {
	rt.err = err
	return rt
}
func (rt *RequestTracker) SetAfter(after int) *RequestTracker {
	rt.after = after
	return rt
}
func (rt *RequestTracker) SetRequests(requests int) *RequestTracker {
	rt.requests = requests
	return rt
}
func (rt *RequestTracker) GetRequests() int {
	return rt.requests
}
func (rt *RequestTracker) GetError() error {
	return rt.err
}
func WatchForObject(informer cache.SharedIndexInformer, q workqueue.Interface) {
	enqueueFn := func(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
		}
		q.Add(key)
	}
	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: enqueueFn,
		UpdateFunc: func(_, cur interface{}) {
			enqueueFn(cur)
		},
		DeleteFunc: enqueueFn,
	})
}
type GetControllerFn func(ns, name string) (runtime.Object, error)