func (c *Controller) enqueueClusterHelmRequest(obj interface{}, name string) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		utilruntime.HandleError(err)
		return
	}
	key = fmt.Sprintf("%s/%s", name, key)
	klog.Infof("enqueue helmrequest: %s", key)
	c.clusterWorkQueues[name].Add(key)
}
enqueueHelmRequest takes a HelmRequest resource and converts it into a namespace/name
string which is then put onto the work queue. This method should *not* be
passed resources of any type other than HelmRequest.
package controller
import (
	"fmt"
	"os"
	"strings"
	"time"
	"github.com/alauda/captain/pkg/cluster"
	"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"
	hrScheme "github.com/alauda/helm-crds/pkg/client/clientset/versioned/scheme"
	informers "github.com/alauda/helm-crds/pkg/client/informers/externalversions"
	corev1 "k8s.io/api/core/v1"
	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/kubernetes/scheme"
	"k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog"
)
func (c *Controller) initClusterWatches(stopCh <-chan struct{}) error {
	clusters, err := c.getAllClusters()
	if err != nil {
		return err
	}
	hrScheme.AddToScheme(scheme.Scheme)
	for _, cluster := range clusters {
		if cluster.Name == "global" {
			continue
		}
		if err := c.initWatchForCluster(stopCh, cluster); err != nil {
			continue
		}
	}
	return nil
}
func (c *Controller) initWatchForCluster(stopCh <-chan struct{}, cluster *cluster.Info) error {
	cfg := cluster.ToRestConfig()
	client, err := clientset.NewForConfig(cfg)
	if err != nil {
		klog.Warningf("init client for cluster %s error: %s", cluster.Name, err.Error())
		return err
	}
	coreClient, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		klog.Warningf("init core client for cluster %s error: %s", cluster.Name, err.Error())
		return err
	}
	if err := util.InstallHelmRequestCRD(cfg); err != nil {
		klog.Warningf("install helmrequest crd for cluster %s error: %s", cluster.Name, err.Error())
		return err
	}
	informerFactory := informers.NewSharedInformerFactory(client, defaultResyncDuration)
	informer := informerFactory.App().V1alpha1().HelmRequests()
	c.clusterHelmRequestListers[cluster.Name] = informer.Lister()
	c.clusterHelmRequestSynced[cluster.Name] = informer.Informer().HasSynced
	c.clusterWorkQueues[cluster.Name] = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), cluster.Name)
	c.clusterClients[cluster.Name] = client
	c.clusterRecorders[cluster.Name] = c.createEventRecorder(cluster.Name, coreClient)
	informer.Informer().AddEventHandler(c.newClusterHelmRequestHandler(cluster.Name))
	informerFactory.Start(stopCh)
	klog.Info("init watch config for cluster: ", cluster.Name)
	return nil
}
func (c *Controller) CleanupClusterWatch(name string) {
	c.clusterHelmRequestListers[name] = nil
	c.clusterHelmRequestSynced[name] = nil
	c.clusterWorkQueues[name] = nil
	c.clusterClients[name] = nil
	c.clusterRecorders[name] = nil
}
func (c *Controller) IsClusterWatchStarted(name string) bool {
	return c.clusterClients[name] != nil
}
func (c *Controller) restartClusterWatch(cluster *cluster.Info) error {
	if err := c.initWatchForCluster(c.stopCh, cluster); err != nil {
		return err
	}
	if err := c.startClusterWatch(cluster.Name, c.stopCh); err != nil {
		return err
	}
	klog.Info("restart watch for cluster done", cluster.Name)
	return nil
}
func (c *Controller) createEventRecorder(cluster string, client kubernetes.Interface) record.EventRecorder {
	klog.Info("Creating event broadcaster for cluster: ", cluster)
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(klog.Infof)
	eventBroadcaster.StartRecordingToSink(
		&v1.EventSinkImpl{
			Interface: client.CoreV1().Events(""),
		},
	)
	hostIP := os.Getenv("MY_POD_IP")
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: cluster, Host: hostIP})
	return recorder
}
func (c *Controller) startAllClustersWatch(stopCh <-chan struct{}) error {
	if err := c.initClusterWatches(stopCh); err != nil {
		return err
	}
	klog.Info("init cluster helmrequests watches done")
	for k := range c.clusterClients {
		if err := c.startClusterWatch(k, stopCh); err != nil {
			return err
		}
	}
	return nil
}
func (c *Controller) startClusterWatch(name string, stopCh <-chan struct{}) error {
	klog.Info("Starting HelmRequest controller for cluster: ", name)
	klog.Info("Waiting for informer caches to sync: ", name)
	if ok := cache.WaitForCacheSync(stopCh, c.clusterHelmRequestSynced[name]); !ok {
		return fmt.Errorf("failed to wait for caches to sync: %s", name)
	}
	klog.Info("Starting workers for clusters: ", name)
	f := func() {
		c.runClusterWorker(name)
	}
	for i := 0; i < 2; i++ {
		go wait.Until(f, time.Second, stopCh)
	}
	return nil
}
func (c *Controller) runClusterWorker(name string) {
	for c.processNextClusterWorkItem(name) {
	}
}
func (c *Controller) processNextClusterWorkItem(name string) bool {
	queue := c.clusterWorkQueues[name]
	obj, shutdown := queue.Get()
	if shutdown {
		klog.Warning("work queue has closed: ", name)
		return false
	}
	err := func(obj interface{}) error {
		defer queue.Done(obj)
		var key string
		var ok bool
		if key, ok = obj.(string); !ok {
			queue.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) {
				queue.AddRateLimited(key)
			}
			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
		}
		queue.Forget(obj)
		klog.Infof("Successfully synced '%s'", key)
		return nil
	}(obj)
	if err != nil {
		utilruntime.HandleError(err)
		return true
	}
	return true
}