Playing with generic informers

generic/client.go

@@ -1,18 +1,24 @@
 package generic
 
 import (
+	"log"
 	"bytes"
 	"context"
 	"encoding/json"
+	"time"
 
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/rest"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const resyncPeriod = time.Hour
+
 // Ideally, this wouldn't need to take a GVR, and it could just be inferred
 // from the runtime.Object type given.
 //
@@ -27,15 +33,25 @@ import (
 // In the meantime, taking a GVR removes ambiguity at the cost of verbosity.
 // /shrug
 func NewClient[T runtime.Object](gvr schema.GroupVersionResource, config *rest.Config) client[T] {
+	dyn:= dynamic.NewForConfigOrDie(config)
 	return client[T] {
 		gvr: gvr,
-		dyn: dynamic.NewForConfigOrDie(config),
+		dyn: dyn,
+		dsif: dynamicinformer.NewDynamicSharedInformerFactory(dyn, resyncPeriod),
 	}
 }
 
 type client[T runtime.Object] struct {
 	gvr schema.GroupVersionResource
-	dyn dynamic.Interface // TODO: don't depend on dynamic client
+
+	// TODO: don't depend on dynamic client
+	dyn dynamic.Interface 
+	dsif dynamicinformer.DynamicSharedInformerFactory
+}
+
+func (c client[T]) Start(ctx context.Context) {
+	go c.dsif.Start(ctx.Done())
+	c.dsif.WaitForCacheSync(ctx.Done())
 }
 
 func (c client[T]) List(ctx context.Context, namespace string) ([]T, error) {
@@ -76,3 +92,25 @@ func (c client[T]) Create(ctx context.Context, namespace string, t T) error {
 	return err
 }
 
+func (c client[T]) Inform(ctx context.Context) {
+	inf := c.dsif.ForResource(c.gvr).Informer()
+	inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			key, _ := cache.MetaNamespaceKeyFunc(obj)
+			log.Println("--> ADD", key)
+	},
+		UpdateFunc: func(_, obj interface{}) { 
+			key, _ := cache.MetaNamespaceKeyFunc(obj)
+			log.Println("--> UPDATE", key)
+	},
+		DeleteFunc: func(obj interface{}) { 
+			key, _ := cache.MetaNamespaceKeyFunc(obj)
+			log.Println("--> DELETE", key)
+		},
+	})
+	go inf.Run(ctx.Done())
+	if !cache.WaitForNamedCacheSync(c.gvr.String(), ctx.Done(), inf.HasSynced) {
+		log.Println("Failed to wait for caches to sync:"< c.gvr.String())
+		return
+	}
+}

go.mod

@@ -15,6 +15,8 @@ require (
 	github.com/golang/protobuf v1.4.3 // indirect
 	github.com/google/go-cmp v0.5.2 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
+	github.com/googleapis/gnostic v0.4.1 // indirect
+	github.com/hashicorp/golang-lru v0.5.1 // indirect
 	github.com/imdario/mergo v0.3.5 // indirect
 	github.com/json-iterator/go v1.1.10 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect

go.sum

@@ -111,10 +111,12 @@ github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
+github.com/googleapis/gnostic v0.4.1 h1:DLJCy1n/vrD4HPjOvYcT8aYQXpPIzoRZONaYwyycI+I=
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=

main.go

@@ -36,13 +36,18 @@ func main() {
 	if err != nil {
 		log.Fatal("listing pods:", err)
 	}
-	log.Println("PODS")
+	log.Println("LISTING PODS")
 	for _, p := range pods {
 		log.Println("-", p.Name)
 	}
 
-	// Create a ConfigMap, then list ConfigMaps.
 	cmc := generic.NewClient[*corev1.ConfigMap](cmGVR, config)
+	cmc.Start(ctx)
+
+	// Start an informer to log all adds/updates/deletes for ConfigMaps.
+	cmc.Inform(ctx)
+
+	// Create a ConfigMap, then list ConfigMaps.
 	if err := cmc.Create(ctx, "kube-system", &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: "foo-",
@@ -57,7 +62,7 @@ func main() {
 	if err != nil {
 		log.Fatal("listing configmaps:", err)
 	}
-	log.Println("CONFIGMAPS")
+	log.Println("LISTING CONFIGMAPS")
 	for _, cm := range cms {
 		log.Println("-", cm.Name)
 	}