Switch to watcher package to handle watch corner cases

Issue #766 is caused by starting a watch after the change we are watching
for has happened.  The watcher package will handle this cases and others.

Signed-off-by: Darren Shepherd <darren@acorn.io>

go.mod

@@ -17,7 +17,7 @@ require (
 	cuelang.org/go v0.4.3
 	github.com/AlecAivazis/survey/v2 v2.3.6
 	github.com/acorn-io/aml v0.0.0-20220717003025-bc8cb1214693
-	github.com/acorn-io/baaah v0.0.0-20221016060913-92803f5acbdb
+	github.com/acorn-io/baaah v0.0.0-20221017051037-e8c2ee0423d9
 	github.com/acorn-io/mink v0.0.0-20221015002203-a58541a4f675
 	github.com/acorn-io/namegenerator v0.0.0-20220915160418-9e3d5a0ffe78
 	github.com/containerd/console v1.0.3
@@ -32,6 +32,7 @@ require (
 	github.com/moby/buildkit v0.10.4
 	github.com/moby/locker v1.0.1
 	github.com/opencontainers/image-spec v1.0.3-0.20220114050600-8b9d41f48198
+	github.com/pkg/errors v0.9.1
 	github.com/pterm/pterm v0.12.49
 	github.com/rancher/lasso v0.0.0-20220412224715-5f3517291ad4
 	github.com/rancher/wrangler v1.0.1-0.20220520195731-8eeded9bae2a
@@ -154,7 +155,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/otiai10/copy v1.2.0 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.12.1 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect

go.sum

@@ -211,8 +211,8 @@ github.com/acorn-io/aml v0.0.0-20220717003025-bc8cb1214693 h1:5uxUFWpREhhKllLkan
 github.com/acorn-io/aml v0.0.0-20220717003025-bc8cb1214693/go.mod h1:uiNpPSAYWhEBfbGWKpN185+EP+hZr2M/U8Ckr/Jo3Kk=
 github.com/acorn-io/apiserver v0.25.2-ot-1 h1:91Q7+Jd9BeYZhzt0C+38w2sMn8aHFOxfTdlE6MCH9UI=
 github.com/acorn-io/apiserver v0.25.2-ot-1/go.mod h1:8cynBL5SR6CIKypk9nWtZXHzEiJhLVQxeMVZ5CGzkFo=
-github.com/acorn-io/baaah v0.0.0-20221016060913-92803f5acbdb h1:bdy6ZvrW6fdpZBMVJXsLaGjYVBxvA+h972Fh013oDqg=
-github.com/acorn-io/baaah v0.0.0-20221016060913-92803f5acbdb/go.mod h1:WLsKtMvvlNf3k0LXx3V56RpPVp2iajGFbIvLSbGDZZE=
+github.com/acorn-io/baaah v0.0.0-20221017051037-e8c2ee0423d9 h1:35oSP83lZzOMf7FyYIl/llwN2+OmAeZHtptVkU74++A=
+github.com/acorn-io/baaah v0.0.0-20221017051037-e8c2ee0423d9/go.mod h1:WLsKtMvvlNf3k0LXx3V56RpPVp2iajGFbIvLSbGDZZE=
 github.com/acorn-io/component-base v0.25.2-ot-1 h1:xinqJUNbpW2/zsvm8mDv6Q7riLhvXup9x7Kz9eIPM1M=
 github.com/acorn-io/component-base v0.25.2-ot-1/go.mod h1:/5qYr5BXGNPs+cRd6+WL1NfOYtzOstJlm1CMK06cm7s=
 github.com/acorn-io/etcd/server/v3 v3.5.4-ot-1 h1:sQMBy/UImb1923toh9U0oIxlpO7crLrD7eKE/orB/pQ=

pkg/install/check.go

@@ -16,19 +16,19 @@ import (
 	"github.com/acorn-io/acorn/pkg/system"
 	"github.com/acorn-io/baaah/pkg/randomtoken"
 	"github.com/acorn-io/baaah/pkg/restconfig"
+	"github.com/acorn-io/baaah/pkg/watcher"
+	"github.com/pkg/errors"
 	"github.com/rancher/wrangler/pkg/name"
 	authorizationv1 "k8s.io/api/authorization/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	storagev1 "k8s.io/api/storage/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog"
 	klogv2 "k8s.io/klog/v2"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // CheckResult describes the results of a check, making it human-readable
@@ -172,29 +172,14 @@ func CheckExec(ctx context.Context, opts CheckOptions) CheckResult {
 		}
 	}()
 
-	// Wait for pod to be ready
-	var podList corev1.PodList
-	watcher, err := cli.Watch(ctx, &podList, client.InNamespace(*opts.Namespace), client.MatchingFields{"metadata.name": pod.GetName()})
+	_, err = watcher.New[*corev1.Pod](cli).ByObject(ctx, pod, func(pod *corev1.Pod) (bool, error) {
+		return pod.Status.Phase == corev1.PodRunning, nil
+	})
 	if err != nil {
 		result.Passed = false
 		result.Message = fmt.Sprintf("Error creating watcher: %v", err)
 		return result
 	}
-	defer watcher.Stop()
-	for {
-		event := <-watcher.ResultChan()
-		if event.Type == watch.Error {
-			result.Passed = false
-			result.Message = fmt.Sprintf("Error watching pod: %v", event.Object)
-			return result
-		}
-		if event.Type == watch.Added || event.Type == watch.Modified {
-			pod := event.Object.(*corev1.Pod)
-			if pod.Status.Phase == corev1.PodRunning {
-				break
-			}
-		}
-	}
 
 	// Execute command in container
 	// had to reimplement this from the client pkg to avoid an import cycle
@@ -378,7 +363,7 @@ func CheckIngressCapability(ctx context.Context, opts CheckOptions) CheckResult
 		return result
 	}
 	defer func() {
-		if err := cli.Delete(ctx, ep); err != nil && !errors.IsNotFound(err) {
+		if err := cli.Delete(ctx, ep); err != nil && !apierrors.IsNotFound(err) {
 			klog.Errorf("Error deleting endpoint: %v", err)
 		}
 	}()
@@ -405,35 +390,24 @@ func CheckIngressCapability(ctx context.Context, opts CheckOptions) CheckResult
 		}
 	}()
 
-	// Wait for ingress to be ready, 10s timeout
-	ingw := &networkingv1.IngressList{Items: []networkingv1.Ingress{*ing}}
-	w, err := cli.Watch(ctx, ingw, client.InNamespace(*opts.Namespace))
-	if err != nil {
+	nctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+	_, err = watcher.New[*networkingv1.Ingress](cli).ByObject(nctx, ing, func(ing *networkingv1.Ingress) (bool, error) {
+		return ing.Status.LoadBalancer.Ingress != nil, nil
+	})
+	if errors.Is(err, context.DeadlineExceeded) {
+		result.Passed = false
+		result.Message = "Ingress not ready (test timed out after 1 minute)"
+		return result
+	} else if err != nil {
 		result.Passed = false
 		result.Message = fmt.Sprintf("Error watching ingress: %v", err)
 		return result
 	}
 
-	nctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
-	defer cancel()
-	for {
-		select {
-		case <-nctx.Done():
-			result.Passed = false
-			result.Message = "Ingress not ready (test timed out after 1 minute)"
-			return result
-		case f := <-w.ResultChan():
-			if f.Type == watch.Modified {
-				ing := f.Object.(*networkingv1.Ingress)
-				if ing.Status.LoadBalancer.Ingress != nil {
-					result.Passed = true
-					result.Message = "Ingress is ready"
-					return result
-				}
-			}
-		}
-	}
-
+	result.Passed = true
+	result.Message = "Ingress is ready"
+	return result
 }
 
 /*