Merge pull request #225 from jaffee/223-copy-pilosactl

223 copy pilosactl

cmd/pilosactl/main.go

@@ -14,6 +14,7 @@ import (
 	"math/rand"
 	"os"
 	"os/signal"
+	"path"
 	"path/filepath"
 	"sort"
 	"strconv"
@@ -1328,6 +1329,8 @@ type BspawnCommand struct {
 	// agents specified here are used.
 	AgentHosts []string
 
+	CopyBinary bool
+
 	// Benchmarks is a slice of Spawns which specifies all of the bagent
 	// commands to run. These will all be run in parallel, started on each
 	// of the agents in a round robin fashion.
@@ -1370,10 +1373,6 @@ func (cmd *BspawnCommand) ParseFlags(args []string) error {
 	if err != nil {
 		return err
 	}
-
-	// handle pilosa creation
-	// handle agent creation
-
 	return nil
 }
 
@@ -1428,12 +1427,38 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error {
 	}
 }
 
-func (cmd *BspawnCommand) spawnRemote(ctx context.Context, runUUID uuid.UUID) error {
-	agentIndex := 0
-	agentConnections, err := pilosactl.SSHClients(cmd.AgentHosts, cmd.SSHUser, "")
+func copyBinary(fleet pilosactl.SSHFleet, pkg, goos, goarch string) error {
+	bin, err := pilosactl.BuildBinary(pkg, goos, goarch)
 	if err != nil {
 		return err
 	}
+
+	wc, err := fleet.OpenFile(path.Base(pkg), "+x")
+	if err != nil {
+		return err
+	}
+
+	_, err = io.Copy(wc, bin)
+	if err != nil {
+		return err
+	}
+	return wc.Close()
+}
+
+func (cmd *BspawnCommand) spawnRemote(ctx context.Context, runUUID uuid.UUID) error {
+	agentIndex := 0
+	agentConnections, err := pilosactl.SSHClients(cmd.AgentHosts, cmd.SSHUser, "", cmd.Stderr)
+	if err != nil {
+		return err
+	}
+
+	if cmd.CopyBinary {
+		err = copyBinary(agentConnections, "github.com/pilosa/pilosa/cmd/pilosactl", "linux", "amd64")
+		if err != nil {
+			return err
+		}
+	}
+
 	sessions := make([]*ssh.Session, 0)
 	for _, sp := range cmd.Benchmarks {
 		for i := 0; i < sp.Num; i++ {
@@ -1444,7 +1469,7 @@ func (cmd *BspawnCommand) spawnRemote(ctx context.Context, runUUID uuid.UUID) er
 			sessions = append(sessions, sess)
 			sess.Stdout = cmd.Stdout
 			sess.Stderr = cmd.Stderr
-			err = sess.Start("pilosactl bagent -agentNum=" + strconv.Itoa(i) + " -hosts=" + strings.Join(cmd.PilosaHosts, ",") + " -run-uuid=" + runUUID.String() + " " + strings.Join(sp.Args, " "))
+			err = sess.Start("PATH=.:$PATH pilosactl bagent -agentNum=" + strconv.Itoa(i) + " -hosts=" + strings.Join(cmd.PilosaHosts, ",") + " -run-uuid=" + runUUID.String() + " " + strings.Join(sp.Args, " "))
 			if err != nil {
 				return err
 			}

cmd/pilosactl/multidbspawn.json

@@ -1,6 +1,8 @@
 {
-    "CreatorArgs": ["-type", "local", "-serverN", "3", "-replicaN", "1", "-log-file-prefix", "multidblog"],
-    "AgentHosts": ["localhost"],
+    "CreatorArgs": ["-hosts", "pilosa0.jaffee.sandbox.pilosa.com:15000,pilosa1.jaffee.sandbox.pilosa.com:15000,pilosa2.jaffee.sandbox.pilosa.com:15000", "-log-file-prefix", "multidblog", "-ssh-user", "ubuntu"],
+    "AgentHosts": ["agent0.jaffee.sandbox.pilosa.com"],
+    "SSHUser": "ubuntu",
+    "CopyBinary": true,
     "Benchmarks": [
         {
             "Num": 3,

creator/remote.go

@@ -57,21 +57,14 @@ func (c *RemoteCluster) Start() error {
 		conf.DataDir = "~/.pilosa" + port
 
 		// Connect to remote host
-		client, err := pilosactl.NewSSH(host, c.SSHUser, "")
-		if err != nil {
-			return err
-		}
-
-		// Create config file on remote host
-		sess, err := client.NewSession()
+		client, err := pilosactl.NewSSH(host, c.SSHUser, "", c.Stderr)
 		if err != nil {
 			return err
 		}
 		configname := "pilosa" + port + ".conf"
-		w, err := sess.StdinPipe()
-		err = sess.Start("cat > " + configname)
+		w, err := client.OpenFile(configname, "")
 		if err != nil {
-			return err
+			return fmt.Errorf("opening remote config file: %v", err)
 		}
 		enc := toml.NewEncoder(w)
 		err = enc.Encode(conf)
@@ -82,13 +75,9 @@ func (c *RemoteCluster) Start() error {
 		if err != nil {
 			return err
 		}
-		err = sess.Wait()
-		if err != nil {
-			return err
-		}
 
 		// Start pilosa on remote host
-		sess, err = client.NewSession()
+		sess, err := client.NewSession()
 		if err != nil {
 			return err
 		}

pilosactl/build.go

@@ -0,0 +1,29 @@
+package pilosactl
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+)
+
+func BuildBinary(pkg, goos, goarch string) (io.Reader, error) {
+	binFile, err := ioutil.TempFile("", "pilosactl")
+	if err != nil {
+		return nil, fmt.Errorf("build binary: %v", err)
+	}
+	com := exec.Command("go", "build", "-o", binFile.Name(), pkg)
+	com.Env = append([]string{"GOOS=" + goos, "GOARCH=" + goarch}, os.Environ()...)
+
+	err = com.Run()
+	if err != nil {
+		return nil, err
+	}
+
+	f, err := os.Open(binFile.Name())
+	if err != nil {
+		return nil, err
+	}
+	return f, nil
+}

pilosactl/ssh.go

@@ -2,23 +2,27 @@ package pilosactl
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"os"
 	"os/user"
 	"strings"
 
+	"errors"
+
 	"golang.org/x/crypto/ssh"
 	"golang.org/x/crypto/ssh/agent"
 )
 
 type SSH struct {
 	client *ssh.Client
+	Stderr io.Writer
 }
 
 // NewSSH wraps up some of the complexity of using the crypto/ssh pacakge
 // directly assuming you want to connect using public key auth and you can pass
 // a keyfile or your key is accessible through ssh agent.
-func NewSSH(host, username, keyfile string) (*SSH, error) {
+func NewSSH(host, username, keyfile string, stderr io.Writer) (*SSH, error) {
 	if username == "" {
 		user, err := user.Current()
 		if err != nil {
@@ -52,13 +56,15 @@ func NewSSH(host, username, keyfile string) (*SSH, error) {
 		return nil, fmt.Errorf("NewSHH failed Dial: %v ", err)
 	}
 
-	return &SSH{client: client}, nil
+	return &SSH{client: client, Stderr: stderr}, nil
 }
 
-func SSHClients(hosts []string, username, keyfile string) ([]*SSH, error) {
+type SSHFleet []*SSH
+
+func SSHClients(hosts []string, username, keyfile string, stderr io.Writer) (SSHFleet, error) {
 	clients := make([]*SSH, len(hosts))
 	for i, host := range hosts {
-		client, err := NewSSH(host, username, keyfile)
+		client, err := NewSSH(host, username, keyfile, stderr)
 		if err != nil {
 			return nil, err
 		}
@@ -70,3 +76,94 @@ func SSHClients(hosts []string, username, keyfile string) ([]*SSH, error) {
 func (s *SSH) NewSession() (*ssh.Session, error) {
 	return s.client.NewSession()
 }
+
+type remoteFile struct {
+	w    io.WriteCloser
+	sess *ssh.Session
+}
+
+func (r *remoteFile) Write(p []byte) (n int, err error) {
+	return r.w.Write(p)
+}
+
+func (r *remoteFile) Close() error {
+	errc := r.w.Close()
+	errw := r.sess.Wait()
+	if errc != nil || errw != nil {
+		return fmt.Errorf("error closing remote file - close: '%v', wait: '%v'", errc, errw)
+	}
+	return nil
+}
+
+// OpenFile creates or truncates an existing file of the given name on the
+// remote host, and returns a WriteCloser which will write to that file. perm
+// will be passed directly to chmod to set the file permissions. rm, touch,
+// chmod, cat and support for semicolons, double ampersand, and output
+// redirection (>>) must be available in the remote shell.
+func (s *SSH) OpenFile(name, perm string) (io.WriteCloser, error) {
+	sess, err := s.NewSession()
+	if err != nil {
+		return nil, err
+	}
+	w, err := sess.StdinPipe()
+	if err != nil {
+		return nil, err
+	}
+	if perm == "" {
+		perm = "0664"
+	}
+	err = sess.Start(fmt.Sprintf("rm %v; touch %v && chmod %v %v && cat >> %v", name, name, perm, name, name))
+	if err != nil {
+		return nil, err
+	}
+
+	return &remoteFile{w: w, sess: sess}, nil
+}
+
+type multiWriteCloser struct {
+	wcs []io.WriteCloser
+	ws  []io.Writer
+}
+
+func newMultiWriteCloser() *multiWriteCloser {
+	return &multiWriteCloser{
+		wcs: make([]io.WriteCloser, 0),
+		ws:  make([]io.Writer, 0),
+	}
+}
+
+func (mwc *multiWriteCloser) add(wc io.WriteCloser) {
+	mwc.wcs = append(mwc.wcs, wc)
+	mwc.ws = append(mwc.ws, wc)
+}
+
+func (mwc *multiWriteCloser) Write(p []byte) (n int, err error) {
+	mw := io.MultiWriter(mwc.ws...)
+	return mw.Write(p)
+}
+
+func (mwc *multiWriteCloser) Close() error {
+	errStr := ""
+	for _, wc := range mwc.wcs {
+		err := wc.Close()
+		if err != nil {
+			errStr = errStr + "; " + err.Error()
+		}
+	}
+	if errStr != "" {
+		return errors.New(errStr)
+	}
+	return nil
+}
+
+func (sf SSHFleet) OpenFile(name, perm string) (io.WriteCloser, error) {
+	writers := newMultiWriteCloser()
+	for _, cli := range sf {
+		wc, err := cli.OpenFile(name, "+x")
+		if err != nil {
+			return nil, err
+		}
+		writers.add(wc)
+	}
+	return writers, nil
+}