watcher: Clean up test

logs/pod_watcher.go

@@ -159,6 +159,9 @@ 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,11 +109,6 @@ 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()
 
@@ -126,6 +121,10 @@ 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)
@@ -140,14 +139,12 @@ func TestWatcherWithPodWatcher(t *testing.T) {
 			podsWatcher.Add(pod)
 			time.Sleep(time.Millisecond * 250)
 		}
+
+		watcher.Close()
 	}()
 
-	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())
+	err := watcher.Watch(context.Background())
+	require.NoError(t, err)
 	lines := bufToLines(&buf)
 	require.Len(t, lines, 2)
 	assert.ElementsMatch(t, []string{"[foo] fake logs", "[bar] fake logs"}, lines)