Switch kubectl to use watch.Until

pkg/kubectl/cmd/run.go

@@ -28,16 +28,19 @@ import (
 	"github.com/docker/distribution/reference"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/meta"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	batchv1 "k8s.io/kubernetes/pkg/apis/batch/v1"
 	"k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 	coreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
+	conditions "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/kubectl"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/kubectl/resource"
 	"k8s.io/kubernetes/pkg/runtime"
 	uexec "k8s.io/kubernetes/pkg/util/exec"
+	"k8s.io/kubernetes/pkg/util/interrupt"
 	"k8s.io/kubernetes/pkg/watch"
 )
 
@@ -372,90 +375,90 @@ func contains(resourcesList map[string]*unversioned.APIResourceList, resource un
 
 // waitForPod watches the given pod until the exitCondition is true. Each two seconds
 // the tick function is called e.g. for progress output.
-func waitForPod(podClient coreclient.PodsGetter, ns, name string, exitCondition func(*api.Pod) bool, tick func(*api.Pod)) (*api.Pod, error) {
-	pod, err := podClient.Pods(ns).Get(name)
-	if err != nil {
-		return nil, err
-	}
-	if exitCondition(pod) {
-		return pod, nil
-	}
-
-	tick(pod)
-
-	w, err := podClient.Pods(ns).Watch(api.SingleObject(api.ObjectMeta{Name: pod.Name, ResourceVersion: pod.ResourceVersion}))
+func waitForPod(podClient coreclient.PodsGetter, ns, name string, exitCondition watch.ConditionFunc, tick func(*api.Pod)) (*api.Pod, error) {
+	w, err := podClient.Pods(ns).Watch(api.SingleObject(api.ObjectMeta{Name: name}))
 	if err != nil {
 		return nil, err
 	}
 
-	t := time.NewTicker(2 * time.Second)
-	defer t.Stop()
+	pods := make(chan *api.Pod) // observed pods passed to the exitCondition
+	defer close(pods)
+
+	// wait for the first event, then start the 2 sec ticker and loop
 	go func() {
-		for range t.C {
-			tick(pod)
+		pod := <-pods
+		if pod == nil {
+			return
+		}
+		tick(pod)
+
+		t := time.NewTicker(2 * time.Second)
+		defer t.Stop()
+
+		for {
+			select {
+			case pod = <-pods:
+				if pod == nil {
+					return
+				}
+			case _, ok := <-t.C:
+				if !ok {
+					return
+				}
+				tick(pod)
+			}
 		}
 	}()
 
-	err = nil
-	result := pod
-	kubectl.WatchLoop(w, func(ev watch.Event) error {
-		switch ev.Type {
-		case watch.Added, watch.Modified:
-			pod = ev.Object.(*api.Pod)
-			if exitCondition(pod) {
-				result = pod
-				w.Stop()
+	intr := interrupt.New(nil, w.Stop)
+	var result *api.Pod
+	intr.Run(func() error {
+		ev, err := watch.Until(0, w, func(ev watch.Event) (bool, error) {
+			c, err := exitCondition(ev)
+			if c == false && err == nil {
+				pods <- ev.Object.(*api.Pod) // send to ticker
 			}
-		case watch.Deleted:
-			w.Stop()
-		case watch.Error:
-			result = nil
-			err = fmt.Errorf("failed to watch pod %s/%s", ns, name)
-			w.Stop()
-		}
-		return nil
+			return c, err
+		})
+		result = ev.Object.(*api.Pod)
+		return err
 	})
-
 	return result, err
 }
 
 func waitForPodRunning(podClient coreclient.PodsGetter, ns, name string, out io.Writer, quiet bool) (*api.Pod, error) {
-	exitCondition := func(pod *api.Pod) bool {
-		switch pod.Status.Phase {
-		case api.PodRunning:
-			for _, status := range pod.Status.ContainerStatuses {
-				if !status.Ready {
-					return false
-				}
-			}
-			return true
-		case api.PodSucceeded, api.PodFailed:
-			return true
-		default:
-			return false
-		}
-	}
-	return waitForPod(podClient, ns, name, exitCondition, func(pod *api.Pod) {
+	pod, err := waitForPod(podClient, ns, name, conditions.PodRunningAndReady, func(pod *api.Pod) {
 		if !quiet {
 			fmt.Fprintf(out, "Waiting for pod %s/%s to be running, status is %s, pod ready: false\n", pod.Namespace, pod.Name, pod.Status.Phase)
 		}
 	})
+
+	// fix generic not found error with empty name in PodRunningAndReady
+	if err != nil && errors.IsNotFound(err) {
+		return nil, errors.NewNotFound(api.Resource("pods"), name)
+	}
+
+	return pod, err
 }
 
 func waitForPodTerminated(podClient coreclient.PodsGetter, ns, name string, out io.Writer, quiet bool) (*api.Pod, error) {
-	exitCondition := func(pod *api.Pod) bool {
-		return pod.Status.Phase == api.PodSucceeded || pod.Status.Phase == api.PodFailed
-	}
-	return waitForPod(podClient, ns, name, exitCondition, func(pod *api.Pod) {
+	pod, err := waitForPod(podClient, ns, name, conditions.PodCompleted, func(pod *api.Pod) {
 		if !quiet {
 			fmt.Fprintf(out, "Waiting for pod %s/%s to terminate, status is %s\n", pod.Namespace, pod.Name, pod.Status.Phase)
 		}
 	})
+
+	// fix generic not found error with empty name in PodCompleted
+	if err != nil && errors.IsNotFound(err) {
+		return nil, errors.NewNotFound(api.Resource("pods"), name)
+	}
+
+	return pod, err
 }
 
 func handleAttachPod(f *cmdutil.Factory, podClient coreclient.PodsGetter, ns, name string, opts *AttachOptions, quiet bool) error {
 	pod, err := waitForPodRunning(podClient, ns, name, opts.Out, quiet)
-	if err != nil {
+	if err != nil && err != conditions.ErrPodCompleted {
 		return err
 	}
 	ctrName, err := opts.GetContainerName(pod)