Merge pull request #359 from buildpacks-community/issue-278-import-cmd-retry

[Issue-278] Waiter uses watchTools.UntilWithSync instead of Until to handle resource version old error

Author: tomkennedy513

pkg/commands/import/import.go

@@ -5,7 +5,6 @@ package _import
 
 import (
 	"io"
-	"io/ioutil"
 	"os"
 
 	"github.com/spf13/cobra"
@@ -170,7 +169,7 @@ func readDescriptor(cmd *cobra.Command, filename string) (string, error) {
 	)
 
 	if filename == "-" {
-		reader = ioutil.NopCloser(cmd.InOrStdin())
+		reader = io.NopCloser(cmd.InOrStdin())
 	} else {
 		reader, err = os.Open(filename)
 		if err != nil {
@@ -179,7 +178,7 @@ func readDescriptor(cmd *cobra.Command, filename string) (string, error) {
 	}
 	defer reader.Close()
 
-	buf, err := ioutil.ReadAll(reader)
+	buf, err := io.ReadAll(reader)
 	if err != nil {
 		return "", err
 	}

pkg/commands/waiter.go

@@ -16,6 +16,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
 	watchTools "k8s.io/client-go/tools/watch"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/apis/duck"
@@ -59,18 +60,24 @@ func (w *Waiter) wait(ctx context.Context, ob runtime.Object, condition watchToo
 		return err
 	}
 
-	if !done {
-		refable, ok := ob.(kmeta.OwnerRefable)
-		if !ok {
-			return errors.New("unexpected type")
-		}
+	refable, ok := ob.(kmeta.OwnerRefable)
+	if !ok {
+		return errors.New("unexpected type")
+	}
 
-		rv := refable.GetObjectMeta().GetResourceVersion()
-		watchOne := newWatchOneWatcher(ctx, refable, w.dynamicClient)
+	listerWatcherOne := newWatchOneWatcher(ctx, refable, w.dynamicClient)
 
-		ctx, cancel := context.WithTimeout(context.Background(), w.timeout)
-		defer cancel()
-		e, err = watchTools.Until(ctx, rv, watchOne, filterErrors(cfs)...)
+	ctx, cancel := context.WithTimeout(context.Background(), w.timeout)
+	defer cancel()
+
+	if !done {
+		e, err = watchTools.UntilWithSync(ctx,
+			&listerWatcherOne,
+			&unstructured.Unstructured{},
+			func(store cache.Store) (bool, error) {
+				return false, nil
+			},
+			filterErrors(cfs)...)
 		if err != nil {
 			return err
 		}
@@ -158,6 +165,11 @@ func (w watchOneWatcher) Watch(options metav1.ListOptions) (watch.Interface, err
 	return w.dynamicClient.Resource(w.gvr).Namespace(w.namespace).Watch(w.ctx, options)
 }
 
+func (w watchOneWatcher) List(options metav1.ListOptions) (runtime.Object, error) {
+	options.FieldSelector = fmt.Sprintf("metadata.name=%s", w.name)
+	return w.dynamicClient.Resource(w.gvr).Namespace(w.namespace).List(w.ctx, options)
+}
+
 func filterErrors(conditions []watchTools.ConditionFunc) []watchTools.ConditionFunc {
 	cfs := []watchTools.ConditionFunc{}
 	for _, c := range conditions {

pkg/commands/waiter_test.go

@@ -15,6 +15,8 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -81,13 +83,19 @@ func testWaiter(t *testing.T, when spec.G, it spec.S) {
 				Status: conditionReady(corev1.ConditionFalse, generation-1),
 			}
 
+			builderObj := &v1alpha2.Builder{
+				TypeMeta:   resourceToWatch.TypeMeta,
+				ObjectMeta: resourceToWatch.ObjectMeta,
+				Status:     v1alpha2.BuilderStatus{Status: conditionReady(corev1.ConditionTrue, generation)},
+			}
+
+			content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(builderObj)
+			if err != nil {
+				panic(err)
+			}
 			watcher.addEvent(watch.Event{
-				Type: watch.Modified,
-				Object: &v1alpha2.Builder{
-					TypeMeta:   resourceToWatch.TypeMeta,
-					ObjectMeta: resourceToWatch.ObjectMeta,
-					Status:     v1alpha2.BuilderStatus{Status: conditionReady(corev1.ConditionTrue, generation)},
-				},
+				Type:   watch.Modified,
+				Object: &unstructured.Unstructured{Object: content},
 			})
 
 			require.NoError(t, waiter.Wait(context.Background(), resourceToWatch))
@@ -99,18 +107,56 @@ func testWaiter(t *testing.T, when spec.G, it spec.S) {
 				Status: conditionReady(corev1.ConditionFalse, generation-1),
 			}
 
+			builderObj := &v1alpha2.Builder{
+				TypeMeta:   resourceToWatch.TypeMeta,
+				ObjectMeta: resourceToWatch.ObjectMeta,
+				Status:     v1alpha2.BuilderStatus{Status: conditionReady(corev1.ConditionTrue, generation)},
+			}
+
+			content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(builderObj)
+			if err != nil {
+				panic(err)
+			}
 			watcher.addEvent(watch.Event{
-				Type: watch.Modified,
-				Object: &v1alpha2.Builder{
-					TypeMeta:   resourceToWatch.TypeMeta,
-					ObjectMeta: resourceToWatch.ObjectMeta,
-					Status:     v1alpha2.BuilderStatus{Status: conditionReady(corev1.ConditionTrue, generation)},
-				},
+				Type:   watch.Modified,
+				Object: &unstructured.Unstructured{Object: content},
 			})
 
 			require.NoError(t, waiter.Wait(context.Background(), resourceToWatch, fakeConditionChecker.conditionCheck))
 			require.True(t, fakeConditionChecker.called)
 		})
+
+		it("recovers from too old resource version error", func() {
+			watcher.addEvent(watch.Event{
+				Type: watch.Error,
+				Object: &v1.Status{
+					TypeMeta: v1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "Status",
+					},
+					Status:  "Failure",
+					Message: "too old resource version: 23358 (23360)",
+					Reason:  "Expired",
+					Code:    410,
+				},
+			})
+
+			builderObj := &v1alpha2.Builder{
+				TypeMeta:   resourceToWatch.TypeMeta,
+				ObjectMeta: resourceToWatch.ObjectMeta,
+				Status:     v1alpha2.BuilderStatus{Status: conditionReady(corev1.ConditionTrue, generation)},
+			}
+			content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(builderObj)
+			if err != nil {
+				panic(err)
+			}
+			watcher.addEvent(watch.Event{
+				Type:   watch.Modified,
+				Object: &unstructured.Unstructured{Object: content},
+			})
+
+			require.NoError(t, waiter.Wait(context.Background(), resourceToWatch))
+		})
 	})
 }
 
@@ -121,7 +167,6 @@ type fakeConditionChecker struct {
 func (cc *fakeConditionChecker) conditionCheck(_ watch.Event) (bool, error) {
 	cc.called = true
 	return true, nil
-
 }
 
 func conditionReady(status corev1.ConditionStatus, generation int64) corev1alpha1.Status {
@@ -159,10 +204,6 @@ func (t *TestWatcher) watchReactor(action clientgotesting.Action) (handled bool,
 	}
 
 	watchAction := action.(clientgotesting.WatchAction)
-	if watchAction.GetWatchRestrictions().ResourceVersion != t.expectedResource.GetObjectMeta().GetResourceVersion() {
-		return true, nil, errors.New("expected watch on resource version")
-	}
-
 	if watchAction.GetNamespace() != t.expectedResource.GetObjectMeta().GetNamespace() {
 		return true, nil, errors.New("expected watch on namespace")
 	}