diff --git a/pilosactl/build.go b/build/build.go similarity index 85% rename from pilosactl/build.go rename to build/build.go index d69a89335..eb25af446 100644 --- a/pilosactl/build.go +++ b/build/build.go @@ -1,4 +1,4 @@ -package pilosactl +package build import ( "fmt" @@ -8,7 +8,7 @@ import ( "os/exec" ) -func BuildBinary(pkg, goos, goarch string) (io.Reader, error) { +func Binary(pkg, goos, goarch string) (io.Reader, error) { binFile, err := ioutil.TempFile("", "pilosactl") if err != nil { return nil, fmt.Errorf("build binary: %v", err) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index e0f2a82d0..3bdeb098e 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -27,9 +27,11 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/bench" + "github.com/pilosa/pilosa/build" "github.com/pilosa/pilosa/creator" "github.com/pilosa/pilosa/pilosactl" "github.com/pilosa/pilosa/roaring" + pssh "github.com/pilosa/pilosa/ssh" "github.com/satori/go.uuid" "golang.org/x/crypto/ssh" @@ -1001,6 +1003,10 @@ type CreateCommand struct { SSHUser string + CopyBinary bool + GOOS string + GOARCH string + // Standard input/output Stdin io.Reader Stdout io.Writer @@ -1028,6 +1034,9 @@ func (cmd *CreateCommand) ParseFlags(args []string) error { fs.IntVar(&cmd.GoMaxProcs, "gomaxprocs", 0, "") fs.StringVar(&hosts, "hosts", "", "") fs.StringVar(&cmd.SSHUser, "ssh-user", "", "") + fs.BoolVar(&cmd.CopyBinary, "copy-binary", false, "") + fs.StringVar(&cmd.GOOS, "goos", "linux", "") + fs.StringVar(&cmd.GOARCH, "goarch", "amd64", "") if err := fs.Parse(args); err != nil { return err @@ -1071,6 +1080,16 @@ The following flags are allowed: when starting a cluster on remote hosts, this will set the value of GOMAXPROCS. + -copy-binary + controls whether or not to build and copy pilosa to agents + + -goos + when using copy-binary, GOOS to use while building binary + + -goarch + when using copy-binary, GOARCH to use while building binary + + `) } @@ -1097,6 +1116,9 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { SSHUser: cmd.SSHUser, Stderr: cmd.Stderr, GoMaxProcs: cmd.GoMaxProcs, + CopyBinary: cmd.CopyBinary, + GOOS: cmd.GOOS, + GOARCH: cmd.GOARCH, } default: return fmt.Errorf("Unknown cluster type %v", cmd.Type) @@ -1303,7 +1325,6 @@ func (cmd *BagentCommand) Run(ctx context.Context) error { if err != nil { fmt.Fprintln(cmd.Stderr, err) } - // fmt.Fprintln(cmd.Stdout, res) return nil } @@ -1326,15 +1347,17 @@ type BspawnCommand struct { // locally. AgentHosts []string - // If this is true, build and copy pilosactl binary to agent hosts. - CopyBinary bool - // Makes output human readable Human bool // Result destination, ["stdout", "s3"] Output string + // If this is true, build and copy pilosactl binary to agent hosts. + CopyBinary bool + GOOS string + GOARCH string + // 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,12 +1393,15 @@ func (cmd *BspawnCommand) ParseFlags(args []string) error { fs.SetOutput(ioutil.Discard) creatorHosts := fs.String("creator.hosts", "", "") creatorLFP := fs.String("creator.log-file-prefix", "", "") + creatorCopyBinary := fs.Bool("creator.copy-binary", false, "") pilosaHosts := fs.String("pilosa-hosts", "", "") agentHosts := fs.String("agent-hosts", "", "") sshUser := fs.String("ssh-user", "", "") fs.BoolVar(&cmd.Human, "human", false, "") fs.StringVar(&cmd.Output, "output", "stdout", "") fs.BoolVar(&cmd.CopyBinary, "copy-binary", false, "") + fs.StringVar(&cmd.GOOS, "goos", "linux", "") + fs.StringVar(&cmd.GOARCH, "goarch", "amd64", "") err := fs.Parse(args) if err != nil { @@ -1404,7 +1430,10 @@ func (cmd *BspawnCommand) ParseFlags(args []string) error { } // TODO support all creator args - just checking for creatorHosts here won't be sufficient if *creatorHosts != "" { - cmd.CreatorArgs = []string{"-hosts=" + *creatorHosts, "-log-file-prefix=" + *creatorLFP, "-ssh-user=" + cmd.SSHUser} + cmd.CreatorArgs = []string{"-hosts=" + *creatorHosts, "-log-file-prefix=" + *creatorLFP, "-ssh-user=" + cmd.SSHUser, "-goos=" + cmd.GOOS, "-goarch=" + cmd.GOARCH} + if *creatorCopyBinary { + cmd.CreatorArgs = append(cmd.CreatorArgs, "-copy-binary") + } } return nil @@ -1436,14 +1465,22 @@ The following flags are allowed and will override the values in the config file: -ssh-user pilosa hosts to run against (will ignore creator args) - -copy-binary - controls whether or not to build and copy pilosactl to agents - -human toggle human readable output (indented json with formatted times) + -copy-binary + controls whether or not to build and copy pilosactl to agents + -output string to select output destination, "stdout" or "s3" + + -goos + when using copy-binary, GOOS to use while building binary + + -goarch + when using copy-binary, GOARCH to use while building binary + + `) } @@ -1493,8 +1530,8 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { } else { return fmt.Errorf("invalid bspawn output destination") } - enc := json.NewEncoder(writer) + enc := json.NewEncoder(writer) if cmd.Human { enc.SetIndent("", " ") output = bench.Prettify(output) @@ -1502,33 +1539,21 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { return enc.Encode(output) } -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) (map[string]interface{}, error) { agentIndex := 0 - agentConnections, err := pilosactl.SSHClients(cmd.AgentHosts, cmd.SSHUser, "", cmd.Stderr) + agentFleet, err := pssh.SSHClients(cmd.AgentHosts, cmd.SSHUser, "", cmd.Stderr) if err != nil { return nil, err } if cmd.CopyBinary { - err = copyBinary(agentConnections, "github.com/pilosa/pilosa/cmd/pilosactl", "linux", "amd64") + pkg := "github.com/pilosa/pilosa/cmd/pilosactl" + bin, err := build.Binary(pkg, cmd.GOOS, cmd.GOARCH) + if err != nil { + return nil, err + } + + err = agentFleet.WriteFile(path.Base(pkg), "+x", bin) if err != nil { return nil, err } @@ -1541,7 +1566,7 @@ func (cmd *BspawnCommand) spawnRemote(ctx context.Context) (map[string]interface for _, sp := range cmd.Benchmarks { results[sp.Name] = make(map[int]interface{}) for i := 0; i < sp.Num; i++ { - sess, err := agentConnections[agentIndex].NewSession() + sess, err := agentFleet[agentIndex].NewSession() if err != nil { return nil, err } diff --git a/creator/remote.go b/creator/remote.go index d951d7d4d..bdd18b7d5 100644 --- a/creator/remote.go +++ b/creator/remote.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "net" + "path" "strconv" "sync" @@ -11,7 +12,8 @@ import ( "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/pilosactl" + "github.com/pilosa/pilosa/build" + pssh "github.com/pilosa/pilosa/ssh" "golang.org/x/crypto/ssh" ) @@ -22,6 +24,9 @@ type RemoteCluster struct { Keyfile string Key []byte GoMaxProcs int + CopyBinary bool + GOOS string + GOARCH string Stderr io.Writer wg *sync.WaitGroup logs []io.Reader @@ -38,8 +43,26 @@ func (c *RemoteCluster) Start() error { if len(c.ClusterHosts) == 0 { return fmt.Errorf("no type or hosts specified - cannot continue") } - // TODO: build pilosa - // TODO: copy binary to hosts + + if c.CopyBinary { + fleet, err := pssh.SSHClients(c.ClusterHosts, c.SSHUser, c.Keyfile, c.Stderr) + if err != nil { + return fmt.Errorf("copying binary: %v", err) + } + + pkg := "github.com/pilosa/pilosa/cmd/pilosa" + bin, err := build.Binary(pkg, c.GOOS, c.GOARCH) + if err != nil { + return fmt.Errorf("building binary: %v", err) + } + + err = fleet.WriteFile(path.Base(pkg), "+x", bin) + if err != nil { + return fmt.Errorf("writing binary to fleet: %v", err) + } + + } + // build config conf := pilosa.NewConfigForHosts(c.ClusterHosts) conf.Cluster.ReplicaN = c.ReplicaN @@ -51,15 +74,15 @@ func (c *RemoteCluster) Start() error { // Set up config for this host host, port, err := net.SplitHostPort(hostport) if err != nil { - return err + return fmt.Errorf("splitting hostport: %v", err) } conf.Host = hostport conf.DataDir = "~/.pilosa" + port // Connect to remote host - client, err := pilosactl.NewSSH(host, c.SSHUser, "", c.Stderr) + client, err := pssh.NewSSH(host, c.SSHUser, "", c.Stderr) if err != nil { - return err + return fmt.Errorf("connecting to host: %v", err) } configname := "pilosa" + port + ".conf" w, err := client.OpenFile(configname, "") @@ -109,7 +132,7 @@ func (c *RemoteCluster) Start() error { gomaxprocsString = "GOMAXPROCS=" + strconv.Itoa(c.GoMaxProcs) + " " } - err = sess.Start(gomaxprocsString + "pilosa -config " + configname) + err = sess.Start("PATH=.:$PATH " + gomaxprocsString + "pilosa -config " + configname) if err != nil { return err } diff --git a/pilosactl/ssh.go b/ssh/ssh.go similarity index 78% rename from pilosactl/ssh.go rename to ssh/ssh.go index b595dcd82..62c156156 100644 --- a/pilosactl/ssh.go +++ b/ssh/ssh.go @@ -1,4 +1,4 @@ -package pilosactl +package ssh import ( "fmt" @@ -53,7 +53,7 @@ func NewSSH(host, username, keyfile string, stderr io.Writer) (*SSH, error) { client, err := ssh.Dial("tcp", host, config) if err != nil { - return nil, fmt.Errorf("NewSHH failed Dial - host: %v, config: %v, err: %v ", host, config, err) + return nil, fmt.Errorf("NewSSH failed Dial - host: %v, config: %v, err: %v ", host, config, err) } return &SSH{client: client, Stderr: stderr}, nil @@ -62,6 +62,7 @@ func NewSSH(host, username, keyfile string, stderr io.Writer) (*SSH, error) { type SSHFleet []*SSH func SSHClients(hosts []string, username, keyfile string, stderr io.Writer) (SSHFleet, error) { + hosts = DedupHosts(hosts) clients := make([]*SSH, len(hosts)) for i, host := range hosts { client, err := NewSSH(host, username, keyfile, stderr) @@ -73,6 +74,24 @@ func SSHClients(hosts []string, username, keyfile string, stderr io.Writer) (SSH return clients, nil } +// DedupHosts takes a slice of hosts, strips off any specified ports, and +// returns a de-duplicated slice of hosts. +func DedupHosts(hosts []string) []string { + seenHosts := make(map[string]bool) + ret := []string{} + for _, h := range hosts { + colonIdx := strings.Index(h, ":") + if colonIdx != -1 { + h = h[:colonIdx] + } + if !seenHosts[h] { + ret = append(ret, h) + } + seenHosts[h] = true + } + return ret +} + func (s *SSH) NewSession() (*ssh.Session, error) { return s.client.NewSession() } @@ -167,3 +186,20 @@ func (sf SSHFleet) OpenFile(name, perm string) (io.WriteCloser, error) { } return writers, nil } + +// WriteFile writes all of data (until EOF) into a file with the given name on +// each of the hosts in the fleet. It sets the permissions on the file to +// which is any valid input to chmod. If is the empty string, it defaults +// to 0664 +func (sf SSHFleet) WriteFile(name, perm string, data io.Reader) error { + wc, err := sf.OpenFile(name, perm) + if err != nil { + return err + } + + _, err = io.Copy(wc, data) + if err != nil { + return err + } + return wc.Close() +}