sql: silence informer errors on unwatchable resources (#111)

* sql: silence informer errors on unwatchable resources Listable-but-unwatchable resources such as nodes metrics.k8s.io/v1beta1 allow an Informer to be started, but it will by default log all watch failures. Add a boolean flag to silence those errors. Signed-off-by: Silvio Moioli <[email protected]> * adapt tests Signed-off-by: Silvio Moioli <[email protected]> --------- Signed-off-by: Silvio Moioli <[email protected]>
rancher · Dec 2, 2024 · 04649f3 · 04649f3
1 parent d19dc78
commit 04649f3
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 8 deletions.
diff --git a/pkg/cache/sql/informer/factory/informer_factory.go b/pkg/cache/sql/informer/factory/informer_factory.go
@@ -14,6 +14,7 @@ import (
 	"github.com/rancher/lasso/pkg/cache/sql/informer"
 	sqlStore "github.com/rancher/lasso/pkg/cache/sql/store"
 	"github.com/rancher/lasso/pkg/log"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
@@ -88,7 +89,7 @@ func NewCacheFactory() (*CacheFactory, error) {
 
 // CacheFor returns an informer for given GVK, using sql store indexed with fields, using the specified client. For virtual fields, they must be added by the transform function
 // and specified by fields to be used for later fields.
-func (f *CacheFactory) CacheFor(fields [][]string, transform cache.TransformFunc, client dynamic.ResourceInterface, gvk schema.GroupVersionKind, namespaced bool) (Cache, error) {
+func (f *CacheFactory) CacheFor(fields [][]string, transform cache.TransformFunc, client dynamic.ResourceInterface, gvk schema.GroupVersionKind, namespaced bool, watchable bool) (Cache, error) {
 	// First of all block Reset() until we are done
 	f.mutex.RLock()
 	defer f.mutex.RUnlock()
@@ -129,6 +130,17 @@ func (f *CacheFactory) CacheFor(fields [][]string, transform cache.TransformFunc
 			return Cache{}, err
 		}
 
+		err = i.SetWatchErrorHandler(func(r *cache.Reflector, err error) {
+			if !watchable && errors.IsMethodNotSupported(err) {
+				// expected, continue without logging
+				return
+			}
+			cache.DefaultWatchErrorHandler(r, err)
+		})
+		if err != nil {
+			return Cache{}, err
+		}
+
 		f.wg.StartWithChannel(f.stopCh, i.Run)
 
 		gi.informer = i

diff --git a/pkg/cache/sql/informer/factory/informer_factory_test.go b/pkg/cache/sql/informer/factory/informer_factory_test.go
@@ -65,6 +65,7 @@ func TestCacheFor(t *testing.T) {
 		sii := NewMockSharedIndexInformer(gomock.NewController(t))
 		sii.EXPECT().HasSynced().Return(true).AnyTimes()
 		sii.EXPECT().Run(gomock.Any()).MinTimes(1)
+		sii.EXPECT().SetWatchErrorHandler(gomock.Any())
 		i := &informer.Informer{
 			// need to set this so Run function is not nil
 			SharedIndexInformer: sii,
@@ -94,12 +95,12 @@ func TestCacheFor(t *testing.T) {
 		}()
 		var c Cache
 		var err error
-		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false)
+		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false, true)
 		assert.Nil(t, err)
 		assert.Equal(t, expectedC, c)
 		// this sleep is critical to the test. It ensure there has been enough time for expected function like Run to be invoked in their go routines.
 		time.Sleep(1 * time.Second)
-		c2, err := f.CacheFor(fields, nil, dynamicClient, expectedGVK, false)
+		c2, err := f.CacheFor(fields, nil, dynamicClient, expectedGVK, false, true)
 		assert.Nil(t, err)
 		assert.Equal(t, c, c2)
 	}})
@@ -112,6 +113,7 @@ func TestCacheFor(t *testing.T) {
 		sii := NewMockSharedIndexInformer(gomock.NewController(t))
 		sii.EXPECT().HasSynced().Return(false).AnyTimes()
 		sii.EXPECT().Run(gomock.Any())
+		sii.EXPECT().SetWatchErrorHandler(gomock.Any())
 		expectedI := &informer.Informer{
 			// need to set this so Run function is not nil
 			SharedIndexInformer: sii,
@@ -136,7 +138,7 @@ func TestCacheFor(t *testing.T) {
 			close(f.stopCh)
 		}()
 		var err error
-		_, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false)
+		_, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false, true)
 		assert.NotNil(t, err)
 		time.Sleep(2 * time.Second)
 	}})
@@ -150,6 +152,7 @@ func TestCacheFor(t *testing.T) {
 		sii.EXPECT().HasSynced().Return(true).AnyTimes()
 		// may or may not call run initially
 		sii.EXPECT().Run(gomock.Any()).MaxTimes(1)
+		sii.EXPECT().SetWatchErrorHandler(gomock.Any())
 		i := &informer.Informer{
 			// need to set this so Run function is not nil
 			SharedIndexInformer: sii,
@@ -175,7 +178,7 @@ func TestCacheFor(t *testing.T) {
 		close(f.stopCh)
 		var c Cache
 		var err error
-		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false)
+		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false, true)
 		assert.Nil(t, err)
 		assert.Equal(t, expectedC, c)
 		time.Sleep(1 * time.Second)
@@ -188,6 +191,7 @@ func TestCacheFor(t *testing.T) {
 		sii := NewMockSharedIndexInformer(gomock.NewController(t))
 		sii.EXPECT().HasSynced().Return(true)
 		sii.EXPECT().Run(gomock.Any()).MinTimes(1).AnyTimes()
+		sii.EXPECT().SetWatchErrorHandler(gomock.Any())
 		i := &informer.Informer{
 			// need to set this so Run function is not nil
 			SharedIndexInformer: sii,
@@ -217,7 +221,7 @@ func TestCacheFor(t *testing.T) {
 		}()
 		var c Cache
 		var err error
-		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false)
+		c, err = f.CacheFor(fields, nil, dynamicClient, expectedGVK, false, true)
 		assert.Nil(t, err)
 		assert.Equal(t, expectedC, c)
 		time.Sleep(1 * time.Second)
@@ -230,6 +234,7 @@ func TestCacheFor(t *testing.T) {
 		sii := NewMockSharedIndexInformer(gomock.NewController(t))
 		sii.EXPECT().HasSynced().Return(true)
 		sii.EXPECT().Run(gomock.Any()).MinTimes(1)
+		sii.EXPECT().SetWatchErrorHandler(gomock.Any())
 		transformFunc := func(input interface{}) (interface{}, error) {
 			return "someoutput", nil
 		}
@@ -270,7 +275,7 @@ func TestCacheFor(t *testing.T) {
 		}()
 		var c Cache
 		var err error
-		c, err = f.CacheFor(fields, transformFunc, dynamicClient, expectedGVK, false)
+		c, err = f.CacheFor(fields, transformFunc, dynamicClient, expectedGVK, false, true)
 		assert.Nil(t, err)
 		assert.Equal(t, expectedC, c)
 		time.Sleep(1 * time.Second)

diff --git a/pkg/cache/sql/integration_test.go b/pkg/cache/sql/integration_test.go
@@ -323,7 +323,7 @@ func (i *IntegrationSuite) createCacheAndFactory(fields [][]string, transformFun
 		Resource: "configmaps",
 	}
 	dynamicResource := dynamicClient.Resource(configMapGVR).Namespace(testNamespace)
-	cache, err := cacheFactory.CacheFor(fields, transformFunc, dynamicResource, configMapGVK, true)
+	cache, err := cacheFactory.CacheFor(fields, transformFunc, dynamicResource, configMapGVK, true, true)
 	if err != nil {
 		return nil, nil, fmt.Errorf("unable to make cache: %w", err)
 	}