Integrate file sync events into dev command

integration/sync_test.go

@@ -98,12 +98,34 @@ func TestDevSyncAPITrigger(t *testing.T) {
 	skaffold.Build().InDir("testdata/file-sync").WithConfig("skaffold-manual.yaml").InNs(ns.Name).RunOrFail(t)
 
 	rpcAddr := randomPort()
-	client, shutdown := setupRPCClient(t, rpcAddr)
-	defer shutdown()
 
 	stop := skaffold.Dev("--auto-sync=false", "--rpc-port", rpcAddr).InDir("testdata/file-sync").WithConfig("skaffold-manual.yaml").InNs(ns.Name).RunBackground(t)
 	defer stop()
 
+	client, shutdown := setupRPCClient(t, rpcAddr)
+	defer shutdown()
+
+	stream, err := readEventAPIStream(client, t, readRetries)
+	if stream == nil {
+		t.Fatalf("error retrieving event log: %v\n", err)
+	}
+
+	// throw away first 5 entries of log (from first run of dev loop)
+	for i := 0; i < 5; i++ {
+		stream.Recv()
+	}
+
+	// read entries from the log
+	entries := make(chan *proto.LogEntry)
+	go func() {
+		for {
+			entry, _ := stream.Recv()
+			if entry != nil {
+				entries <- entry
+			}
+		}
+	}()
+
 	k8sclient.WaitForPodsReady("test-file-sync")
 
 	ioutil.WriteFile("testdata/file-sync/foo", []byte("foo"), 0644)
@@ -115,9 +137,23 @@ func TestDevSyncAPITrigger(t *testing.T) {
 		},
 	})
 
-	err := wait.PollImmediate(time.Millisecond*500, 1*time.Minute, func() (bool, error) {
+	// Ensure we see a file sync in progress triggered in the event log
+	err = wait.PollImmediate(time.Millisecond*500, 2*time.Minute, func() (bool, error) {
+		e := <-entries
+		return e.GetEvent().GetFileSyncEvent().GetStatus() == "In Progress", nil
+	})
+	testutil.CheckError(t, false, err)
+
+	err = wait.PollImmediate(time.Millisecond*500, 1*time.Minute, func() (bool, error) {
 		out, _ := exec.Command("kubectl", "exec", "test-file-sync", "-n", ns.Name, "--", "cat", "foo").Output()
 		return string(out) == "foo", nil
 	})
 	testutil.CheckError(t, false, err)
+
+	// Ensure we see a file sync succeeded triggered in the event log
+	err = wait.PollImmediate(time.Millisecond*500, 2*time.Minute, func() (bool, error) {
+		e := <-entries
+		return e.GetEvent().GetFileSyncEvent().GetStatus() == "Succeeded", nil
+	})
+	testutil.CheckError(t, false, err)
 }

pkg/skaffold/runner/dev.go

@@ -34,6 +34,13 @@ import (
 // ErrorConfigurationChanged is a special error that's returned when the skaffold configuration was changed.
 var ErrorConfigurationChanged = errors.New("configuration changed")
 
+var (
+	// For testing
+	fileSyncInProgress = event.FileSyncInProgress
+	fileSyncFailed     = event.FileSyncFailed
+	fileSyncSucceeded  = event.FileSyncSucceeded
+)
+
 func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer) error {
 	if r.changeSet.needsReload {
 		return ErrorConfigurationChanged
@@ -59,12 +66,17 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer) error {
 		}()
 
 		for _, s := range r.changeSet.needsResync {
-			color.Default.Fprintf(out, "Syncing %d files for %s\n", len(s.Copy)+len(s.Delete), s.Image)
+			fileCount := len(s.Copy) + len(s.Delete)
+			color.Default.Fprintf(out, "Syncing %d files for %s\n", fileCount, s.Image)
+			fileSyncInProgress(fileCount, s.Image)
 
 			if err := r.syncer.Sync(ctx, s); err != nil {
 				logrus.Warnln("Skipping deploy due to sync error:", err)
+				fileSyncFailed(fileCount, s.Image, err)
 				return nil
 			}
+
+			fileSyncSucceeded(fileCount, s.Image)
 		}
 	}
 

pkg/skaffold/runner/dev_test.go

@@ -276,11 +276,18 @@ func TestDev(t *testing.T) {
 }
 
 func TestDevSync(t *testing.T) {
+	type fileSyncEventCalls struct {
+		InProgress int
+		Failed     int
+		Succeeded  int
+	}
+
 	tests := []struct {
-		description     string
-		testBench       *TestBench
-		watchEvents     []filemon.Events
-		expectedActions []Actions
+		description                string
+		testBench                  *TestBench
+		watchEvents                []filemon.Events
+		expectedActions            []Actions
+		expectedFileSyncEventCalls fileSyncEventCalls
 	}{
 		{
 			description: "sync",
@@ -298,6 +305,11 @@ func TestDevSync(t *testing.T) {
 					Synced: []string{"img1:1"},
 				},
 			},
+			expectedFileSyncEventCalls: fileSyncEventCalls{
+				InProgress: 1,
+				Failed:     0,
+				Succeeded:  1,
+			},
 		},
 		{
 			description: "sync twice",
@@ -319,10 +331,19 @@ func TestDevSync(t *testing.T) {
 					Synced: []string{"img1:1"},
 				},
 			},
+			expectedFileSyncEventCalls: fileSyncEventCalls{
+				InProgress: 2,
+				Failed:     0,
+				Succeeded:  2,
+			},
 		},
 	}
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
+			var actualFileSyncEventCalls fileSyncEventCalls
+			t.Override(&fileSyncInProgress, func(int, string) { actualFileSyncEventCalls.InProgress++ })
+			t.Override(&fileSyncFailed, func(int, string, error) { actualFileSyncEventCalls.Failed++ })
+			t.Override(&fileSyncSucceeded, func(int, string) { actualFileSyncEventCalls.Succeeded++ })
 			t.SetupFakeKubernetesContext(api.Config{CurrentContext: "cluster1"})
 			t.Override(&sync.WorkingDir, func(string, map[string]bool) (string, error) { return "/", nil })
 			test.testBench.cycles = len(test.watchEvents)
@@ -346,6 +367,7 @@ func TestDevSync(t *testing.T) {
 
 			t.CheckNoError(err)
 			t.CheckDeepEqual(test.expectedActions, test.testBench.Actions())
+			t.CheckDeepEqual(test.expectedFileSyncEventCalls, actualFileSyncEventCalls)
 		})
 	}
 }