introduce --watch capability for oc rsync

Author: bparees

contrib/completions/bash/oc

@@ -2622,6 +2622,8 @@ _oc_rsync()
     flags+=("--quiet")
     flags+=("-q")
     flags+=("--strategy=")
+    flags+=("--watch")
+    flags+=("-w")
     flags+=("--api-version=")
     flags+=("--certificate-authority=")
     flags_with_completion+=("--certificate-authority")

contrib/completions/bash/openshift

@@ -6171,6 +6171,8 @@ _openshift_cli_rsync()
     flags+=("--quiet")
     flags+=("-q")
     flags+=("--strategy=")
+    flags+=("--watch")
+    flags+=("-w")
     flags+=("--api-version=")
     flags+=("--certificate-authority=")
     flags_with_completion+=("--certificate-authority")

pkg/cmd/cli/cmd/rsync/copyrsync.go

@@ -28,7 +28,8 @@ type rsyncStrategy struct {
 	podChecker     podChecker
 }
 
-var rshExcludeFlags = sets.NewString("delete", "strategy", "quiet", "include", "exclude", "progress", "no-perms")
+// rshExcludeFlags are flags that are passed to oc rsync, and should not be passed on to the underlying command being invoked via oc rsh.
+var rshExcludeFlags = sets.NewString("delete", "strategy", "quiet", "include", "exclude", "progress", "no-perms", "watch")
 
 func newRsyncStrategy(f *clientcmd.Factory, c *cobra.Command, o *RsyncOptions) (copyStrategy, error) {
 	// Determine the rsh command to pass to the local rsync command

pkg/cmd/cli/cmd/rsync/rsync.go

@@ -4,11 +4,16 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sync"
+	"time"
 
+	"github.com/fsnotify/fsnotify"
+	"github.com/golang/glog"
 	"github.com/spf13/cobra"
 	kcmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 
 	"github.com/openshift/origin/pkg/cmd/util/clientcmd"
+	"github.com/openshift/origin/pkg/util/fsnotification"
 )
 
 const (
@@ -70,6 +75,7 @@ type RsyncOptions struct {
 	StrategyName  string
 	Quiet         bool
 	Delete        bool
+	Watch         bool
 
 	RsyncInclude  string
 	RsyncExclude  string
@@ -111,6 +117,7 @@ func NewCmdRsync(name, parent string, f *clientcmd.Factory, out, errOut io.Write
 	cmd.Flags().StringVar(&o.RsyncInclude, "include", "", "rsync - include files matching specified pattern")
 	cmd.Flags().BoolVar(&o.RsyncProgress, "progress", false, "rsync - show progress during transfer")
 	cmd.Flags().BoolVar(&o.RsyncNoPerms, "no-perms", false, "rsync - do not transfer permissions")
+	cmd.Flags().BoolVarP(&o.Watch, "watch", "w", false, "Watch directory for changes and resync automatically")
 	return cmd
 }
 
@@ -229,7 +236,92 @@ func (o *RsyncOptions) Validate() error {
 
 // RunRsync copies files from source to destination
 func (o *RsyncOptions) RunRsync() error {
-	return o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut)
+	if err := o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut); err != nil {
+		return err
+	}
+
+	if !o.Watch {
+		return nil
+	}
+	return o.WatchAndSync()
+}
+
+// WatchAndSync sets up a recursive filesystem watch on the sync path
+// and invokes rsync each time the path changes.
+func (o *RsyncOptions) WatchAndSync() error {
+
+	// these variables must be accessed while holding the changeLock
+	// mutex as they are shared between goroutines to communicate
+	// sync state/events.
+	var (
+		changeLock sync.Mutex
+		dirty      bool
+		lastChange time.Time
+		watchError error
+	)
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return fmt.Errorf("error setting up filesystem watcher: %v", err)
+	}
+	defer watcher.Close()
+
+	go func() {
+		for {
+			select {
+			case event := <-watcher.Events:
+				changeLock.Lock()
+				glog.V(5).Infof("filesystem watch event: %s", event)
+				lastChange = time.Now()
+				dirty = true
+				if event.Op&fsnotify.Remove == fsnotify.Remove {
+					if e := watcher.Remove(event.Name); e != nil {
+						glog.V(5).Infof("error removing watch for %s: %v", event.Name, e)
+					}
+				} else {
+					if e := fsnotification.AddRecursiveWatch(watcher, event.Name); e != nil && watchError == nil {
+						watchError = e
+					}
+				}
+				changeLock.Unlock()
+			case err := <-watcher.Errors:
+				changeLock.Lock()
+				watchError = fmt.Errorf("error watching filesystem for changes: %v", err)
+				changeLock.Unlock()
+			}
+		}
+	}()
+
+	err = fsnotification.AddRecursiveWatch(watcher, o.Source.Path)
+	if err != nil {
+		return fmt.Errorf("error watching source path %s: %v", o.Source.Path, err)
+	}
+
+	delay := 2 * time.Second
+	ticker := time.NewTicker(delay)
+	defer ticker.Stop()
+	for {
+		changeLock.Lock()
+		if watchError != nil {
+			return watchError
+		}
+		// if a change happened more than 'delay' seconds ago, sync it now.
+		// if a change happened less than 'delay' seconds ago, sleep for 'delay' seconds
+		// and see if more changes happen, we don't want to sync when
+		// the filesystem is in the middle of changing due to a massive
+		// set of changes (such as a local build in progress).
+		if dirty && time.Now().After(lastChange.Add(delay)) {
+			glog.V(1).Info("Synchronizing filesystem changes...")
+			err = o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut)
+			if err != nil {
+				return err
+			}
+			glog.V(1).Info("Done.")
+			dirty = false
+		}
+		changeLock.Unlock()
+		<-ticker.C
+	}
 }
 
 // PodName returns the name of the pod as specified in either the

pkg/util/fsnotification/doc.go

@@ -0,0 +1,3 @@
+// Package fsnotification provides helper functions that wrap the fsnotify filesystem
+// notification package.
+package fsnotification

pkg/util/fsnotification/fsnotification.go

@@ -0,0 +1,52 @@
+package fsnotification
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/golang/glog"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+// AddRecursiveWatch handles adding watches recursively for the path provided
+// and its subdirectories.  If a non-directory is specified, this call is a no-op.
+// Recursive logic from https://github.com/bronze1man/kmg/blob/master/fsnotify/Watcher.go
+func AddRecursiveWatch(watcher *fsnotify.Watcher, path string) error {
+	file, err := os.Stat(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return fmt.Errorf("error introspecting path %s: %v", path, err)
+	}
+	if !file.IsDir() {
+		return nil
+	}
+
+	folders, err := getSubFolders(path)
+	for _, v := range folders {
+		glog.V(5).Infof("adding watch on path %s", v)
+		err = watcher.Add(v)
+		if err != nil {
+			return fmt.Errorf("error adding watcher for path %s: %v", v, err)
+		}
+	}
+	return nil
+}
+
+// getSubFolders recursively retrieves all subfolders of the specified path.
+func getSubFolders(path string) (paths []string, err error) {
+	err = filepath.Walk(path, func(newPath string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			paths = append(paths, newPath)
+		}
+		return nil
+	})
+	return paths, err
+}

test/extended/cli/rsync.go

@@ -1,17 +1,20 @@
 package cli
 
 import (
+	"bufio"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	g "github.com/onsi/ginkgo"
 	o "github.com/onsi/gomega"
 
 	kapi "k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/kubernetes/test/e2e"
 
 	exutil "github.com/openshift/origin/test/extended/util"
 )
@@ -47,8 +50,144 @@ var _ = g.Describe("[cli][Slow] can use rsync to upload files to pods", func() {
 		podName = pods.Items[0].Name
 	})
 
-	g.Describe("copy by strategy", func() {
+	g.Describe("using a watch", func() {
+		g.It("should watch for changes and rsync them", func() {
+			g.By("Creating a local temporary directory")
+			tempDir, err := ioutil.TempDir("", "rsync")
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By("creating a subdirectory in that temp directory")
+			subdir1 := filepath.Join(tempDir, "subdir1")
+			err = os.Mkdir(subdir1, 0777)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By("creating a file in the subdirectory")
+			subdir1file1 := filepath.Join(subdir1, "file1")
+			_, err = os.Create(subdir1file1)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By("Creating a scratch directory in the pod")
+			_, err = oc.Run("rsh").Args(podName, "mkdir", "/tmp/rsync").Output()
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By(fmt.Sprintf("Calling oc rsync %s/ %s:/tmp/rsync --delete --watch", tempDir, podName))
+			cmd, stdout, stderr, err := oc.Run("rsync").Args(
+				fmt.Sprintf("%s/", tempDir),
+				fmt.Sprintf("%s:/tmp/rsync", podName),
+				"--loglevel=5",
+				"--delete",
+				"--watch").Background()
+
+			failed := true
+			defer cmd.Process.Kill()
+			defer func() {
+				if failed {
+					writer := cmd.Stdout.(*bufio.Writer)
+					writer.Flush()
+					writer2 := cmd.Stderr.(*bufio.Writer)
+					writer2.Flush()
+					fmt.Fprintf(g.GinkgoWriter, "Dumping rsync output: \n%s\n%s\n", stdout.String(), stderr.String())
+				}
+			}()
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			var result string
+			found := false
+			for i := 0; i < 12; i++ {
+				g.By("Verifying that files are copied to the container")
+				result, _ = oc.Run("rsh").Args(podName, "ls", "/tmp/rsync/subdir1").Output()
+				if strings.Contains(result, "file1") {
+					found = true
+					break
+				}
+				time.Sleep(5 * time.Second)
+			}
+			if !found {
+				e2e.Failf("Directory does not contain expected files: \n%s", result)
+			}
+
+			g.By("renaming file1 to file2")
+			subdir1file2 := filepath.Join(subdir1, "file2")
+			err = os.Rename(subdir1file1, subdir1file2)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			found = false
+			for i := 0; i < 12; i++ {
+				g.By("Verifying that files are copied to the container")
+				result, _ = oc.Run("rsh").Args(podName, "ls", "/tmp/rsync/subdir1").Output()
+				if strings.Contains(result, "file2") && !strings.Contains(result, "file1") {
+					found = true
+					break
+				}
+				time.Sleep(5 * time.Second)
+			}
+			if !found {
+				e2e.Failf("Directory does not contain expected files: \n%s", result)
+			}
+
+			g.By("removing file2")
+			err = os.Remove(subdir1file2)
+			o.Expect(err).NotTo(o.HaveOccurred())
 
+			found = false
+			for i := 0; i < 12; i++ {
+				g.By("Verifying that files are copied to the container")
+				result, _ = oc.Run("rsh").Args(podName, "ls", "/tmp/rsync/subdir1").Output()
+				if !strings.Contains(result, "file2") {
+					found = true
+					break
+				}
+				time.Sleep(5 * time.Second)
+			}
+			if !found {
+				e2e.Failf("Directory does not contain expected files: \n%s", result)
+			}
+
+			g.By("renaming subdir1 to subdir2")
+			subdir2 := filepath.Join(tempDir, "subdir2")
+			err = os.Rename(subdir1, subdir2)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			g.By("creating a file in the subdir2")
+			subdir2file1 := filepath.Join(subdir2, "file1")
+			_, err = os.Create(subdir2file1)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			found = false
+			for i := 0; i < 12; i++ {
+				g.By("Verifying that files are copied to the container")
+				result, _ = oc.Run("rsh").Args(podName, "ls", "/tmp/rsync/subdir2").Output()
+				if !strings.Contains(result, "file1") {
+					found = true
+					break
+				}
+				time.Sleep(5 * time.Second)
+			}
+			if !found {
+				e2e.Failf("Directory does not contain expected files: \n%s", result)
+			}
+
+			g.By("removing subdir2")
+			err = os.RemoveAll(subdir2)
+			o.Expect(err).NotTo(o.HaveOccurred())
+
+			found = false
+			for i := 0; i < 12; i++ {
+				g.By("Verifying that files are copied to the container")
+				result, _ = oc.Run("rsh").Args(podName, "ls", "/tmp/rsync").Output()
+				if !strings.Contains(result, "subdir2") {
+					found = true
+					break
+				}
+				time.Sleep(5 * time.Second)
+			}
+			if !found {
+				e2e.Failf("Directory does not contain expected files: \n%s", result)
+			}
+			failed = false
+		})
+	})
+	g.Describe("copy by strategy", func() {
 		testRsyncFn := func(strategy string) func() {
 			return func() {
 				g.By(fmt.Sprintf("Calling oc rsync %s %s:/tmp --strategy=%s", sourcePath1, podName, strategy))
@@ -219,5 +358,4 @@ var _ = g.Describe("[cli][Slow] can use rsync to upload files to pods", func() {
 			o.Expect(err).NotTo(o.HaveOccurred())
 		})
 	})
-
 })