func (c *Controller) getDeployCluster(hr *alpha1.HelmRequest) string {
	if hr.Spec.ClusterName != "" {
		return hr.Spec.ClusterName
	}
	return hr.ClusterName
}
getDeployCluster returns the cluster name which the target Release lives in
package controller
import (
	"context"
	"fmt"
	"time"
	clusterclientset "github.com/alauda/captain/pkg/clusterregistry/client/clientset/versioned"
	"github.com/alauda/captain/pkg/config"
	"github.com/alauda/captain/pkg/helm"
	"github.com/alauda/captain/pkg/util"
	alpha1 "github.com/alauda/helm-crds/pkg/apis/app/v1alpha1"
	clientset "github.com/alauda/helm-crds/pkg/client/clientset/versioned"
	informers "github.com/alauda/helm-crds/pkg/client/informers/externalversions"
	listers "github.com/alauda/helm-crds/pkg/client/listers/app/v1alpha1"
	commoncache "github.com/patrickmn/go-cache"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"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/manager"
)
const (
	defaultResyncDuration = 10 * time.Minute
)
type clusterConfig struct {
	clusterClient clusterclientset.Interface
	clusterNamespace string
	globalClusterName string
}
type Controller struct {
	kubeClient kubernetes.Interface
	appClientSet clientset.Interface
	clusterConfig clusterConfig
	systemNamespace string
	restConfig *rest.Config
	helmRequestLister listers.HelmRequestLister
	helmRequestSynced cache.InformerSynced
	ClusterCache *commoncache.Cache
	workQueue workqueue.RateLimitingInterface
	recorder record.EventRecorder
	clusterHelmRequestListers map[string]listers.HelmRequestLister
	clusterHelmRequestSynced  map[string]cache.InformerSynced
	clusterWorkQueues         map[string]workqueue.RateLimitingInterface
	clusterClients            map[string]clientset.Interface
	clusterRecorders          map[string]record.EventRecorder
	stopCh <-chan struct{}
}
func NewController(mgr manager.Manager, opt *config.Options, ctx context.Context) (*Controller, error) {
	cfg := mgr.GetConfig()
	kubeClient, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	appClient, err := clientset.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	clusterClient, err := clusterclientset.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	appInformerFactory := informers.NewSharedInformerFactory(appClient, time.Second*30)
	informer := appInformerFactory.App().V1alpha1().HelmRequests()
	controller := &Controller{
		kubeClient:   kubeClient,
		appClientSet: appClient,
		clusterConfig: clusterConfig{
			clusterNamespace:  opt.ClusterNamespace,
			clusterClient:     clusterClient,
			globalClusterName: opt.GlobalClusterName,
		},
		systemNamespace:   opt.ChartRepoNamespace,
		restConfig:        cfg,
		recorder:          mgr.GetEventRecorderFor(util.ComponentName),
		helmRequestLister: informer.Lister(),
		helmRequestSynced: informer.Informer().HasSynced,
		workQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "HelmRequests"),
		ClusterCache: commoncache.New(1*time.Minute, 5*time.Minute),
		clusterHelmRequestListers: make(map[string]listers.HelmRequestLister),
		clusterHelmRequestSynced:  make(map[string]cache.InformerSynced),
		clusterWorkQueues:         make(map[string]workqueue.RateLimitingInterface),
		clusterClients:            make(map[string]clientset.Interface),
		clusterRecorders:          make(map[string]record.EventRecorder),
		stopCh: ctx.Done(),
	}
	klog.Info("Setting up event handlers")
	informer.Informer().AddEventHandler(controller.newHelmRequestHandler())
	klog.V(7).Infof("cluster rest config is : %+v", cfg)
	appInformerFactory.Start(ctx.Done())
	return controller, mgr.Add(controller)
}
func (c *Controller) GetClusterClient() clusterclientset.Interface {
	return c.clusterConfig.clusterClient
}
func (c *Controller) Start(ctx context.Context) error {
	defer utilruntime.HandleCrash()
	defer c.workQueue.ShutDown()
	klog.Info("Starting HelmRequest controller")
	if err := c.startAllClustersWatch(ctx.Done()); err != nil {
		return err
	}
	klog.Info("Waiting for informer caches to sync")
	if ok := cache.WaitForCacheSync(ctx.Done(), c.helmRequestSynced); !ok {
		return fmt.Errorf("failed to wait for caches to sync")
	}
	klog.Info("Starting workers")
	for i := 0; i < 2; i++ {
		go wait.Until(c.runWorker, time.Second, ctx.Done())
	}
	klog.Info("Started workers")
	<-ctx.Done()
	klog.Info("Shutting down workers")
	for _, v := range c.clusterWorkQueues {
		v.ShutDown()
	}
	return nil
}
func (c *Controller) runWorker() {
	for c.processNextWorkItem() {
	}
}
func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.workQueue.Get()
	if shutdown {
		return false
	}
	err := func(obj interface{}) error {
		defer c.workQueue.Done(obj)
		var key string
		var ok bool
		if key, ok = obj.(string); !ok {
			c.workQueue.Forget(obj)
			utilruntime.HandleError(fmt.Errorf("expected string in workQueue but got %#v", obj))
			return nil
		}
		if err := c.syncHandler(key); err != nil {
			if !apierrors.IsForbidden(err) {
				c.workQueue.AddRateLimited(key)
			}
			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
		}
		c.workQueue.Forget(obj)
		klog.Infof("Successfully synced '%s'", key)
		return nil
	}(obj)
	if err != nil {
		utilruntime.HandleError(err)
		return true
	}
	return true
}
func (c *Controller) updateHelmRequestStatus(helmRequest *alpha1.HelmRequest) error {
	h := helm.GenUniqueHash(helmRequest)
	request := helmRequest.DeepCopy()
	request.Status.LastSpecHash = h
	request.Status.Reason = ""
	request.Status.Phase = alpha1.HelmRequestSynced
	return helm.UpdateHelmRequestStatus(c.getAppClient(helmRequest), request)
}
func (c *Controller) setPartialSyncedStatus(helmRequest *alpha1.HelmRequest) error {
	h := helm.GenUniqueHash(helmRequest)
	request := helmRequest.DeepCopy()
	request.Status.LastSpecHash = h
	request.Status.Reason = ""
	return c.updateHelmRequestPhase(request, alpha1.HelmRequestPartialSynced)
}
func (c *Controller) setSyncFailedStatus(helmRequest *alpha1.HelmRequest, err error) error {
	c.sendFailedSyncEvent(helmRequest, err)
	request := helmRequest.DeepCopy()
	request.Status.Reason = err.Error()
	return c.updateHelmRequestPhase(request, alpha1.HelmRequestFailed)
}
func (c *Controller) setPendingStatus(helmRequest *alpha1.HelmRequest) error {
	return c.updateHelmRequestPhase(helmRequest, alpha1.HelmRequestPending)
}
func (c *Controller) getAppClient(hr *alpha1.HelmRequest) clientset.Interface {
	if hr.ClusterName == "" {
		return c.appClientSet
	}
	return c.clusterClients[hr.ClusterName]
}
func (c *Controller) getClusterAppClient(name string) clientset.Interface {
	if name == "" {
		return c.appClientSet
	}
	return c.clusterClients[name]
}
func (c *Controller) getAppClientForRelease(hr *alpha1.HelmRequest) clientset.Interface {
	if hr.Spec.ClusterName == "" {
		return c.getAppClient(hr)
	}
	if hr.Spec.ClusterName == "global" {
		return c.appClientSet
	}
	return c.clusterClients[hr.Spec.ClusterName]
}
func (c *Controller) getHelmRequestLister(name string) listers.HelmRequestLister {
	if name == "" {
		return c.helmRequestLister
	}
	return c.clusterHelmRequestListers[name]
}
func (c *Controller) getEventRecorder(hr *alpha1.HelmRequest) record.EventRecorder {
	if hr.ClusterName == "" {
		return c.recorder
	}
	return c.clusterRecorders[hr.ClusterName]
}