Merge pull request #22 from gabemontero/buildrun-cancel

add buildrun cancel

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/mailru/easyjson v0.7.1 // indirect
 	github.com/onsi/gomega v1.10.3
 	github.com/pkg/errors v0.9.1
-	github.com/shipwright-io/build v0.5.0
+	github.com/shipwright-io/build v0.5.2-0.20210715083206-5d8fb411a1eb
 	github.com/spf13/cobra v1.1.3
 	github.com/spf13/pflag v1.0.5
 	github.com/texttheater/golang-levenshtein/levenshtein v0.0.0-20200805054039-cae8b0eaed6c

pkg/shp/cmd/build/run.go

@@ -6,13 +6,17 @@ import (
 	"time"
 
 	buildv1alpha1 "github.com/shipwright-io/build/pkg/apis/build/v1alpha1"
+	buildclientset "github.com/shipwright-io/build/pkg/client/clientset/versioned"
+
 	"github.com/shipwright-io/cli/pkg/shp/cmd/runner"
 	"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/shipwright-io/cli/pkg/shp/resource"
 	"github.com/shipwright-io/cli/pkg/shp/tail"
+
 	"github.com/spf13/cobra"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
@@ -28,9 +32,11 @@ type RunCommand struct {
 	logTail         *tail.Tail                   // follow container logs
 	tailLogsStarted map[string]bool              // controls tail instance per container
 
-	buildName    string                      // build name
+	buildName    string // build name
+	buildRunName string
 	buildRunSpec *buildv1alpha1.BuildRunSpec // stores command-line flags
-	Follow       bool                        // flag to tail pod logs
+	shpClientset buildclientset.Interface
+	follow       bool // flag to tail pod logs
 }
 
 const buildRunLongDesc = `
@@ -94,9 +100,23 @@ func (r *RunCommand) onEvent(pod *corev1.Pod) error {
 		// start tailing container logs
 		r.tailLogs(pod)
 	case corev1.PodFailed:
-		fmt.Fprintf(r.ioStreams.Out, "Pod '%s' has failed!\n", pod.GetName())
+		msg := ""
+		br, err := r.shpClientset.ShipwrightV1alpha1().BuildRuns(pod.Namespace).Get(r.cmd.Context(), r.buildRunName, metav1.GetOptions{})
+		switch {
+		case err == nil && br.IsCanceled():
+			msg = fmt.Sprintf("BuildRun '%s' has been canceled.\n", br.Name)
+		case err == nil && br.DeletionTimestamp != nil:
+			msg = fmt.Sprintf("BuildRun '%s' has been deleted.\n", br.Name)
+		case pod.DeletionTimestamp != nil:
+			msg = fmt.Sprintf("Pod '%s' has been deleted.\n", pod.GetName())
+		default:
+			msg = fmt.Sprintf("Pod '%s' has failed!\n", pod.GetName())
+			err = fmt.Errorf("build pod '%s' has failed", pod.GetName())
+		}
+		// see if because of deletion or cancelation
+		fmt.Fprintf(r.ioStreams.Out, msg)
 		r.stop()
-		return fmt.Errorf("build pod '%s' has failed", pod.GetName())
+		return err
 	case corev1.PodSucceeded:
 		fmt.Fprintf(r.ioStreams.Out, "Pod '%s' has succeeded!\n", pod.GetName())
 		r.stop()
@@ -137,11 +157,16 @@ func (r *RunCommand) Run(params *params.Params, ioStreams *genericclioptions.IOS
 		return err
 	}
 
-	if !r.Follow {
+	if !r.follow {
 		fmt.Fprintf(ioStreams.Out, "BuildRun created %q for build %q\n", br.GetName(), r.buildName)
 		return nil
 	}
 
+	r.buildRunName = br.Name
+	if r.shpClientset, err = params.ShipwrightClientSet(); err != nil {
+		return err
+	}
+
 	clientset, err := params.ClientSet()
 	if err != nil {
 		return err
@@ -178,6 +203,6 @@ func runCmd() runner.SubCommand {
 		buildRunSpec:    flags.BuildRunSpecFromFlags(cmd.Flags()),
 		tailLogsStarted: make(map[string]bool),
 	}
-	cmd.Flags().BoolVarP(&runCommand.Follow, "follow", "F", runCommand.Follow, "Start a build and watch its log until it completes or fails.")
+	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/buildrun/buildrun.go

@@ -26,6 +26,7 @@ func Command(p *params.Params, ioStreams *genericclioptions.IOStreams) *cobra.Co
 		runner.NewRunner(p, ioStreams, logsCmd()).Cmd(),
 		runner.NewRunner(p, ioStreams, logsCmd()).Cmd(),
 		runner.NewRunner(p, ioStreams, createCmd()).Cmd(),
+		runner.NewRunner(p, ioStreams, cancelCmd()).Cmd(),
 	)
 	return command
 }

pkg/shp/cmd/buildrun/cancel.go

@@ -0,0 +1,73 @@
+package buildrun
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+
+	buildv1alpha1 "github.com/shipwright-io/build/pkg/apis/build/v1alpha1"
+	"github.com/shipwright-io/cli/pkg/shp/cmd/runner"
+	"github.com/shipwright-io/cli/pkg/shp/params"
+	"github.com/shipwright-io/cli/pkg/shp/resource"
+)
+
+// CancelCommand contains data input from user for delete sub-command
+type CancelCommand struct {
+	cmd *cobra.Command
+
+	name string
+}
+
+func cancelCmd() runner.SubCommand {
+	return &CancelCommand{
+		cmd: &cobra.Command{
+			Use:   "cancel <name>",
+			Short: "Cancel BuildRun",
+			Args:  cobra.ExactArgs(1),
+		},
+	}
+}
+
+// Cmd returns cobra command object
+func (c *CancelCommand) Cmd() *cobra.Command {
+	return c.cmd
+}
+
+// Complete fills in data provided by user
+func (c *CancelCommand) Complete(params *params.Params, args []string) error {
+	c.name = args[0]
+
+	return nil
+}
+
+// Validate validates data input by user
+func (c *CancelCommand) Validate() error {
+	return nil
+}
+
+// Run executes cancel sub-command logic
+func (c *CancelCommand) Run(params *params.Params, ioStreams *genericclioptions.IOStreams) error {
+	brr := resource.GetBuildRunResource(params)
+
+	br := &buildv1alpha1.BuildRun{}
+	if err := brr.Get(c.cmd.Context(), c.name, br); err != nil {
+		return fmt.Errorf("failed to retrieve BuildRun %s: %s", c.name, err.Error())
+	}
+	//TODO replace with br.IsDone() when that is available and vendored in
+	cond := br.Status.GetCondition(buildv1alpha1.Succeeded)
+	if cond != nil && cond.GetStatus() != corev1.ConditionUnknown {
+		return fmt.Errorf("failed to cancel BuildRun %s: execution has already finished", c.name)
+	}
+
+	//TODO use constant when vendor in api changes
+	if err := brr.Patch(c.cmd.Context(), c.name, "replace", "/spec/state", "BuildRunCanceled"); err != nil {
+		return err
+	}
+
+	fmt.Fprintf(ioStreams.Out, "BuildRun successfully canceled '%v'\n", c.name)
+
+	return nil
+}

pkg/shp/params/params.go

@@ -3,6 +3,9 @@ package params
 import (
 	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
+
+	buildclientset "github.com/shipwright-io/build/pkg/client/clientset/versioned"
+
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
@@ -11,8 +14,9 @@ import (
 // Params is a place for Shipwright CLI to store its runtime parameters including configured dynamic
 // client and global flags.
 type Params struct {
-	client    dynamic.Interface
-	clientset kubernetes.Interface
+	client       dynamic.Interface
+	clientset    kubernetes.Interface
+	shpClientset buildclientset.Interface
 
 	configFlags *genericclioptions.ConfigFlags
 	namespace   string
@@ -70,6 +74,23 @@ func (p *Params) ClientSet() (kubernetes.Interface, error) {
 	return p.clientset, nil
 }
 
+// ShipwrightClientSet returns a Shipwright Clientset
+func (p *Params) ShipwrightClientSet() (buildclientset.Interface, error) {
+	if p.shpClientset != nil {
+		return p.shpClientset, nil
+	}
+	clientConfig := p.configFlags.ToRawKubeConfigLoader()
+	config, err := clientConfig.ClientConfig()
+	if err != nil {
+		return nil, err
+	}
+	p.shpClientset, err = buildclientset.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return p.shpClientset, nil
+}
+
 // Namespace returns kubernetes namespace with alle the overrides
 // from command line and kubernetes config
 func (p *Params) Namespace() string {

pkg/shp/reactor/pod_watcher.go

@@ -2,6 +2,7 @@ package reactor
 
 import (
 	"context"
+	"sync"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -13,9 +14,11 @@ import (
 // state modifications, should work as a helper to build business logic based on the build POD
 // changes.
 type PodWatcher struct {
-	ctx     context.Context
-	stopCh  chan bool       // stops the event loop execution
-	watcher watch.Interface // client watch instance
+	ctx      context.Context
+	stopCh   chan bool // stops the event loop execution
+	stopLock sync.Mutex
+	stopped  bool
+	watcher  watch.Interface // client watch instance
 
 	skipPodFn       SkipPodFn
 	onPodAddedFn    OnPodEventFn
@@ -116,7 +119,14 @@ func (p *PodWatcher) Start() (*corev1.Pod, error) {
 
 // Stop closes the stop channel, and stops the execution loop.
 func (p *PodWatcher) Stop() {
-	close(p.stopCh)
+	// employ sync because of observed 'panic: close of closed channel' when running build run log following
+	// along with canceling of builds
+	p.stopLock.Lock()
+	defer p.stopLock.Unlock()
+	if !p.stopped {
+		close(p.stopCh)
+		p.stopped = true
+	}
 }
 
 // NewPodWatcher instantiate PodWatcher event-loop.
@@ -130,5 +140,5 @@ func NewPodWatcher(
 	if err != nil {
 		return nil, err
 	}
-	return &PodWatcher{ctx: ctx, watcher: w, stopCh: make(chan bool)}, nil
+	return &PodWatcher{ctx: ctx, watcher: w, stopCh: make(chan bool), stopLock: sync.Mutex{}}, nil
 }

pkg/shp/resource/resource.go

@@ -57,6 +57,15 @@ func (r *Resource) Update(ctx context.Context, name string, obj interface{}) err
 	return util.UpdateObject(ctx, ri, name, r.gv.WithKind(r.kind), obj)
 }
 
+func (r *Resource) Patch(ctx context.Context, name, op, path, value string) error {
+	ri, err := r.getResourceInterface()
+	if err != nil {
+		return err
+	}
+
+	return util.PatchObject(ctx, ri, name, op, path, value)
+}
+
 // Delete deletes the object identified by name
 func (r *Resource) Delete(ctx context.Context, name string) error {
 	ri, err := r.getResourceInterface()

pkg/shp/tail/tail.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"strings"
+	"sync"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -18,6 +19,8 @@ type Tail struct {
 	ctx       context.Context      // global context
 	clientset kubernetes.Interface // kubernetes client instance
 	stopCh    chan bool            // stop channel
+	stopLock  sync.Mutex
+	stopped   bool
 
 	stdout io.Writer
 	stderr io.Writer
@@ -60,13 +63,20 @@ func (t *Tail) Start(ns, podName, container string) {
 	}()
 	go func() {
 		<-t.ctx.Done()
-		close(t.stopCh)
+		t.Stop()
 	}()
 }
 
 // Stop closes stop channel to stop log streaming.
 func (t *Tail) Stop() {
-	close(t.stopCh)
+	// employ sync because of observed 'panic: close of closed channel' when running build run log following
+	// along with canceling of builds
+	t.stopLock.Lock()
+	defer t.stopLock.Unlock()
+	if !t.stopped {
+		close(t.stopCh)
+		t.stopped = true
+	}
 }
 
 // NewTail instantiate Tail, using by default regular stdout and stderr.
@@ -75,6 +85,7 @@ func NewTail(ctx context.Context, clientset kubernetes.Interface) *Tail {
 		ctx:       ctx,
 		clientset: clientset,
 		stopCh:    make(chan bool, 1),
+		stopLock:  sync.Mutex{},
 		stdout:    os.Stdout,
 		stderr:    os.Stderr,
 	}

pkg/shp/util/crud.go

@@ -2,11 +2,13 @@ package util
 
 import (
 	"context"
+	"encoding/json"
 
 	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/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
 )
 
@@ -83,3 +85,24 @@ func ListObjectWithOptions(ctx context.Context, resource dynamic.ResourceInterfa
 
 	return fromUnstructured(u.UnstructuredContent(), result)
 }
+
+type patchStringValue struct {
+	Op    string `json:"op"`
+	Path  string `json:"path"`
+	Value string `json:"value"`
+}
+
+func PatchObject(ctx context.Context, resource dynamic.ResourceInterface, name, op, path, value string) error {
+	payload := []patchStringValue{{
+		Op:    op,
+		Path:  path,
+		Value: value,
+	}}
+	var data []byte
+	var err error
+	if data, err = json.Marshal(payload); err != nil {
+		return err
+	}
+	_, err = resource.Patch(ctx, name, types.JSONPatchType, data, v1.PatchOptions{})
+	return err
+}