From bd78a7c1b9ea36f9f6272bae32d0c75e3f063d7f Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 12 Dec 2016 11:32:56 -0600 Subject: [PATCH] json output and local cluster log handling --- cmd/pilosactl/main.go | 96 +++++++++++++++++++++++-------------------- creator/creator.go | 35 +++++++--------- 2 files changed, 65 insertions(+), 66 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 6408794df..a70a5a791 100644 --- a/cmd/pilosactl/main.go +++ b/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 diff --git a/creator/creator.go b/creator/creator.go index b07e4b467..4fb9a51c8 100644 --- a/creator/creator.go +++ b/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 {