Merge pull request #56 from gabemontero/fix-follow-no-pod

various run build -F/--follow fixes around timeout, lack of pods, and data races

pkg/shp/cmd/build/create.go

@@ -35,7 +35,7 @@ func (c *CreateCommand) Cmd() *cobra.Command {
 }
 
 // Complete fills internal subcommand structure for future work with user input
-func (c *CreateCommand) Complete(params *params.Params, args []string) error {
+func (c *CreateCommand) Complete(params *params.Params, io *genericclioptions.IOStreams, args []string) error {
 	switch len(args) {
 	case 1:
 		c.name = args[0]

pkg/shp/cmd/build/delete.go

@@ -41,7 +41,7 @@ func (c *DeleteCommand) Cmd() *cobra.Command {
 }
 
 // Complete fills DeleteSubCommand structure with data obtained from cobra command
-func (c *DeleteCommand) Complete(params *params.Params, args []string) error {
+func (c *DeleteCommand) Complete(params *params.Params, io *genericclioptions.IOStreams, args []string) error {
 	c.name = args[0]
 
 	return nil

pkg/shp/cmd/build/list.go

@@ -39,7 +39,7 @@ func (c *ListCommand) Cmd() *cobra.Command {
 }
 
 // Complete fills object with user input data
-func (c *ListCommand) Complete(params *params.Params, args []string) error {
+func (c *ListCommand) Complete(params *params.Params, io *genericclioptions.IOStreams, args []string) error {
 	return nil
 }
 

pkg/shp/cmd/build/run.go

@@ -32,12 +32,14 @@ type RunCommand struct {
 	logTail         *tail.Tail                   // follow container logs
 	tailLogsStarted map[string]bool              // controls tail instance per container
 
-	buildName    string // build name
+	logLock sync.Mutex
+
+	buildName    string
 	buildRunName string
+	namespace    string
 	buildRunSpec *buildv1alpha1.BuildRunSpec // stores command-line flags
 	shpClientset buildclientset.Interface
 	follow       bool // flag to tail pod logs
-	watchLock    sync.Mutex
 }
 
 const buildRunLongDesc = `
@@ -53,7 +55,7 @@ func (r *RunCommand) Cmd() *cobra.Command {
 }
 
 // Complete picks the build resource name from arguments, and instantiate additional components.
-func (r *RunCommand) Complete(params *params.Params, args []string) error {
+func (r *RunCommand) Complete(params *params.Params, io *genericclioptions.IOStreams, args []string) error {
 	switch len(args) {
 	case 1:
 		r.buildName = args[0]
@@ -66,6 +68,31 @@ func (r *RunCommand) Complete(params *params.Params, args []string) error {
 		return err
 	}
 	r.logTail = tail.NewTail(r.Cmd().Context(), clientset)
+	r.ioStreams = io
+	r.namespace = params.Namespace()
+	if r.follow {
+		if r.shpClientset, err = params.ShipwrightClientSet(); err != nil {
+			return err
+		}
+
+		kclientset, err := params.ClientSet()
+		if err != nil {
+			return err
+		}
+		to, err := params.RequestTimeout()
+		if err != nil {
+			return err
+		}
+		r.pw, err = reactor.NewPodWatcher(r.Cmd().Context(), to, kclientset, params.Namespace())
+		if err != nil {
+			return err
+		}
+
+		r.pw.WithOnPodModifiedFn(r.onEvent)
+		r.pw.WithTimeoutPodFn(r.onTimeout)
+		r.pw.WithNoPodEventsYetFn(r.onNoPodEventsYet)
+
+	}
 
 	// overwriting build-ref name to use what's on arguments
 	return r.Cmd().Flags().Set(flags.BuildrefNameFlag, r.buildName)
@@ -92,11 +119,49 @@ func (r *RunCommand) tailLogs(pod *corev1.Pod) {
 	}
 }
 
+// onNoPodEventsYet reacts to the pod watcher telling us it has not received any pod events for our build run
+func (r *RunCommand) onNoPodEventsYet() {
+	r.Log(fmt.Sprintf("BuildRun %q log following has not observed any pod events yet.", r.buildRunName))
+	br, err := r.shpClientset.ShipwrightV1alpha1().BuildRuns(r.namespace).Get(r.cmd.Context(), r.buildRunName, metav1.GetOptions{})
+	if err != nil {
+		r.Log(fmt.Sprintf("error accessing BuildRun %q: %s", r.buildRunName, err.Error()))
+		return
+	}
+
+	c := br.Status.GetCondition(buildv1alpha1.Succeeded)
+	giveUp := false
+	msg := ""
+	switch {
+	case c != nil && c.Status == corev1.ConditionTrue:
+		giveUp = true
+		msg = fmt.Sprintf("BuildRun '%s' has been marked as successful.\n", br.Name)
+	case c != nil && c.Status == corev1.ConditionFalse:
+		giveUp = true
+		msg = fmt.Sprintf("BuildRun '%s' has been marked as failed.\n", br.Name)
+	case br.IsCanceled():
+		giveUp = true
+		msg = fmt.Sprintf("BuildRun '%s' has been canceled.\n", br.Name)
+	case br.DeletionTimestamp != nil:
+		giveUp = true
+		msg = fmt.Sprintf("BuildRun '%s' has been deleted.\n", br.Name)
+	case !br.HasStarted():
+		r.Log(fmt.Sprintf("BuildRun '%s' has been marked as failed.\n", br.Name))
+	}
+	if giveUp {
+		r.Log(msg)
+		r.Log(fmt.Sprintf("exiting 'ship build run --follow' for BuildRun %q", br.Name))
+		r.stop()
+	}
+
+}
+
+// onTimeout reacts to either the context or request timeout causing the pod watcher to exit
+func (r *RunCommand) onTimeout(msg string) {
+	r.Log(fmt.Sprintf("BuildRun %q log following has stopped because: %q\n", r.buildRunName, msg))
+}
+
 // onEvent reacts on pod state changes, to start and stop tailing container logs.
 func (r *RunCommand) onEvent(pod *corev1.Pod) error {
-	// found more data races during unit testing with concurrent events coming in
-	r.watchLock.Lock()
-	defer r.watchLock.Unlock()
 	switch pod.Status.Phase {
 	case corev1.PodRunning:
 		// graceful time to wait for container start
@@ -118,14 +183,14 @@ func (r *RunCommand) onEvent(pod *corev1.Pod) error {
 			err = fmt.Errorf("build pod '%s' has failed", pod.GetName())
 		}
 		// see if because of deletion or cancelation
-		fmt.Fprintf(r.ioStreams.Out, msg)
+		r.Log(msg)
 		r.stop()
 		return err
 	case corev1.PodSucceeded:
-		fmt.Fprintf(r.ioStreams.Out, "Pod '%s' has succeeded!\n", pod.GetName())
+		r.Log(fmt.Sprintf("Pod '%s' has succeeded!\n", pod.GetName()))
 		r.stop()
 	default:
-		fmt.Fprintf(r.ioStreams.Out, "Pod '%s' is in state %q...\n", pod.GetName(), string(pod.Status.Phase))
+		r.Log(fmt.Sprintf("Pod '%s' is in state %q...\n", pod.GetName(), string(pod.Status.Phase)))
 		// handle any issues with pulling images that may fail
 		for _, c := range pod.Status.Conditions {
 			if c.Type == corev1.PodInitialized || c.Type == corev1.ContainersReady {
@@ -146,10 +211,7 @@ func (r *RunCommand) stop() {
 
 // Run creates a BuildRun resource based on Build's name informed on arguments.
 func (r *RunCommand) Run(params *params.Params, ioStreams *genericclioptions.IOStreams) error {
-	// ran into some data race conditions during unit test with this starting up, but pod events
-	// coming in before we completed initialization below
-	r.watchLock.Lock()
-	// resource using GenerateName, which will provice a unique instance
+	// resource using GenerateName, which will provide a unique instance
 	br := &buildv1alpha1.BuildRun{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: fmt.Sprintf("%s-", r.buildName),
@@ -162,7 +224,7 @@ func (r *RunCommand) Run(params *params.Params, ioStreams *genericclioptions.IOS
 	if err != nil {
 		return err
 	}
-	br, err = clientset.ShipwrightV1alpha1().BuildRuns(params.Namespace()).Create(r.cmd.Context(), br, metav1.CreateOptions{})
+	br, err = clientset.ShipwrightV1alpha1().BuildRuns(r.namespace).Create(r.cmd.Context(), br, metav1.CreateOptions{})
 	if err != nil {
 		return err
 	}
@@ -172,15 +234,7 @@ func (r *RunCommand) Run(params *params.Params, ioStreams *genericclioptions.IOS
 		return nil
 	}
 
-	r.ioStreams = ioStreams
-	kclientset, err := params.ClientSet()
-	if err != nil {
-		return err
-	}
 	r.buildRunName = br.Name
-	if r.shpClientset, err = params.ShipwrightClientSet(); err != nil {
-		return err
-	}
 
 	// instantiating a pod watcher with a specific label-selector to find the indented pod where the
 	// actual build started by this subcommand is being executed, including the randomized buildrun
@@ -190,19 +244,17 @@ func (r *RunCommand) Run(params *params.Params, ioStreams *genericclioptions.IOS
 		r.buildName,
 		br.GetName(),
 	)}
-	r.pw, err = reactor.NewPodWatcher(r.Cmd().Context(), kclientset, listOpts, params.Namespace())
-	if err != nil {
-		return err
-	}
-
-	r.pw.WithOnPodModifiedFn(r.onEvent)
-	// cannot defer with unlock up top because r.pw.Start() blocks;  but the erroring out above kills the
-	// cli invocation, so it does not matter
-	r.watchLock.Unlock()
-	_, err = r.pw.Start()
+	_, err = r.pw.Start(listOpts)
 	return err
 }
 
+func (r *RunCommand) Log(msg string) {
+	// concurrent fmt.Fprintf(r.ioStream.Out...) calls need locking to avoid data races, as we 'write' to the stream
+	r.logLock.Lock()
+	defer r.logLock.Unlock()
+	fmt.Fprintf(r.ioStreams.Out, msg)
+}
+
 // runCmd instantiate the "build run" sub-command using common BuildRun flags.
 func runCmd() runner.SubCommand {
 	cmd := &cobra.Command{
@@ -214,7 +266,7 @@ func runCmd() runner.SubCommand {
 		cmd:             cmd,
 		buildRunSpec:    flags.BuildRunSpecFromFlags(cmd.Flags()),
 		tailLogsStarted: make(map[string]bool),
-		watchLock:       sync.Mutex{},
+		logLock:         sync.Mutex{},
 	}
 	cmd.Flags().BoolVarP(&runCommand.follow, "follow", "F", runCommand.follow, "Start a build and watch its log until it completes or fails.")
 	return runCommand

pkg/shp/cmd/build/run_test.go

@@ -1,21 +1,21 @@
 package build
 
 import (
-	"runtime"
+	"bytes"
 	"strings"
 	"sync"
 	"testing"
-	"time"
 
 	buildv1alpha1 "github.com/shipwright-io/build/pkg/apis/build/v1alpha1"
 	shpfake "github.com/shipwright-io/build/pkg/client/clientset/versioned/fake"
 	"github.com/shipwright-io/cli/pkg/shp/flags"
 	"github.com/shipwright-io/cli/pkg/shp/params"
+	"github.com/shipwright-io/cli/pkg/shp/reactor"
 	"github.com/spf13/cobra"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kruntime "k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/client-go/kubernetes/fake"
 	fakekubetesting "k8s.io/client-go/testing"
@@ -26,6 +26,8 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 		name       string
 		phase      corev1.PodPhase
 		logText    string
+		to         string
+		noPodYet   bool
 		cancelled  bool
 		brDeleted  bool
 		podDeleted bool
@@ -76,6 +78,16 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 			// k8s folks to "be careful" with it; fortunately, what we do for tail and pod_watcher so far is within
 			// the realm of reliable.
 		},
+		{
+			name:    "timeout",
+			to:      "1s",
+			logText: reactor.RequestTimeoutMessage,
+		},
+		{
+			name:     "no pod yet",
+			noPodYet: true,
+			logText:  "has not observed any pod events yet",
+		},
 	}
 
 	for _, test := range tests {
@@ -103,6 +115,7 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 			},
 		}
 		shpclientset := shpfake.NewSimpleClientset()
+
 		// need this reactor since the Run method uses the ObjectMeta.GenerateName k8s feature to generate the random
 		// name for the BuildRun.  However, for our purposes with unit testing, we want to control the name of the BuildRun
 		// to facilitate the list/selector via labels that is also employed by the Run method.
@@ -116,7 +129,10 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 			return true, br, nil
 		}
 		shpclientset.PrependReactor("get", "buildruns", getReactorFunc)
-		kclientset := fake.NewSimpleClientset(pod)
+		kclientset := fake.NewSimpleClientset()
+		if !test.noPodYet {
+			kclientset = fake.NewSimpleClientset(pod)
+		}
 		ccmd := &cobra.Command{}
 		cmd := &RunCommand{
 			cmd:             ccmd,
@@ -125,12 +141,16 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 			follow:          true,
 			shpClientset:    shpclientset,
 			tailLogsStarted: make(map[string]bool),
-			watchLock:       sync.Mutex{},
+			logLock:         sync.Mutex{},
 		}
 
 		// set up context
 		cmd.Cmd().ExecuteC()
-		param := params.NewParamsForTest(kclientset, shpclientset, nil, metav1.NamespaceDefault)
+		pm := genericclioptions.NewConfigFlags(true)
+		if len(test.to) > 0 {
+			pm.Timeout = &test.to
+		}
+		param := params.NewParamsForTest(kclientset, shpclientset, pm, metav1.NamespaceDefault)
 
 		ioStreams, _, out, _ := genericclioptions.NewTestIOStreams()
 
@@ -143,7 +163,13 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 			pod.DeletionTimestamp = &metav1.Time{}
 		}
 
-		cmd.Complete(param, []string{name})
+		cmd.Complete(param, &ioStreams, []string{name})
+		if len(test.to) > 0 {
+			cmd.Run(param, &ioStreams)
+			checkLog(test.name, test.logText, cmd, out, t)
+			continue
+		}
+
 		go func() {
 			err := cmd.Run(param, &ioStreams)
 			if err != nil {
@@ -152,35 +178,22 @@ func TestStartBuildRunFollowLog(t *testing.T) {
 
 		}()
 
-		// yield the processor, so the initialization in Run can occur; afterward, the watchLock should allow
-		// coordination between Run and onEvent
-		runtime.Gosched()
-
-		// even with our release of the context above with Gosched(), repeated runs in CI have surfaced occasional timing issues between
-		// cmd.Run() finishing initialization and cmd.onEvent trying to used struct variables, resulting in panics; so we employ the lock here
-		// to insure the required initializations have run; this is still better than a generic "sleep log enough for
-		// the init to occur.
-		cmd.watchLock.Lock()
-		err := wait.PollImmediate(1*time.Second, 3*time.Second, func() (done bool, err error) {
-			// check any of the vars on RunCommand that are used in onEvent and make sure they are set;
-			// we are verifying the initialization done in Run() on RunCommand is complete
-			if cmd.pw != nil && cmd.ioStreams != nil && cmd.shpClientset != nil {
-				cmd.watchLock.Unlock()
-				return true, nil
-			}
-			return false, nil
-		})
-		if err != nil {
-			cmd.watchLock.Unlock()
-			t.Errorf("Run initialization did not complete in time")
-		}
-
-		// mimic watch events, bypassing k8s fake client watch hoopla whose plug points are not always useful;
-		pod.Status.Phase = test.phase
-		cmd.onEvent(pod)
-		if !strings.Contains(out.String(), test.logText) {
-			t.Errorf("test %s: unexpected output: %s", test.name, out.String())
+		if !test.noPodYet {
+			// mimic watch events, bypassing k8s fake client watch hoopla whose plug points are not always useful;
+			pod.Status.Phase = test.phase
+			cmd.onEvent(pod)
+		} else {
+			cmd.onNoPodEventsYet()
 		}
+		checkLog(test.name, test.logText, cmd, out, t)
+	}
+}
 
+func checkLog(name, text string, cmd *RunCommand, out *bytes.Buffer, t *testing.T) {
+	// need to employ log lock since accessing same iostream out used by Run cmd
+	cmd.logLock.Lock()
+	defer cmd.logLock.Unlock()
+	if !strings.Contains(out.String(), text) {
+		t.Errorf("test %s: unexpected output: %s", name, out.String())
 	}
 }

pkg/shp/cmd/buildrun/cancel.go

@@ -38,7 +38,7 @@ func (c *CancelCommand) Cmd() *cobra.Command {
 }
 
 // Complete fills in data provided by user
-func (c *CancelCommand) Complete(params *params.Params, args []string) error {
+func (c *CancelCommand) Complete(params *params.Params, io *genericclioptions.IOStreams, args []string) error {
 	c.name = args[0]
 
 	return nil