Add check in informer to make controllers work when process invalid resources (#880)

* #2047 sources-controller can't recover from faulty sink (in eventing)

* #2047 sources-controller can't recover from faulty sink (in eventing)

* add unit test

* change after review

* change after review-2

Co-Authored-By: Victor Agababov <vagababov@gmail.com>

apis/duck/typed.go

@@ -46,6 +46,12 @@ var _ InformerFactory = (*TypedInformerFactory)(nil)
 
 // Get implements InformerFactory.
 func (dif *TypedInformerFactory) Get(gvr schema.GroupVersionResource) (cache.SharedIndexInformer, cache.GenericLister, error) {
+	// Avoid error cases, like the GVR does not exist.
+	// It is not a full check. Some RBACs might sneak by, but the window is very small.
+	if _, err := dif.Client.Resource(gvr).List(metav1.ListOptions{}); err != nil {
+		return nil, nil, err
+	}
+
 	listObj := dif.Type.GetListType()
 	lw := &cache.ListWatch{
 		ListFunc:  asStructuredLister(dif.Client.Resource(gvr).List, listObj),

apis/duck/typed_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/fake"
 
 	"knative.dev/pkg/apis/duck"
@@ -94,6 +95,25 @@ func TestSimpleList(t *testing.T) {
 	// TODO(mattmoor): Access through informer
 }
 
+func TestInvalidResource(t *testing.T) {
+	client := &invalidResourceClient{}
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	tif := &duck.TypedInformerFactory{
+		Client:       client,
+		Type:         &duckv1alpha1.AddressableType{},
+		ResyncPeriod: 1 * time.Second,
+		StopChannel:  stopCh,
+	}
+
+	_, _, got := tif.Get(SchemeGroupVersion.WithResource("resources"))
+
+	if got != testErr {
+		t.Errorf("Error = %v, want: %v", got, testErr)
+	}
+}
+
 func TestAsStructuredWatcherNestedError(t *testing.T) {
 	want := errors.New("this is what we expect")
 	nwf := func(lo metav1.ListOptions) (watch.Interface, error) {
@@ -275,3 +295,21 @@ func (bo *badObject) GetObjectKind() schema.ObjectKind {
 func (bo *badObject) DeepCopyObject() runtime.Object {
 	return &badObject{}
 }
+
+var testErr = errors.New("failed to get list")
+
+type invalidResourceClient struct {
+	*fake.FakeDynamicClient
+}
+
+func (*invalidResourceClient) Resource(resource schema.GroupVersionResource) dynamic.NamespaceableResourceInterface {
+	return &invalidResource{}
+}
+
+type invalidResource struct {
+	dynamic.NamespaceableResourceInterface
+}
+
+func (*invalidResource) List(options metav1.ListOptions) (*unstructured.UnstructuredList, error) {
+	return nil, testErr
+}