From 73c0ea2f0cc69e3dca01ecb6fbbbdc0c960132f0 Mon Sep 17 00:00:00 2001 From: jaffee Date: Thu, 22 Dec 2016 09:23:55 -0600 Subject: [PATCH 1/4] refactor/cleanup remote file creation --- cmd/pilosactl/multidbspawn.json | 2 +- creator/remote.go | 19 +++----------- pilosactl/ssh.go | 45 ++++++++++++++++++++++++++++++--- 3 files changed, 46 insertions(+), 20 deletions(-) diff --git a/cmd/pilosactl/multidbspawn.json b/cmd/pilosactl/multidbspawn.json index 83d6d8a4c..6be8b40f3 100644 --- a/cmd/pilosactl/multidbspawn.json +++ b/cmd/pilosactl/multidbspawn.json @@ -1,5 +1,5 @@ { - "CreatorArgs": ["-type", "local", "-serverN", "3", "-replicaN", "1", "-log-file-prefix", "multidblog"], + "CreatorArgs": ["-hosts", "localhost:19444,localhost:19445,localhost:19446", "-log-file-prefix", "multidblog"], "AgentHosts": ["localhost"], "Benchmarks": [ { diff --git a/creator/remote.go b/creator/remote.go index dd8baa564..41c1c9ca7 100644 --- a/creator/remote.go +++ b/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 } diff --git a/pilosactl/ssh.go b/pilosactl/ssh.go index 38898dea3..98ab6d18a 100644 --- a/pilosactl/ssh.go +++ b/pilosactl/ssh.go @@ -2,6 +2,7 @@ package pilosactl import ( "fmt" + "io" "net" "os" "os/user" @@ -13,12 +14,13 @@ import ( 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 +54,13 @@ 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) { +func SSHClients(hosts []string, username, keyfile string, stderr io.Writer) ([]*SSH, 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 +72,38 @@ 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 +} + +func (s *SSH) OpenFile(name string) (io.WriteCloser, error) { + sess, err := s.NewSession() + if err != nil { + return nil, err + } + w, err := sess.StdinPipe() + if err != nil { + return nil, err + } + err = sess.Start("cat > " + name) + if err != nil { + return nil, err + } + + return &remoteFile{w: w, sess: sess}, nil +} From 537b3878effdddf121eb7af7f07173dba21c78db Mon Sep 17 00:00:00 2001 From: jaffee Date: Thu, 22 Dec 2016 12:19:18 -0600 Subject: [PATCH 2/4] build and copy pilosactl to remote agents --- cmd/pilosactl/main.go | 42 +++++++++++++++++++++++++++++++-- cmd/pilosactl/multidbspawn.json | 6 +++-- creator/remote.go | 2 +- pilosactl/ssh.go | 12 ++++++++-- 4 files changed, 55 insertions(+), 7 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index c45b01a16..5494e7841 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -13,7 +13,9 @@ import ( "log" "math/rand" "os" + "os/exec" "os/signal" + "path" "path/filepath" "sort" "strconv" @@ -1328,6 +1330,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. @@ -1430,10 +1434,44 @@ 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, "") + agentConnections, err := pilosactl.SSHClients(cmd.AgentHosts, cmd.SSHUser, "", cmd.Stderr) if err != nil { return err } + if cmd.CopyBinary { + binLoc := path.Join(os.TempDir(), "pilosactl") + cmd := exec.Command("go", "build", "-o", binLoc, "github.com/pilosa/pilosa/cmd/pilosactl") + cmd.Env = append([]string{"GOOS=linux", "GOARCH=amd64"}, os.Environ()...) + err := cmd.Run() + if err != nil { + return err + } + agentWriters := make([]io.Writer, len(agentConnections)) + for i, conn := range agentConnections { + agentWriters[i], err = conn.OpenFile("pilosactl", "+x") + if err != nil { + return err + } + } + f, err := os.Open(binLoc) + if err != nil { + return err + } + _, err = io.Copy(io.MultiWriter(agentWriters...), f) + if err != nil { + return err + } + + for _, w := range agentWriters { + if wc, ok := w.(io.WriteCloser); ok { + err = wc.Close() + if err != nil { + return err + } + } + } + } + sessions := make([]*ssh.Session, 0) for _, sp := range cmd.Benchmarks { for i := 0; i < sp.Num; i++ { @@ -1444,7 +1482,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 } diff --git a/cmd/pilosactl/multidbspawn.json b/cmd/pilosactl/multidbspawn.json index 6be8b40f3..379fec0dd 100644 --- a/cmd/pilosactl/multidbspawn.json +++ b/cmd/pilosactl/multidbspawn.json @@ -1,6 +1,8 @@ { - "CreatorArgs": ["-hosts", "localhost:19444,localhost:19445,localhost:19446", "-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, diff --git a/creator/remote.go b/creator/remote.go index 41c1c9ca7..d951d7d4d 100644 --- a/creator/remote.go +++ b/creator/remote.go @@ -62,7 +62,7 @@ func (c *RemoteCluster) Start() error { return err } configname := "pilosa" + port + ".conf" - w, err := client.OpenFile(configname) + w, err := client.OpenFile(configname, "") if err != nil { return fmt.Errorf("opening remote config file: %v", err) } diff --git a/pilosactl/ssh.go b/pilosactl/ssh.go index 98ab6d18a..a42b776fc 100644 --- a/pilosactl/ssh.go +++ b/pilosactl/ssh.go @@ -91,7 +91,12 @@ func (r *remoteFile) Close() error { return nil } -func (s *SSH) OpenFile(name string) (io.WriteCloser, error) { +// 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 string, perm string) (io.WriteCloser, error) { sess, err := s.NewSession() if err != nil { return nil, err @@ -100,7 +105,10 @@ func (s *SSH) OpenFile(name string) (io.WriteCloser, error) { if err != nil { return nil, err } - err = sess.Start("cat > " + name) + 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 } From 9ffbe241e8c3635a06f124ba402c4551299ec291 Mon Sep 17 00:00:00 2001 From: jaffee Date: Thu, 22 Dec 2016 15:03:18 -0600 Subject: [PATCH 3/4] factor copying binary to remote hosts into separate function --- cmd/pilosactl/main.go | 73 +++++++++++++++++++++++++------------------ 1 file changed, 43 insertions(+), 30 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 5494e7841..e2db1c39e 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -1432,43 +1432,56 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { } } +func copyBinary(agentConnections []*pilosactl.SSH, pkg, goos, goarch string) error { + binName := path.Base(pkg) + binLoc := path.Join(os.TempDir(), binName) + com := exec.Command("go", "build", "-o", binLoc, pkg) + com.Env = append([]string{"GOOS=" + goos, "GOARCH=" + goarch}, os.Environ()...) + + err := com.Run() + if err != nil { + return err + } + + agentWriters := make([]io.Writer, len(agentConnections)) + for i, conn := range agentConnections { + agentWriters[i], err = conn.OpenFile(binName, "+x") + if err != nil { + return err + } + } + + f, err := os.Open(binLoc) + if err != nil { + return err + } + _, err = io.Copy(io.MultiWriter(agentWriters...), f) + if err != nil { + return err + } + + for _, w := range agentWriters { + if wc, ok := w.(io.WriteCloser); ok { + err = wc.Close() + if err != nil { + return err + } + } + } + return nil +} + 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 { - binLoc := path.Join(os.TempDir(), "pilosactl") - cmd := exec.Command("go", "build", "-o", binLoc, "github.com/pilosa/pilosa/cmd/pilosactl") - cmd.Env = append([]string{"GOOS=linux", "GOARCH=amd64"}, os.Environ()...) - err := cmd.Run() - if err != nil { - return err - } - agentWriters := make([]io.Writer, len(agentConnections)) - for i, conn := range agentConnections { - agentWriters[i], err = conn.OpenFile("pilosactl", "+x") - if err != nil { - return err - } - } - f, err := os.Open(binLoc) - if err != nil { - return err - } - _, err = io.Copy(io.MultiWriter(agentWriters...), f) - if err != nil { - return err - } - for _, w := range agentWriters { - if wc, ok := w.(io.WriteCloser); ok { - err = wc.Close() - if err != nil { - return err - } - } + if cmd.CopyBinary { + err = copyBinary(agentConnections, "github.com/pilosa/pilosa/cmd/pilosactl", "linux", "amd64") + if err != nil { + return err } } From a126df414cc8dc0a266d92e8df337a3774bf33fc Mon Sep 17 00:00:00 2001 From: jaffee Date: Fri, 23 Dec 2016 11:49:33 -0600 Subject: [PATCH 4/4] further refactor of building and copying pilosactl --- cmd/pilosactl/main.go | 40 ++++++------------------------- pilosactl/build.go | 29 ++++++++++++++++++++++ pilosactl/ssh.go | 56 +++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 90 insertions(+), 35 deletions(-) create mode 100644 pilosactl/build.go diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index e2db1c39e..d5b36153d 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -13,7 +13,6 @@ import ( "log" "math/rand" "os" - "os/exec" "os/signal" "path" "path/filepath" @@ -1374,10 +1373,6 @@ func (cmd *BspawnCommand) ParseFlags(args []string) error { if err != nil { return err } - - // handle pilosa creation - // handle agent creation - return nil } @@ -1432,43 +1427,22 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { } } -func copyBinary(agentConnections []*pilosactl.SSH, pkg, goos, goarch string) error { - binName := path.Base(pkg) - binLoc := path.Join(os.TempDir(), binName) - com := exec.Command("go", "build", "-o", binLoc, pkg) - com.Env = append([]string{"GOOS=" + goos, "GOARCH=" + goarch}, os.Environ()...) - - err := com.Run() +func copyBinary(fleet pilosactl.SSHFleet, pkg, goos, goarch string) error { + bin, err := pilosactl.BuildBinary(pkg, goos, goarch) if err != nil { return err } - agentWriters := make([]io.Writer, len(agentConnections)) - for i, conn := range agentConnections { - agentWriters[i], err = conn.OpenFile(binName, "+x") - if err != nil { - return err - } - } - - f, err := os.Open(binLoc) - if err != nil { - return err - } - _, err = io.Copy(io.MultiWriter(agentWriters...), f) + wc, err := fleet.OpenFile(path.Base(pkg), "+x") if err != nil { return err } - for _, w := range agentWriters { - if wc, ok := w.(io.WriteCloser); ok { - err = wc.Close() - if err != nil { - return err - } - } + _, err = io.Copy(wc, bin) + if err != nil { + return err } - return nil + return wc.Close() } func (cmd *BspawnCommand) spawnRemote(ctx context.Context, runUUID uuid.UUID) error { diff --git a/pilosactl/build.go b/pilosactl/build.go new file mode 100644 index 000000000..d69a89335 --- /dev/null +++ b/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 +} diff --git a/pilosactl/ssh.go b/pilosactl/ssh.go index a42b776fc..4e6074157 100644 --- a/pilosactl/ssh.go +++ b/pilosactl/ssh.go @@ -8,6 +8,8 @@ import ( "os/user" "strings" + "errors" + "golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh/agent" ) @@ -57,7 +59,9 @@ func NewSSH(host, username, keyfile string, stderr io.Writer) (*SSH, error) { return &SSH{client: client, Stderr: stderr}, nil } -func SSHClients(hosts []string, username, keyfile string, stderr io.Writer) ([]*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, stderr) @@ -96,7 +100,7 @@ func (r *remoteFile) Close() error { // 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 string, perm string) (io.WriteCloser, error) { +func (s *SSH) OpenFile(name, perm string) (io.WriteCloser, error) { sess, err := s.NewSession() if err != nil { return nil, err @@ -115,3 +119,51 @@ func (s *SSH) OpenFile(name string, perm string) (io.WriteCloser, error) { 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 +}