Use k8s cache indexer to look for pod status

Author: Keerthan Mala

pkg/gitreceive/build.go

@@ -183,7 +183,12 @@ func build(
 		return fmt.Errorf("creating builder pod (%s)", err)
 	}
 
-	if err := waitForPod(kubeClient, newPod.Namespace, newPod.Name, conf.SessionIdleInterval(), conf.BuilderPodTickDuration(), conf.BuilderPodWaitDuration()); err != nil {
+	pw := k8s.NewPodWatcher(kubeClient, "deis")
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	go pw.Controller.Run(stopCh)
+
+	if err := waitForPod(pw, newPod.Namespace, newPod.Name, conf.SessionIdleInterval(), conf.BuilderPodTickDuration(), conf.BuilderPodWaitDuration()); err != nil {
 		return fmt.Errorf("watching events for builder pod startup (%s)", err)
 	}
 
@@ -213,7 +218,7 @@ func build(
 	)
 	// check the state and exit code of the build pod.
 	// if the code is not 0 return error
-	if err := waitForPodEnd(kubeClient, newPod.Namespace, newPod.Name, conf.BuilderPodTickDuration(), conf.BuilderPodWaitDuration()); err != nil {
+	if err := waitForPodEnd(pw, newPod.Namespace, newPod.Name, conf.BuilderPodTickDuration(), conf.BuilderPodWaitDuration()); err != nil {
 		return fmt.Errorf("error getting builder pod status (%s)", err)
 	}
 	log.Debug("Done")

pkg/gitreceive/k8s_util.go

@@ -4,10 +4,10 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/deis/builder/pkg/k8s"
 	"github.com/pborman/uuid"
 	"k8s.io/kubernetes/pkg/api"
-	apierrs "k8s.io/kubernetes/pkg/api/errors"
-	client "k8s.io/kubernetes/pkg/client/unversioned"
+	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/util/wait"
 )
 
@@ -171,7 +171,7 @@ func addEnvToPod(pod api.Pod, key, value string) {
 }
 
 // waitForPod waits for a pod in state running, succeeded or failed
-func waitForPod(c *client.Client, ns, podName string, ticker, interval, timeout time.Duration) error {
+func waitForPod(pw *k8s.PodWatcher, ns, podName string, ticker, interval, timeout time.Duration) error {
 	condition := func(pod *api.Pod) (bool, error) {
 		if pod.Status.Phase == api.PodRunning {
 			return true, nil
@@ -186,14 +186,14 @@ func waitForPod(c *client.Client, ns, podName string, ticker, interval, timeout
 	}
 
 	quit := progress("...", ticker)
-	err := waitForPodCondition(c, ns, podName, condition, interval, timeout)
+	err := waitForPodCondition(pw, ns, podName, condition, interval, timeout)
 	quit <- true
 	<-quit
 	return err
 }
 
 // waitForPodEnd waits for a pod in state succeeded or failed
-func waitForPodEnd(c *client.Client, ns, podName string, interval, timeout time.Duration) error {
+func waitForPodEnd(pw *k8s.PodWatcher, ns, podName string, interval, timeout time.Duration) error {
 	condition := func(pod *api.Pod) (bool, error) {
 		if pod.Status.Phase == api.PodSucceeded {
 			return true, nil
@@ -204,21 +204,19 @@ func waitForPodEnd(c *client.Client, ns, podName string, interval, timeout time.
 		return false, nil
 	}
 
-	return waitForPodCondition(c, ns, podName, condition, interval, timeout)
+	return waitForPodCondition(pw, ns, podName, condition, interval, timeout)
 }
 
 // waitForPodCondition waits for a pod in state defined by a condition (func)
-func waitForPodCondition(c *client.Client, ns, podName string, condition func(pod *api.Pod) (bool, error),
+func waitForPodCondition(pw *k8s.PodWatcher, ns, podName string, condition func(pod *api.Pod) (bool, error),
 	interval, timeout time.Duration) error {
 	return wait.PollImmediate(interval, timeout, func() (bool, error) {
-		pod, err := c.Pods(ns).Get(podName)
-		if err != nil {
-			if apierrs.IsNotFound(err) {
-				return false, nil
-			}
+		pods, err := pw.Store.List(labels.Set{"heritage": podName}.AsSelector())
+		if err != nil || len(pods) == 0 {
+			return false, nil
 		}
 
-		done, err := condition(pod)
+		done, err := condition(pods[0])
 		if err != nil {
 			return false, err
 		}

pkg/k8s/watch.go

@@ -0,0 +1,57 @@
+package k8s
+
+import (
+	"time"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/cache"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+	"k8s.io/kubernetes/pkg/controller/framework"
+	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/watch"
+)
+
+var (
+	resyncPeriod = 30 * time.Second
+)
+
+//PodWatcher is a struct which holds the return values of (k8s.io/kubernetes/pkg/controller/framework).NewIndexerInformer together.
+type PodWatcher struct {
+	Store      cache.StoreToPodLister
+	Controller *framework.Controller
+}
+
+//NewPodWatcher creates a new BuildPodWatcher useful to list the pods using a cache which gets updated based on the watch func.
+func NewPodWatcher(c *client.Client, ns string) *PodWatcher {
+	pw := &PodWatcher{}
+
+	pw.Store.Store, pw.Controller = framework.NewIndexerInformer(
+		&cache.ListWatch{
+			ListFunc:  podListFunc(c, ns),
+			WatchFunc: podWatchFunc(c, ns),
+		},
+		&api.Pod{},
+		resyncPeriod,
+		framework.ResourceEventHandlerFuncs{},
+		cache.Indexers{},
+	)
+
+	return pw
+}
+
+func podListFunc(c *client.Client, ns string) func(options api.ListOptions) (runtime.Object, error) {
+	return func(opts api.ListOptions) (runtime.Object, error) {
+		return c.Pods(ns).List(api.ListOptions{
+			LabelSelector: labels.Everything(),
+		})
+	}
+}
+
+func podWatchFunc(c *client.Client, ns string) func(options api.ListOptions) (watch.Interface, error) {
+	return func(opts api.ListOptions) (watch.Interface, error) {
+		return c.Pods(ns).Watch(api.ListOptions{
+			LabelSelector: labels.Everything(),
+		})
+	}
+}