Use ssh package to save backups

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

Author: JoshVanL

cmd/tarmak/cmd/cluster_snapshot_etcd_restore.go

@@ -15,7 +15,7 @@ var clusterSnapshotEtcdRestoreCmd = &cobra.Command{
 	Short: "restore etcd cluster with source snapshot",
 	PreRunE: func(cmd *cobra.Command, args []string) error {
 		if len(args) != 1 {
-			return fmt.Errorf("expecting single source path, got=%d", len(args))
+			return fmt.Errorf("expecting single target path, got=%d", len(args))
 		}
 
 		return nil

cmd/tarmak/cmd/cluster_snapshot_etcd_save.go

@@ -2,8 +2,6 @@
 package cmd
 
 import (
-	"fmt"
-
 	"github.com/spf13/cobra"
 
 	"github.com/jetstack/tarmak/pkg/tarmak"
@@ -13,14 +11,10 @@ import (
 var clusterSnapshotEtcdSaveCmd = &cobra.Command{
 	Use:   "save [target path prefix]",
 	Short: "save etcd snapshot to target path prefix, i.e 'backup-'",
-	PreRunE: func(cmd *cobra.Command, args []string) error {
-		if len(args) != 1 {
-			return fmt.Errorf("expecting single target path, got=%d", len(args))
-		}
-
-		return nil
-	},
 	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) == 0 {
+			args = []string{""}
+		}
 		t := tarmak.New(globalFlags)
 		s := etcd.New(t, args[0])
 		t.CancellationContext().WaitOrCancel(t.NewCmdSnapshot(cmd.Flags(), args, s).Save)

cmd/tarmak/cmd/cluster_ssh.go

@@ -3,6 +3,7 @@ package cmd
 
 import (
 	"errors"
+	"strings"
 
 	"github.com/spf13/cobra"
 
@@ -20,7 +21,7 @@ var clusterSshCmd = &cobra.Command{
 	},
 	Run: func(cmd *cobra.Command, args []string) {
 		t := tarmak.New(globalFlags)
-		t.Perform(t.SSHPassThrough(args[0], args[1:]))
+		t.Perform(t.SSHPassThrough(args[0], strings.Join(args[1:], " ")))
 	},
 }
 

pkg/tarmak/environment/bastion.go

@@ -61,7 +61,7 @@ func (e *Environment) VerifyBastionAvailable() error {
 	executeSSH := func() error {
 		retCode, err := ssh.Execute(
 			"bastion",
-			[]string{"/bin/true"},
+			"/bin/true",
 			nil, nil, stderrW,
 		)
 

pkg/tarmak/interfaces/interfaces.go

@@ -208,9 +208,9 @@ type Terraform interface {
 
 type SSH interface {
 	WriteConfig(Cluster) error
-	PassThrough(host string, additionalArguments []string) error
+	PassThrough(host string, additionalArguments string) error
 	Tunnel(destination, destinationPort, localPort string, daemonize bool) Tunnel
-	Execute(host string, cmd []string, stdin io.Reader, stdout, stderr io.Writer) (returnCode int, err error)
+	Execute(host string, cmd string, stdin io.Reader, stdout, stderr io.Writer) (returnCode int, err error)
 	Validate() error
 	Cleanup()
 }
@@ -296,4 +296,6 @@ type CancellationContext interface {
 type Snapshot interface {
 	Save() error
 	Restore() error
+	Log() *logrus.Entry
+	SSH() SSH
 }

pkg/tarmak/snapshot/consul/consul.go

@@ -2,7 +2,6 @@
 package consul
 
 import (
-	"bufio"
 	"fmt"
 	"io"
 	"os"
@@ -55,25 +54,26 @@ func (c *Consul) Save() error {
 
 	c.log.Infof("saving snapshot from instance %s", aliases[0])
 
-	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.GZipCCmd, hostPath), " ")...)
-
-	err = c.sshCmd(
-		aliases[0],
-		cmdArgs,
-		nil,
-		writer,
-	)
+	cmdArgs := fmt.Sprintf(`
+export CONSUL_HTTP_TOKEN=$(sudo cat /etc/consul/consul.json | jq -r '.acl_master_token');
+export DATACENTER=$(sudo cat /etc/consul/consul.json | jq -r '.datacenter');
+/usr/local/bin/consul snapshot save -datacenter $DATACENTER %s;
+/usr/local/bin/consul snapshot inspect %s`, hostPath, hostPath)
+
+	err = snapshot.SSHCmd(c, aliases[0], cmdArgs, nil, nil, nil)
+	if err != nil {
+		return err
+	}
+
+	reader, writer := io.Pipe()
+	err = snapshot.TarFromStream(func() error {
+		err := snapshot.SSHCmd(c, aliases[0], fmt.Sprintf(snapshot.GZipCCmd, hostPath),
+			nil, writer, nil)
+		writer.Close()
+		return err
+	}, reader, c.path)
 	if err != nil {
 		return err
 	}
@@ -107,11 +107,11 @@ func (c *Consul) Restore() error {
 		reader, writer := io.Pipe()
 		go snapshot.WriteTarToStream(c.path, writer, result, errLock)
 
-		err = c.sshCmd(
-			a,
-			cmdArgs,
+		err = snapshot.SSHCmd(c, a,
+			strings.Join(cmdArgs, " "),
 			reader,
 			os.Stdout,
+			nil,
 		)
 		if err != nil {
 			return err
@@ -123,20 +123,10 @@ func (c *Consul) Restore() error {
 	return nil
 }
 
-func (c *Consul) sshCmd(host string, args []string, stdin io.Reader, stdout io.Writer) error {
-	readerE, writerE := io.Pipe()
-	scannerE := bufio.NewScanner(readerE)
-
-	go func() {
-		for scannerE.Scan() {
-			c.log.WithField("std", "err").Warn(scannerE.Text())
-		}
-	}()
-
-	ret, err := c.ssh.ExecuteWithPipe(host, args[0], args[1:], stdin, stdout, writerE)
-	if ret != 0 {
-		return fmt.Errorf("command [%s] returned non-zero: %d", strings.Join(args, " "), ret)
-	}
+func (c *Consul) Log() *logrus.Entry {
+	return c.log
+}
 
-	return err
+func (c *Consul) SSH() interfaces.SSH {
+	return c.ssh
 }

pkg/tarmak/snapshot/etcd/etcd.go

@@ -2,7 +2,7 @@
 package etcd
 
 import (
-	"bufio"
+	//"bufio"
 	"fmt"
 	"io"
 	"strings"
@@ -20,7 +20,7 @@ import (
 var _ interfaces.Snapshot = &Etcd{}
 
 const (
-	etcdctlCmd = "/opt/bin/etcdctl snapshot %s %s > /dev/null;"
+	etcdctlCmd = `/opt/bin/etcdctl snapshot %s %s`
 )
 
 var (
@@ -30,13 +30,13 @@ var (
 		{"store": "overlay", "file": "overlay", "port": "2359"},
 	}
 
-	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",
-		"ETCDCTL_API=3",
-		"ETCDCTL_ENDPOINTS=https://127.0.0.1:{{port}}",
-	}
+	envCmd = `
+export ETCDCTL_CERT=/etc/etcd/ssl/etcd-{{file}}.pem;
+export ETCDCTL_KEY=/etc/etcd/ssl/etcd-{{file}}-key.pem;
+export ETCDCTL_CACERT=/etc/etcd/ssl/etcd-{{file}}-ca.pem;
+export ETCDCTL_API=3;
+export ETCDCTL_ENDPOINTS=https://127.0.0.1:{{port}}
+`
 )
 
 type Etcd struct {
@@ -75,23 +75,12 @@ func (e *Etcd) Save() error {
 	saveFunc := func(store map[string]string) {
 		defer wg.Done()
 
-		targetPath := fmt.Sprintf("%s%s.db", e.path, store["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(snapshot.GZipCCmd, hostPath), " ")...)
-
-		err = e.sshCmd(
-			aliases[0],
-			cmdArgs,
-			writer,
-		)
+
+		cmdArgs := fmt.Sprintf(`sudo /bin/bash -c "%s; %s"`, e.template(envCmd, store),
+			fmt.Sprintf(etcdctlCmd, "save", hostPath))
+		err = snapshot.SSHCmd(e, aliases[0], cmdArgs, nil, nil, nil)
 		if err != nil {
 
 			errLock.Lock()
@@ -101,6 +90,22 @@ func (e *Etcd) Save() error {
 			return
 		}
 
+		targetPath := fmt.Sprintf("%s%s.db", e.path, store["store"])
+		reader, writer := io.Pipe()
+		err = snapshot.TarFromStream(func() error {
+			err := snapshot.SSHCmd(e, aliases[0], fmt.Sprintf(snapshot.GZipCCmd, hostPath),
+				nil, writer, nil)
+			writer.Close()
+			return err
+		}, reader, targetPath)
+		if err != nil {
+			errLock.Lock()
+			result = multierror.Append(result, err)
+			errLock.Unlock()
+
+			return
+		}
+
 		e.log.Infof("etcd %s snapshot saved to %s", store["store"], targetPath)
 
 		select {
@@ -111,7 +116,6 @@ func (e *Etcd) Save() error {
 	}
 
 	wg.Add(len(stores))
-
 	for _, store := range stores {
 		go saveFunc(store)
 	}
@@ -137,31 +141,18 @@ func (e *Etcd) Restore() error {
 	return nil
 }
 
-func (e *Etcd) sshCmd(host string, args []string, stdout io.Writer) error {
-	readerE, writerE := io.Pipe()
-	scannerE := bufio.NewScanner(readerE)
-
-	go func() {
-		for scannerE.Scan() {
-			e.log.WithField("std", "err").Warn(scannerE.Text())
-		}
-	}()
-
-	args = append([]string{"sudo"}, args...)
-	ret, err := e.ssh.ExecuteWithPipe(host, args[0], args[1:], nil, stdout, writerE)
-	if ret != 0 {
-		return fmt.Errorf("command [%s] returned non-zero: %d", strings.Join(args, " "), ret)
+func (e *Etcd) template(args string, vars map[string]string) string {
+	for k, v := range vars {
+		args = strings.Replace(args, fmt.Sprintf("{{%s}}", k), v, -1)
 	}
 
-	return err
+	return args
 }
 
-func (e *Etcd) template(args []string, vars map[string]string) []string {
-	for i := range args {
-		for k, v := range vars {
-			args[i] = strings.Replace(args[i], fmt.Sprintf("{{%s}}", k), v, -1)
-		}
-	}
+func (e *Etcd) Log() *logrus.Entry {
+	return e.log
+}
 
-	return args
+func (e *Etcd) SSH() interfaces.SSH {
+	return e.ssh
 }