Merge pull request #216 from jaffee/213-remote-agents

213 remote agents

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

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

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 {

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

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 {

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 {

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 {

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 {

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
 }

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 {

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,

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()
 }