create - start cluster over ssh when given hosts

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) {

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

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

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
+}

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 {

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