fix intermittent race on closing tail/watch channels

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/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,
 	}