From bd78a7c1b9ea36f9f6272bae32d0c75e3f063d7f Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 12 Dec 2016 11:32:56 -0600 Subject: [PATCH 1/6] 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 { From e72e82ca7fa36ce7ca7eb41334091266c913b12c Mon Sep 17 00:00:00 2001 From: jaffee Date: Mon, 12 Dec 2016 16:29:40 -0600 Subject: [PATCH 2/6] remove separate create() and get bspawn working again --- cmd/pilosactl/main.go | 34 +++++++++++++++------------------ cmd/pilosactl/multidbspawn.json | 2 +- 2 files changed, 16 insertions(+), 20 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index a70a5a791..b66d84452 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -1040,19 +1040,6 @@ The following flags are allowed: `) } -// create separates creation from running for use programmatically by other -// commands like bspawn. -func (cmd *CreateCommand) create() (creator.Cluster, error) { - switch cmd.Type { - case "local": - return creator.NewLocalCluster(cmd.ReplicaN, cmd.ServerN) - case "AWS": - return nil, fmt.Errorf("unimplemented create type: %v", cmd.Type) - default: - return nil, fmt.Errorf("unsupported create type: %v", cmd.Type) - } -} - type CreateOutput struct { Hosts []string `json:"hosts"` LogFiles []string `json:"log-files"` @@ -1065,10 +1052,11 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { switch cmd.Type { case "local": var err error - clus, err = cmd.create() + clus, err = creator.NewLocalCluster(cmd.ReplicaN, cmd.ServerN) if err != nil { return fmt.Errorf("running create command: %v", err) } + defer clus.Shutdown() output.Hosts = clus.Hosts() logReaders := clus.Logs() @@ -1328,14 +1316,22 @@ pilosactl spawn configfile func (cmd *BspawnCommand) Run(ctx context.Context) error { if len(cmd.PilosaHosts) == 0 { // must create cluster - createCmd := NewCreateCommand(cmd.Stdin, cmd.Stdout, cmd.Stderr) + r, w := io.Pipe() + createCmd := NewCreateCommand(cmd.Stdin, w, cmd.Stderr) createCmd.ParseFlags(cmd.CreatorArgs) - clus, err := createCmd.create() + go func() { + err := createCmd.Run(ctx) + if err != nil { + fmt.Fprintf(cmd.Stderr, "Cluster creation error while spawning: %v", err) + } + }() + clus := &CreateOutput{} + dec := json.NewDecoder(r) + err := dec.Decode(clus) if err != nil { - return fmt.Errorf("Cluster creation error while spawning: %v", err) + return err } - defer clus.Shutdown() - cmd.PilosaHosts = clus.Hosts() + cmd.PilosaHosts = clus.Hosts } switch cmd.Agents.Type { case "local": diff --git a/cmd/pilosactl/multidbspawn.json b/cmd/pilosactl/multidbspawn.json index 627c6af94..08c00f776 100644 --- a/cmd/pilosactl/multidbspawn.json +++ b/cmd/pilosactl/multidbspawn.json @@ -1,5 +1,5 @@ { - "CreatorArgs": ["-type", "local", "-serverN", "3", "-replicaN", "1"], + "CreatorArgs": ["-type", "local", "-serverN", "3", "-replicaN", "1", "-log-file-prefix", "multidblog"], "Agents": { "Type": "local" }, "Benchmarks": [ { From 6829f7cdd0ca28ac2f018d36e9a7c95e613cd690 Mon Sep 17 00:00:00 2001 From: jaffee Date: Tue, 13 Dec 2016 10:14:09 -0600 Subject: [PATCH 3/6] create - start cluster over ssh when given hosts --- bench/import_test.go | 3 -- cmd/pilosa/main.go | 71 +-------------------------------- cmd/pilosactl/main.go | 91 +++++++++++++++++++++++++++++++++++++++++-- config.go | 84 +++++++++++++++++++++++++++++++++++++++ creator/creator.go | 10 ++--- pilosactl/ssh.go | 60 ++++++++++++++++++++++++++++ 6 files changed, 239 insertions(+), 80 deletions(-) create mode 100644 config.go create mode 100644 pilosactl/ssh.go diff --git a/bench/import_test.go b/bench/import_test.go index 6ef13527b..07074f125 100644 --- a/bench/import_test.go +++ b/bench/import_test.go @@ -2,7 +2,6 @@ package bench_test import ( "bytes" - "log" "testing" "io/ioutil" @@ -60,8 +59,6 @@ func TestImportInit(t *testing.T) { if string(bytes) != expected { t.Fatalf("unexpected result: %v", string(bytes)) } - - log.Println(imp) } func TestGenerateImportCSVNonRand(t *testing.T) { diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 5821a475b..24d2a9fec 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -31,9 +31,6 @@ func init() { const ( // DefaultDataDir is the default data directory. DefaultDataDir = "~/.pilosa" - - // DefaultHost is the default hostname and port to use. - DefaultHost = "localhost:15000" ) func main() { @@ -92,7 +89,7 @@ type Main struct { // Configuration options. ConfigPath string - Config *Config + Config *pilosa.Config // Profiling options. CPUProfile string @@ -108,7 +105,7 @@ type Main struct { func NewMain() *Main { return &Main{ Server: pilosa.NewServer(), - Config: NewConfig(), + Config: pilosa.NewConfig(), Stdin: os.Stdin, Stdout: os.Stdout, @@ -191,67 +188,3 @@ func (m *Main) ParseFlags(args []string) error { return nil } - -// Config represents the configuration for the command. -type Config struct { - DataDir string `toml:"data-dir"` - Host string `toml:"host"` - - Cluster struct { - ReplicaN int `toml:"replicas"` - Nodes []*ConfigNode `toml:"node"` - PollingInterval Duration `toml:"polling-interval"` - } `toml:"cluster"` - - Plugins struct { - Path string `toml:"path"` - } `toml:"plugins"` - - AntiEntropy struct { - Interval Duration `toml:"interval"` - } `toml:"anti-entropy"` -} - -type ConfigNode struct { - Host string `toml:"host"` -} - -// NewConfig returns an instance of Config with default options. -func NewConfig() *Config { - c := &Config{ - Host: DefaultHost, - } - c.Cluster.ReplicaN = pilosa.DefaultReplicaN - c.Cluster.PollingInterval = Duration(pilosa.DefaultPollingInterval) - c.AntiEntropy.Interval = Duration(pilosa.DefaultAntiEntropyInterval) - return c -} - -// PilosaCluster returns a new instance of pilosa.Cluster based on the config. -func (c *Config) PilosaCluster() *pilosa.Cluster { - cluster := pilosa.NewCluster() - cluster.ReplicaN = c.Cluster.ReplicaN - - for _, n := range c.Cluster.Nodes { - cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: n.Host}) - } - - return cluster -} - -// Duration is a TOML wrapper type for time.Duration. -type Duration time.Duration - -// String returns the string representation of the duration. -func (d Duration) String() string { return time.Duration(d).String() } - -// UnmarshalText parses a TOML value into a duration value. -func (d *Duration) UnmarshalText(text []byte) error { - v, err := time.ParseDuration(string(text)) - if err != nil { - return err - } - - *d = Duration(v) - return nil -} diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index b66d84452..119357d95 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "encoding/csv" + "encoding/json" "errors" "flag" "fmt" @@ -11,6 +12,7 @@ import ( "io/ioutil" "log" "math/rand" + "net" "os" "path/filepath" "sort" @@ -22,8 +24,7 @@ import ( "time" "unsafe" - "encoding/json" - + "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/bench" "github.com/pilosa/pilosa/creator" @@ -984,6 +985,9 @@ type CreateCommand struct { ServerN int ReplicaN int LogFilePrefix string + Hosts []string + + SSHUser string // Standard input/output Stdin io.Reader @@ -1004,14 +1008,18 @@ 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", "") + fs.StringVar(&cmd.Type, "type", "", "") fs.IntVar(&cmd.ServerN, "serverN", 3, "") fs.IntVar(&cmd.ReplicaN, "replicaN", 1, "") fs.StringVar(&cmd.LogFilePrefix, "log-file-prefix", "", "") + var hosts string + fs.StringVar(&hosts, "hosts", "", "") + fs.StringVar(&cmd.SSHUser, "ssh-user", "", "") if err := fs.Parse(args); err != nil { return err } + cmd.Hosts = strings.Split(hosts, ",") return nil } @@ -1033,10 +1041,18 @@ The following flags are allowed: -replicaN replication factor for cluster + -hosts + Comma separated host:port list. Instead of + creating hosts, just start pilosa on these + pre-existing hosts. The same host may be + listed multiple times with different ports. + -log-file-prefix output from the started cluster will go into files with this prefix (one per node) + -ssh-user + username to use when contacting remote hosts `) } @@ -1090,6 +1106,75 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { select {} case "AWS": return fmt.Errorf("AWS cluster type is not yet implemented") + case "": + if len(cmd.Hosts) == 0 { + return fmt.Errorf("no type or hosts specified - cannot continue") + } + // TODO: build pilosa + // TODO: copy binary to hosts + // build config + conf := pilosa.NewConfigForHosts(cmd.Hosts) + conf.Cluster.ReplicaN = cmd.ReplicaN + + // copy config to remote hosts and start pilosa + waitall := &sync.WaitGroup{} + for _, hostport := range cmd.Hosts { + host, port, err := net.SplitHostPort(hostport) + if err != nil { + return err + } + conf.Host = hostport + conf.DataDir = "~/.pilosa" + port + + client, err := pilosactl.NewSSH(host, cmd.SSHUser, "") + if err != nil { + return err + } + sess, err := client.NewSession() + if err != nil { + return err + } + configname := "pilosa" + port + ".conf" + w, err := sess.StdinPipe() + err = sess.Start("cat > " + configname) + if err != nil { + return err + } + enc := toml.NewEncoder(w) + err = enc.Encode(conf) + if err != nil { + return fmt.Errorf("encoding config: %v", err) + } + err = w.Close() + if err != nil { + return err + } + err = sess.Wait() + if err != nil { + return err + } + + sess, err = client.NewSession() + if err != nil { + return err + } + sess.Stdout = cmd.Stdout + sess.Stderr = cmd.Stderr + err = sess.Start("pilosa -config " + configname) + if err != nil { + return err + } + waitall.Add(1) + go func() { + defer waitall.Done() + err = sess.Wait() + if err != nil { + fmt.Fprintf(cmd.Stderr, "problem with remote pilosa process: %v", err) + } + }() + } + waitall.Wait() + return nil default: return fmt.Errorf("Unknown cluster type %v", cmd.Type) } diff --git a/config.go b/config.go new file mode 100644 index 000000000..3b8924c96 --- /dev/null +++ b/config.go @@ -0,0 +1,84 @@ +package pilosa + +import "time" + +const ( + // DefaultHost is the default hostname and port to use. + DefaultHost = "localhost:15000" +) + +// Config represents the configuration for the command. +type Config struct { + DataDir string `toml:"data-dir"` + Host string `toml:"host"` + + Cluster struct { + ReplicaN int `toml:"replicas"` + Nodes []*ConfigNode `toml:"node"` + PollingInterval Duration `toml:"polling-interval"` + } `toml:"cluster"` + + Plugins struct { + Path string `toml:"path"` + } `toml:"plugins"` + + AntiEntropy struct { + Interval Duration `toml:"interval"` + } `toml:"anti-entropy"` +} + +type ConfigNode struct { + Host string `toml:"host"` +} + +// NewConfig returns an instance of Config with default options. +func NewConfig() *Config { + c := &Config{ + Host: DefaultHost, + } + c.Cluster.ReplicaN = DefaultReplicaN + c.Cluster.PollingInterval = Duration(DefaultPollingInterval) + c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) + return c +} + +func NewConfigForHosts(hosts []string) *Config { + conf := NewConfig() + for _, hostport := range hosts { + conf.Cluster.Nodes = append(conf.Cluster.Nodes, &ConfigNode{Host: hostport}) + } + return conf +} + +// PilosaCluster returns a new instance of Cluster based on the config. +func (c *Config) PilosaCluster() *Cluster { + cluster := NewCluster() + cluster.ReplicaN = c.Cluster.ReplicaN + + for _, n := range c.Cluster.Nodes { + cluster.Nodes = append(cluster.Nodes, &Node{Host: n.Host}) + } + + return cluster +} + +// Duration is a TOML wrapper type for time.Duration. +type Duration time.Duration + +// String returns the string representation of the duration. +func (d Duration) String() string { return time.Duration(d).String() } + +// UnmarshalText parses a TOML value into a duration value. +func (d *Duration) UnmarshalText(text []byte) error { + v, err := time.ParseDuration(string(text)) + if err != nil { + return err + } + + *d = Duration(v) + return nil +} + +func (d Duration) MarshalText() (text []byte, err error) { + return []byte(d.String()), nil +} diff --git a/creator/creator.go b/creator/creator.go index 4fb9a51c8..a6b0f54f4 100644 --- a/creator/creator.go +++ b/creator/creator.go @@ -18,7 +18,7 @@ type Cluster interface { Logs() []io.Reader } -type cluster struct { +type localcluster struct { hosts []string logs []io.Reader servers []*pilosa.Server @@ -29,7 +29,7 @@ type cluster struct { func NewLocalCluster(replicaN, serverN int) (Cluster, error) { BasePort := 19327 - localCluster := &cluster{ + localCluster := &localcluster{ hosts: make([]string, serverN), servers: make([]*pilosa.Server, serverN), logs: make([]io.Reader, serverN), @@ -81,9 +81,9 @@ func NewLocalCluster(replicaN, serverN int) (Cluster, error) { return localCluster, nil } -func (c *cluster) Hosts() []string { return c.hosts } -func (c *cluster) Logs() []io.Reader { return c.logs } -func (c *cluster) Shutdown() error { +func (c *localcluster) Hosts() []string { return c.hosts } +func (c *localcluster) Logs() []io.Reader { return c.logs } +func (c *localcluster) Shutdown() error { errs := "" for _, s := range c.servers { if err := s.Close(); err != nil { diff --git a/pilosactl/ssh.go b/pilosactl/ssh.go new file mode 100644 index 000000000..95795c381 --- /dev/null +++ b/pilosactl/ssh.go @@ -0,0 +1,60 @@ +package pilosactl + +import ( + "fmt" + "net" + "os" + "os/user" + "strings" + + "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/agent" +) + +type SSH struct { + client *ssh.Client +} + +// NewSSH wraps up some of the complexity of using the crypto/ssh pacakge +// directly assuming you want to connect using public key auth and you can pass +// a keyfile or your key is accessible through ssh agent. +func NewSSH(host, username, keyfile string) (*SSH, error) { + if username == "" { + user, err := user.Current() + if err != nil { + return nil, err + } + username = user.Username + } + + var auth ssh.AuthMethod + if keyfile == "" { + sshAgent, err := net.Dial("unix", os.Getenv("SSH_AUTH_SOCK")) + if err != nil { + return nil, err + } + auth = ssh.PublicKeysCallback(agent.NewClient(sshAgent).Signers) + } else { + return nil, fmt.Errorf("using a keyfile is unimplemented") + } + + config := &ssh.ClientConfig{ + User: username, + Auth: []ssh.AuthMethod{auth}, + } + + if strings.Index(host, ":") == -1 { + host = host + ":22" + } + + client, err := ssh.Dial("tcp", host, config) + if err != nil { + return nil, fmt.Errorf("NewSHH failed Dial: %v ", err) + } + + return &SSH{client: client}, nil +} + +func (s *SSH) NewSession() (*ssh.Session, error) { + return s.client.NewSession() +} From 968e82d1ad309d6b412f9ee40a619a747afabf2d Mon Sep 17 00:00:00 2001 From: jaffee Date: Tue, 13 Dec 2016 16:36:52 -0600 Subject: [PATCH 4/6] clean up remote create, kill remote cluster Had to jump through some hoops to kill the remote cluster reliably. Not all ssh server implementations respect signals, so in order to kill the cluster, we request a pty for the ssh session. This allows us to send the byte 0x03 which is effectively the same has hitting Ctrl-c in an interactive ssh session. With a pty, just closing the session also seems to kill the remote process. --- cmd/pilosactl/main.go | 169 ++++++++++++++++-------------------------- creator/creator.go | 93 +---------------------- creator/local.go | 94 +++++++++++++++++++++++ creator/remote.go | 162 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 321 insertions(+), 197 deletions(-) create mode 100644 creator/local.go create mode 100644 creator/remote.go diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 119357d95..0853b8172 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -12,8 +12,8 @@ import ( "io/ioutil" "log" "math/rand" - "net" "os" + "os/signal" "path/filepath" "sort" "strconv" @@ -24,7 +24,6 @@ import ( "time" "unsafe" - "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/bench" "github.com/pilosa/pilosa/creator" @@ -1064,120 +1063,78 @@ type CreateOutput struct { // 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 - clus, err = creator.NewLocalCluster(cmd.ReplicaN, cmd.ServerN) - if err != nil { - return fmt.Errorf("running create command: %v", err) + clus = &creator.LocalCluster{ + ReplicaN: cmd.ReplicaN, + ServerN: cmd.ServerN, } - defer clus.Shutdown() - 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") case "": - if len(cmd.Hosts) == 0 { - return fmt.Errorf("no type or hosts specified - cannot continue") + clus = &creator.RemoteCluster{ + ClusterHosts: cmd.Hosts, + ReplicaN: cmd.ReplicaN, + SSHUser: cmd.SSHUser, + Stderr: cmd.Stderr, } - // TODO: build pilosa - // TODO: copy binary to hosts - // build config - conf := pilosa.NewConfigForHosts(cmd.Hosts) - conf.Cluster.ReplicaN = cmd.ReplicaN - - // copy config to remote hosts and start pilosa - waitall := &sync.WaitGroup{} - for _, hostport := range cmd.Hosts { - host, port, err := net.SplitHostPort(hostport) - if err != nil { - return err - } - conf.Host = hostport - conf.DataDir = "~/.pilosa" + port - - client, err := pilosactl.NewSSH(host, cmd.SSHUser, "") - if err != nil { - return err - } - sess, err := client.NewSession() - if err != nil { - return err - } - configname := "pilosa" + port + ".conf" - w, err := sess.StdinPipe() - err = sess.Start("cat > " + configname) - if err != nil { - return err - } - enc := toml.NewEncoder(w) - err = enc.Encode(conf) - if err != nil { - return fmt.Errorf("encoding config: %v", err) - } - err = w.Close() - if err != nil { - return err - } - err = sess.Wait() - if err != nil { - return err - } - - sess, err = client.NewSession() - if err != nil { - return err - } - sess.Stdout = cmd.Stdout - sess.Stderr = cmd.Stderr - err = sess.Start("pilosa -config " + configname) - if err != nil { - return err - } - waitall.Add(1) - go func() { - defer waitall.Done() - err = sess.Wait() - if err != nil { - fmt.Fprintf(cmd.Stderr, "problem with remote pilosa process: %v", err) - } - }() - } - waitall.Wait() - return nil default: return fmt.Errorf("Unknown cluster type %v", cmd.Type) } + + err := clus.Start() + if err != nil { + return fmt.Errorf("starting cluster: %v", err) + } + + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt) + go func() { + for range c { + fmt.Fprintf(cmd.Stderr, "caught signal\n") + err := clus.Shutdown() + code := 0 + if err != nil { + code = 1 + } + os.Exit(code) + } + }() + + defer clus.Shutdown() + output := &CreateOutput{} + 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 {} + } // BagentCommand represents a command for running a benchmark agent. A benchmark diff --git a/creator/creator.go b/creator/creator.go index a6b0f54f4..a7840dad2 100644 --- a/creator/creator.go +++ b/creator/creator.go @@ -1,100 +1,11 @@ // creator contains code for standing up pilosa clusters package creator -import ( - "fmt" - "io" - "io/ioutil" - "os" - "path/filepath" - "strconv" - - "github.com/pilosa/pilosa" -) +import "io" type Cluster interface { + Start() error Hosts() []string Shutdown() error Logs() []io.Reader } - -type localcluster struct { - hosts []string - logs []io.Reader - servers []*pilosa.Server - cluster *pilosa.Cluster - path string -} - -func NewLocalCluster(replicaN, serverN int) (Cluster, error) { - BasePort := 19327 - - localCluster := &localcluster{ - hosts: make([]string, serverN), - servers: make([]*pilosa.Server, serverN), - logs: make([]io.Reader, serverN), - } - path, err := ioutil.TempDir("", "pilosa-bench-") - if err != nil { - return localCluster, err - } - localCluster.path = path - - // Build cluster configuration. - cluster := pilosa.NewCluster() - cluster.ReplicaN = replicaN - - for i := 0; i < serverN; i++ { - cluster.Nodes = append(cluster.Nodes, &pilosa.Node{ - Host: fmt.Sprintf("localhost:%d", BasePort+i), - }) - } - localCluster.cluster = cluster - - // Build 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 - } - - // Build server. - s := pilosa.NewServer() - s.Host = fmt.Sprintf("localhost:%d", BasePort+i) - s.Cluster = cluster - s.Index.Path = filepath.Join(path, strconv.Itoa(i), "data") - - // Create log stream - localCluster.logs[i], s.LogOutput = io.Pipe() - - localCluster.servers[i] = s - } - - // Open all servers. - for i, s := range localCluster.servers { - if err := s.Open(); err != nil { - return localCluster, err - } - localCluster.hosts[i] = s.Host - } - - return localCluster, nil -} - -func (c *localcluster) Hosts() []string { return c.hosts } -func (c *localcluster) Logs() []io.Reader { return c.logs } -func (c *localcluster) Shutdown() error { - errs := "" - for _, s := range c.servers { - if err := s.Close(); err != nil { - errs = errs + err.Error() + "; " - } - } - if err := os.RemoveAll(c.path); err != nil { - errs = errs + err.Error() + ";" - } - if errs != "" { - return fmt.Errorf(errs) - } - return nil -} diff --git a/creator/local.go b/creator/local.go new file mode 100644 index 000000000..4d4b6b8ef --- /dev/null +++ b/creator/local.go @@ -0,0 +1,94 @@ +package creator + +import ( + "fmt" + "io" + "io/ioutil" + "os" + "path/filepath" + "strconv" + + "github.com/pilosa/pilosa" +) + +type LocalCluster struct { + ReplicaN int + ServerN int + hosts []string + logs []io.Reader + servers []*pilosa.Server + cluster *pilosa.Cluster + path string +} + +func (localCluster *LocalCluster) Start() error { + BasePort := 19327 + + localCluster.hosts = make([]string, localCluster.ServerN) + localCluster.servers = make([]*pilosa.Server, localCluster.ServerN) + localCluster.logs = make([]io.Reader, localCluster.ServerN) + + path, err := ioutil.TempDir("", "pilosa-bench-") + if err != nil { + return err + } + localCluster.path = path + + // Build cluster configuration. + cluster := pilosa.NewCluster() + cluster.ReplicaN = localCluster.ReplicaN + + for i := 0; i < localCluster.ServerN; i++ { + cluster.Nodes = append(cluster.Nodes, &pilosa.Node{ + Host: fmt.Sprintf("localhost:%d", BasePort+i), + }) + } + localCluster.cluster = cluster + + // Build servers. + for i := range localCluster.servers { + // Make server work directory. + if err := os.MkdirAll(filepath.Join(path, strconv.Itoa(i)), 0777); err != nil { + return err + } + + // Build server. + s := pilosa.NewServer() + s.Host = fmt.Sprintf("localhost:%d", BasePort+i) + s.Cluster = cluster + s.Index.Path = filepath.Join(path, strconv.Itoa(i), "data") + + // Create log stream + localCluster.logs[i], s.LogOutput = io.Pipe() + + localCluster.servers[i] = s + } + + // Open all servers. + for i, s := range localCluster.servers { + if err := s.Open(); err != nil { + return err + } + localCluster.hosts[i] = s.Host + } + + return nil +} + +func (c *LocalCluster) Hosts() []string { return c.hosts } +func (c *LocalCluster) Logs() []io.Reader { return c.logs } +func (c *LocalCluster) Shutdown() error { + errs := "" + for _, s := range c.servers { + if err := s.Close(); err != nil { + errs = errs + err.Error() + "; " + } + } + if err := os.RemoveAll(c.path); err != nil { + errs = errs + err.Error() + ";" + } + if errs != "" { + return fmt.Errorf(errs) + } + return nil +} diff --git a/creator/remote.go b/creator/remote.go new file mode 100644 index 000000000..84cf9fa89 --- /dev/null +++ b/creator/remote.go @@ -0,0 +1,162 @@ +package creator + +import ( + "fmt" + "io" + "net" + "sync" + + "time" + + "github.com/BurntSushi/toml" + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/pilosactl" + "golang.org/x/crypto/ssh" +) + +type RemoteCluster struct { + ClusterHosts []string + ReplicaN int + SSHUser string + Keyfile string + Key []byte + Stderr io.Writer + wg *sync.WaitGroup + logs []io.Reader + sessions []*ssh.Session + pipeRs []*io.PipeReader + pipeWs []*io.PipeWriter + stdins []io.WriteCloser +} + +func (c *RemoteCluster) Start() error { + c.logs = make([]io.Reader, 0) + if len(c.ClusterHosts) == 0 { + return fmt.Errorf("no type or hosts specified - cannot continue") + } + // TODO: build pilosa + // TODO: copy binary to hosts + // build config + conf := pilosa.NewConfigForHosts(c.ClusterHosts) + conf.Cluster.ReplicaN = c.ReplicaN + + // copy config to remote hosts and start pilosa + c.wg = &sync.WaitGroup{} + for _, hostport := range c.ClusterHosts { + + // Set up config for this host + host, port, err := net.SplitHostPort(hostport) + if err != nil { + return err + } + conf.Host = hostport + conf.DataDir = "~/.pilosa" + port + + // Connect to remote host + client, err := pilosactl.NewSSH(host, c.SSHUser, "") + if err != nil { + return err + } + + // Create config file on remote host + sess, err := client.NewSession() + if err != nil { + return err + } + configname := "pilosa" + port + ".conf" + w, err := sess.StdinPipe() + err = sess.Start("cat > " + configname) + if err != nil { + return err + } + enc := toml.NewEncoder(w) + err = enc.Encode(conf) + if err != nil { + return fmt.Errorf("encoding config: %v", err) + } + err = w.Close() + if err != nil { + return err + } + err = sess.Wait() + if err != nil { + return err + } + + // Start pilosa on remote host + sess, err = client.NewSession() + if err != nil { + return err + } + // Have to request pty in order to be able to kill remote process + // reliably. + modes := ssh.TerminalModes{ + ssh.ISIG: 1, + ssh.ECHO: 0, + } + err = sess.RequestPty("vt100", 40, 80, modes) + if err != nil { + return fmt.Errorf("request pty error: %v", err) + } + pipeR, pipeW := io.Pipe() + sess.Stdout = pipeW + sess.Stderr = pipeW + inpipe, err := sess.StdinPipe() + if err != nil { + return err + } + c.logs = append(c.logs, pipeR) + c.sessions = append(c.sessions, sess) + c.pipeRs = append(c.pipeRs, pipeR) + c.pipeWs = append(c.pipeWs, pipeW) + c.stdins = append(c.stdins, inpipe) + + err = sess.Start("pilosa -config " + configname) + if err != nil { + return err + } + c.wg.Add(1) + go func() { + defer c.wg.Done() + fmt.Fprintf(c.Stderr, "start waiting on %v\n", sess) + err = sess.Wait() + fmt.Fprintf(c.Stderr, "done waiting on session\n") + if err != nil { + fmt.Fprintf(c.Stderr, "problem with remote pilosa process: %v", err) + } + }() + } + return nil +} + +func (c *RemoteCluster) Hosts() []string { return c.ClusterHosts } +func (c *RemoteCluster) Logs() []io.Reader { return c.logs } +func (c *RemoteCluster) Shutdown() error { + for i, sess := range c.sessions { + var err error + _, err = c.stdins[i].Write([]byte{3}) // Send Control C + if err != nil { + fmt.Fprintf(c.Stderr, "Error write-signaling remote process: %v\n", err) + } + err = sess.Signal(ssh.SIGINT) + if err != nil { + fmt.Fprintf(c.Stderr, "Error signaling remote process: %v\n", err) + } + err = sess.Close() + if err != nil { + fmt.Fprintf(c.Stderr, "Error closing remote session: %v\n", err) + } + } + done := make(chan struct{}, 1) + go func() { + fmt.Fprintf(c.Stderr, "Waiting\n") + c.wg.Wait() + done <- struct{}{} + }() + select { + case <-done: + return nil + case <-time.After(time.Second * 5): + return fmt.Errorf("timed out waiting for remote processes to exit") + } +} From 9d5acdeecd21cf13100c9d0db5362ce57e36009e Mon Sep 17 00:00:00 2001 From: jaffee Date: Wed, 14 Dec 2016 10:53:14 -0600 Subject: [PATCH 5/6] cleanup cluster shutdown and extra logging --- cmd/pilosactl/main.go | 2 +- creator/remote.go | 20 ++++++++++++-------- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 0853b8172..a220212f0 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -1091,7 +1091,7 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { signal.Notify(c, os.Interrupt) go func() { for range c { - fmt.Fprintf(cmd.Stderr, "caught signal\n") + fmt.Fprintf(cmd.Stderr, "\ncaught signal - shutting down\n") err := clus.Shutdown() code := 0 if err != nil { diff --git a/creator/remote.go b/creator/remote.go index 84cf9fa89..724b05459 100644 --- a/creator/remote.go +++ b/creator/remote.go @@ -29,6 +29,8 @@ type RemoteCluster struct { stdins []io.WriteCloser } +// Start creates a configuration for each host in the cluster, copies it to the +// node, and starts the pilosa process on the remote host. func (c *RemoteCluster) Start() error { c.logs = make([]io.Reader, 0) if len(c.ClusterHosts) == 0 { @@ -118,9 +120,7 @@ func (c *RemoteCluster) Start() error { c.wg.Add(1) go func() { defer c.wg.Done() - fmt.Fprintf(c.Stderr, "start waiting on %v\n", sess) err = sess.Wait() - fmt.Fprintf(c.Stderr, "done waiting on session\n") if err != nil { fmt.Fprintf(c.Stderr, "problem with remote pilosa process: %v", err) } @@ -138,25 +138,29 @@ func (c *RemoteCluster) Shutdown() error { if err != nil { fmt.Fprintf(c.Stderr, "Error write-signaling remote process: %v\n", err) } + // signaling isn't supported by many ssh servers - hence the hack above err = sess.Signal(ssh.SIGINT) if err != nil { fmt.Fprintf(c.Stderr, "Error signaling remote process: %v\n", err) } - err = sess.Close() - if err != nil { - fmt.Fprintf(c.Stderr, "Error closing remote session: %v\n", err) - } } + done := make(chan struct{}, 1) go func() { - fmt.Fprintf(c.Stderr, "Waiting\n") c.wg.Wait() - done <- struct{}{} + close(done) }() select { case <-done: return nil case <-time.After(time.Second * 5): + for _, sess := range c.sessions { + err := sess.Close() + if err != nil { + fmt.Fprintf(c.Stderr, "Error closing remote session: %v\n", err) + } + } return fmt.Errorf("timed out waiting for remote processes to exit") } + } From 0c6e6e4f4447f269d7108ba72bfa9a4b86a7a839 Mon Sep 17 00:00:00 2001 From: jaffee Date: Wed, 14 Dec 2016 15:05:50 -0600 Subject: [PATCH 6/6] creator can set GOMAXPROCS for remote pilosa nodes --- cmd/pilosactl/main.go | 8 ++++++++ creator/remote.go | 9 ++++++++- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index a220212f0..68ff61cc0 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -985,6 +985,7 @@ type CreateCommand struct { ReplicaN int LogFilePrefix string Hosts []string + GoMaxProcs int SSHUser string @@ -1012,6 +1013,7 @@ func (cmd *CreateCommand) ParseFlags(args []string) error { fs.IntVar(&cmd.ReplicaN, "replicaN", 1, "") fs.StringVar(&cmd.LogFilePrefix, "log-file-prefix", "", "") var hosts string + fs.IntVar(&cmd.GoMaxProcs, "gomaxprocs", 0, "") fs.StringVar(&hosts, "hosts", "", "") fs.StringVar(&cmd.SSHUser, "ssh-user", "", "") @@ -1052,6 +1054,11 @@ The following flags are allowed: -ssh-user username to use when contacting remote hosts + + -gomaxprocs + when starting a cluster on remote hosts, this + will set the value of GOMAXPROCS. + `) } @@ -1077,6 +1084,7 @@ func (cmd *CreateCommand) Run(ctx context.Context) error { ReplicaN: cmd.ReplicaN, SSHUser: cmd.SSHUser, Stderr: cmd.Stderr, + GoMaxProcs: cmd.GoMaxProcs, } default: return fmt.Errorf("Unknown cluster type %v", cmd.Type) diff --git a/creator/remote.go b/creator/remote.go index 724b05459..dd8baa564 100644 --- a/creator/remote.go +++ b/creator/remote.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "net" + "strconv" "sync" "time" @@ -20,6 +21,7 @@ type RemoteCluster struct { SSHUser string Keyfile string Key []byte + GoMaxProcs int Stderr io.Writer wg *sync.WaitGroup logs []io.Reader @@ -113,7 +115,12 @@ func (c *RemoteCluster) Start() error { c.pipeWs = append(c.pipeWs, pipeW) c.stdins = append(c.stdins, inpipe) - err = sess.Start("pilosa -config " + configname) + gomaxprocsString := "" + if c.GoMaxProcs != 0 { + gomaxprocsString = "GOMAXPROCS=" + strconv.Itoa(c.GoMaxProcs) + " " + } + + err = sess.Start(gomaxprocsString + "pilosa -config " + configname) if err != nil { return err }