func (c *Controller) enqueueBuildJob(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		runtime.HandleError(err)
		return
	}
	c.workqueue.AddRateLimited(key)
}
enqueueBuildJob takes a BuildJob 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 BuildJob.
package controller
import (
	"context"
	"fmt"
	"time"
	"github.com/golang/glog"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	batchlisters "k8s.io/client-go/listers/batch/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	cbiv1alpha1 "github.com/containerbuilding/cbi/pkg/apis/cbi/v1alpha1"
	"github.com/containerbuilding/cbi/pkg/cbid/pluginselector"
	clientset "github.com/containerbuilding/cbi/pkg/client/clientset/versioned"
	cbischeme "github.com/containerbuilding/cbi/pkg/client/clientset/versioned/scheme"
	informers "github.com/containerbuilding/cbi/pkg/client/informers/externalversions"
	listers "github.com/containerbuilding/cbi/pkg/client/listers/cbi/v1alpha1"
)
const controllerAgentName = "cbid"
const (
	SuccessSynced = "Synced"
	ErrResourceExists = "ErrResourceExists"
	MessageResourceExists = "Resource %q already exists and is not managed by BuildJob"
	MessageResourceSynced = "BuildJob synced successfully"
)
type Controller struct {
	kubeclientset kubernetes.Interface
	cbiclientset    clientset.Interface
	jobsLister      batchlisters.JobLister
	jobsSynced      cache.InformerSynced
	buildJobsLister listers.BuildJobLister
	buildJobsSynced cache.InformerSynced
	workqueue workqueue.RateLimitingInterface
	recorder record.EventRecorder
	pluginSelector *pluginselector.PluginSelector
}
func New(
	kubeclientset kubernetes.Interface,
	cbiclientset clientset.Interface,
	kubeInformerFactory kubeinformers.SharedInformerFactory,
	cbiInformerFactory informers.SharedInformerFactory,
	pluginSelector *pluginselector.PluginSelector) *Controller {
	jobInformer := kubeInformerFactory.Batch().V1().Jobs()
	buildJobInformer := cbiInformerFactory.Cbi().V1alpha1().BuildJobs()
	cbischeme.AddToScheme(scheme.Scheme)
	glog.V(4).Info("Creating event broadcaster")
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
	controller := &Controller{
		kubeclientset:   kubeclientset,
		cbiclientset:    cbiclientset,
		jobsLister:      jobInformer.Lister(),
		jobsSynced:      jobInformer.Informer().HasSynced,
		buildJobsLister: buildJobInformer.Lister(),
		buildJobsSynced: buildJobInformer.Informer().HasSynced,
		workqueue:       workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "BuildJobs"),
		recorder:        recorder,
		pluginSelector:  pluginSelector,
	}
	glog.Info("Setting up event handlers")
	buildJobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueBuildJob,
		UpdateFunc: func(old, new interface{}) {
			controller.enqueueBuildJob(new)
		},
	})
	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newJob := new.(*batchv1.Job)
			oldJob := old.(*batchv1.Job)
			if newJob.ResourceVersion == oldJob.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	return controller
}
func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
	defer runtime.HandleCrash()
	defer c.workqueue.ShutDown()
	glog.Info("Starting BuildJob controller")
	glog.Info("Waiting for informer caches to sync")
	if ok := cache.WaitForCacheSync(stopCh, c.jobsSynced, c.buildJobsSynced); !ok {
		return fmt.Errorf("failed to wait for caches to sync")
	}
	glog.Info("Starting workers")
	for i := 0; i < threadiness; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}
	glog.Info("Started workers")
	<-stopCh
	glog.Info("Shutting down workers")
	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)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}
		if err := c.syncHandler(key); err != nil {
			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
		}
		c.workqueue.Forget(obj)
		glog.Infof("Successfully synced '%s'", key)
		return nil
	}(obj)
	if err != nil {
		runtime.HandleError(err)
		return true
	}
	return true
}
func (c *Controller) syncHandler(key string) error {
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}
	buildJob, err := c.buildJobsLister.BuildJobs(namespace).Get(name)
	if err != nil {
		if errors.IsNotFound(err) {
			runtime.HandleError(fmt.Errorf("BuildJob '%s' in work queue no longer exists", key))
			return nil
		}
		return err
	}
	if buildJob.Name == "" {
		runtime.HandleError(fmt.Errorf("%s: invalid BuildJob spec", key))
		return nil
	}
	pluginClient := c.pluginSelector.Select(*buildJob)
	if pluginClient == nil {
		runtime.HandleError(fmt.Errorf("%s: no plugin support this spec", key))
		return nil
	}
	jobManifest, err := newJob(context.TODO(), pluginClient, buildJob)
	if err != nil {
		runtime.HandleError(fmt.Errorf("%s: invalid BuildJob spec: %v", key, err))
		return nil
	}
	job, err := c.jobsLister.Jobs(buildJob.Namespace).Get(jobManifest.ObjectMeta.Name)
	if errors.IsNotFound(err) {
		job, err = c.kubeclientset.BatchV1().Jobs(buildJob.Namespace).Create(jobManifest)
	}
	if err != nil {
		return err
	}
	if !metav1.IsControlledBy(job, buildJob) {
		msg := fmt.Sprintf(MessageResourceExists, job.Name)
		c.recorder.Event(buildJob, corev1.EventTypeWarning, ErrResourceExists, msg)
		return fmt.Errorf(msg)
	}
	updateJob := false  
	if updateJob {
		job, err = c.kubeclientset.BatchV1().Jobs(buildJob.Namespace).Update(jobManifest)
	}
	if err != nil {
		return err
	}
	err = c.updateBuildJobStatus(buildJob, job)
	if err != nil {
		return err
	}
	c.recorder.Event(buildJob, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
	return nil
}
func (c *Controller) updateBuildJobStatus(buildJob *cbiv1alpha1.BuildJob, job *batchv1.Job) error {
	buildJobCopy := buildJob.DeepCopy()
	buildJobCopy.Status.Job = job.Name
	_, err := c.cbiclientset.CbiV1alpha1().BuildJobs(buildJob.Namespace).Update(buildJobCopy)
	return err
}