NK9: Controller is a separate module now

controller/controller.go

@@ -1,37 +1,56 @@
-package main
+package controller
 
 import (
 	"time"
 	"fmt"
 
+	"k8s.io/sample-controller/pkg/signals"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/client-go/tools/cache"
 
 	clientset "nirmata/kube-policy/pkg/client/clientset/versioned"
-	informer "nirmata/kube-policy/pkg/client/informers/externalversions/policy/v1alpha1"
+	informers "nirmata/kube-policy/pkg/client/informers/externalversions"
 	lister "nirmata/kube-policy/pkg/client/listers/policy/v1alpha1"
 )
 
 // Controller for CRD
 type Controller struct {
 	policyClientset clientset.Interface
+	policyInformerFactory informers.SharedInformerFactory
 	policyLister lister.PolicyLister
 	policiesSynced cache.InformerSynced
 	workqueue workqueue.RateLimitingInterface
 }
 
-// NewController is used to create Controller
+// NewController from cmd args
-func NewController(clientset clientset.Interface, informer informer.PolicyInformer) *Controller {
+func NewController(masterURL, kubeconfigPath string) (*Controller, error) {
-    controller := &Controller {
+	cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfigPath)
-		policyClientset: clientset,
+	if err != nil {
-		policyLister: informer.Lister(),
+		fmt.Printf("Error building kubeconfig: %v\n", err)
-		policiesSynced: informer.Informer().HasSynced,
+		return nil, err
+	}
+
+	policyClientset, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		fmt.Printf("Error building policy clientset: %v\n", err)
+		return nil, err
+	}
+
+	policyInformerFactory := informers.NewSharedInformerFactory(policyClientset, time.Second*30)
+	policyInformer := policyInformerFactory.Nirmata().V1alpha1().Policies()
+	
+	controller := &Controller {
+		policyClientset: policyClientset,
+		policyInformerFactory: policyInformerFactory,
+		policyLister: policyInformer.Lister(),
+		policiesSynced: policyInformer.Informer().HasSynced,
 		workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Policies"),
 	}
 
 	// Set up an event handler for when Foo resources change
-	informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+	policyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: controller.enqueueFoo,
 		UpdateFunc: func(old, new interface{}) {
 			controller.enqueueFoo(new)
@@ -39,11 +58,13 @@ func NewController(clientset clientset.Interface, informer informer.PolicyInform
 		DeleteFunc: controller.enqueueFoo,
 	})
 
-	return controller
+	return controller, nil
 }
 
 // Run is main controller thread
-func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
+func (c *Controller) Run(threadiness int) error {
+	stopCh := signals.SetupSignalHandler()
+	c.policyInformerFactory.Start(stopCh)
 
 	defer c.workqueue.ShutDown()
 

main.go

@@ -1,49 +1,33 @@
 package main
 
 import (
-	"time"
+    "runtime"
-	"flag"
+    "flag"
-	"fmt"
+    "fmt"
-	
-	"k8s.io/sample-controller/pkg/signals"
-	"k8s.io/client-go/tools/clientcmd"
 
-	clientset "nirmata/kube-policy/pkg/client/clientset/versioned"
+    controller "nirmata/kube-policy/controller"
-	informers "nirmata/kube-policy/pkg/client/informers/externalversions"
 )
 
 var (
-	masterURL  string
+    masterURL  string
-	kubeconfig string
+    kubeconfig string
 )
 
 func main() {
-	flag.Parse()
+    flag.Parse()
 
-	stopCh := signals.SetupSignalHandler()
+    controller, err := controller.NewController(masterURL, kubeconfig)
+    if err != nil {
+        fmt.Println("Error running Controller!")
+    }
 
-	cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
+    err = controller.Run(runtime.NumCPU())
-	if err != nil {
+    if err != nil {
-		fmt.Printf("Error building kubeconfig: %v\n", err)
+        fmt.Println("Error running Controller!")
-	}
+    }
-
-	exampleClient, err := clientset.NewForConfig(cfg)
-	if err != nil {
-		fmt.Printf("Error building example clientset: %v\n", err)
-	}
-
-	fmt.Println("Hello from Policy Controller!")
-
-	exampleInformerFactory := informers.NewSharedInformerFactory(exampleClient, time.Second*30)
-
-	controller := NewController(exampleClient, exampleInformerFactory.Nirmata().V1alpha1().Policies())
-	exampleInformerFactory.Start(stopCh)
-	if err = controller.Run(4, stopCh); err != nil {
-		fmt.Println("Error running Controller!")
-	}
 }
 
 func init() {
-	flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
+    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
-	flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+    flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
 }