Implement a (broken) informer/watcher

Author: jbeda

.gitignore

@@ -14,7 +14,7 @@
 .glide/
 
 /kubeconfig
-/main
+/tgik-controller
 
 /bin
 /.go

controller.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"log"
+
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	listercorev1 "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+)
+
+type TGIKController struct {
+	podGetter       corev1.PodsGetter
+	podLister       listercorev1.PodLister
+	podListerSynced cache.InformerSynced
+}
+
+func NewTGIKController(client *kubernetes.Clientset, informerFactory informers.SharedInformerFactory) *TGIKController {
+	informer := informerFactory.Core().V1().Pods()
+	c := &TGIKController{
+		podGetter:       client.CoreV1(),
+		podLister:       informer.Lister(),
+		podListerSynced: informer.Informer().HasSynced,
+	}
+
+	informer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				c.onAdd(obj)
+			},
+			UpdateFunc: func(oldObj, newObj interface{}) {
+				c.onUpdate(oldObj, newObj)
+			},
+			DeleteFunc: func(obj interface{}) {
+				c.onDelete(obj)
+			},
+		},
+	)
+
+	return c
+}
+
+func (c *TGIKController) Run(stop <-chan struct{}) {
+	log.Print("waiting for cache sync")
+	if !cache.WaitForCacheSync(stop, c.podListerSynced) {
+		log.Print("timed out waiting for cache sync")
+		return
+	}
+	log.Print("caches are synced")
+
+	// wait until we're told to stop
+	log.Print("waiting for stop signal")
+	<-stop
+	log.Print("received stop signal")
+}
+
+func (c *TGIKController) onAdd(obj interface{}) {
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		log.Printf("onAdd: error getting key for %#v: %v", obj, err)
+		runtime.HandleError(err)
+	}
+	log.Printf("onAdd: %v", key)
+}
+
+func (c *TGIKController) onUpdate(oldObj, _ interface{}) {
+	key, err := cache.MetaNamespaceKeyFunc(oldObj)
+	if err != nil {
+		log.Printf("onUpdate: error getting key for %#v: %v", oldObj, err)
+		runtime.HandleError(err)
+	}
+	log.Printf("onUpdate: %v", key)
+}
+
+func (c *TGIKController) onDelete(obj interface{}) {
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		runtime.HandleError(err)
+	}
+	log.Printf("onDelete: %v", key)
+}

tgik-controller.go

@@ -3,15 +3,20 @@ package main
 import (
 	"flag"
 	"fmt"
+	"log"
 	"os"
+	"time"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"github.com/jbeda/tgik-controller/version"
+
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
 
 func main() {
+	log.Printf("tgik-controller version %s", version.VERSION)
 	kubeconfig := ""
 	flag.StringVar(&kubeconfig, "kubeconfig", kubeconfig, "kubeconfig file")
 	flag.Parse()
@@ -32,12 +37,9 @@ func main() {
 		os.Exit(1)
 	}
 	client := kubernetes.NewForConfigOrDie(config)
-	list, err := client.CoreV1().Nodes().List(metav1.ListOptions{})
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "error listing nodes: %v", err)
-		os.Exit(1)
-	}
-	for _, node := range list.Items {
-		fmt.Printf("Node: %s\n", node.Name)
-	}
+
+	sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)
+
+	tgikController := NewTGIKController(client, sharedInformers)
+	tgikController.Run(nil)
 }