scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-06 00:25:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #208 from jaffee/171-cluster-creation

Browse source 
#171 cluster creation
This commit is contained in:
Matthew Jaffee 2016-12-19 11:12:38 -06:00 • committed by GitHub
commit bde6f03c81
9 changed files with 533 additions and 236 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
bench/import_test.go
View file

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

71
cmd/pilosa/main.go
View file

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

180
cmd/pilosactl/main.go
View file

@ -4,6 +4,7 @@ import (
"bufio"
"context"
"encoding/csv"
"encoding/json"
"errors"
"flag"
"fmt"
@ -12,7 +13,7 @@ import (
"log"
"math/rand"
"os"
"os/exec"
"os/signal"
"path/filepath"
"sort"
"strconv"
@ -23,8 +24,6 @@ import (
"time"
"unsafe"
"encoding/json"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/bench"
"github.com/pilosa/pilosa/creator"
@ -981,17 +980,14 @@ 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
Type string
ServerN int
ReplicaN int
LogFilePrefix string
Hosts []string
GoMaxProcs int
// 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
SSHUser string
// Standard input/output
Stdin io.Reader
@ -1012,14 +1008,19 @@ 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", "", "")
fs.IntVar(&cmd.ServerN, "serverN", 3, "")
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", "", "")
if err := fs.Parse(args); err != nil {
return err
}
cmd.Hosts = strings.Split(hosts, ",")
return nil
}
@ -1040,20 +1041,30 @@ 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
-gomaxprocs
when starting a cluster on remote hosts, this
will set the value of GOMAXPROCS.
`)
}
// 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"`
}
// Run executes cluster creation.
@ -1061,44 +1072,77 @@ func (cmd *CreateCommand) Run(ctx context.Context) error {
var clus creator.Cluster
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 {}
clus = &creator.LocalCluster{
ReplicaN: cmd.ReplicaN,
ServerN: cmd.ServerN,
}
args := append(os.Args, "-run")
subcmd := exec.Command(args[0], args[1:]...)
pipeR, err := subcmd.StdoutPipe()
if err != nil {
return fmt.Errorf("Couldn't get pipe for subcmd stdout: %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()
case "AWS":
return fmt.Errorf("AWS cluster type is not yet implemented")
case "":
clus = &creator.RemoteCluster{
ClusterHosts: cmd.Hosts,
ReplicaN: cmd.ReplicaN,
SSHUser: cmd.SSHUser,
Stderr: cmd.Stderr,
GoMaxProcs: cmd.GoMaxProcs,
}
default:
return fmt.Errorf("Unknown cluster type %v", cmd.Type)
}
return nil
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, "\ncaught signal - shutting down\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
@ -1322,14 +1366,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":

2
cmd/pilosactl/multidbspawn.json
View file

@ -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": [
{

84
config.go Normal file
View file

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

102
creator/creator.go
View file

@ -1,107 +1,11 @@
// creator contains code for standing up pilosa clusters
package creator
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strconv"
"github.com/pilosa/pilosa"
)
import "io"
type Cluster interface {
Start() error
Hosts() []string
Shutdown() error
}
type cluster struct {
hosts []string
servers []*pilosa.Server
cluster *pilosa.Cluster
path string
}
func NewLocalCluster(replicaN, serverN int) (Cluster, error) {
BasePort := 19327
localCluster := &cluster{
hosts: make([]string, 0),
servers: make([]*pilosa.Server, 0),
}
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.
servers := make([]*pilosa.Server, serverN)
for i := range 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 file.
f, err := os.Create(filepath.Join(path, strconv.Itoa(i), "log"))
if err != nil {
return localCluster, err
}
// Set log and optionally write out to stderr as well.
s.LogOutput = f
servers[i] = s
}
localCluster.servers = servers
// Open all servers.
for _, s := range servers {
if err := s.Open(); err != nil {
return localCluster, err
}
}
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) 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
Logs() []io.Reader
}

94
creator/local.go Normal file
View file

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

173
creator/remote.go Normal file
View file

@ -0,0 +1,173 @@
package creator
import (
"fmt"
"io"
"net"
"strconv"
"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
GoMaxProcs int
Stderr io.Writer
wg *sync.WaitGroup
logs []io.Reader
sessions []*ssh.Session
pipeRs []*io.PipeReader
pipeWs []*io.PipeWriter
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 {
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)
gomaxprocsString := ""
if c.GoMaxProcs != 0 {
gomaxprocsString = "GOMAXPROCS=" + strconv.Itoa(c.GoMaxProcs) + " "
}
err = sess.Start(gomaxprocsString + "pilosa -config " + configname)
if err != nil {
return err
}
c.wg.Add(1)
go func() {
defer c.wg.Done()
err = sess.Wait()
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)
}
// 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)
}
}
done := make(chan struct{}, 1)
go func() {
c.wg.Wait()
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")
}
}

60
pilosactl/ssh.go Normal file
View file

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