@@ -1178,7 +1178,7 @@ _oc_rsync()

     flags+=("--delete")

     flags+=("--quiet")

     flags+=("-q")

-    flags+=("--use-tar")

+    flags+=("--strategy=")

     must_have_one_flag=()

     must_have_one_noun=()

@@ -2986,7 +2986,7 @@ _openshift_cli_rsync()

     flags+=("--delete")

     flags+=("--quiet")

     flags+=("-q")

-    flags+=("--use-tar")

+    flags+=("--strategy=")

     must_have_one_flag=()

     must_have_one_noun=()

@@ -762,7 +762,7 @@ Start a shell session in a pod

 == oc rsync

-Copy local files to a pod

+Copy files between local filesystem and a pod

 ====

@@ -771,6 +771,9 @@ Copy local files to a pod

   # Synchronize a local directory with a pod directory

   $ openshift cli rsync ./local/dir/ POD:/remote/dir

+  

+  # Synchronize a pod directory with a local directory

+  $ openshift cli rsync POD:/remote/dir/ ./local/dir

 ----

 ====

@@ -14,6 +14,7 @@ import (

 	kubecmd "k8s.io/kubernetes/pkg/kubectl/cmd"

 	"github.com/openshift/origin/pkg/cmd/cli/cmd"

+	"github.com/openshift/origin/pkg/cmd/cli/cmd/rsync"

 	"github.com/openshift/origin/pkg/cmd/cli/policy"

 	"github.com/openshift/origin/pkg/cmd/cli/secrets"

 	"github.com/openshift/origin/pkg/cmd/flagtypes"

@@ -111,7 +112,7 @@ func NewCommandCLI(name, fullName string, in io.Reader, out, errout io.Writer) *

 				cmd.NewCmdExplain(fullName, f, out),

 				cmd.NewCmdLogs(fullName, f, out),

 				cmd.NewCmdRsh(cmd.RshRecommendedName, fullName, f, in, out, errout),

-				cmd.NewCmdRsync(cmd.RsyncRecommendedName, fullName, f, out, errout),

+				rsync.NewCmdRsync(rsync.RsyncRecommendedName, fullName, f, out, errout),

 				cmd.NewCmdExec(fullName, f, in, out, errout),

 				cmd.NewCmdPortForward(fullName, f),

 				cmd.NewCmdProxy(fullName, f, out),

deleted file mode 100644

@@ -1,413 +0,0 @@

-package cmd

-

-import (

-	"bytes"

-	"fmt"

-	"io"

-	"io/ioutil"

-	"os"

-	"os/exec"

-	"path/filepath"

-	"runtime"

-	"strings"

-

-	"github.com/golang/glog"

-	"github.com/openshift/source-to-image/pkg/tar"

-	"github.com/spf13/cobra"

-	kvalidation "k8s.io/kubernetes/pkg/api/validation"

-	kclient "k8s.io/kubernetes/pkg/client/unversioned"

-	kubecmd "k8s.io/kubernetes/pkg/kubectl/cmd"

-	kcmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"

-

-	"github.com/openshift/origin/pkg/cmd/util/clientcmd"

-)

-

-const (

-	// RsyncRecommendedName is the recommended name for the rsync command

-	RsyncRecommendedName = "rsync"

-

-	rsyncLong = `

-Copy local files to a container

-

-This command will attempt to copy local files to a remote container. It will default to the

-first container if none is specified, and will attempt to use 'rsync' if available locally and in the

-container. If 'rsync' is not present, it will attempt to use 'tar' to send files to the container.`

-

-	rsyncExample = `

-  # Synchronize a local directory with a pod directory

-  $ %[1]s ./local/dir/ POD:/remote/dir`

-

-	defaultRsyncExecutable = "rsync"

-	defaultTarExecutable   = "tar"

-)

-

-var (

-	testRsyncCommand = []string{"rsync", "--version"}

-	testTarCommand   = []string{"tar", "--version"}

-)

-

-// RsyncOptions holds the options to execute the sync command

-type RsyncOptions struct {

-	Namespace      string

-	PodName        string

-	ContainerName  string

-	Source         string

-	Destination    string

-	DestinationDir string

-	RshCommand     string

-	UseTar         bool

-	Quiet          bool

-	Delete         bool

-

-	Out    io.Writer

-	ErrOut io.Writer

-

-	LocalExecutor  executor

-	RemoteExecutor executor

-	Tar            tar.Tar

-	PodClient      kclient.PodInterface

-}

-

-// executor executes commands

-type executor interface {

-	Execute(command []string, in io.Reader, out, err io.Writer) error

-}

-

-// NewCmdRsync creates a new sync command

-func NewCmdRsync(name, parent string, f *clientcmd.Factory, out, errOut io.Writer) *cobra.Command {

-	tarHelper := tar.New()

-	tarHelper.SetExclusionPattern(nil)

-	o := RsyncOptions{

-		Out:           out,

-		ErrOut:        errOut,

-		LocalExecutor: &defaultLocalExecutor{},

-		Tar:           tarHelper,

-	}

-	cmd := &cobra.Command{

-		Use:     fmt.Sprintf("%s SOURCE_DIR POD:DESTINATION_DIR", name),

-		Short:   "Copy local files to a pod",

-		Long:    rsyncLong,

-		Example: fmt.Sprintf(rsyncExample, parent+" "+name),

-		Run: func(c *cobra.Command, args []string) {

-			kcmdutil.CheckErr(o.Complete(f, c, args))

-			kcmdutil.CheckErr(o.Validate())

-			kcmdutil.CheckErr(o.RunRsync())

-		},

-	}

-

-	cmd.Flags().StringVarP(&o.ContainerName, "container", "c", "", "Container within the pod")

-	cmd.Flags().BoolVarP(&o.Quiet, "quiet", "q", false, "Quiet copy")

-	cmd.Flags().BoolVar(&o.Delete, "delete", false, "Delete files not present in source")

-	cmd.Flags().BoolVar(&o.UseTar, "use-tar", false, "Use tar instead of rsync")

-	return cmd

-}

-

-func parseDestination(destination string) (string, string, error) {

-	parts := strings.SplitN(destination, ":", 2)

-	if len(parts) < 2 || len(parts[0]) == 0 || len(parts[1]) == 0 {

-		return "", "", fmt.Errorf("invalid destination %s: must be of the form PODNAME:DESTINATION_DIR", destination)

-	}

-	valid, msg := kvalidation.ValidatePodName(parts[0], false)

-	if !valid {

-		return "", "", fmt.Errorf("invalid pod name %s: %s", parts[0], msg)

-	}

-	return parts[0], parts[1], nil

-}

-

-// Complete verifies command line arguments and loads data from the command environment

-func (o *RsyncOptions) Complete(f *clientcmd.Factory, cmd *cobra.Command, args []string) error {

-	switch n := len(args); {

-	case n == 0:

-		cmd.Help()

-		fallthrough

-	case n < 2:

-		return kcmdutil.UsageError(cmd, "SOURCE_DIR and POD:DESTINATION_DIR are required arguments")

-	case n > 2:

-		return kcmdutil.UsageError(cmd, "only SOURCE_DIR and POD:DESTINATION_DIR should be specified as arguments")

-	}

-

-	// Set main command arguments

-	o.Source = args[0]

-	o.Destination = args[1]

-

-	// Determine pod name

-	var err error

-	o.PodName, o.DestinationDir, err = parseDestination(o.Destination)

-	if err != nil {

-		return kcmdutil.UsageError(cmd, err.Error())

-	}

-

-	// Use tar if running on windows

-	// TODO: Figure out how to use rsync in windows so that I/O can be

-	// redirected from the openshift native command to the cygwin rsync command

-	if runtime.GOOS == "windows" {

-		o.UseTar = true

-	}

-

-	namespace, _, err := f.DefaultNamespace()

-	if err != nil {

-		return err

-	}

-	o.Namespace = namespace

-

-	// Determine the Rsh command to use for rsync

-	if !o.UseTar {

-		rsh := siblingCommand(cmd, "rsh")

-		rshCmd := []string{rsh, "-n", o.Namespace}

-		if len(o.ContainerName) > 0 {

-			rshCmd = append(rshCmd, "-c", o.ContainerName)

-		}

-		o.RshCommand = strings.Join(rshCmd, " ")

-		glog.V(4).Infof("Rsh command: %s", o.RshCommand)

-	}

-

-	config, err := f.ClientConfig()

-	if err != nil {

-		return err

-	}

-

-	client, err := f.Client()

-	if err != nil {

-		return err

-	}

-

-	o.RemoteExecutor = &defaultRemoteExecutor{

-		Namespace:     o.Namespace,

-		PodName:       o.PodName,

-		ContainerName: o.ContainerName,

-		Config:        config,

-		Client:        client,

-	}

-

-	o.PodClient = client.Pods(namespace)

-

-	return nil

-}

-

-// sibling command returns a sibling command to the current command

-func siblingCommand(cmd *cobra.Command, name string) string {

-	c := cmd.Parent()

-	command := []string{}

-	for c != nil {

-		glog.V(5).Infof("Found parent command: %s", c.Name())

-		command = append([]string{c.Name()}, command...)

-		c = c.Parent()

-	}

-	// Replace the root command with what was actually used

-	// in the command line

-	glog.V(4).Infof("Setting root command to: %s", os.Args[0])

-	command[0] = os.Args[0]

-

-	// Append the sibling command

-	command = append(command, name)

-	glog.V(4).Infof("The sibling command is: %s", strings.Join(command, " "))

-

-	return strings.Join(command, " ")

-}

-

-// Validate checks that SyncOptions has all necessary fields

-func (o *RsyncOptions) Validate() error {

-	if len(o.PodName) == 0 {

-		return fmt.Errorf("pod name must be provided")

-	}

-	if len(o.Source) == 0 {

-		return fmt.Errorf("local source must be provided")

-	}

-	if len(o.Destination) == 0 {

-		return fmt.Errorf("remote destination must be provided")

-	}

-	if o.UseTar && len(o.DestinationDir) == 0 {

-		return fmt.Errorf("destination directory must be provided if using tar")

-	}

-	if !o.UseTar && len(o.RshCommand) == 0 {

-		return fmt.Errorf("rsh command must be provided when not using tar")

-	}

-	if o.Out == nil || o.ErrOut == nil {

-		return fmt.Errorf("output and error streams must be specified")

-	}

-	if o.LocalExecutor == nil || o.RemoteExecutor == nil {

-		return fmt.Errorf("local and remote executors must be provided")

-	}

-	if o.PodClient == nil {

-		return fmt.Errorf("pod client must be provided")

-	}

-	return nil

-}

-

-func (o *RsyncOptions) copyWithRsync(out, errOut io.Writer) error {

-	glog.V(3).Infof("Copying files with rsync")

-	flags := "-a"

-	if !o.Quiet {

-		flags += "v"

-	}

-	cmd := []string{defaultRsyncExecutable, flags}

-	if o.Delete {

-		cmd = append(cmd, "--delete")

-	}

-	cmd = append(cmd, "-e", o.RshCommand, o.Source, o.Destination)

-	glog.V(4).Infof("Local command: %s", strings.Join(cmd, " "))

-	return o.LocalExecutor.Execute(cmd, nil, out, errOut)

-}

-

-func (o *RsyncOptions) copyWithTar(out, errOut io.Writer) error {

-	glog.V(3).Infof("Copying files with tar")

-	if o.Delete {

-		// Implement the rsync --delete flag as a separate call to first delete directory contents

-		deleteCmd := []string{"sh", "-c", fmt.Sprintf("rm -rf %s", filepath.Join(o.DestinationDir, "*"))}

-		err := executeWithLogging(o.RemoteExecutor, deleteCmd)

-		if err != nil {

-			return fmt.Errorf("unable to delete files in destination: %v", err)

-		}

-	}

-	tmp, err := ioutil.TempFile("", "rsync")

-	if err != nil {

-		return fmt.Errorf("cannot create local temporary file for tar: %v", err)

-	}

-	defer os.Remove(tmp.Name())

-

-	err = tarLocal(o.Tar, o.Source, tmp)

-	if err != nil {

-		return fmt.Errorf("error creating tar of source directory: %v", err)

-	}

-	err = tmp.Close()

-	if err != nil {

-		return fmt.Errorf("error closing temporary tar file %s: %v", tmp.Name(), err)

-	}

-	tmp, err = os.Open(tmp.Name())

-	if err != nil {

-		return fmt.Errorf("cannot open temporary tar file %s: %v", tmp.Name(), err)

-	}

-	flags := "x"

-	if !o.Quiet {

-		flags += "v"

-	}

-	remoteCmd := []string{defaultTarExecutable, flags, "-C", o.DestinationDir}

-	errBuf := &bytes.Buffer{}

-	return o.RemoteExecutor.Execute(remoteCmd, tmp, out, errBuf)

-}

-

-func tarLocal(tar tar.Tar, sourceDir string, w io.Writer) error {

-	glog.V(4).Infof("Tarring %s locally", sourceDir)

-	// includeParent mimics rsync's behavior. When the source path ends in a path

-	// separator, then only the contents of the directory are copied. Otherwise,

-	// the directory itself is copied.

-	includeParent := true

-	if strings.HasSuffix(sourceDir, string(filepath.Separator)) {

-		includeParent = false

-		sourceDir = sourceDir[:len(sourceDir)-1]

-	}

-	return tar.CreateTarStream(sourceDir, includeParent, w)

-}

-

-// RunRsync copies files from source to destination

-func (o *RsyncOptions) RunRsync() error {

-	// If not going straight to tar and rsync exists locally, attempt to copy with rsync

-	if !o.UseTar && o.checkLocalRsync() {

-		errBuf := &bytes.Buffer{}

-		err := o.copyWithRsync(o.Out, errBuf)

-		// If no error occurred, we're done

-		if err == nil {

-			return nil

-		}

-		// If an error occurred, check whether rsync exists on the container.

-		// If it doesn't, fallback to tar

-		if o.checkRemoteRsync() {

-			// If remote rsync does exist, simply report the error

-			io.Copy(o.ErrOut, errBuf)

-			return err

-		}

-	}

-	return o.copyWithTar(o.Out, o.ErrOut)

-}

-

-func executeWithLogging(e executor, cmd []string) error {

-	w := &bytes.Buffer{}

-	err := e.Execute(cmd, nil, w, w)

-	glog.V(4).Infof("%s", w.String())

-	glog.V(4).Infof("error: %v", err)

-	return err

-}

-

-func (o *RsyncOptions) checkLocalRsync() bool {

-	_, err := exec.LookPath("rsync")

-	if err != nil {

-		glog.Warningf("rsync not found in local computer")

-		return false

-	}

-	return true

-}

-

-func (o *RsyncOptions) checkRemoteRsync() bool {

-	err := executeWithLogging(o.RemoteExecutor, testRsyncCommand)

-	if err != nil {

-		glog.Warningf("rsync not found in container %s of pod %s", o.containerName(), o.PodName)

-		return false

-	}

-	return true

-}

-

-func (o *RsyncOptions) containerName() string {

-	if len(o.ContainerName) > 0 {

-		return o.ContainerName

-	}

-	pod, err := o.PodClient.Get(o.PodName)

-	if err != nil {

-		glog.V(1).Infof("Error getting pod %s: %v", o.PodName, err)

-		return "[unknown]"

-	}

-	return pod.Spec.Containers[0].Name

-}

-

-// defaultLocalExecutor will execute commands on the local machine

-type defaultLocalExecutor struct{}

-

-// Execute will run a command locally

-func (*defaultLocalExecutor) Execute(command []string, in io.Reader, out, errOut io.Writer) error {

-	glog.V(3).Infof("Local executor running command: %s", strings.Join(command, " "))

-	cmd := exec.Command(command[0], command[1:]...)

-	cmd.Stdout = out

-	cmd.Stderr = errOut

-	cmd.Stdin = in

-	err := cmd.Run()

-	if err != nil {

-		glog.V(4).Infof("Error from local command execution: %v", err)

-	}

-	return err

-}

-

-// defaultRemoteExecutor will execute commands on a given pod/container by using the kube Exec command

-type defaultRemoteExecutor struct {

-	Namespace     string

-	PodName       string

-	ContainerName string

-	Client        *kclient.Client

-	Config        *kclient.Config

-}

-

-// Execute will run a command in a pod

-func (e *defaultRemoteExecutor) Execute(command []string, in io.Reader, out, errOut io.Writer) error {

-	glog.V(3).Infof("Remote executor running command: %s", strings.Join(command, " "))

-	execOptions := &kubecmd.ExecOptions{

-		In:            in,

-		Out:           out,

-		Err:           errOut,

-		Stdin:         in != nil,

-		Executor:      &kubecmd.DefaultRemoteExecutor{},

-		Client:        e.Client,

-		Config:        e.Config,

-		PodName:       e.PodName,

-		ContainerName: e.ContainerName,

-		Namespace:     e.Namespace,

-		Command:       command,

-	}

-	err := execOptions.Validate()

-	if err != nil {

-		glog.V(4).Infof("Error from remote command validation: %v", err)

-		return err

-	}

-	err = execOptions.Run()

-	if err != nil {

-		glog.V(4).Infof("Error from remote execution: %v", err)

-	}

-	return err

-}

new file mode 100644

@@ -0,0 +1,63 @@

+package rsync

+

+import (

+	"bytes"

+	"fmt"

+	"io"

+	"strings"

+

+	"github.com/golang/glog"

+	"k8s.io/kubernetes/pkg/util/errors"

+)

+

+// copyStrategies is an ordered list of copyStrategy objects that behaves as a single

+// strategy.

+type copyStrategies []copyStrategy

+

+// ensure copyStrategies implements the copyStrategy interface

+var _ copyStrategy = copyStrategies{}

+

+// Copy will call copy for strategies in list order. If a strategySetupError results from a copy,

+// the next strategy will be attempted. Otherwise the error is returned.

+func (ss copyStrategies) Copy(source, destination *pathSpec, out, errOut io.Writer) error {

+	var err error

+	for _, s := range ss {

+		errBuf := &bytes.Buffer{}

+		err = s.Copy(source, destination, out, errBuf)

+		if _, isSetupError := err.(strategySetupError); isSetupError {

+			glog.V(4).Infof("Error output:\n%s", errBuf.String())

+			glog.Warningf("Cannot use %s: %v", s.String(), err.Error())

+			continue

+		}

+		io.Copy(errOut, errBuf)

+		break

+	}

+	return err

+}

+

+// Validate will call Validate on all strategies and return an aggregate of their errors

+func (ss copyStrategies) Validate() error {

+	var errs []error

+	for _, s := range ss {

+		err := s.Validate()

+		if err != nil {

+			errs = append(errs, fmt.Errorf("invalid %v strategy: %v", s, err))

+		}

+	}

+	return errors.NewAggregate(errs)

+}

+

+// String will return a comma-separated list of strategy names

+func (ss copyStrategies) String() string {

+	names := []string{}

+	for _, s := range ss {

+		names = append(names, s.String())

+	}

+	return strings.Join(names, ",")

+}

+

+// strategySetupError is an error that occurred while setting up a strategy

+// (as opposed to actually executing a copy and getting an error from normal copy execution)

+type strategySetupError string

+

+func (e strategySetupError) Error() string { return string(e) }

new file mode 100644

@@ -0,0 +1,98 @@

+package rsync

+

+import (

+	"errors"

+	"io"

+	"strings"

+

+	"github.com/golang/glog"

+	"github.com/spf13/cobra"

+	kerrors "k8s.io/kubernetes/pkg/util/errors"

+

+	"github.com/openshift/origin/pkg/cmd/util/clientcmd"

+)

+

+var (

+	testRsyncCommand = []string{"rsync", "--version"}

+)

+

+type rsyncStrategy struct {

+	Flags          []string

+	RshCommand     string

+	LocalExecutor  executor

+	RemoteExecutor executor

+}

+

+func newRsyncStrategy(f *clientcmd.Factory, c *cobra.Command, o *RsyncOptions) (copyStrategy, error) {

+	// Determine the rsh command to pass to the local rsync command

+	rsh := siblingCommand(c, "rsh")

+	rshCmd := []string{rsh, "-n", o.Namespace}

+	if len(o.ContainerName) > 0 {

+		rshCmd = append(rshCmd, "-c", o.ContainerName)

+	}

+	rshCmdStr := strings.Join(rshCmd, " ")

+	glog.V(4).Infof("Rsh command: %s", rshCmdStr)

+

+	remoteExec, err := newRemoteExecutor(f, o)

+	if err != nil {

+		return nil, err

+	}

+

+	// TODO: Expose more flags to send to the rsync command

+	// either as a special argument or any unrecognized arguments.

+	// The blocking-io flag is used to resolve a sync issue when

+	// copying from the pod to the local machine

+	flags := []string{"-a", "--blocking-io", "--omit-dir-times", "--numeric-ids"}

+	if o.Quiet {

+		flags = append(flags, "-q")

+	} else {

+		flags = append(flags, "-v")

+	}

+	if o.Delete {

+		flags = append(flags, "--delete")

+	}

+

+	return &rsyncStrategy{

+		Flags:          flags,

+		RshCommand:     rshCmdStr,

+		RemoteExecutor: remoteExec,

+		LocalExecutor:  newLocalExecutor(),

+	}, nil

+}

+

+func (r *rsyncStrategy) Copy(source, destination *pathSpec, out, errOut io.Writer) error {

+	glog.V(3).Infof("Copying files with rsync")

+	cmd := append([]string{"rsync"}, r.Flags...)

+	cmd = append(cmd, "-e", r.RshCommand, source.RsyncPath(), destination.RsyncPath())

+	err := r.LocalExecutor.Execute(cmd, nil, out, errOut)

+	if isExitError(err) {

+		// Determine whether rsync is present in the pod container

+		testRsyncErr := executeWithLogging(r.RemoteExecutor, testRsyncCommand)

+		if testRsyncErr != nil {

+			glog.V(4).Infof("error testing whether rsync is available: %v", testRsyncErr)

+			return strategySetupError("rsync not available in container")

+		}

+	}

+	return err

+}

+

+func (r *rsyncStrategy) Validate() error {

+	errs := []error{}

+	if len(r.RshCommand) == 0 {

+		errs = append(errs, errors.New("rsh command must be provided"))

+	}

+	if r.LocalExecutor == nil {

+		errs = append(errs, errors.New("local executor must not be nil"))

+	}

+	if r.RemoteExecutor == nil {

+		errs = append(errs, errors.New("remote executor must not be nil"))

+	}

+	if len(errs) > 0 {

+		return kerrors.NewAggregate(errs)

+	}

+	return nil

+}

+

+func (r *rsyncStrategy) String() string {

+	return "rsync"

+}

new file mode 100644

@@ -0,0 +1,282 @@

+package rsync

+

+import (

+	"bytes"

+	"errors"

+	"fmt"

+	"io"

+	"math/rand"

+	"net"

+	"strconv"

+	"strings"

+	"time"

+

+	"github.com/golang/glog"

+	"github.com/spf13/cobra"

+	kerrors "k8s.io/kubernetes/pkg/util/errors"

+	krand "k8s.io/kubernetes/pkg/util/rand"

+

+	"github.com/openshift/origin/pkg/cmd/util/clientcmd"

+)

+

+const (

+	// startDaemonScript is the script that will be run on the container to start the

+	// rsync daemon. It takes 3 format parameters:

+	// 1 - alternate random name for config file

+	// 2 - alternate random name for pid file

+	// 3 - port number to listen on

+	// The output of the script is the name of a file containing the PID for the started daemon

+	startDaemonScript = `set -e

+TMPDIR=${TMPDIR:-/tmp}

+CONFIGFILE=$(mktemp 2> /dev/null || (echo -n "" > ${TMPDIR}/%[1]s && echo ${TMPDIR}/%[1]s))

+PIDFILE=$(mktemp 2> /dev/null || (echo -n "" > ${TMPDIR}/%[2]s && echo ${TMPDIR}/%[2]s))

+rm $PIDFILE

+printf "pid file = ${PIDFILE}\n[root]\n  path = /\n  use chroot = no\n  read only = no" > $CONFIGFILE

+rsync --daemon --config=${CONFIGFILE} --port=%[3]d

+echo ${PIDFILE}

+`

+	portRangeFrom = 30000

+	portRangeTo   = 60000

+	remoteLabel   = "root"

+)

+

+var (

+	random = rand.New(rand.NewSource(time.Now().UTC().UnixNano()))

+)

+

+type rsyncDaemonStrategy struct {

+	Flags          []string

+	RemoteExecutor executor

+	PortForwarder  forwarder

+	LocalExecutor  executor

+

+	daemonPIDFile   string

+	daemonPort      int

+	localPort       int

+	portForwardChan chan struct{}

+}

+

+func localRsyncURL(port int, label string, path string) string {

+	return fmt.Sprintf("rsync://127.0.0.1:%d/%s/%s", port, label, strings.TrimPrefix(path, "/"))

+}

+

+func getUsedPorts(data string) map[int]struct{} {

+	ports := map[int]struct{}{}

+	lines := strings.Split(data, "\n")

+	for _, line := range lines {

+		parts := strings.Fields(line)

+		if len(parts) < 2 {

+			continue

+		}

+		// discard lines that don't contain connection data

+		if !strings.Contains(parts[0], ":") {

+			continue

+		}

+		glog.V(5).Infof("Determining port in use from: %s", line)

+		localAddress := strings.Split(parts[1], ":")

+		if len(localAddress) < 2 {

+			continue

+		}

+		port, err := strconv.ParseInt(localAddress[1], 16, 0)

+		if err == nil {

+			ports[int(port)] = struct{}{}

+		}

+	}

+	glog.V(2).Infof("Used ports in container: %#v", ports)

+	return ports

+}

+

+func randomPort() int {

+	return portRangeFrom + random.Intn(portRangeTo-portRangeFrom)

+}

+

+func localPort() (int, error) {

+	l, err := net.Listen("tcp", ":0")

+	if err != nil {

+		glog.V(1).Infof("Could not determine local free port: %v", err)

+		return 0, err

+	}

+	defer l.Close()

+	glog.V(1).Infof("Found listener port at: %s", l.Addr().String())

+	addr := strings.Split(l.Addr().String(), ":")

+	port, err := strconv.Atoi(addr[len(addr)-1])

+	if err != nil {

+		glog.V(1).Infof("Could not parse listener address %#v: %v", addr, err)

+		return 0, err

+	}

+	return port, nil

+}

+

+func (s *rsyncDaemonStrategy) getFreePort() (int, error) {

+	cmd := []string{"cat", "/proc/net/tcp", "/proc/net/tcp6"}

+	tcpData := &bytes.Buffer{}

+	cmdErr := &bytes.Buffer{}

+	usedPorts := map[int]struct{}{}

+	err := s.RemoteExecutor.Execute(cmd, nil, tcpData, cmdErr)

+	if err == nil {

+		usedPorts = getUsedPorts(tcpData.String())

+	} else {

+		glog.V(4).Infof("Error getting free port data: %v, Err: %s", err, cmdErr.String())

+	}

+	tries := 0

+	for {

+		tries++

+		if tries > 20 {

+			glog.V(4).Infof("Too many attempts trying to find free port")

+			break

+		}

+		port := randomPort()

+		if _, used := usedPorts[port]; !used {

+			glog.V(4).Infof("Found free container port: %d", port)

+			return port, nil

+		}

+	}

+	return 0, fmt.Errorf("could not find a free port")

+

+}

+

+func (s *rsyncDaemonStrategy) startRemoteDaemon() error {

+	port, err := s.getFreePort()

+	if err != nil {

+		return err

+	}

+	cmdOut := &bytes.Buffer{}

+	cmdErr := &bytes.Buffer{}

+	cmdIn := bytes.NewBufferString(fmt.Sprintf(startDaemonScript, krand.String(32), krand.String(32), port))

+	err = s.RemoteExecutor.Execute([]string{"sh"}, cmdIn, cmdOut, cmdErr)

+	if err != nil {

+		glog.V(1).Infof("Error starting rsync daemon: %v. Out: %s, Err: %s\n", err, cmdOut.String(), cmdErr.String())

+		return err

+	}

+	s.daemonPort = port

+	s.daemonPIDFile = strings.TrimSpace(cmdOut.String())

+	return nil

+}

+

+func (s *rsyncDaemonStrategy) killRemoteDaemon() error {

+	cmd := []string{"sh", "-c", fmt.Sprintf("kill $(cat %s)", s.daemonPIDFile)}

+	cmdOut := &bytes.Buffer{}

+	cmdErr := &bytes.Buffer{}

+	err := s.RemoteExecutor.Execute(cmd, nil, cmdOut, cmdErr)

+	if err != nil {

+		glog.V(1).Infof("Error killing rsync daemon: %v. Out: %s, Err: %s\n", err, cmdOut.String(), cmdErr.String())

+	}

+	return err

+}

+

+func (s *rsyncDaemonStrategy) startPortForward() error {

+	var err error

+	s.localPort, err = localPort()

+	if err != nil {

+		// Attempt with a random port if other methods fail

+		s.localPort = randomPort()

+	}

+	s.portForwardChan = make(chan struct{})

+	return s.PortForwarder.ForwardPorts([]string{fmt.Sprintf("%d:%d", s.localPort, s.daemonPort)}, s.portForwardChan)

+}

+

+func (s *rsyncDaemonStrategy) stopPortForward() {

+	close(s.portForwardChan)

+}

+

+func (s *rsyncDaemonStrategy) copyUsingDaemon(source, destination *pathSpec, out, errOut io.Writer) error {

+	glog.V(3).Infof("Copying files with rsync daemon")

+	cmd := append([]string{"rsync"}, s.Flags...)

+	var sourceArg, destinationArg string

+	if source.Local() {

+		sourceArg = source.RsyncPath()

+	} else {

+		sourceArg = localRsyncURL(s.localPort, remoteLabel, source.Path)

+	}

+	if destination.Local() {

+		destinationArg = destination.RsyncPath()

+	} else {

+		destinationArg = localRsyncURL(s.localPort, remoteLabel, destination.Path)

+	}

+	cmd = append(cmd, sourceArg, destinationArg)

+	err := s.LocalExecutor.Execute(cmd, nil, out, errOut)

+	if err != nil {

+		// Determine whether rsync is present in the pod container

+		testRsyncErr := executeWithLogging(s.RemoteExecutor, testRsyncCommand)

+		if testRsyncErr != nil {

+			return strategySetupError("rsync not available in container")

+		}

+	}

+	return err

+}

+

+func (s *rsyncDaemonStrategy) Copy(source, destination *pathSpec, out, errOut io.Writer) error {

+	err := s.startRemoteDaemon()

+	if err != nil {

+		if isExitError(err) {

+			return strategySetupError(fmt.Sprintf("cannot start remote rsync daemon: %v", err))

+		} else {

+			return err

+		}

+	}

+	defer s.killRemoteDaemon()

+	err = s.startPortForward()

+	if err != nil {

+		if isExitError(err) {

+			return strategySetupError(fmt.Sprintf("cannot start port-forward: %v", err))

+		} else {

+			return err

+		}

+	}

+	defer s.stopPortForward()

+

+	err = s.copyUsingDaemon(source, destination, out, errOut)

+	return err

+}

+

+func (s *rsyncDaemonStrategy) Validate() error {

+	errs := []error{}

+	if s.PortForwarder == nil {