support unwatchables in vai (#458)

* Create and use a synthetic watcher for non-watchable resources.

* Write unit tests for the synthetic watcher.

* Make the refresh interval for synthetic watchers configurable.

The default is to call `client.List(...)` every 5 seconds for each unwatchable GVK.

There are currently only 3 such GVKs right now so this will be informative
enough but not really noticeable.

* Pass the context into the synthetic watch func.

* Restore changes lost in rebasing.

---------

Co-authored-by: Tom Lebreux <tom.lebreux@suse.com>

Author: ericpromislow

pkg/sqlcache/informer/factory/informer_factory.go

@@ -44,7 +44,7 @@ type guardedInformer struct {
 	mutex    *sync.Mutex
 }
 
-type newInformer func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespace bool) (*informer.Informer, error)
+type newInformer func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespace bool, watchable bool) (*informer.Informer, error)
 
 type Cache struct {
 	informer.ByOptionsLister
@@ -121,7 +121,7 @@ func (f *CacheFactory) CacheFor(ctx context.Context, fields [][]string, transfor
 
 		_, encryptResourceAlways := defaultEncryptedResourceTypes[gvk]
 		shouldEncrypt := f.encryptAll || encryptResourceAlways
-		i, err := f.newInformer(ctx, client, fields, transform, gvk, f.dbClient, shouldEncrypt, namespaced)
+		i, err := f.newInformer(ctx, client, fields, transform, gvk, f.dbClient, shouldEncrypt, namespaced, watchable)
 		if err != nil {
 			return Cache{}, err
 		}

pkg/sqlcache/informer/factory/informer_factory_test.go

@@ -74,7 +74,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -119,7 +119,7 @@ func TestCacheFor(t *testing.T) {
 			// need to set this so Run function is not nil
 			SharedIndexInformer: sii,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -161,7 +161,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -200,7 +200,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -248,7 +248,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -295,7 +295,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt, namespaced, watchable bool) (*informer.Informer, error) {
 			assert.Equal(t, client, dynamicClient)
 			assert.Equal(t, fields, fields)
 			assert.Equal(t, expectedGVK, gvk)
@@ -342,7 +342,7 @@ func TestCacheFor(t *testing.T) {
 		expectedC := Cache{
 			ByOptionsLister: i,
 		}
-		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool) (*informer.Informer, error) {
+		testNewInformer := func(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool, watchable bool) (*informer.Informer, error) {
 			// we can't test func == func, so instead we check if the output was as expected
 			input := "someinput"
 			ouput, err := transform(input)

pkg/sqlcache/informer/informer.go

@@ -20,6 +20,8 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
+var defaultRefreshTime = 5 * time.Second
+
 // Informer is a SQLite-backed cache.SharedIndexInformer that can execute queries on listprocessor structs
 type Informer struct {
 	cache.SharedIndexInformer
@@ -35,15 +37,22 @@ var newInformer = cache.NewSharedIndexInformer
 
 // NewInformer returns a new SQLite-backed Informer for the type specified by schema in unstructured.Unstructured form
 // using the specified client
-func NewInformer(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool) (*Informer, error) {
+func NewInformer(ctx context.Context, client dynamic.ResourceInterface, fields [][]string, transform cache.TransformFunc, gvk schema.GroupVersionKind, db db.Client, shouldEncrypt bool, namespaced bool, watchable bool) (*Informer, error) {
+	watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
+		return client.Watch(ctx, options)
+	}
+	if !watchable {
+		watchFunc = func(options metav1.ListOptions) (watch.Interface, error) {
+			ctx, cancel := context.WithCancel(ctx)
+			return newSyntheticWatcher(ctx, cancel).watch(client, options, defaultRefreshTime)
+		}
+	}
 	listWatcher := &cache.ListWatch{
 		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 			a, err := client.List(ctx, options)
 			return a, err
 		},
-		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-			return client.Watch(ctx, options)
-		},
+		WatchFunc: watchFunc,
 	}
 
 	example := &unstructured.Unstructured{}
@@ -97,6 +106,11 @@ func (i *Informer) ListByOptions(ctx context.Context, lo ListOptions, partitions
 	return i.ByOptionsLister.ListByOptions(ctx, lo, partitions, namespace)
 }
 
+// SetSyntheticWatchableInterval - call this function to override the default interval time of 5 seconds
+func SetSyntheticWatchableInterval(interval time.Duration) {
+	defaultRefreshTime = interval
+}
+
 func informerNameFromGVK(gvk schema.GroupVersionKind) string {
 	return gvk.Group + "_" + gvk.Version + "_" + gvk.Kind
 }

pkg/sqlcache/informer/informer_test.go

@@ -79,7 +79,7 @@ func TestNewInformer(t *testing.T) {
 				}
 			})
 
-		informer, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true)
+		informer, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true, true)
 		assert.Nil(t, err)
 		assert.NotNil(t, informer.ByOptionsLister)
 		assert.NotNil(t, informer.SharedIndexInformer)
@@ -103,7 +103,7 @@ func TestNewInformer(t *testing.T) {
 				}
 			})
 
-		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true)
+		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true, true)
 		assert.NotNil(t, err)
 	}})
 	tests = append(tests, testCase{description: "NewInformer() with errors returned from NewIndexer(), should return an error", test: func(t *testing.T) {
@@ -138,7 +138,7 @@ func TestNewInformer(t *testing.T) {
 				}
 			})
 
-		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true)
+		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true, true)
 		assert.NotNil(t, err)
 	}})
 	tests = append(tests, testCase{description: "NewInformer() with errors returned from NewListOptionIndexer(), should return an error", test: func(t *testing.T) {
@@ -190,7 +190,7 @@ func TestNewInformer(t *testing.T) {
 				}
 			})
 
-		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true)
+		_, err := NewInformer(context.Background(), dynamicClient, fields, nil, gvk, dbClient, false, true, true)
 		assert.NotNil(t, err)
 	}})
 	tests = append(tests, testCase{description: "NewInformer() with transform func", test: func(t *testing.T) {
@@ -253,7 +253,7 @@ func TestNewInformer(t *testing.T) {
 		transformFunc := func(input interface{}) (interface{}, error) {
 			return "someoutput", nil
 		}
-		informer, err := NewInformer(context.Background(), dynamicClient, fields, transformFunc, gvk, dbClient, false, true)
+		informer, err := NewInformer(context.Background(), dynamicClient, fields, transformFunc, gvk, dbClient, false, true, true)
 		assert.Nil(t, err)
 		assert.NotNil(t, informer.ByOptionsLister)
 		assert.NotNil(t, informer.SharedIndexInformer)
@@ -289,7 +289,7 @@ func TestNewInformer(t *testing.T) {
 		transformFunc := func(input interface{}) (interface{}, error) {
 			return "someoutput", nil
 		}
-		_, err := NewInformer(context.Background(), dynamicClient, fields, transformFunc, gvk, dbClient, false, true)
+		_, err := NewInformer(context.Background(), dynamicClient, fields, transformFunc, gvk, dbClient, false, true, true)
 		assert.Error(t, err)
 		newInformer = cache.NewSharedIndexInformer
 	}})

pkg/sqlcache/informer/synthetic_watcher.go

@@ -0,0 +1,141 @@
+package informer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
+)
+
+type SyntheticWatcher struct {
+	resultChan   chan watch.Event
+	stopChan     chan struct{}
+	doneChan     chan struct{}
+	stopChanLock sync.Mutex
+	context      context.Context
+	cancelFunc   context.CancelFunc
+}
+
+func newSyntheticWatcher(context context.Context, cancel context.CancelFunc) *SyntheticWatcher {
+	return &SyntheticWatcher{
+		stopChan:   make(chan struct{}),
+		doneChan:   make(chan struct{}),
+		resultChan: make(chan watch.Event, 0),
+		context:    context,
+		cancelFunc: cancel,
+	}
+}
+
+func (rw *SyntheticWatcher) watch(client dynamic.ResourceInterface, options metav1.ListOptions, interval time.Duration) (*SyntheticWatcher, error) {
+	go rw.receive(client, options, interval)
+	return rw, nil
+}
+
+type objectHolder struct {
+	version            string
+	unstructuredObject *unstructured.Unstructured
+}
+
+// receive periodically calls client.List(), and converts the returned items into Watch Events
+func (rw *SyntheticWatcher) receive(client dynamic.ResourceInterface, options metav1.ListOptions, interval time.Duration) {
+	go func() {
+		defer close(rw.doneChan)
+		defer close(rw.resultChan)
+		defer rw.cancelFunc()
+		previousState := make(map[string]objectHolder)
+		ticker := time.NewTicker(interval)
+
+		for {
+			select {
+			case <-ticker.C:
+				list, err := client.List(rw.context, options)
+				if err != nil {
+					logrus.Errorf("synthetic watcher: client.List => error: %s", err)
+					continue
+				}
+				newObjects := make(map[string]objectHolder)
+				for _, uItem := range list.Items {
+					namespace := uItem.GetNamespace()
+					name := uItem.GetName()
+					key := name
+					if namespace != "" {
+						key = fmt.Sprintf("%s/%s", namespace, name)
+					}
+					version := uItem.GetResourceVersion()
+					newObjects[key] = objectHolder{version: version, unstructuredObject: &uItem}
+				}
+				// Now determine whether items were added, deleted, or modified
+				currentState := make(map[string]objectHolder)
+				for key, newObject := range newObjects {
+					currentState[key] = newObject
+					if oldItem, ok := previousState[key]; !ok {
+						w, err := createWatchEvent(watch.Added, newObject.unstructuredObject)
+						if err != nil {
+							logrus.Errorf("can't convert unstructured obj into runtime: %s", err)
+							continue
+						}
+						rw.resultChan <- w
+					} else {
+						delete(previousState, key)
+						if oldItem.version != newObject.version {
+							w, err := createWatchEvent(watch.Modified, oldItem.unstructuredObject)
+							if err != nil {
+								logrus.Errorf("can't convert unstructured obj into runtime: %s", err)
+								continue
+							}
+							rw.resultChan <- w
+						}
+					}
+				}
+				// And anything left  in the previousState didn't show up in currentState and can be deleted.
+				for _, item := range previousState {
+					w, err := createWatchEvent(watch.Deleted, item.unstructuredObject)
+					if err != nil {
+						continue
+					}
+					rw.resultChan <- w
+				}
+				previousState = currentState
+
+			case <-rw.stopChan:
+				rw.cancelFunc()
+				return
+
+			case <-rw.context.Done():
+				return
+			}
+		}
+	}()
+}
+
+func createWatchEvent(event watch.EventType, u *unstructured.Unstructured) (watch.Event, error) {
+	return watch.Event{Type: event, Object: u}, nil
+}
+
+// ResultChan implements [k8s.io/apimachinery/pkg/watch].Interface.
+func (rw *SyntheticWatcher) ResultChan() <-chan watch.Event {
+	return rw.resultChan
+}
+
+// Stop implements [k8s.io/apimachinery/pkg/watch].Interface.
+func (rw *SyntheticWatcher) Stop() {
+	rw.stopChanLock.Lock()
+	defer rw.stopChanLock.Unlock()
+
+	// Prevent closing an already closed channel to prevent a panic
+	select {
+	case <-rw.stopChan:
+	default:
+		close(rw.stopChan)
+	}
+}
+
+func (rw *SyntheticWatcher) Done() <-chan struct{} {
+	return rw.doneChan
+}