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/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/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/import.go b/bench/import.go index 3688d2625..8a249abbb 100644 --- a/bench/import.go +++ b/bench/import.go @@ -22,15 +22,15 @@ 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 + 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 *pilosactl.ImportCommand 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/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..31aad4531 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 + MaxTime time.Duration `json:"max-time"` hosts []string - 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 { 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 } diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index aea1aee0a..5ea23aab6 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" @@ -1150,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. @@ -1278,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 { @@ -1309,24 +1314,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 +1381,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 +1393,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() }