From 6829f7cdd0ca28ac2f018d36e9a7c95e613cd690 Mon Sep 17 00:00:00 2001 From: jaffee Date: Tue, 13 Dec 2016 10:14:09 -0600 Subject: [PATCH] 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() +}