logs/pod_watcher.go

@@ -159,9 +159,6 @@ func (pw *PodWatcher) watchPods(ctx context.Context, wg *sync.WaitGroup) error {
 
 		// logLines is never closed
 		case line := <-logLines:
-			// FIXME: pw.dst is only written to in this goroutine, but should be
-			// expected to be read by the calling code. This creates a likely race
-			// condition (and fails the race checker).
 			if _, err := pw.dst.Write([]byte(line)); err != nil {
 				return fmt.Errorf("error writing logs: %v", err)
 			}

logs/watcher_test.go

@@ -109,6 +109,11 @@ func TestWatcherClosedChannel(t *testing.T) {
 }
 
 func TestWatcherWithPodWatcher(t *testing.T) {
+	// TODO: use watcher.Close() to clean up watcher instead of relying on
+	// context, which leads to occasional race conditions in tests.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
+	defer cancel()
+
 	deploymentsWatcher := watch.NewFake()
 	defer deploymentsWatcher.Stop()
 
@@ -121,10 +126,6 @@ func TestWatcherWithPodWatcher(t *testing.T) {
 	untypedClient.PrependWatchReactor("deployments", k8stest.DefaultWatchReactor(deploymentsWatcher, nil))
 	client := logs.KubernetesClient{Typed: typedClient, Untyped: untypedClient}
 
-	var buf bytes.Buffer
-	params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default"}
-	watcher := logs.NewWatcher(params, client, logs.NewPodWatcher, &buf, discardLogger())
-
 	go func() {
 		deployment := buildDeployment(t, "mydeployment")
 		deploymentsWatcher.Add(deployment)
@@ -139,12 +140,14 @@ func TestWatcherWithPodWatcher(t *testing.T) {
 			podsWatcher.Add(pod)
 			time.Sleep(time.Millisecond * 250)
 		}
-
-		watcher.Close()
 	}()
 
-	err := watcher.Watch(context.Background())
-	require.NoError(t, err)
+	var buf bytes.Buffer
+	params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default"}
+	watcher := logs.NewWatcher(params, client, logs.NewPodWatcher, &buf, discardLogger())
+
+	err := watcher.Watch(ctx)
+	require.EqualError(t, err, context.DeadlineExceeded.Error())
 	lines := bufToLines(&buf)
 	require.Len(t, lines, 2)
 	assert.ElementsMatch(t, []string{"[foo] fake logs", "[bar] fake logs"}, lines)