From c9f80e2b561b4ef0f0de7b9666a1f10812a13643 Mon Sep 17 00:00:00 2001 From: jaffee Date: Fri, 16 Dec 2016 12:21:05 -0600 Subject: [PATCH 1/5] basic implementation of remote agent spawning --- cmd/pilosactl/main.go | 59 ++++++++++++++++++++++++--------- cmd/pilosactl/multidbspawn.json | 2 +- pilosactl/ssh.go | 12 +++++++ 3 files changed, 57 insertions(+), 16 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index aea1aee0a..92d75128b 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -24,6 +24,8 @@ import ( "time" "unsafe" + "golang.org/x/crypto/ssh" + "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/bench" "github.com/pilosa/pilosa/creator" @@ -1309,24 +1311,19 @@ type BspawnCommand struct { // If AgentHosts is specified, Agents is ignored, and the existing // agents specified here are used. AgentHosts []string - // Agents is config for creating a fleet of agents from which to run the - // benchmark. TODO: mostly unimplemented. - Agents AgentConfig // 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. Benchmarks []Spawn + SSHUser string + Stdin io.Reader `json:"-"` Stdout io.Writer `json:"-"` Stderr io.Writer `json:"-"` } -type AgentConfig struct { - Type string -} - // Spawn represents a bagent command run in parallel across Num agents. The // bagent command can run multiple Benchmarks serially within itself. type Spawn struct { @@ -1381,7 +1378,10 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { // must create cluster r, w := io.Pipe() createCmd := NewCreateCommand(cmd.Stdin, w, cmd.Stderr) - createCmd.ParseFlags(cmd.CreatorArgs) + err := createCmd.ParseFlags(cmd.CreatorArgs) + if err != nil { + return err + } go func() { err := createCmd.Run(ctx) if err != nil { @@ -1390,22 +1390,51 @@ func (cmd *BspawnCommand) Run(ctx context.Context) error { }() clus := &CreateOutput{} dec := json.NewDecoder(r) - err := dec.Decode(clus) + err = dec.Decode(clus) if err != nil { return err } cmd.PilosaHosts = clus.Hosts } - switch cmd.Agents.Type { - case "local": + if len(cmd.AgentHosts) > 0 { + return cmd.spawnRemote(ctx) + } else { return cmd.spawnLocal(ctx) - case "remote": - return fmt.Errorf("remote type spawning is unimplemented") - default: - return fmt.Errorf("'%v' is not a supported type of spawn command", cmd.Agents.Type) } } +func (cmd *BspawnCommand) spawnRemote(ctx context.Context) error { + agentIndex := 0 + agentConnections, err := pilosactl.SSHClients(cmd.AgentHosts, cmd.SSHUser, "") + if err != nil { + return err + } + sessions := make([]*ssh.Session, 0) + for _, sp := range cmd.Benchmarks { + for i := 0; i < sp.Num; i++ { + sess, err := agentConnections[agentIndex].NewSession() + if err != nil { + return err + } + 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, ",") + " " + strings.Join(sp.Args, " ")) + if err != nil { + return err + } + } + } + + for _, sess := range sessions { + err = sess.Wait() + if err != nil { + return fmt.Errorf("error waiting for remote bagent: %v", err) + } + } + return nil +} + func (cmd *BspawnCommand) spawnLocal(ctx context.Context) error { agents := []*BagentCommand{} for _, sp := range cmd.Benchmarks { diff --git a/cmd/pilosactl/multidbspawn.json b/cmd/pilosactl/multidbspawn.json index 08c00f776..83d6d8a4c 100644 --- a/cmd/pilosactl/multidbspawn.json +++ b/cmd/pilosactl/multidbspawn.json @@ -1,6 +1,6 @@ { "CreatorArgs": ["-type", "local", "-serverN", "3", "-replicaN", "1", "-log-file-prefix", "multidblog"], - "Agents": { "Type": "local" }, + "AgentHosts": ["localhost"], "Benchmarks": [ { "Num": 3, diff --git a/pilosactl/ssh.go b/pilosactl/ssh.go index 95795c381..38898dea3 100644 --- a/pilosactl/ssh.go +++ b/pilosactl/ssh.go @@ -55,6 +55,18 @@ func NewSSH(host, username, keyfile string) (*SSH, error) { return &SSH{client: client}, nil } +func SSHClients(hosts []string, username, keyfile string) ([]*SSH, error) { + clients := make([]*SSH, len(hosts)) + for i, host := range hosts { + client, err := NewSSH(host, username, keyfile) + if err != nil { + return nil, err + } + clients[i] = client + } + return clients, nil +} + func (s *SSH) NewSession() (*ssh.Session, error) { return s.client.NewSession() } From a1d6187a2b94412f63153f96b2ddfa2d7db861e7 Mon Sep 17 00:00:00 2001 From: jaffee Date: Fri, 16 Dec 2016 13:56:11 -0600 Subject: [PATCH 2/5] have bagent add metadata to results need this to differentiate multiple agents' output when it's all going to stdout. --- bench/diagonal.go | 8 ++++---- bench/multidb.go | 6 +++--- bench/random.go | 15 +++++++-------- cmd/pilosactl/main.go | 17 ++++++++++------- 4 files changed, 24 insertions(+), 22 deletions(-) diff --git a/bench/diagonal.go b/bench/diagonal.go index f8c592027..10ed20368 100644 --- a/bench/diagonal.go +++ b/bench/diagonal.go @@ -13,10 +13,10 @@ import ( // DiagonalSetBits sets bits with increasing profile id and bitmap id. type DiagonalSetBits struct { HasClient - BaseBitmapID int - BaseProfileID int - Iterations int - DB string + BaseBitmapID int `json:"base-bitmap-id"` + BaseProfileID int `json:"base-profile-id"` + Iterations int `json:"iterations"` + DB string `json:"db"` } func (b *DiagonalSetBits) Usage() string { diff --git a/bench/multidb.go b/bench/multidb.go index 65ad9c7a8..b8e4528cd 100644 --- a/bench/multidb.go +++ b/bench/multidb.go @@ -12,9 +12,9 @@ import ( // MultiDBSetBits sets bits with increasing profile id and bitmap id. type MultiDBSetBits struct { HasClient - BaseBitmapID int - BaseProfileID int - Iterations int + BaseBitmapID int `json:"base-bitmap-id"` + BaseProfileID int `json:"base-profile-id"` + Iterations int `json:"iterations"` } func (b *MultiDBSetBits) Usage() string { diff --git a/bench/random.go b/bench/random.go index 4d46484f1..a39d40106 100644 --- a/bench/random.go +++ b/bench/random.go @@ -14,14 +14,13 @@ import ( // RandomSetBits sets bits randomly and deterministically based on a seed. type RandomSetBits struct { HasClient - BaseBitmapID int64 - BaseProfileID int64 - BitmapIDRange int64 - ProfileIDRange int64 - Iterations int // number of bits that will be set - Seed int64 - DB string // DB to use in pilosa. - + BaseBitmapID int64 `json:"base-bitmap-id"` + BaseProfileID int64 `json:"base-profile-id"` + BitmapIDRange int64 `json:"bitmap-id-range"` + ProfileIDRange int64 `json:"profile-id-range"` + Iterations int `json:"iterations"` + Seed int64 `json:"seed"` + DB string `json:"db"` } func (b *RandomSetBits) Usage() string { diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 92d75128b..5ea23aab6 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -1152,18 +1152,20 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { // on the command line. type BagentCommand struct { // Slice of Benchmarks which will be run serially. - Benchmarks []bench.Benchmark + Benchmarks []bench.Benchmark `json:"benchmarks"` // AgentNum will be passed to each benchmark's Run method so that it can // parameterize its behavior. - AgentNum int - // Slice of pilosa hosts to run the Benchmarks against. - Hosts []string + AgentNum int `json:"agent-num"` + // Enable pretty printing of results, for human consumption. HumanReadable bool - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer + // Slice of pilosa hosts to run the Benchmarks against. + Hosts []string `json:"hosts"` + + Stdin io.Reader `json:"-"` + Stdout io.Writer `json:"-"` + Stderr io.Writer `json:"-"` } // NewBagentCommand returns a new instance of BagentCommand. @@ -1280,6 +1282,7 @@ func (cmd *BagentCommand) Run(ctx context.Context) error { } res := sbm.Run(ctx, cmd.AgentNum) + res["metadata"] = cmd enc := json.NewEncoder(cmd.Stdout) enc.SetIndent("", " ") if cmd.HumanReadable { From 810ffa96e16ca75a46666f5b9f9b8eebe3343447 Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 19 Dec 2016 11:11:20 -0600 Subject: [PATCH 3/5] define json for all benchmarks --- bench/import.go | 20 ++++++++++---------- bench/randquery.go | 17 ++++++++--------- bench/sliceheight.go | 20 ++++++++++---------- 3 files changed, 28 insertions(+), 29 deletions(-) diff --git a/bench/import.go b/bench/import.go index 3688d2625..5a584933a 100644 --- a/bench/import.go +++ b/bench/import.go @@ -22,16 +22,16 @@ func NewImport(stdin io.Reader, stdout, stderr io.Writer) *Import { // Import sets bits with increasing profile id and bitmap id. type Import struct { - BaseBitmapID int64 - MaxBitmapID int64 - BaseProfileID int64 - MaxProfileID int64 - RandomBitmapOrder bool - MinBitsPerMap int64 - MaxBitsPerMap int64 - AgentControls string - Seed int64 - numbits int + BaseBitmapID int64 `json:"base-bitmap-id"` + MaxBitmapID int64 `json:"max-bitmap-id"` + BaseProfileID int64 `json:"base-profile-id"` + MaxProfileID int64 `json:"max-profile-id"` + RandomBitmapOrder bool `json:"random-bitmap-order"` + MinBitsPerMap int64 `json:"min-bits-per-map"` + MaxBitsPerMap int64 `json:"max-bits-per-map"` + AgentControls string `json:"agent-controls"` + Seed int64 `json:"seed"` + numbits int `json:""` *pilosactl.ImportCommand } diff --git a/bench/randquery.go b/bench/randquery.go index b844b4e3e..3d6228f74 100644 --- a/bench/randquery.go +++ b/bench/randquery.go @@ -12,15 +12,14 @@ import ( // RandomQuery queries randomly and deterministically based on a seed. type RandomQuery struct { HasClient - MaxDepth int - MaxArgs int - MaxN int - BaseBitmapID int64 - BitmapIDRange int64 - Iterations int // number of queries - Seed int64 - DBs []string // DBs to query. - + MaxDepth int `json:"max-depth"` + MaxArgs int `json:"max-args"` + MaxN int `json:"max-n"` + BaseBitmapID int64 `json:"base-bitmap-id"` + BitmapIDRange int64 `json:"bitmap-id-range"` + Iterations int `json:"iterations"` + Seed int64 `json:"seed"` + DBs []string `json:"dbs"` } func (b *RandomQuery) Usage() string { diff --git a/bench/sliceheight.go b/bench/sliceheight.go index bc3484b8d..d405c89a5 100644 --- a/bench/sliceheight.go +++ b/bench/sliceheight.go @@ -23,18 +23,18 @@ func NewSliceHeight(stdin io.Reader, stdout, stderr io.Writer) *SliceHeight { // SliceHeight benchmark tests the effect of an increasing number of bitmaps in // a single slice on query time. type SliceHeight struct { - MaxTime time.Duration - hosts []string + MaxTime time.Duration `json:"max-time"` + hosts []string `json:""` - MinBitsPerMap int64 - MaxBitsPerMap int64 - Seed int64 - Database string - Frame string + MinBitsPerMap int64 `json:"min-bits-per-map"` + MaxBitsPerMap int64 `json:"max-bits-per-map"` + Seed int64 `json:"seed"` + Database string `json:"database"` + Frame string `json:"frame"` - Stdin io.Reader - Stdout io.Writer - Stderr io.Writer + Stdin io.Reader `json:""` + Stdout io.Writer `json:""` + Stderr io.Writer `json:""` } func (b *SliceHeight) Usage() string { From 0b0ac95052d4a06da7b1c185b5ac26b9746fea15 Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 19 Dec 2016 12:39:05 -0600 Subject: [PATCH 4/5] few more json fixes --- bench/client.go | 3 ++- bench/import.go | 2 +- bench/sliceheight.go | 8 ++++---- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/bench/client.go b/bench/client.go index 0025d93d2..3742bdca3 100644 --- a/bench/client.go +++ b/bench/client.go @@ -2,6 +2,7 @@ package bench import ( "fmt" + "github.com/pilosa/pilosa" ) @@ -22,7 +23,7 @@ func roundRobinClient(hosts []string, agentNum int) (*pilosa.Client, error) { // provides the Init method, a ClientType argument and a cli internal variable. type HasClient struct { cli *pilosa.Client - ClientType string + ClientType string `json:"client-type"` } // Init for HasClient looks at the ClientType field and creates a pilosa client diff --git a/bench/import.go b/bench/import.go index 5a584933a..8a249abbb 100644 --- a/bench/import.go +++ b/bench/import.go @@ -31,7 +31,7 @@ type Import struct { MaxBitsPerMap int64 `json:"max-bits-per-map"` AgentControls string `json:"agent-controls"` Seed int64 `json:"seed"` - numbits int `json:""` + numbits int *pilosactl.ImportCommand } diff --git a/bench/sliceheight.go b/bench/sliceheight.go index d405c89a5..31aad4531 100644 --- a/bench/sliceheight.go +++ b/bench/sliceheight.go @@ -24,7 +24,7 @@ func NewSliceHeight(stdin io.Reader, stdout, stderr io.Writer) *SliceHeight { // a single slice on query time. type SliceHeight struct { MaxTime time.Duration `json:"max-time"` - hosts []string `json:""` + hosts []string MinBitsPerMap int64 `json:"min-bits-per-map"` MaxBitsPerMap int64 `json:"max-bits-per-map"` @@ -32,9 +32,9 @@ type SliceHeight struct { Database string `json:"database"` Frame string `json:"frame"` - Stdin io.Reader `json:""` - Stdout io.Writer `json:""` - Stderr io.Writer `json:""` + Stdin io.Reader `json:"-"` + Stdout io.Writer `json:"-"` + Stderr io.Writer `json:"-"` } func (b *SliceHeight) Usage() string { From d763b92de3013632978865e4043d2e607a565a9a Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 19 Dec 2016 15:12:56 -0600 Subject: [PATCH 5/5] fix rebase issues --- bench/bench.go | 2 -- cmd/pilosa/main_test.go | 4 ++-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/bench/bench.go b/bench/bench.go index 83d078c43..7949ae062 100644 --- a/bench/bench.go +++ b/bench/bench.go @@ -6,8 +6,6 @@ import ( "strconv" "sync" "time" - - "golang.org/x/sync/errgroup" ) // Benchmark is an interface to guide the creation of new pilosa benchmarks or diff --git a/cmd/pilosa/main_test.go b/cmd/pilosa/main_test.go index e4969c1e0..1267d44a5 100644 --- a/cmd/pilosa/main_test.go +++ b/cmd/pilosa/main_test.go @@ -386,8 +386,8 @@ func GenerateSetCommands(n int, rand *rand.Rand) []SetCommand { } // ParseConfig parses s into a Config. -func ParseConfig(s string) (main.Config, error) { - var c main.Config +func ParseConfig(s string) (pilosa.Config, error) { + var c pilosa.Config _, err := toml.Decode(s, &c) return c, err }