Attempt to tar both received and sent snapshots

Signed-off-by: JoshVanL <[email protected]>

Author: JoshVanL

pkg/tarmak/interfaces/interfaces.go

@@ -210,12 +210,10 @@ type SSH interface {
 	WriteConfig(Cluster) error
 	PassThrough([]string)
 	Tunnel(hostname string, destination string, destinationPort int) Tunnel
-	Execute(host string, cmd string, args []string) (returnCode int, err error)
 	Validate() error
 	Cleanup() error
-	ExecuteWithWriter(host string, cmd string, args []string, stdout, stderr io.Writer) (returnCode int, err error)
-	ScpToLocal(path, targetPath, localPath string) (retCode int, err error)
-	ScpToHost(path, localPath, targetPath string) (retCode int, err error)
+	Execute(host, cmd string, args []string) (returnCode int, err error)
+	ExecuteWithPipe(host, cmd string, args []string, stdin io.Reader, stdout, stderr io.Writer) (returnCode int, err error)
 }
 
 type Tunnel interface {

pkg/tarmak/snapshot/consul/consul.go

@@ -5,9 +5,12 @@ import (
 	"bufio"
 	"fmt"
 	"io"
+	"os"
 	"strings"
+	"sync"
 	"time"
 
+	"github.com/hashicorp/go-multierror"
 	"github.com/sirupsen/logrus"
 
 	clusterv1alpha1 "github.com/jetstack/tarmak/pkg/apis/cluster/v1alpha1"
@@ -18,14 +21,11 @@ import (
 var _ interfaces.Snapshot = &Consul{}
 
 const (
-	snapshotTimeLayout = "2006-01-02_15-04-05"
+	consulCmd = "consul snapshot %s %s > /dev/null;"
 )
 
 var (
-	exportCmd = []string{
-		"export",
-		"CONSUL_HTTP_TOKEN=$(sudo cat /etc/consul/consul.json | jq -r '.acl_master_token');",
-	}
+	envCmd = []string{"CONSUL_HTTP_TOKEN=$(sudo cat /etc/consul/consul.json | jq -r '.acl_master_token')"}
 )
 
 type Consul struct {
@@ -54,27 +54,33 @@ func (c *Consul) Save() error {
 	c.aliases = aliases
 
 	c.log.Infof("saving snapshot from instance %s", aliases[0])
-	targetPath := fmt.Sprintf("/tmp/consul-snapshot-%s.snap", time.Now().Format(snapshotTimeLayout))
 
-	cmdArgs := append(exportCmd, "consul", "snapshot", "save", targetPath)
+	var result *multierror.Error
+	var errLock sync.Mutex
+
+	reader, writer := io.Pipe()
+	go snapshot.ReadTarFromStream(c.path, reader, result, errLock)
+
+	hostPath := fmt.Sprintf("/tmp/consul-snapshot-%s.snap",
+		time.Now().Format(snapshot.TimeLayout))
+	cmdArgs := append(envCmd,
+		strings.Split(fmt.Sprintf(consulCmd, "save", hostPath), " ")...)
+	cmdArgs = append(cmdArgs,
+		strings.Split(fmt.Sprintf(snapshot.TarCCmd, hostPath), " ")...)
+
 	err = c.sshCmd(
 		aliases[0],
-		cmdArgs[0],
-		cmdArgs[1:],
+		cmdArgs,
+		nil,
+		writer,
 	)
 	if err != nil {
 		return err
 	}
 
-	ret, err := c.ssh.ScpToLocal(aliases[0], targetPath, c.path)
-	if ret != 0 {
-		cmdStr := fmt.Sprintf("%s", strings.Join(cmdArgs, " "))
-		return fmt.Errorf("command [%s] returned non-zero: %d, %s", cmdStr, ret, err)
-	}
-
 	c.log.Infof("consul snapshot saved to %s", c.path)
 
-	return err
+	return nil
 }
 
 func (c *Consul) Restore() error {
@@ -86,18 +92,27 @@ func (c *Consul) Restore() error {
 
 	for _, a := range aliases {
 		c.log.Infof("restoring snapshot to instance %s", a)
-		targetPath := fmt.Sprintf("/tmp/consul-snapshot-%s.snap", time.Now().Format(snapshotTimeLayout))
 
-		ret, err := c.ssh.ScpToHost(a, c.path, targetPath)
-		if ret != 0 {
-			return fmt.Errorf("command scp returned non-zero: %d, %s", ret, err)
-		}
+		hostPath := fmt.Sprintf("/tmp/consul-snapshot-%s.snap",
+			time.Now().Format(snapshot.TimeLayout))
+
+		cmdArgs := strings.Split(fmt.Sprintf(snapshot.TarXCmd, hostPath), " ")
+		//cmdArgs := strings.Split(snapshot.TarXCmd, " ")
+		//cmdArgs = append(cmdArgs,
+		//	append(envCmd,
+		//		strings.Split(fmt.Sprintf(consulCmd, "restore", hostPath), " ")...)...)
+
+		var result *multierror.Error
+		var errLock sync.Mutex
+
+		reader, writer := io.Pipe()
+		go snapshot.WriteTarToStream(c.path, writer, result, errLock)
 
-		cmdArgs := append(exportCmd, "consul", "snapshot", "restore", targetPath)
 		err = c.sshCmd(
 			a,
-			cmdArgs[0],
-			cmdArgs[1:],
+			cmdArgs,
+			reader,
+			os.Stdout,
 		)
 		if err != nil {
 			return err
@@ -106,31 +121,22 @@ func (c *Consul) Restore() error {
 
 	c.log.Infof("consul snapshot restored from %s", c.path)
 
-	return err
+	return nil
 }
 
-func (c *Consul) sshCmd(host, command string, args []string) error {
-	readerO, writerO := io.Pipe()
+func (c *Consul) sshCmd(host string, args []string, stdin io.Reader, stdout io.Writer) error {
 	readerE, writerE := io.Pipe()
-	scannerO := bufio.NewScanner(readerO)
 	scannerE := bufio.NewScanner(readerE)
 
-	go func() {
-		for scannerO.Scan() {
-			c.log.WithField("std", "out").Debug(scannerO.Text())
-		}
-	}()
-
 	go func() {
 		for scannerE.Scan() {
 			c.log.WithField("std", "err").Warn(scannerE.Text())
 		}
 	}()
 
-	ret, err := c.ssh.ExecuteWithWriter(host, command, args, writerO, writerE)
+	ret, err := c.ssh.ExecuteWithPipe(host, args[0], args[1:], stdin, stdout, writerE)
 	if ret != 0 {
-		cmdStr := fmt.Sprintf("%s %s", command, strings.Join(args, " "))
-		return fmt.Errorf("command [%s] returned non-zero: %d", cmdStr, ret)
+		return fmt.Errorf("command [%s] returned non-zero: %d", strings.Join(args, " "), ret)
 	}
 
 	return err

pkg/tarmak/snapshot/etcd/etcd.go

@@ -3,15 +3,12 @@ package etcd
 
 import (
 	"bufio"
-	"compress/gzip"
 	"fmt"
 	"io"
-	"os"
 	"strings"
 	"sync"
 	"time"
 
-	//"github.com/docker/docker/pkg/archive"
 	"github.com/hashicorp/go-multierror"
 	"github.com/sirupsen/logrus"
 
@@ -23,10 +20,7 @@ import (
 var _ interfaces.Snapshot = &Etcd{}
 
 const (
-	snapshotTimeLayout = "2006-01-02_15-04-05"
-
 	etcdctlCmd = "/opt/bin/etcdctl snapshot %s %s > /dev/null;"
-	tarCmd     = "tar -czPf - %s"
 )
 
 var (
@@ -36,7 +30,7 @@ var (
 		{"store": "overlay", "file": "overlay", "port": "2359"},
 	}
 
-	exportCmd = []string{
+	envCmd = []string{
 		"ETCDCTL_CERT=/etc/etcd/ssl/etcd-{{file}}.pem",
 		"ETCDCTL_KEY=/etc/etcd/ssl/etcd-{{file}}-key.pem",
 		"ETCDCTL_CACERT=/etc/etcd/ssl/etcd-{{file}}-ca.pem",
@@ -53,7 +47,6 @@ type Etcd struct {
 
 	path    string
 	aliases []string
-	errLock sync.Mutex // prevent multiple writes to result error
 }
 
 func New(tarmak interfaces.Tarmak, path string) *Etcd {
@@ -77,21 +70,22 @@ func (e *Etcd) Save() error {
 
 	var wg sync.WaitGroup
 	var result *multierror.Error
+	var errLock sync.Mutex
 
 	saveFunc := func(store map[string]string) {
 		defer wg.Done()
 
-		hostPath := fmt.Sprintf("/tmp/etcd-snapshot-%s-%s.db",
-			store["store"], time.Now().Format(snapshotTimeLayout))
 		targetPath := fmt.Sprintf("%s%s.db", e.path, store["store"])
 
-		cmdArgs := append(e.template(exportCmd, store),
+		reader, writer := io.Pipe()
+		go snapshot.ReadTarFromStream(targetPath, reader, result, errLock)
+
+		hostPath := fmt.Sprintf("/tmp/etcd-snapshot-%s-%s.db",
+			store["store"], time.Now().Format(snapshot.TimeLayout))
+		cmdArgs := append(e.template(envCmd, store),
 			strings.Split(fmt.Sprintf(etcdctlCmd, "save", hostPath), " ")...)
 		cmdArgs = append(cmdArgs,
-			strings.Split(fmt.Sprintf(tarCmd, hostPath), " ")...)
-
-		reader, writer := io.Pipe()
-		go e.readTarFromStream(targetPath, reader, result)
+			strings.Split(fmt.Sprintf(snapshot.TarCCmd, hostPath), " ")...)
 
 		err = e.sshCmd(
 			aliases[0],
@@ -100,9 +94,9 @@ func (e *Etcd) Save() error {
 		)
 		if err != nil {
 
-			e.errLock.Lock()
+			errLock.Lock()
 			result = multierror.Append(result, err)
-			e.errLock.Unlock()
+			errLock.Unlock()
 
 			return
 		}
@@ -154,46 +148,14 @@ func (e *Etcd) sshCmd(host string, args []string, stdout io.Writer) error {
 	}()
 
 	args = append([]string{"sudo"}, args...)
-	ret, err := e.ssh.ExecuteWithWriter(host, args[0], args[1:], stdout, writerE)
+	ret, err := e.ssh.ExecuteWithPipe(host, args[0], args[1:], nil, stdout, writerE)
 	if ret != 0 {
-		cmdStr := fmt.Sprintf("%s", strings.Join(args, " "))
-		return fmt.Errorf("command [%s] returned non-zero: %d", cmdStr, ret)
+		return fmt.Errorf("command [%s] returned non-zero: %d", strings.Join(args, " "), ret)
 	}
 
 	return err
 }
 
-func (e *Etcd) readTarFromStream(dest string, stream io.Reader, result *multierror.Error) {
-	gzr, err := gzip.NewReader(stream)
-	if err != nil {
-
-		e.errLock.Lock()
-		result = multierror.Append(result, err)
-		e.errLock.Unlock()
-
-		return
-	}
-
-	f, err := os.Create(dest)
-	if err != nil {
-
-		e.errLock.Lock()
-		result = multierror.Append(result, err)
-		e.errLock.Unlock()
-
-		return
-	}
-
-	if _, err := io.Copy(f, gzr); err != nil {
-
-		e.errLock.Lock()
-		result = multierror.Append(result, err)
-		e.errLock.Unlock()
-
-		return
-	}
-}
-
 func (e *Etcd) template(args []string, vars map[string]string) []string {
 	for i := range args {
 		for k, v := range vars {

pkg/tarmak/snapshot/snapshot.go

@@ -2,14 +2,25 @@
 package snapshot
 
 import (
+	"compress/gzip"
 	"fmt"
+	"io"
+	"os"
+	"sync"
 
 	"github.com/hashicorp/go-multierror"
 
 	"github.com/jetstack/tarmak/pkg/tarmak/interfaces"
 	"github.com/jetstack/tarmak/pkg/tarmak/utils"
 )
 
+const (
+	TimeLayout = "2006-01-02_15-04-05"
+	TarCCmd    = "tar -czPf - %s"
+	//TarXCmd    = "cat | tar -xz | cat > /tmp/foo" //| cat > %s"
+	TarXCmd = "cat > %s"
+)
+
 func Prepare(tarmak interfaces.Tarmak, role string) (aliases []string, err error) {
 	if err := tarmak.SSH().WriteConfig(tarmak.Cluster()); err != nil {
 		return nil, err
@@ -47,3 +58,58 @@ func Prepare(tarmak interfaces.Tarmak, role string) (aliases []string, err error
 
 	return aliases, result.ErrorOrNil()
 }
+
+func ReadTarFromStream(dest string, stream io.Reader, result *multierror.Error, errLock sync.Mutex) {
+	gzr, err := gzip.NewReader(stream)
+	if err != nil {
+
+		errLock.Lock()
+		result = multierror.Append(result, err)
+		errLock.Unlock()
+
+		return
+	}
+
+	f, err := os.Create(dest)
+	if err != nil {
+
+		errLock.Lock()
+		result = multierror.Append(result, err)
+		errLock.Unlock()
+
+		return
+	}
+
+	if _, err := io.Copy(f, gzr); err != nil {
+
+		errLock.Lock()
+		result = multierror.Append(result, err)
+		errLock.Unlock()
+
+		return
+	}
+}
+
+func WriteTarToStream(src string, stream io.WriteCloser, result *multierror.Error, errLock sync.Mutex) {
+	//defer stream.Close()
+
+	f, err := os.Open(src)
+	if err != nil {
+
+		errLock.Lock()
+		result = multierror.Append(result, err)
+		errLock.Unlock()
+
+		return
+	}
+
+	gzw := gzip.NewWriter(stream)
+	if _, err := io.Copy(gzw, f); err != nil {
+
+		errLock.Lock()
+		result = multierror.Append(result, err)
+		errLock.Unlock()
+
+		return
+	}
+}