json output and local cluster log handling

cmd/pilosactl/main.go

@@ -12,7 +12,6 @@ import (
 	"log"
 	"math/rand"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"sort"
 	"strconv"
@@ -981,17 +980,10 @@ func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) e
 
 // CreateCommand represents a command for creating a pilosa cluster.
 type CreateCommand struct {
-	// Type can be AWS, local, etc.
-	Type string
-
-	// ServerN is the number of pilosa hosts in the cluster
-	ServerN int
-
-	// ReplicaN is the replication number for the cluster
-	ReplicaN int
-
-	// run is used internally by local cluster to signal that the cluster should be run and not exit
-	run bool
+	Type          string
+	ServerN       int
+	ReplicaN      int
+	LogFilePrefix string
 
 	// Standard input/output
 	Stdin  io.Reader
@@ -1012,10 +1004,10 @@ func NewCreateCommand(stdin io.Reader, stdout, stderr io.Writer) *CreateCommand
 func (cmd *CreateCommand) ParseFlags(args []string) error {
 	fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError)
 	fs.SetOutput(ioutil.Discard)
-	fs.StringVar(&cmd.Type, "type", "local", "Type of cluster - local, AWS, etc.")
-	fs.IntVar(&cmd.ServerN, "serverN", 3, "Number of hosts in cluster")
-	fs.IntVar(&cmd.ReplicaN, "replicaN", 1, "Replication factor for cluster")
-	fs.BoolVar(&cmd.run, "run", false, "run, don't exit")
+	fs.StringVar(&cmd.Type, "type", "local", "")
+	fs.IntVar(&cmd.ServerN, "serverN", 3, "")
+	fs.IntVar(&cmd.ReplicaN, "replicaN", 1, "")
+	fs.StringVar(&cmd.LogFilePrefix, "log-file-prefix", "", "")
 
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -1040,6 +1032,11 @@ The following flags are allowed:
 
 	-replicaN
 		replication factor for cluster
+
+	-log-file-prefix
+		output from the started cluster will go
+		into files with this prefix (one per node)
+
 `)
 }
 
@@ -1056,49 +1053,58 @@ func (cmd *CreateCommand) create() (creator.Cluster, error) {
 	}
 }
 
+type CreateOutput struct {
+	Hosts    []string `json:"hosts"`
+	LogFiles []string `json:"log-files"`
+}
+
 // Run executes cluster creation.
 func (cmd *CreateCommand) Run(ctx context.Context) error {
 	var clus creator.Cluster
+	output := &CreateOutput{}
 	switch cmd.Type {
 	case "local":
 		var err error
-		if cmd.run {
-			clus, err = cmd.create()
-			if err != nil {
-				return fmt.Errorf("running create command: %v", err)
-			}
-			fmt.Fprintln(cmd.Stdout, strings.Join(clus.Hosts(), ","))
-			select {}
-		}
-		args := append(os.Args, "-run")
-		subcmd := exec.Command(args[0], args[1:]...)
-		pipeR, err := subcmd.StdoutPipe()
+		clus, err = cmd.create()
 		if err != nil {
-			return fmt.Errorf("Couldn't get pipe for subcmd stdout: %v", err)
+			return fmt.Errorf("running create command: %v", err)
 		}
-		if subcmdOut, err := ioutil.TempFile("", "pilosactl-create"); err == nil {
-			subcmd.Stderr = subcmdOut
-			fmt.Fprintln(cmd.Stderr, subcmdOut.Name())
-		} else {
-			fmt.Fprintf(cmd.Stderr, "Error creating file for pilosa output - discarding: %v", err)
-		}
-		scanner := bufio.NewScanner(pipeR)
-		err = subcmd.Start()
-		if err != nil {
-			return fmt.Errorf("error kicking off local cluster: %v", err)
-		}
-		scanner.Scan()
-		fmt.Fprintln(cmd.Stdout, scanner.Text())
-		subcmd.Stdout = subcmd.Stderr
-		pipeR.Close()
+		output.Hosts = clus.Hosts()
 
+		logReaders := clus.Logs()
+		if cmd.LogFilePrefix != "" {
+			output.LogFiles = make([]string, len(clus.Hosts()))
+		}
+		for i, _ := range clus.Hosts() {
+			var f io.Writer = cmd.Stderr
+			var err error
+			if cmd.LogFilePrefix != "" {
+				f, err = os.Create(cmd.LogFilePrefix + strconv.Itoa(i))
+				if err != nil {
+					return err
+				}
+				output.LogFiles[i] = f.(*os.File).Name()
+			}
+
+			go func(i int, f io.Writer) {
+				_, err := io.Copy(f, logReaders[i])
+				if err != nil {
+					fmt.Fprintf(cmd.Stderr, "Error copying cluster logs: '%v'", err)
+				}
+			}(i, f)
+		}
+
+		enc := json.NewEncoder(cmd.Stdout)
+		err = enc.Encode(output)
+		if err != nil {
+			return err
+		}
+		select {}
 	case "AWS":
 		return fmt.Errorf("AWS cluster type is not yet implemented")
 	default:
 		return fmt.Errorf("Unknown cluster type %v", cmd.Type)
 	}
-
-	return nil
 }
 
 // BagentCommand represents a command for running a benchmark agent. A benchmark

creator/creator.go

@@ -3,6 +3,7 @@ package creator
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -14,10 +15,12 @@ import (
 type Cluster interface {
 	Hosts() []string
 	Shutdown() error
+	Logs() []io.Reader
 }
 
 type cluster struct {
 	hosts   []string
+	logs    []io.Reader
 	servers []*pilosa.Server
 	cluster *pilosa.Cluster
 	path    string
@@ -27,8 +30,9 @@ func NewLocalCluster(replicaN, serverN int) (Cluster, error) {
 	BasePort := 19327
 
 	localCluster := &cluster{
-		hosts:   make([]string, 0),
-		servers: make([]*pilosa.Server, 0),
+		hosts:   make([]string, serverN),
+		servers: make([]*pilosa.Server, serverN),
+		logs:    make([]io.Reader, serverN),
 	}
 	path, err := ioutil.TempDir("", "pilosa-bench-")
 	if err != nil {
@@ -48,8 +52,7 @@ func NewLocalCluster(replicaN, serverN int) (Cluster, error) {
 	localCluster.cluster = cluster
 
 	// Build servers.
-	servers := make([]*pilosa.Server, serverN)
-	for i := range servers {
+	for i := range localCluster.servers {
 		// Make server work directory.
 		if err := os.MkdirAll(filepath.Join(path, strconv.Itoa(i)), 0777); err != nil {
 			return localCluster, err
@@ -61,35 +64,25 @@ func NewLocalCluster(replicaN, serverN int) (Cluster, error) {
 		s.Cluster = cluster
 		s.Index.Path = filepath.Join(path, strconv.Itoa(i), "data")
 
-		// Create log file.
-		f, err := os.Create(filepath.Join(path, strconv.Itoa(i), "log"))
-		if err != nil {
-			return localCluster, err
-		}
+		// Create log stream
+		localCluster.logs[i], s.LogOutput = io.Pipe()
 
-		// Set log and optionally write out to stderr as well.
-		s.LogOutput = f
-
-		servers[i] = s
+		localCluster.servers[i] = s
 	}
-	localCluster.servers = servers
 
 	// Open all servers.
-	for _, s := range servers {
+	for i, s := range localCluster.servers {
 		if err := s.Open(); err != nil {
 			return localCluster, err
 		}
+		localCluster.hosts[i] = s.Host
 	}
 
-	hosts := make([]string, 0)
-	for _, s := range servers {
-		hosts = append(hosts, s.Host)
-	}
-	localCluster.hosts = hosts
 	return localCluster, nil
 }
 
-func (c *cluster) Hosts() []string { return c.hosts }
+func (c *cluster) Hosts() []string   { return c.hosts }
+func (c *cluster) Logs() []io.Reader { return c.logs }
 func (c *cluster) Shutdown() error {
 	errs := ""
 	for _, s := range c.servers {