✨ Adding timeout to unset context for waiting for cache to sync.

Waiting for cache to sync happens when getting a new informer
in the informer map.

Author: djzager

pkg/cache/informer_cache.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"time"
 
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -57,7 +58,12 @@ func (ip *informerCache) Get(ctx context.Context, key client.ObjectKey, out runt
 		return err
 	}
 
-	started, cache, err := ip.InformersMap.Get(gvk, out)
+	ctx, cancelFunc := addTimeout(ctx)
+	if cancelFunc != nil {
+		defer cancelFunc()
+	}
+	cache, err := ip.InformersMap.Get(ctx, gvk, out)
+	started, cache, err := ip.InformersMap.Get(ctx, gvk, out)
 	if err != nil {
 		return err
 	}
@@ -101,7 +107,11 @@ func (ip *informerCache) List(ctx context.Context, out runtime.Object, opts ...c
 		}
 	}
 
-	started, cache, err := ip.InformersMap.Get(gvk, cacheTypeObj)
+	ctx, cancelFunc := addTimeout(ctx)
+	if cancelFunc != nil {
+		defer cancelFunc()
+	}
+	started, cache, err := ip.InformersMap.Get(ctx, gvk, cacheTypeObj)
 	if err != nil {
 		return err
 	}
@@ -113,27 +123,47 @@ func (ip *informerCache) List(ctx context.Context, out runtime.Object, opts ...c
 	return cache.Reader.List(ctx, out, opts...)
 }
 
+// addTimeout adds a default 30 second timeout to a child contxt
+// if one does not exist.
+func addTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
+	var cancelFunc context.CancelFunc
+	if _, ok := ctx.Deadline(); !ok {
+		ctx, cancelFunc = context.WithTimeout(ctx, 30*time.Second)
+	}
+	return ctx, cancelFunc
+}
+
 // GetInformerForKind returns the informer for the GroupVersionKind
+// Will timeout after 30 seconds if the informer is new and can not be
+// synced.
 func (ip *informerCache) GetInformerForKind(gvk schema.GroupVersionKind) (Informer, error) {
 	// Map the gvk to an object
 	obj, err := ip.Scheme.New(gvk)
 	if err != nil {
 		return nil, err
 	}
-	_, i, err := ip.InformersMap.Get(gvk, obj)
+	ctx, cancelFunc := addTimeout(context.TODO())
+	// There will never be a nil cancelFunc
+	defer cancelFunc()
+	_, i, err := ip.InformersMap.Get(ctx, gvk, obj)
 	if err != nil {
 		return nil, err
 	}
 	return i.Informer, err
 }
 
 // GetInformer returns the informer for the obj
+// Will timeout after 30 seconds if the informer is new and can not be
+// synced.
 func (ip *informerCache) GetInformer(obj runtime.Object) (Informer, error) {
 	gvk, err := apiutil.GVKForObject(obj, ip.Scheme)
 	if err != nil {
 		return nil, err
 	}
-	_, i, err := ip.InformersMap.Get(gvk, obj)
+	ctx, cancelFunc := addTimeout(context.TODO())
+	// There will never be a nil cancelFunc
+	defer cancelFunc()
+	_, i, err := ip.InformersMap.Get(ctx, gvk, obj)
 	if err != nil {
 		return nil, err
 	}

pkg/cache/internal/deleg_map.go

@@ -17,6 +17,7 @@ limitations under the License.
 package internal
 
 import (
+	"context"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -79,16 +80,16 @@ func (m *InformersMap) WaitForCacheSync(stop <-chan struct{}) bool {
 
 // Get will create a new Informer and add it to the map of InformersMap if none exists.  Returns
 // the Informer from the map.
-func (m *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
+func (m *InformersMap) Get(ctx context.Context, gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
 	_, isUnstructured := obj.(*unstructured.Unstructured)
 	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
 	isUnstructured = isUnstructured || isUnstructuredList
 
 	if isUnstructured {
-		return m.unstructured.Get(gvk, obj)
+		return m.unstructured.Get(ctx, gvk, obj)
 	}
 
-	return m.structured.Get(gvk, obj)
+	return m.structured.Get(ctx, gvk, obj)
 }
 
 // newStructuredInformersMap creates a new InformersMap for structured objects.

pkg/cache/internal/informers_map.go

@@ -17,6 +17,7 @@ limitations under the License.
 package internal
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"sync"
@@ -163,7 +164,7 @@ func (ip *specificInformersMap) HasSyncedFuncs() []cache.InformerSynced {
 
 // Get will create a new Informer and add it to the map of specificInformersMap if none exists.  Returns
 // the Informer from the map.
-func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
+func (ip *specificInformersMap) Get(ctx context.Context, gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
 	// Return the informer if it is found
 	i, started, ok := func() (*MapEntry, bool, bool) {
 		ip.mu.RLock()
@@ -180,9 +181,22 @@ func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Obj
 	}
 
 	if started && !i.Informer.HasSynced() {
+		syncReturn := make(chan bool)
+		done := make(chan struct{})
+		go func() {
+			syncReturn <- cache.WaitForCacheSync(done, i.Informer.HasSynced)
+		}()
+
 		// Wait for it to sync before returning the Informer so that folks don't read from a stale cache.
-		if !cache.WaitForCacheSync(ip.stop, i.Informer.HasSynced) {
-			return started, nil, fmt.Errorf("failed waiting for %T Informer to sync", obj)
+		select {
+		case <-ctx.Done():
+			//end the polling for cache to sync
+			done <- struct{}{}
+			return started, nil, fmt.Errorf("timeout waiting for %T Informer to sync", obj)
+		case syncSuccess := <-syncReturn:
+			if !syncSuccess {
+				return started, nil, fmt.Errorf("failed waiting for %T Informer to sync", obj)
+			}
 		}
 	}