Fix race conditions and run tests with a race detector

pkg/skaffold/build/cache/retrieve.go

@@ -52,41 +52,42 @@ func (c *Cache) RetrieveCachedArtifacts(ctx context.Context, out io.Writer, arti
 	start := time.Now()
 	color.Default.Fprintln(out, "Checking cache...")
 
-	var needToBuild []*latest.Artifact
-	var built []build.Artifact
+	var (
+		needToBuild []*latest.Artifact
+		built       []build.Artifact
 
-	var wg sync.WaitGroup
-	wg.Add(len(artifacts))
+		wg   sync.WaitGroup
+		lock sync.Mutex
+	)
 
-	var canceled bool
+	wg.Add(len(artifacts))
 
 	for _, a := range artifacts {
 		a := a
 		go func() {
 			defer wg.Done()
-			select {
-			case <-ctx.Done():
-				canceled = true
-			default:
-				artifact, err := c.resolveCachedArtifact(ctx, out, a)
-				if err != nil {
-					logrus.Debugf("error retrieving cached artifact for %s: %v\n", a.ImageName, err)
-					color.Red.Fprintf(out, "Unable to retrieve %s from cache; this image will be rebuilt.\n", a.ImageName)
-					needToBuild = append(needToBuild, a)
-					return
-				}
-				if artifact == nil {
-					needToBuild = append(needToBuild, a)
-					return
-				}
-				built = append(built, *artifact)
+
+			artifact, err := c.resolveCachedArtifact(ctx, out, a)
+
+			lock.Lock()
+			defer lock.Unlock()
+
+			if err != nil {
+				logrus.Debugf("error retrieving cached artifact for %s: %v\n", a.ImageName, err)
+				color.Red.Fprintf(out, "Unable to retrieve %s from cache; this image will be rebuilt.\n", a.ImageName)
+
+				needToBuild = append(needToBuild, a)
+				return
+			}
+			if artifact == nil {
+				needToBuild = append(needToBuild, a)
+				return
 			}
+
+			built = append(built, *artifact)
 		}()
 	}
 	wg.Wait()
-	if canceled {
-		return nil, nil, context.Canceled
-	}
 
 	color.Default.Fprintln(out, "Cache check complete in", time.Since(start))
 	return needToBuild, built, nil

pkg/skaffold/event/event.go

@@ -53,11 +53,13 @@ type eventHandler struct {
 	state     proto.State
 	stateLock sync.Mutex
 
-	listeners []chan proto.LogEntry
+	listeners []listener
 }
 
-func (ev *eventHandler) RegisterListener(listener chan proto.LogEntry) {
-	ev.listeners = append(ev.listeners, listener)
+type listener struct {
+	callback func(*proto.LogEntry) error
+	errors   chan error
+	closed   bool
 }
 
 func (ev *eventHandler) getState() proto.State {
@@ -75,37 +77,43 @@ func (ev *eventHandler) getState() proto.State {
 func (ev *eventHandler) logEvent(entry proto.LogEntry) {
 	ev.logLock.Lock()
 
-	for _, c := range ev.listeners {
-		c <- entry
+	for _, listener := range ev.listeners {
+		if listener.closed {
+			continue
+		}
+
+		if err := listener.callback(&entry); err != nil {
+			listener.errors <- err
+			listener.closed = true
+		}
 	}
 	ev.eventLog = append(ev.eventLog, entry)
 
 	ev.logLock.Unlock()
 }
 
 func (ev *eventHandler) forEachEvent(callback func(*proto.LogEntry) error) error {
-	c := make(chan proto.LogEntry)
+	listener := listener{
+		callback: callback,
+		errors:   make(chan error),
+	}
 
 	ev.logLock.Lock()
 
 	oldEvents := make([]proto.LogEntry, len(ev.eventLog))
 	copy(oldEvents, ev.eventLog)
-	ev.RegisterListener(c)
+	ev.listeners = append(ev.listeners, listener)
 
 	ev.logLock.Unlock()
 
 	for i := range oldEvents {
 		if err := callback(&oldEvents[i]); err != nil {
+			// listener should maybe be closed
 			return err
 		}
 	}
 
-	for {
-		entry := <-c
-		if err := callback(&entry); err != nil {
-			return err
-		}
-	}
+	return <-listener.errors
 }
 
 func emptyState(build *latest.BuildConfig) proto.State {

pkg/skaffold/event/event_test.go

@@ -26,32 +26,26 @@ import (
 )
 
 func TestGetLogEvents(t *testing.T) {
-	for step := 0; step < 100000; step++ {
+	for step := 0; step < 10000; step++ {
 		ev := &eventHandler{}
 
-		ev.logEvent(proto.LogEntry{Entry: "OLD1"})
-
-		var done int32
+		ev.logEvent(proto.LogEntry{Entry: "OLD"})
 		go func() {
 			ev.logEvent(proto.LogEntry{Entry: "FRESH"})
-
-			for atomic.LoadInt32(&done) == 0 {
-				ev.logEvent(proto.LogEntry{Entry: "POISON PILL"})
-			}
+			ev.logEvent(proto.LogEntry{Entry: "POISON PILL"})
 		}()
 
-		received := 0
+		var received int32
 		ev.forEachEvent(func(e *proto.LogEntry) error {
 			if e.Entry == "POISON PILL" {
 				return errors.New("Done")
 			}
 
-			received++
+			atomic.AddInt32(&received, 1)
 			return nil
 		})
-		atomic.StoreInt32(&done, int32(1))
 
-		if received != 2 {
+		if atomic.LoadInt32(&received) != 2 {
 			t.Fatalf("Expected %d events, Got %d (Step: %d)", 2, received, step)
 		}
 	}

test.sh

@@ -21,7 +21,7 @@ GREEN='\033[0;32m'
 RESET='\033[0m'
 
 echo "Running go tests..."
-go test -count=1 -cover -short -timeout 60s -coverprofile=out/coverage.txt -covermode=atomic ./... | grep -v 'no test files' | sed ''/PASS/s//$(printf "${GREEN}PASS${RESET}")/'' | sed ''/FAIL/s//$(printf "${RED}FAIL${RESET}")/''
+go test -count=1 -race -cover -short -timeout 60s -coverprofile=out/coverage.txt -covermode=atomic ./... | grep -v 'no test files' | sed ''/PASS/s//$(printf "${GREEN}PASS${RESET}")/'' | sed ''/FAIL/s//$(printf "${RED}FAIL${RESET}")/''
 GO_TEST_EXIT_CODE=${PIPESTATUS[0]}
 if [[ $GO_TEST_EXIT_CODE -ne 0 ]]; then
     exit $GO_TEST_EXIT_CODE