diff --git a/client.go b/client.go index 3d94599a3..fbd0d3717 100644 --- a/client.go +++ b/client.go @@ -31,33 +31,59 @@ import ( "strconv" "time" + "crypto/tls" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" ) +// ClientOptions represents the configuration for a Client +type ClientOptions struct { + TLS *tls.Config +} + // Client represents a client to the Pilosa cluster. type Client struct { - host string + host *URI + options *ClientOptions // The client to use for HTTP communication. - // Defaults to the http.DefaultClient. HTTPClient *http.Client } // NewClient returns a new instance of Client to connect to host. -func NewClient(host string) (*Client, error) { +func NewClient(host string, options *ClientOptions) (*Client, error) { if host == "" { return nil, ErrHostRequired } + uri, err := NewURIFromAddress(host) + if err != nil { + return nil, err + } + + return NewClientFromURI(uri, options) +} + +func NewClientFromURI(uri *URI, options *ClientOptions) (*Client, error) { + if uri == nil { + return nil, ErrHostRequired + } + if options == nil { + options = &ClientOptions{} + } + transport := &http.Transport{} + if options.TLS != nil { + transport.TLSClientConfig = options.TLS + } + client := &http.Client{Transport: transport} return &Client{ - host: host, - HTTPClient: http.DefaultClient, + host: uri, + HTTPClient: client, }, nil } // Host returns the host the client was initialized with. -func (c *Client) Host() string { return c.host } +func (c *Client) Host() *URI { return c.host } // MaxSliceByIndex returns the number of slices on a server by index. func (c *Client) MaxSliceByIndex(ctx context.Context) (map[string]uint64, error) { @@ -72,14 +98,10 @@ func (c *Client) MaxInverseSliceByIndex(ctx context.Context) (map[string]uint64, // maxSliceByIndex returns the number of slices on a server by index. func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string]uint64, error) { // Execute request against the host. - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: "/slices/max", - RawQuery: (&url.Values{ - "inverse": {strconv.FormatBool(inverse)}, - }).Encode(), - } + u := uriPathToURL(c.host, "/slices/max") + u.RawQuery = (&url.Values{ + "inverse": {strconv.FormatBool(inverse)}, + }).Encode() // Build request. req, err := http.NewRequest("GET", u.String(), nil) @@ -109,11 +131,7 @@ func (c *Client) maxSliceByIndex(ctx context.Context, inverse bool) (map[string] // Schema returns all index and frame schema information. func (c *Client) Schema(ctx context.Context) ([]*IndexInfo, error) { // Execute request against the host. - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: "/schema", - } + u := uriPathToURL(c.host, "/schema") // Build request. req, err := http.NewRequest("GET", u.String(), nil) @@ -150,7 +168,7 @@ func (c *Client) CreateIndex(ctx context.Context, index string, opt IndexOptions } // Create URL & HTTP request. - u := url.URL{Scheme: "http", Host: c.host, Path: fmt.Sprintf("/index/%s", index)} + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s", index)) req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return err @@ -187,12 +205,8 @@ func (c *Client) CreateIndex(ctx context.Context, index string, opt IndexOptions // FragmentNodes returns a list of nodes that own a slice. func (c *Client) FragmentNodes(ctx context.Context, index string, slice uint64) ([]*Node, error) { // Execute request against the host. - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: "/fragment/nodes", - RawQuery: (url.Values{"index": {index}, "slice": {strconv.FormatUint(slice, 10)}}).Encode(), - } + u := uriPathToURL(c.host, "/fragment/nodes") + u.RawQuery = (url.Values{"index": {index}, "slice": {strconv.FormatUint(slice, 10)}}).Encode() // Build request. req, err := http.NewRequest("GET", u.String(), nil) @@ -237,11 +251,7 @@ func (c *Client) ExecuteQuery(ctx context.Context, index, query string, allowRed } // Create URL & HTTP request. - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: fmt.Sprintf("/index/%s/query", index), - } + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/query", index)) req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return nil, err @@ -278,14 +288,8 @@ func (c *Client) ExecuteQuery(ctx context.Context, index, query string, allowRed // ExecutePQL executes query string against index on the server. func (c *Client) ExecutePQL(ctx context.Context, index, query string) (interface{}, error) { - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: "/query", - RawQuery: url.Values{ - "index": {index}, - }.Encode(), - } + u := uriPathToURL(c.host, "/query") + u.RawQuery = url.Values{"index": {index}}.Encode() req, err := http.NewRequest("POST", u.String(), bytes.NewReader([]byte(query))) if err != nil { @@ -380,7 +384,7 @@ func MarshalImportPayload(index, frame string, slice uint64, bits []Bit) ([]byte // importNode sends a pre-marshaled import request to a node. func (c *Client) importNode(ctx context.Context, node *Node, buf []byte) error { // Create URL & HTTP request. - u := url.URL{Scheme: "http", Host: node.Host, Path: "/import"} + u := nodePathToURL(node, "/import") req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return err @@ -468,7 +472,7 @@ func MarshalImportValuePayload(index, frame, field string, slice uint64, vals [] // importValueNode sends a pre-marshaled import request to a node. func (c *Client) importValueNode(ctx context.Context, node *Node, buf []byte) error { // Create URL & HTTP request. - u := url.URL{Scheme: "http", Host: node.Host, Path: "/import-value"} + u := nodePathToURL(node, "/import-value") req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return err @@ -538,17 +542,13 @@ func (c *Client) ExportCSV(ctx context.Context, index, frame, view string, slice // exportNode copies a CSV export from a node to w. func (c *Client) exportNodeCSV(ctx context.Context, node *Node, index, frame, view string, slice uint64, w io.Writer) error { // Create URL. - u := url.URL{ - Scheme: "http", - Host: node.Host, - Path: "/export", - RawQuery: url.Values{ - "index": {index}, - "frame": {frame}, - "view": {view}, - "slice": {strconv.FormatUint(slice, 10)}, - }.Encode(), - } + u := nodePathToURL(node, "/export") + u.RawQuery = url.Values{ + "index": {index}, + "frame": {frame}, + "view": {view}, + "slice": {strconv.FormatUint(slice, 10)}, + }.Encode() // Generate HTTP request. req, err := http.NewRequest("GET", u.String(), nil) @@ -682,17 +682,13 @@ func (c *Client) BackupSlice(ctx context.Context, index, frame, view string, sli } func (c *Client) backupSliceNode(ctx context.Context, index, frame, view string, slice uint64, node *Node) (io.ReadCloser, error) { - u := url.URL{ - Scheme: "http", - Host: node.Host, - Path: "/fragment/data", - RawQuery: url.Values{ - "index": {index}, - "frame": {frame}, - "view": {view}, - "slice": {strconv.FormatUint(slice, 10)}, - }.Encode(), - } + u := nodePathToURL(node, "/fragment/data") + u.RawQuery = url.Values{ + "index": {index}, + "frame": {frame}, + "view": {view}, + "slice": {strconv.FormatUint(slice, 10)}, + }.Encode() // Build request. req, err := http.NewRequest("GET", u.String(), nil) @@ -769,17 +765,13 @@ func (c *Client) restoreSliceFrom(ctx context.Context, buf []byte, index, frame, // Restore slice to each owner. for _, node := range nodes { - u := url.URL{ - Scheme: "http", - Host: node.Host, - Path: "/fragment/data", - RawQuery: url.Values{ - "index": {index}, - "frame": {frame}, - "view": {view}, - "slice": {strconv.FormatUint(slice, 10)}, - }.Encode(), - } + u := nodePathToURL(node, "/fragment/data") + u.RawQuery = url.Values{ + "index": {index}, + "frame": {frame}, + "view": {view}, + "slice": {strconv.FormatUint(slice, 10)}, + }.Encode() // Build request. req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) @@ -819,7 +811,7 @@ func (c *Client) CreateFrame(ctx context.Context, index, frame string, opt Frame } // Create URL & HTTP request. - u := url.URL{Scheme: "http", Host: c.host, Path: fmt.Sprintf("/index/%s/frame/%s", index, frame)} + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s", index, frame)) req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) if err != nil { return err @@ -855,14 +847,10 @@ func (c *Client) CreateFrame(ctx context.Context, index, frame string, opt Frame // RestoreFrame restores an entire frame from a host in another cluster. func (c *Client) RestoreFrame(ctx context.Context, host, index, frame string) error { - u := url.URL{ - Scheme: "http", - Host: c.Host(), - Path: fmt.Sprintf("/index/%s/frame/%s/restore", index, frame), - RawQuery: url.Values{ - "host": {host}, - }.Encode(), - } + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/restore", index, frame)) + u.RawQuery = url.Values{ + "host": {host}, + }.Encode() // Build request. req, err := http.NewRequest("POST", u.String(), nil) @@ -890,11 +878,7 @@ func (c *Client) RestoreFrame(ctx context.Context, host, index, frame string) er // FrameViews returns a list of view names for a frame. func (c *Client) FrameViews(ctx context.Context, index, frame string) ([]string, error) { // Create URL & HTTP request. - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: fmt.Sprintf("/index/%s/frame/%s/views", index, frame), - } + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/views", index, frame)) req, err := http.NewRequest("GET", u.String(), nil) if err != nil { return nil, err @@ -930,17 +914,13 @@ func (c *Client) FrameViews(ctx context.Context, index, frame string) ([]string, // FragmentBlocks returns a list of block checksums for a fragment on a host. // Only returns blocks which contain data. func (c *Client) FragmentBlocks(ctx context.Context, index, frame, view string, slice uint64) ([]FragmentBlock, error) { - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: "/fragment/blocks", - RawQuery: url.Values{ - "index": {index}, - "frame": {frame}, - "view": {view}, - "slice": {strconv.FormatUint(slice, 10)}, - }.Encode(), - } + u := uriPathToURL(c.host, "/fragment/blocks") + u.RawQuery = url.Values{ + "index": {index}, + "frame": {frame}, + "view": {view}, + "slice": {strconv.FormatUint(slice, 10)}, + }.Encode() // Build request. req, err := http.NewRequest("GET", u.String(), nil) @@ -987,7 +967,7 @@ func (c *Client) BlockData(ctx context.Context, index, frame, view string, slice return nil, nil, err } - u := url.URL{Scheme: "http", Host: c.host, Path: "/fragment/block/data"} + u := uriPathToURL(c.host, "/fragment/block/data") req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf)) if err != nil { return nil, nil, err @@ -1024,11 +1004,7 @@ func (c *Client) BlockData(ctx context.Context, index, frame, view string, slice // ColumnAttrDiff returns data from differing blocks on a remote host. func (c *Client) ColumnAttrDiff(ctx context.Context, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: fmt.Sprintf("/index/%s/attr/diff", index), - } + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/attr/diff", index)) // Encode request. buf, err := json.Marshal(postIndexAttrDiffRequest{Blocks: blks}) @@ -1068,11 +1044,7 @@ func (c *Client) ColumnAttrDiff(ctx context.Context, index string, blks []AttrBl // RowAttrDiff returns data from differing blocks on a remote host. func (c *Client) RowAttrDiff(ctx context.Context, index, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { - u := url.URL{ - Scheme: "http", - Host: c.host, - Path: fmt.Sprintf("/index/%s/frame/%s/attr/diff", index, frame), - } + u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/attr/diff", index, frame)) // Encode request. buf, err := json.Marshal(postFrameAttrDiffRequest{Blocks: blks}) @@ -1241,3 +1213,19 @@ func (p BitsByPos) Less(i, j int) bool { } return p0 < p1 } + +func uriPathToURL(uri *URI, path string) url.URL { + return url.URL{ + Scheme: uri.Scheme(), + Host: uri.HostPort(), + Path: path, + } +} + +func nodePathToURL(node *Node, path string) url.URL { + return url.URL{ + Scheme: node.Scheme, + Host: node.Host, + Path: path, + } +} diff --git a/client_test.go b/client_test.go index f9880e33e..70ea6c669 100644 --- a/client_test.go +++ b/client_test.go @@ -35,7 +35,7 @@ func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) { for i := 0; i < numNodes; i++ { hldr[i] = test.MustOpenHolder() server[i] = test.NewServer() - server[i].Handler.Host = server[i].Host() + server[i].Handler.URI = server[i].HostURI() server[i].Handler.Cluster = c server[i].Handler.Cluster.Nodes[i].Host = server[i].Host() server[i].Handler.Holder = hldr[i].Holder @@ -54,22 +54,25 @@ func TestClient_MultiNode(t *testing.T) { } s[0].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor() + e := pilosa.NewExecutor(nil) e.Holder = hldr[0].Holder + e.Scheme = cluster.Nodes[0].Scheme e.Host = cluster.Nodes[0].Host e.Cluster = cluster return e.Execute(ctx, index, query, slices, opt) } s[1].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor() + e := pilosa.NewExecutor(nil) e.Holder = hldr[1].Holder + e.Scheme = cluster.Nodes[1].Scheme e.Host = cluster.Nodes[1].Host e.Cluster = cluster return e.Execute(ctx, index, query, slices, opt) } s[2].Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor() + e := pilosa.NewExecutor(nil) e.Holder = hldr[2].Holder + e.Scheme = cluster.Nodes[2].Scheme e.Host = cluster.Nodes[2].Host e.Cluster = cluster return e.Execute(ctx, index, query, slices, opt) @@ -207,7 +210,7 @@ func TestClient_Import(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -258,7 +261,7 @@ func TestClient_ImportInverseEnabled(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -307,7 +310,7 @@ func TestClient_ImportValue(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -345,7 +348,7 @@ func TestClient_BackupRestore(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -410,7 +413,7 @@ func TestClient_BackupInverseView(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -447,7 +450,7 @@ func TestClient_BackupInvalidView(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -476,7 +479,7 @@ func TestClient_FragmentBlocks(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.URI = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder diff --git a/cluster.go b/cluster.go index 6b5dddd38..820349ded 100644 --- a/cluster.go +++ b/cluster.go @@ -38,7 +38,8 @@ const ( // Node represents a node in the cluster. type Node struct { - Host string `json:"host"` + Scheme string `json:"scheme"` + Host string `json:"host"` status *internal.NodeStatus `json:"status"` } @@ -56,6 +57,16 @@ func (n *Node) SetState(s string) { n.status.State = s } +// URI returns the pilosa.URI corresponding to this node +func (n *Node) URI() (*URI, error) { + uri, err := NewURIFromAddress(n.Host) + if err != nil { + return nil, err + } + uri.SetScheme(n.Scheme) + return uri, nil +} + // Nodes represents a list of nodes. type Nodes []*Node diff --git a/cmd/backup.go b/cmd/backup.go index 1dd3a50d6..4bbbf70c2 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -19,9 +19,8 @@ import ( "io" "os" - "github.com/spf13/cobra" - "github.com/pilosa/pilosa/ctl" + "github.com/spf13/cobra" ) var Backuper *ctl.BackupCommand @@ -47,6 +46,7 @@ Backs up the view from across the cluster into a single file. flags.StringVarP(&Backuper.Frame, "frame", "f", "", "Frame to backup.") flags.StringVarP(&Backuper.View, "view", "v", "", "View to backup.") flags.StringVarP(&Backuper.Path, "output-file", "o", "", "File to write backup to - default stdout") + ctl.SetTLSConfig(flags, &Backuper.TLS.CertificatePath, &Backuper.TLS.CertificateKeyPath, &Backuper.TLS.SkipVerify) return backupCmd } diff --git a/cmd/bench.go b/cmd/bench.go index 0ea32d22a..d4b2b8580 100644 --- a/cmd/bench.go +++ b/cmd/bench.go @@ -47,6 +47,7 @@ Executes a benchmark for a given operation against the index. flags.StringVarP(&Bencher.Frame, "frame", "f", "", "Frame to benchmark.") flags.StringVarP(&Bencher.Op, "operation", "o", "set-bit", "Operation to perform: choose from [set-bit]") flags.IntVarP(&Bencher.N, "num", "n", 0, "Number of operations to perform.") + ctl.SetTLSConfig(flags, &Bencher.TLS.CertificatePath, &Bencher.TLS.CertificateKeyPath, &Bencher.TLS.SkipVerify) return benchCmd } diff --git a/cmd/export.go b/cmd/export.go index e55f84e59..c5907fbea 100644 --- a/cmd/export.go +++ b/cmd/export.go @@ -55,6 +55,7 @@ The file does not contain any headers. flags.StringVarP(&Exporter.Frame, "frame", "f", "", "Frame to export") flags.StringVarP(&Exporter.View, "view", "v", "standard", "View to export - default standard") flags.StringVarP(&Exporter.Path, "output-file", "o", "", "File to write export to - default stdout") + ctl.SetTLSConfig(flags, &Exporter.TLS.CertificatePath, &Exporter.TLS.CertificateKeyPath, &Exporter.TLS.SkipVerify) return exportCmd } diff --git a/cmd/import.go b/cmd/import.go index 6832480a3..f998bd4b3 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -65,6 +65,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.BoolVar(&Importer.FrameOptions.RangeEnabled, "frame-range-enabled", false, "Enabled range encoded frame") flags.StringVar(&Importer.FrameOptions.CacheType, "frame-cache-type", pilosa.CacheTypeRanked, "Cache type for the frame; valid values: none, lru, ranked") flags.Uint32Var(&Importer.FrameOptions.CacheSize, "frame-cache-size", 50000, "Cache size for the frame") + ctl.SetTLSConfig(flags, &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.SkipVerify) return importCmd } diff --git a/cmd/restore.go b/cmd/restore.go index 80f0bdd74..e5169be12 100644 --- a/cmd/restore.go +++ b/cmd/restore.go @@ -48,6 +48,7 @@ Restores a view to the cluster from a backup file. flags.StringVarP(&Restorer.Frame, "frame", "f", "", "Frame to restore into.") flags.StringVarP(&Restorer.View, "view", "v", "", "View to restore into.") flags.StringVarP(&Restorer.Path, "input-file", "d", "", "File to restore data from.") + ctl.SetTLSConfig(flags, &Restorer.TLS.CertificatePath, &Restorer.TLS.CertificateKeyPath, &Restorer.TLS.SkipVerify) return restoreCmd } diff --git a/cmd/server_test.go b/cmd/server_test.go index 40200e2d0..53841a5d9 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -44,7 +44,7 @@ func TestServerConfig(t *testing.T) { tests := []commandTest{ // TEST 0 { - args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "example.com:10111,example.com:10110", "--bind", "example.com:10111"}, + args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "localhost:10111,localhost:10110", "--bind", "localhost:10111"}, env: map[string]string{"PILOSA_DATA_DIR": "/tmp/myEnvDatadir", "PILOSA_CLUSTER_POLL_INTERVAL": "3m2s"}, cfgFileContent: ` data-dir = "/tmp/myFileDatadir" @@ -61,9 +61,9 @@ func TestServerConfig(t *testing.T) { validation: func() error { v := validator{} v.Check(cmd.Server.Config.DataDir, actualDataDir) - v.Check(cmd.Server.Config.Bind, "example.com:10111") + v.Check(cmd.Server.Config.Bind, "localhost:10111") v.Check(cmd.Server.Config.Cluster.ReplicaN, 2) - v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10111", "example.com:10110"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:10111", "localhost:10110"}) v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Second*182)) return v.Error() }, @@ -71,7 +71,7 @@ func TestServerConfig(t *testing.T) { // TEST 1 { args: []string{"server", "--anti-entropy.interval", "9m0s"}, - env: map[string]string{"PILOSA_CLUSTER_HOSTS": "example.com:1110,example.com:1111", "PILOSA_BIND": "example.com:1110"}, + env: map[string]string{"PILOSA_CLUSTER_HOSTS": "localhost:1110,localhost:1111", "PILOSA_BIND": "localhost:1110"}, cfgFileContent: ` bind = "localhost:0" data-dir = "` + actualDataDir + `" @@ -85,7 +85,7 @@ func TestServerConfig(t *testing.T) { `, validation: func() error { v := validator{} - v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:1110", "example.com:1111"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:1110", "localhost:1111"}) v.Check(cmd.Server.Config.Plugins.Path, "/var/sloth") v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*9)) return v.Error() diff --git a/config.go b/config.go index 9743d4c1b..fe93be19e 100644 --- a/config.go +++ b/config.go @@ -14,7 +14,9 @@ package pilosa -import "time" +import ( + "time" +) // Cluster types. const ( @@ -46,6 +48,16 @@ const ( // ClusterTypes set of cluster types. var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterGossip} +// TLSConfig contains TLS configuration +type TLSConfig struct { + // CertificatePath contains the path to the certificate (.crt or .pem file) + CertificatePath string `toml:"certificate-path"` + // CertificateKeyPath contains the path to the certificate key (.key file) + CertificateKeyPath string `toml:"certificate-key-path"` + // SkipVerify disables verification for self-signed certificates + SkipVerify bool `toml:"skip-verify"` +} + // Config represents the configuration for the command. type Config struct { DataDir string `toml:"data-dir"` @@ -80,6 +92,8 @@ type Config struct { Host string `toml:"host"` PollInterval Duration `toml:"poll-interval"` } `toml:"metric"` + + TLS TLSConfig } // NewConfig returns an instance of Config with default options. @@ -94,6 +108,7 @@ func NewConfig() *Config { c.Cluster.Hosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) c.Metric.Service = DefaultMetrics + c.TLS = TLSConfig{} return c } @@ -109,7 +124,7 @@ func (c *Config) Validate() error { if err != nil { return err } - if !foundItem(c.Cluster.Hosts, bindWithDefaults) { + if !c.foundHost(bindWithDefaults) { return ErrConfigHostsMissing } } @@ -118,6 +133,19 @@ func (c *Config) Validate() error { return nil } +func (c *Config) foundHost(host *URI) bool { + for _, clusterHost := range c.Cluster.Hosts { + uri, err := NewURIFromAddress(clusterHost) + if err != nil { + continue + } + if host.Equals(uri) { + return true + } + } + return false +} + // Duration is a TOML wrapper type for time.Duration. type Duration time.Duration diff --git a/ctl/backup.go b/ctl/backup.go index 7b8465637..8e725b41e 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -38,6 +38,8 @@ type BackupCommand struct { // Standard input/output *pilosa.CmdIO + + TLS pilosa.TLSConfig } // NewBackupCommand returns a new instance of BackupCommand. @@ -55,7 +57,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) error { } // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) + client, err := CommandClient(cmd) if err != nil { return err } @@ -81,3 +83,11 @@ func (cmd *BackupCommand) Run(ctx context.Context) error { return nil } + +func (cmd *BackupCommand) TLSHost() string { + return cmd.Host +} + +func (cmd *BackupCommand) TLSConfiguration() pilosa.TLSConfig { + return cmd.TLS +} diff --git a/ctl/backup_test.go b/ctl/backup_test.go index 1c292685b..3db4b3b73 100644 --- a/ctl/backup_test.go +++ b/ctl/backup_test.go @@ -46,7 +46,11 @@ func TestBackupCommand_Run(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + t.Fatal(err) + } + s.Handler.URI = uri s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder diff --git a/ctl/bench.go b/ctl/bench.go index 9e7b5d386..24e6e7658 100644 --- a/ctl/bench.go +++ b/ctl/bench.go @@ -40,6 +40,8 @@ type BenchCommand struct { // Standard input/output *pilosa.CmdIO + + TLS pilosa.TLSConfig } // NewBenchCommand returns a new instance of BenchCommand. @@ -52,7 +54,7 @@ func NewBenchCommand(stdin io.Reader, stdout, stderr io.Writer) *BenchCommand { // Run executes the bench command. func (cmd *BenchCommand) Run(ctx context.Context) error { // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) + client, err := CommandClient(cmd) if err != nil { return err } @@ -100,3 +102,11 @@ func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) e return nil } + +func (cmd *BenchCommand) TLSHost() string { + return cmd.Host +} + +func (cmd *BenchCommand) TLSConfiguration() pilosa.TLSConfig { + return cmd.TLS +} diff --git a/ctl/common.go b/ctl/common.go new file mode 100644 index 000000000..975f19898 --- /dev/null +++ b/ctl/common.go @@ -0,0 +1,42 @@ +package ctl + +import ( + "crypto/tls" + "github.com/pilosa/pilosa" + "github.com/spf13/pflag" +) + +// CommandWithTLSSupport is the interface for commands which has TLS settings +type CommandWithTLSSupport interface { + TLSHost() string + TLSConfiguration() pilosa.TLSConfig +} + +// SetTLSConfig creates common TLS flags +func SetTLSConfig(flags *pflag.FlagSet, certificatePath *string, certificateKeyPath *string, skipVerify *bool) { + flags.StringVarP(certificatePath, "tls.certificate", "", "", "TLS certificate path (usually has the .crt or .pem extension") + flags.StringVarP(certificateKeyPath, "tls.key", "", "", "TLS certificate key path (usually has the .key extension") + flags.BoolVarP(skipVerify, "tls.skip-verify", "", false, "Skip TLS certificate verification (not secure)") +} + +// CommandClient returns a pilosa.Client for the command +func CommandClient(cmd CommandWithTLSSupport) (*pilosa.Client, error) { + tlsConfig := cmd.TLSConfiguration() + var clientOptions *pilosa.ClientOptions + if tlsConfig.CertificatePath != "" && tlsConfig.CertificateKeyPath != "" { + cert, err := tls.LoadX509KeyPair(tlsConfig.CertificatePath, tlsConfig.CertificateKeyPath) + if err != nil { + return nil, err + } + TLSConfig := &tls.Config{ + Certificates: []tls.Certificate{cert}, + InsecureSkipVerify: tlsConfig.SkipVerify, + } + clientOptions = &pilosa.ClientOptions{TLS: TLSConfig} + } + client, err := pilosa.NewClient(cmd.TLSHost(), clientOptions) + if err != nil { + return nil, err + } + return client, err +} diff --git a/ctl/export.go b/ctl/export.go index 295e10039..84b91edec 100644 --- a/ctl/export.go +++ b/ctl/export.go @@ -37,6 +37,8 @@ type ExportCommand struct { // Standard input/output *pilosa.CmdIO + + TLS pilosa.TLSConfig } // NewExportCommand returns a new instance of ExportCommand. @@ -73,7 +75,7 @@ func (cmd *ExportCommand) Run(ctx context.Context) error { } // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) + client, err := CommandClient(cmd) if err != nil { return err } @@ -107,3 +109,11 @@ func (cmd *ExportCommand) Run(ctx context.Context) error { return nil } + +func (cmd *ExportCommand) TLSHost() string { + return cmd.Host +} + +func (cmd *ExportCommand) TLSConfiguration() pilosa.TLSConfig { + return cmd.TLS +} diff --git a/ctl/export_test.go b/ctl/export_test.go index 4f9d9b184..401b902e9 100644 --- a/ctl/export_test.go +++ b/ctl/export_test.go @@ -59,7 +59,11 @@ func TestExportCommand_Run(t *testing.T) { defer hldr.Close() s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + t.Fatal(err) + } + s.Handler.URI = uri s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -71,7 +75,7 @@ func TestExportCommand_Run(t *testing.T) { cm.Index = "i" cm.Frame = "f" cm.View = pilosa.ViewStandard - err := cm.Run(context.Background()) + err = cm.Run(context.Background()) if err != nil { t.Fatalf("Export Run doesn't work: %s", err) } diff --git a/ctl/import.go b/ctl/import.go index c9cefbbbe..e3eda260e 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -62,6 +62,8 @@ type ImportCommand struct { // Standard input/output *pilosa.CmdIO + + TLS pilosa.TLSConfig } // NewImportCommand returns a new instance of ImportCommand. @@ -86,7 +88,7 @@ func (cmd *ImportCommand) Run(ctx context.Context) error { return errors.New("path required") } // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) + client, err := CommandClient(cmd) if err != nil { return err } @@ -338,3 +340,11 @@ func (cmd *ImportCommand) importFieldValues(ctx context.Context, vals []pilosa.F return nil } + +func (cmd *ImportCommand) TLSHost() string { + return cmd.Host +} + +func (cmd *ImportCommand) TLSConfiguration() pilosa.TLSConfig { + return cmd.TLS +} diff --git a/ctl/import_test.go b/ctl/import_test.go index c9c36d5c8..5979bdbee 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -65,7 +65,11 @@ func TestImportCommand_Run(t *testing.T) { defer hldr.Close() s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + t.Fatal(err) + } + s.Handler.URI = uri s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -101,7 +105,11 @@ func TestImportCommand_RunValue(t *testing.T) { defer hldr.Close() s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + t.Fatal(err) + } + s.Handler.URI = uri s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder diff --git a/ctl/restore.go b/ctl/restore.go index 676e81331..38528c2b5 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -38,6 +38,8 @@ type RestoreCommand struct { // Standard input/output *pilosa.CmdIO + + TLS pilosa.TLSConfig } // NewRestoreCommand returns a new instance of RestoreCommand. @@ -55,7 +57,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error { } // Create a client to the server. - client, err := pilosa.NewClient(cmd.Host) + client, err := CommandClient(cmd) if err != nil { return err } @@ -74,3 +76,11 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error { return nil } + +func (cmd *RestoreCommand) TLSHost() string { + return cmd.Host +} + +func (cmd *RestoreCommand) TLSConfiguration() pilosa.TLSConfig { + return cmd.TLS +} diff --git a/ctl/restore_test.go b/ctl/restore_test.go index 39ac92388..9dd2d3661 100644 --- a/ctl/restore_test.go +++ b/ctl/restore_test.go @@ -48,7 +48,11 @@ func TestRestoreCommand_Run(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + t.Fatal(err) + } + s.Handler.URI = uri s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder diff --git a/ctl/server.go b/ctl/server.go index e3a2c4156..a0c080729 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -42,4 +42,5 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", "nop", "Default URI on which pilosa should listen.") flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", "", "Default URI to send metrics.") flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") + SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify) } diff --git a/docs/configuration.md b/docs/configuration.md index db67c9e7c..6741bc74b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -206,6 +206,42 @@ Any flag that has a value that is a comma separated list on the command line bec poll-interval = "0m15s" ``` +##### TLS Certificate + +* Description: Path to the TLS certificate to use for serving HTTPS. Usually has one of`.crt` or `.pem` extensions. +* Flag: `tls.certificate=/srv/pilosa/certs/server.crt` +* Env: `PILOSA_TLS_CERTIFICATE=/srv/pilosa/certs/server.crt` +* Config: + + ```toml + [tls] + certificate = "/srv/pilosa/certs/server.crt" + ``` + +##### TLS Certificate Key + +* Description: Path to the TLS certificate key to use for serving HTTPS. Usually has the `.key` extension. +* Flag: `tls.key=/srv/pilosa/certs/server.key` +* Env: `PILOSA_TLS_KEY=/srv/pilosa/certs/server.key` +* Config: + + ```toml + [tls] + key = "/srv/pilosa/certs/server.key" + ``` + +##### TLS Skip Verify + +* Description: Disables verification for checking TLS certificates. This configuration item is mainly useful for using self-signed certificates for a Pilosa cluster. Do not use in production since it makes man-in-the-middle attacks trivial. +* Flag: `tls.skip-verify` +* Env: `PILOSA_TLS_SKIP_VERIFY` +* Config: + + ```toml + [tls] + skip-verify = true + ``` + ### Example Cluster Configuration A three node cluster could be minimally configured as follows: @@ -245,3 +281,56 @@ A three node cluster could be minimally configured as follows: replicas = 1 type = "gossip" hosts = ["node0.pilosa.com:10101","node1.pilosa.com:10101","node2.pilosa.com:10101"] + + +### Example Cluster Configuration (HTTPS) + +The same cluster which uses HTTPS instead of HTTP can be configured as follows. Note that we explicitly specify `https` as the protocol in `bind` and `cluster.hosts` configuration: + +#### Node 0 + + data-dir = "/home/pilosa/data" + bind = "https://node0.pilosa.com:10101" + gossip-port = 12000 + gossip-seed = "node0.pilosa.com:12000" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://node0.pilosa.com:10101","https://node1.pilosa.com:10101","https://node2.pilosa.com:10101"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" + +#### Node 1 + + data-dir = "/home/pilosa/data" + bind = "https://node1.pilosa.com:10101" + gossip-port = 12000 + gossip-seed = "node0.pilosa.com:12000" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://node0.pilosa.com:10101","https://node1.pilosa.com:10101","https://node2.pilosa.com:10101"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" + +#### Node 2 + + data-dir = "/home/pilosa/data" + bind = "https://node2.pilosa.com:10101" + gossip-port = 12000 + gossip-seed = "node0.pilosa.com:12000" + + [cluster] + replicas = 1 + type = "gossip" + hosts = ["https://node0.pilosa.com:10101","https://node1.pilosa.com:10101","https://node2.pilosa.com:10101"] + + [tls] + certificate = "/home/pilosa/private/server.crt" + key = "/home/pilosa/private/server.key" diff --git a/executor.go b/executor.go index a620a1145..64a068149 100644 --- a/executor.go +++ b/executor.go @@ -21,7 +21,6 @@ import ( "fmt" "io/ioutil" "net/http" - "net/url" "sort" "time" @@ -44,6 +43,7 @@ type Executor struct { Holder *Holder // Local hostname & cluster configuration. + Scheme string Host string Cluster *Cluster @@ -55,9 +55,17 @@ type Executor struct { } // NewExecutor returns a new instance of Executor. -func NewExecutor() *Executor { +func NewExecutor(clientOptions *ClientOptions) *Executor { + if clientOptions == nil { + clientOptions = &ClientOptions{} + } + transport := &http.Transport{} + if clientOptions.TLS != nil { + transport.TLSClientConfig = clientOptions.TLS + } + client := &http.Client{Transport: transport} return &Executor{ - HTTPClient: http.DefaultClient, + HTTPClient: client, } } @@ -1375,11 +1383,9 @@ func (e *Executor) exec(ctx context.Context, node *Node, index string, q *pql.Qu } // Create HTTP request. - req, err := http.NewRequest("POST", (&url.URL{ - Scheme: "http", - Host: node.Host, - Path: fmt.Sprintf("/index/%s/query", index), - }).String(), bytes.NewReader(buf)) + u := nodePathToURL(node, fmt.Sprintf("/index/%s/query", index)) + u.Scheme = e.Scheme + req, err := http.NewRequest("POST", (&u).String(), bytes.NewReader(buf)) if err != nil { return nil, err } diff --git a/executor_test.go b/executor_test.go index 3ccac27ec..bf7550b25 100644 --- a/executor_test.go +++ b/executor_test.go @@ -75,7 +75,6 @@ func TestExecutor_Execute_Bitmap(t *testing.T) { } else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{3, SliceWidth + 1}) { t.Fatalf("unexpected bits: %+v", bits) } else if attrs := res[0].(*pilosa.Bitmap).Attrs; !reflect.DeepEqual(attrs, map[string]interface{}{}) { - fmt.Println("ATTRS", attrs) t.Fatalf("unexpected attrs: %s", spew.Sdump(attrs)) } }) @@ -871,6 +870,7 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) { // Create secondary server and update second cluster node. s := test.NewServer() defer s.Close() + c.Nodes[1].Scheme = "http" c.Nodes[1].Host = s.Host() // Mock secondary server's executor to verify arguments and return a bitmap. diff --git a/fragment.go b/fragment.go index 263966ad0..67f9ff000 100644 --- a/fragment.go +++ b/fragment.go @@ -1677,8 +1677,9 @@ func (h *blockHasher) WriteValue(v uint64) { type FragmentSyncer struct { Fragment *Fragment - Host string - Cluster *Cluster + Host string + Cluster *Cluster + ClientOptions *ClientOptions Closing <-chan struct{} } @@ -1713,7 +1714,7 @@ func (s *FragmentSyncer) SyncFragment() error { } // Retrieve remote blocks. - client, err := NewClient(node.Host) + client, err := NewClient(node.Host, s.ClientOptions) if err != nil { return err } @@ -1792,7 +1793,7 @@ func (s *FragmentSyncer) syncBlock(id int) error { return nil } - client, err := NewClient(node.Host) + client, err := NewClient(node.Host, s.ClientOptions) if err != nil { return err } diff --git a/gossip/gossip.go b/gossip/gossip.go index e2e43a00b..eade109a7 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -48,7 +48,7 @@ type GossipNodeSet struct { func (g *GossipNodeSet) Nodes() []*pilosa.Node { a := make([]*pilosa.Node, 0, g.memberlist.NumMembers()) for _, n := range g.memberlist.Members() { - a = append(a, &pilosa.Node{Host: n.Name}) + a = append(a, &pilosa.Node{Scheme: "gossip", Host: n.Name}) } return a } @@ -77,7 +77,7 @@ func (g *GossipNodeSet) Open() error { } // attach to gossip seed node - nodes := []*pilosa.Node{&pilosa.Node{Host: g.config.gossipSeed}} //TODO: support a list of seeds + nodes := []*pilosa.Node{&pilosa.Node{Scheme: "gossip", Host: g.config.gossipSeed}} //TODO: support a list of seeds err = g.joinWithRetry(pilosa.Nodes(nodes).Hosts()) if err != nil { return err diff --git a/handler.go b/handler.go index bae161515..016e3a348 100644 --- a/handler.go +++ b/handler.go @@ -56,8 +56,9 @@ type Handler struct { StatusHandler StatusHandler // Local hostname & cluster configuration. - Host string - Cluster *Cluster + URI *URI + Cluster *Cluster + ClientOptions *ClientOptions Router *mux.Router @@ -1167,8 +1168,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { } // Validate that this handler owns the slice. - if !h.Cluster.OwnsFragment(h.Host, req.Index, req.Slice) { - mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.Index, req.Slice) + if !h.Cluster.OwnsFragment(h.URI.HostPort(), req.Index, req.Slice) { + mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, req.Index, req.Slice) http.Error(w, mesg, http.StatusPreconditionFailed) return } @@ -1237,8 +1238,8 @@ func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request) } // Validate that this handler owns the slice. - if !h.Cluster.OwnsFragment(h.Host, req.Index, req.Slice) { - mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.Index, req.Slice) + if !h.Cluster.OwnsFragment(h.URI.HostPort(), req.Index, req.Slice) { + mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, req.Index, req.Slice) http.Error(w, mesg, http.StatusPreconditionFailed) return } @@ -1303,8 +1304,8 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) { } // Validate that this handler owns the slice. - if !h.Cluster.OwnsFragment(h.Host, index, slice) { - mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, index, slice) + if !h.Cluster.OwnsFragment(h.URI.HostPort(), index, slice) { + mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.URI, index, slice) http.Error(w, mesg, http.StatusPreconditionFailed) return } @@ -1491,16 +1492,21 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request) frameName := mux.Vars(r)["frame"] q := r.URL.Query() - host := q.Get("host") + hostStr := q.Get("host") // Validate query parameters. - if host == "" { + if hostStr == "" { http.Error(w, "host required", http.StatusBadRequest) return } + host, err := NewURIFromAddress(hostStr) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + } + // Create a client for the remote cluster. - client, err := NewClient(host) + client, err := NewClientFromURI(host, h.ClientOptions) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return @@ -1530,7 +1536,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request) // Loop over each slice and import it if this node owns it. for slice := uint64(0); slice <= maxSlices[indexName]; slice++ { // Ignore this slice if we don't own it. - if !h.Cluster.OwnsFragment(h.Host, indexName, slice) { + if !h.Cluster.OwnsFragment(h.URI.HostPort(), indexName, slice) { continue } diff --git a/handler_test.go b/handler_test.go index 1524a30b8..fcae35a28 100644 --- a/handler_test.go +++ b/handler_test.go @@ -1183,7 +1183,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) { h.ServeHTTP(w, r) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if w.Body.String() != `[{"host":"host2"},{"host":"host0"}]`+"\n" { + } else if w.Body.String() != `[{"scheme":"http","host":"host2"},{"scheme":"http","host":"host0"}]`+"\n" { t.Fatalf("unexpected body: %q", w.Body.String()) } } diff --git a/holder.go b/holder.go index 0b091045f..f3713faa2 100644 --- a/holder.go +++ b/holder.go @@ -427,8 +427,9 @@ func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.Lstd type HolderSyncer struct { Holder *Holder - Host string - Cluster *Cluster + URI *URI + Cluster *Cluster + ClientOptions *ClientOptions // Signals that the sync should stop. Closing <-chan struct{} @@ -477,7 +478,7 @@ func (s *HolderSyncer) SyncHolder() error { for slice := uint64(0); slice <= s.Holder.Index(di.Name).MaxSlice(); slice++ { // Ignore slices that this host doesn't own. - if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) { + if !s.Cluster.OwnsFragment(s.URI.HostPort(), di.Name, slice) { continue } @@ -513,8 +514,8 @@ func (s *HolderSyncer) syncIndex(index string) error { } // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { - client, err := NewClient(node.Host) + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.URI.HostPort()) { + client, err := NewClient(node.Host, s.ClientOptions) if err != nil { return err } @@ -558,8 +559,8 @@ func (s *HolderSyncer) syncFrame(index, name string) error { } // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { - client, err := NewClient(node.Host) + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.URI.HostPort()) { + client, err := NewClient(node.Host, s.ClientOptions) if err != nil { return err } @@ -612,10 +613,11 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err // Sync fragments together. fs := FragmentSyncer{ - Fragment: frag, - Host: s.Host, - Cluster: s.Cluster, - Closing: s.Closing, + Fragment: frag, + Host: s.URI.HostPort(), + Cluster: s.Cluster, + Closing: s.Closing, + ClientOptions: s.ClientOptions, } if err := fs.SyncFragment(); err != nil { return err diff --git a/holder_test.go b/holder_test.go index b3947abb8..104bd1bb0 100644 --- a/holder_test.go +++ b/holder_test.go @@ -314,8 +314,9 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { defer s.Close() s.Handler.Holder = hldr1.Holder s.Handler.Executor.ExecuteFn = func(ctx context.Context, index string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor() + e := pilosa.NewExecutor(nil) e.Holder = hldr1.Holder + e.Scheme = cluster.Nodes[1].Scheme e.Host = cluster.Nodes[1].Host e.Cluster = cluster return e.Execute(ctx, index, query, slices, opt) @@ -376,9 +377,13 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { hldr0.Index("y").SetRemoteMaxSlice(3) // Set up syncer. + uri, err := cluster.Nodes[0].URI() + if err != nil { + t.Fatal(err) + } syncer := pilosa.HolderSyncer{ Holder: hldr0.Holder, - Host: cluster.Nodes[0].Host, + URI: uri, Cluster: cluster, } diff --git a/pilosa.go b/pilosa.go index dfa8a615f..aa1378a36 100644 --- a/pilosa.go +++ b/pilosa.go @@ -172,15 +172,6 @@ func ContainsSubstring(a string, list []string) bool { return false } -// NormalizeAddress converts addr into a valid "IP4:port" string. -func NormalizeAddress(addr string) (string, error) { - host, port, err := hostPortWithDefaults(addr) - if err != nil { - return "", err - } - return net.JoinHostPort(HostToIP(host), port), nil -} - // HostToIP converts host to an IP4 address based on net.LookupIP(). func HostToIP(host string) string { // if host is not an IP addr, check net.LookupIP() @@ -201,39 +192,10 @@ func HostToIP(host string) string { // AddressWithDefaults converts addr into a valid address, // using defaults when necessary. -func AddressWithDefaults(addr string) (string, error) { - host, port, err := hostPortWithDefaults(addr) - if err != nil { - return "", err +func AddressWithDefaults(addr string) (*URI, error) { + if addr == "" { + return DefaultURI(), nil + } else { + return NewURIFromAddress(addr) } - return net.JoinHostPort(host, port), nil -} - -// hostPortWithDefaults returns the host and port portions of addr -// using defaults when necessary. -func hostPortWithDefaults(addr string) (host, port string, err error) { - // check for a colon between host and port - if !hasPort(addr) { - addr += ":" - } - - // break into host, port - host, port, err = net.SplitHostPort(addr) - if err != nil { - return host, port, err - } - - // use defaults when not provided - if host == "" { - host = DefaultHost - } - if port == "" { - port = DefaultPort - } - - return host, port, nil -} - -func hasPort(s string) bool { - return strings.LastIndex(s, ":") > strings.LastIndex(s, "]") } diff --git a/pilosa_test.go b/pilosa_test.go index 8034451eb..b4d9cc2e2 100644 --- a/pilosa_test.go +++ b/pilosa_test.go @@ -1,7 +1,6 @@ package pilosa_test import ( - "reflect" "strings" "testing" @@ -81,40 +80,6 @@ func TestContainsSubstring(t *testing.T) { } } -func TestNormalizeAddress(t *testing.T) { - tests := []struct { - addr string - expected string - err string - }{ - {addr: "", expected: "127.0.0.1:10101"}, - {addr: ":", expected: "127.0.0.1:10101"}, - {addr: "localhost", expected: "127.0.0.1:10101"}, - {addr: "localhost:", expected: "127.0.0.1:10101"}, - {addr: "127.0.0.1:10101", expected: "127.0.0.1:10101"}, - {addr: "127.0.0.1:", expected: "127.0.0.1:10101"}, - {addr: ":10101", expected: "127.0.0.1:10101"}, - {addr: ":55555", expected: "127.0.0.1:55555"}, - {addr: "1.2.3.4", expected: "1.2.3.4:10101"}, - {addr: "1.2.3.4:", expected: "1.2.3.4:10101"}, - {addr: "1.2.3.4:55555", expected: "1.2.3.4:55555"}, - // The following tests check the error conditions. - {addr: "[invalid][addr]:port", err: "missing port in address"}, - } - for _, test := range tests { - actual, err := pilosa.NormalizeAddress(test.addr) - if err != nil { - if !strings.Contains(err.Error(), test.err) { - t.Errorf("expected error: %v, but got: %v", test.err, err) - } - } else { - if !reflect.DeepEqual(actual, test.expected) { - t.Errorf("expected: %v, but got: %v", test.expected, actual) - } - } - } -} - func TestAddressWithDefaults(t *testing.T) { tests := []struct { addr string @@ -133,7 +98,7 @@ func TestAddressWithDefaults(t *testing.T) { {addr: "1.2.3.4:", expected: "1.2.3.4:10101"}, {addr: "1.2.3.4:55555", expected: "1.2.3.4:55555"}, // The following tests check the error conditions. - {addr: "[invalid][addr]:port", err: "missing port in address"}, + {addr: "[invalid][addr]:port", err: "invalid address"}, } for _, test := range tests { actual, err := pilosa.AddressWithDefaults(test.addr) @@ -142,7 +107,7 @@ func TestAddressWithDefaults(t *testing.T) { t.Errorf("expected error: %v, but got: %v", test.err, err) } } else { - if !reflect.DeepEqual(actual, test.expected) { + if actual.HostPort() != test.expected { t.Errorf("expected: %v, but got: %v", test.expected, actual) } } diff --git a/server.go b/server.go index b785cdda1..773cf9025 100644 --- a/server.go +++ b/server.go @@ -15,6 +15,7 @@ package pilosa import ( + "crypto/tls" "errors" "fmt" "io" @@ -59,7 +60,7 @@ type Server struct { // Cluster configuration. // Host is replaced with actual host after opening if port is ":0". Network string - Host string + URI *URI Cluster *Cluster // Background monitoring intervals. @@ -67,10 +68,15 @@ type Server struct { PollingInterval time.Duration MetricInterval time.Duration + // TLS configuration + TLS *tls.Config + // Misc options. MaxWritesPerRequest int LogOutput io.Writer + + defaultClient *http.Client } // NewServer returns a new instance of Server. @@ -99,27 +105,38 @@ func NewServer() *Server { // Open opens and initializes the server. func (s *Server) Open() error { - // Require a port in the hostname. - host, port, err := net.SplitHostPort(s.Host) - if err != nil { - return err - } else if port == "" { - port = DefaultPort + var ln net.Listener + var err error + + // If bind URI has the https scheme, enable TLS + if s.URI.Scheme() == "https" && s.TLS != nil { + ln, err = tls.Listen("tcp", s.URI.HostPort(), s.TLS) + if err != nil { + return err + } + } else if s.URI.Scheme() == "http" { + // Open HTTP listener to determine port (if specified as :0). + ln, err = net.Listen(s.Network, s.URI.HostPort()) + if err != nil { + return fmt.Errorf("net.Listen: %v", err) + } + } else { + return fmt.Errorf("unsupported scheme: %s", s.URI.Scheme()) } - // Open HTTP listener to determine port (if specified as :0). - ln, err := net.Listen(s.Network, ":"+port) - if err != nil { - return fmt.Errorf("net.Listen: %v", err) - } s.ln = ln - // Determine hostname based on listening port. - s.Host = net.JoinHostPort(host, strconv.Itoa(s.ln.Addr().(*net.TCPAddr).Port)) + if s.URI.Port() == 0 { + // If the port is 0, it is set automatically. + // Find out automatically set port and update the host. + s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port)) + } // Create local node if no cluster is specified. if len(s.Cluster.Nodes) == 0 { - s.Cluster.Nodes = []*Node{{Host: s.Host}} + s.Cluster.Nodes = []*Node{ + {Scheme: s.URI.Scheme(), Host: s.URI.HostPort()}, + } } for i, n := range s.Cluster.Nodes { @@ -143,17 +160,21 @@ func (s *Server) Open() error { return fmt.Errorf("opening NodeSet: %v", err) } + // Create default HTTP client + s.createDefaultClient() + // Create executor for executing queries. - e := NewExecutor() + e := NewExecutor(&ClientOptions{TLS: s.TLS}) e.Holder = s.Holder - e.Host = s.Host + e.Scheme = s.URI.Scheme() + e.Host = s.URI.HostPort() e.Cluster = s.Cluster e.MaxWritesPerRequest = s.MaxWritesPerRequest // Initialize HTTP handler. s.Handler.Broadcaster = s.Broadcaster s.Handler.StatusHandler = s - s.Handler.Host = s.Host + s.Handler.URI = s.URI s.Handler.Cluster = s.Cluster s.Handler.Executor = e s.Handler.LogOutput = s.LogOutput @@ -225,9 +246,10 @@ func (s *Server) monitorAntiEntropy() { // Initialize syncer with local holder and remote client. var syncer HolderSyncer syncer.Holder = s.Holder - syncer.Host = s.Host + syncer.URI = s.URI syncer.Cluster = s.Cluster syncer.Closing = s.closing + syncer.ClientOptions = &ClientOptions{TLS: s.TLS} // Sync holders. if err := syncer.SyncHolder(); err != nil { @@ -261,8 +283,8 @@ func (s *Server) monitorMaxSlices() { oldmaxslices := s.Holder.MaxSlices() for _, node := range s.Cluster.Nodes { - if s.Host != node.Host { - maxSlices, _ := checkMaxSlices(node.Host) + if s.URI.HostPort() != node.Host { + maxSlices, _ := s.checkMaxSlices(node.Scheme, node.Host) for index, newmax := range maxSlices { // if we don't know about an index locally, log an error because // indexes should be created and synced prior to slice creation @@ -364,14 +386,14 @@ func (s *Server) LocalStatus() (proto.Message, error) { } ns := internal.NodeStatus{ - Host: s.Host, + Host: s.URI.HostPort(), State: NodeStateUp, Indexes: EncodeIndexes(s.Holder.Indexes()), } // Append Slice list per this Node's indexes for _, index := range ns.Indexes { - index.Slices = s.Cluster.OwnsSlices(index.Name, index.MaxSlice, s.Host) + index.Slices = s.Cluster.OwnsSlices(index.Name, index.MaxSlice, s.URI.HostPort()) } return &ns, nil @@ -384,7 +406,7 @@ func (s *Server) ClusterStatus() (proto.Message, error) { if err != nil { return nil, err } - node := s.Cluster.NodeByHost(s.Host) + node := s.Cluster.NodeByHost(s.URI.HostPort()) node.SetStatus(ns.(*internal.NodeStatus)) // Update NodeState for all nodes. @@ -394,7 +416,7 @@ func (s *Server) ClusterStatus() (proto.Message, error) { // the local node as UP. // TODO: we should be able to remove this check if/when cluster.Nodes and // cluster.NodeSet are unified. - if host == s.Host { + if host == s.URI.HostPort() { nodeState = NodeStateUp } node := s.Cluster.NodeByHost(host) @@ -441,11 +463,11 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { return nil } -func checkMaxSlices(hostport string) (map[string]uint64, error) { +func (s *Server) checkMaxSlices(scheme string, hostPort string) (map[string]uint64, error) { // Create HTTP request. req, err := http.NewRequest("GET", (&url.URL{ - Scheme: "http", - Host: hostport, + Scheme: scheme, + Host: hostPort, Path: "/slices/max", }).String(), nil) @@ -458,8 +480,7 @@ func checkMaxSlices(hostport string) (map[string]uint64, error) { req.Header.Set("Content-Type", "application/x-protobuf") req.Header.Set("User-Agent", "pilosa/"+Version) - // Send request to remote node. - resp, err := http.DefaultClient.Do(req) + resp, err := s.defaultClient.Do(req) if err != nil { return nil, err } @@ -529,6 +550,14 @@ func (s *Server) monitorRuntime() { } } +func (s *Server) createDefaultClient() { + transport := &http.Transport{} + if s.TLS != nil { + transport.TLSClientConfig = s.TLS + } + s.defaultClient = &http.Client{Transport: transport} +} + // CountOpenFiles on opperating systems that support lsof func CountOpenFiles() int { count := 0 diff --git a/server/server.go b/server/server.go index 157fc69f3..d1063f306 100644 --- a/server/server.go +++ b/server/server.go @@ -23,13 +23,13 @@ import ( "fmt" "io" "math/rand" - "net" "os" "path/filepath" "strconv" "strings" "time" + "crypto/tls" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/statsd" @@ -100,7 +100,7 @@ func (m *Command) Run(args ...string) (err error) { return fmt.Errorf("server.Open: %v", err) } - m.Server.Logger().Printf("Listening as http://%s\n", m.Server.Host) + m.Server.Logger().Printf("Listening as %s\n", m.Server.URI.Normalize()) return nil } @@ -111,11 +111,24 @@ func (m *Command) SetupServer() error { return err } + uri, err := pilosa.AddressWithDefaults(m.Config.Bind) + if err != nil { + return err + } + m.Server.URI = uri + cluster := pilosa.NewCluster() cluster.ReplicaN = m.Config.Cluster.ReplicaN - for _, hostport := range m.Config.Cluster.Hosts { - cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: hostport}) + for _, address := range m.Config.Cluster.Hosts { + uri, err := pilosa.NewURIFromAddress(address) + if err != nil { + return err + } + cluster.Nodes = append(cluster.Nodes, &pilosa.Node{ + Scheme: uri.Scheme(), + Host: uri.HostPort(), + }) } m.Server.Cluster = cluster @@ -139,11 +152,24 @@ func (m *Command) SetupServer() error { // Copy configuration flags. m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest - bindWithDefaults, err := pilosa.AddressWithDefaults(m.Config.Bind) - if err != nil { - return err + // Setup TLS + if uri.Scheme() == "https" { + if m.Config.TLS.CertificatePath == "" { + return errors.New("certificate path is required for TLS sockets") + } + if m.Config.TLS.CertificateKeyPath == "" { + return errors.New("certificate key path is required for TLS sockets") + } + cert, err := tls.LoadX509KeyPair(m.Config.TLS.CertificatePath, m.Config.TLS.CertificateKeyPath) + if err != nil { + return err + } + m.Server.TLS = &tls.Config{ + Certificates: []tls.Certificate{cert}, + InsecureSkipVerify: m.Config.TLS.SkipVerify, + } + m.Server.Handler.ClientOptions = &pilosa.ClientOptions{TLS: m.Server.TLS} } - m.Server.Host = bindWithDefaults // Set internal port (string). gossipPortStr := pilosa.DefaultGossipPort @@ -163,11 +189,8 @@ func (m *Command) SetupServer() error { } // get the host portion of addr to use for binding - gossipHost, _, err := net.SplitHostPort(bindWithDefaults) - if err != nil { - gossipHost = m.Config.Bind - } - gossipNodeSet := gossip.NewGossipNodeSet(bindWithDefaults, gossipHost, gossipPort, gossipSeed, m.Server) + gossipHost := uri.Host() + gossipNodeSet := gossip.NewGossipNodeSet(uri.HostPort(), gossipHost, gossipPort, gossipSeed, m.Server) m.Server.Cluster.NodeSet = gossipNodeSet m.Server.Broadcaster = gossipNodeSet m.Server.BroadcastReceiver = gossipNodeSet diff --git a/server/server_test.go b/server/server_test.go index c1a90121d..a5de5d68c 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -53,7 +53,7 @@ func TestMain_Set_Quick(t *testing.T) { defer m.Close() // Create client. - client, err := pilosa.NewClient(m.Server.Host) + client, err := pilosa.NewClient(m.Server.URI.HostPort(), nil) if err != nil { t.Fatal(err) } @@ -288,8 +288,8 @@ func TestMain_FrameRestore(t *testing.T) { // Update cluster config. m0.Server.Cluster.Nodes = []*pilosa.Node{ - {Host: m0.Server.Host}, - {Host: m1.Server.Host}, + {Scheme: "http", Host: m0.Server.URI.HostPort()}, + {Scheme: "http", Host: m1.Server.URI.HostPort()}, } m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes @@ -326,14 +326,14 @@ func TestMain_FrameRestore(t *testing.T) { defer m2.Close() // Import from first cluster. - client, err := pilosa.NewClient(m2.Server.Host) + client, err := pilosa.NewClient(m2.Server.URI.HostPort(), nil) if err != nil { t.Fatal(err) } else if err := m2.Client().CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) } else if err := m2.Client().CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil { t.Fatal(err) - } else if err := client.RestoreFrame(context.Background(), m0.Server.Host, "i", "f"); err != nil { + } else if err := client.RestoreFrame(context.Background(), m0.Server.URI.HostPort(), "i", "f"); err != nil { t.Fatal(err) } @@ -427,17 +427,17 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Update cluster config m0.Server.Cluster.Nodes = []*pilosa.Node{ - {Host: m0.Server.Host}, - {Host: m1.Server.Host}, + {Host: m0.Server.URI.HostPort()}, + {Host: m1.Server.URI.HostPort()}, } m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes // Configure node0 // get the host portion of addr to use for binding - gossipHost, _, err := net.SplitHostPort(m0.Server.Host) + gossipHost, _, err := net.SplitHostPort(m0.Server.URI.HostPort()) if err != nil { - gossipHost = m0.Server.Host + gossipHost = m0.Server.URI.HostPort() } gossipPort, err := strconv.Atoi(freePorts[0]) if err != nil { @@ -445,7 +445,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { } gossipSeed := gossipHost + ":" + freePorts[0] - gossipNodeSet0 := gossip.NewGossipNodeSet(m0.Server.Host, gossipHost, gossipPort, gossipSeed, m0.Server) + gossipNodeSet0 := gossip.NewGossipNodeSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server) m0.Server.Cluster.NodeSet = gossipNodeSet0 m0.Server.Broadcaster = gossipNodeSet0 m0.Server.Handler.Broadcaster = m0.Server.Broadcaster @@ -463,16 +463,16 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Configure node1 // get the host portion of addr to use for binding - gossipHost, _, err = net.SplitHostPort(m1.Server.Host) + gossipHost, _, err = net.SplitHostPort(m1.Server.URI.HostPort()) if err != nil { - gossipHost = m1.Server.Host + gossipHost = m1.Server.URI.HostPort() } gossipPort, err = strconv.Atoi(freePorts[1]) if err != nil { t.Fatal(err) } - gossipNodeSet1 := gossip.NewGossipNodeSet(m1.Server.Host, gossipHost, gossipPort, gossipSeed, m1.Server) + gossipNodeSet1 := gossip.NewGossipNodeSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server) m1.Server.Cluster.NodeSet = gossipNodeSet1 m1.Server.Broadcaster = gossipNodeSet1 m1.Server.Handler.Broadcaster = m1.Server.Broadcaster @@ -697,7 +697,7 @@ func (m *Main) URL() string { return "http://" + m.Server.Addr().String() } // Client returns a client to connect to the program. func (m *Main) Client() *pilosa.Client { - client, err := pilosa.NewClient(m.Server.Host) + client, err := pilosa.NewClient(m.Server.URI.HostPort(), nil) if err != nil { panic(err) } diff --git a/test/client.go b/test/client.go index 1afb8df1b..10ef6368e 100644 --- a/test/client.go +++ b/test/client.go @@ -11,7 +11,7 @@ type Client struct { // MustNewClient returns a new instance of Client. Panic on error. func MustNewClient(host string) *Client { - c, err := pilosa.NewClient(host) + c, err := pilosa.NewClient(host, nil) if err != nil { panic(err) } diff --git a/test/cluster.go b/test/cluster.go index 557aff559..c45b1c989 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -14,6 +14,7 @@ func NewCluster(n int) *pilosa.Cluster { for i := 0; i < n; i++ { c.Nodes = append(c.Nodes, &pilosa.Node{ + Scheme: "http", Host: fmt.Sprintf("host%d", i), }) } diff --git a/test/executor.go b/test/executor.go index af5a248f6..73445a1cd 100644 --- a/test/executor.go +++ b/test/executor.go @@ -15,9 +15,10 @@ type Executor struct { // NewExecutor returns a new instance of Executor. // The executor always matches the hostname of the first cluster node. func NewExecutor(holder *pilosa.Holder, cluster *pilosa.Cluster) *Executor { - e := &Executor{Executor: pilosa.NewExecutor()} + e := &Executor{Executor: pilosa.NewExecutor(nil)} e.Holder = holder e.Cluster = cluster + e.Scheme = cluster.Nodes[0].Scheme e.Host = cluster.Nodes[0].Host return e } diff --git a/test/handler.go b/test/handler.go index 70ac9e4e0..aa5a56acb 100644 --- a/test/handler.go +++ b/test/handler.go @@ -62,7 +62,11 @@ func NewServer() *Server { s.Server = httptest.NewServer(s.Handler.Handler) // Update handler to use hostname. - s.Handler.Host = s.Host() + uri, err := pilosa.NewURIFromAddress(s.Host()) + if err != nil { + panic(err) + } + s.Handler.URI = uri // Handler test messages can no-op. s.Handler.Broadcaster = pilosa.NopBroadcaster @@ -81,14 +85,14 @@ func (s *Server) LocalStatus() (proto.Message, error) { } ns := internal.NodeStatus{ - Host: s.Handler.Handler.Host, + Host: s.Handler.Handler.URI.HostPort(), State: pilosa.NodeStateUp, Indexes: pilosa.EncodeIndexes(s.Handler.Holder.Indexes()), } // Append Slice list per this Node's indexes for _, index := range ns.Indexes { - index.Slices = s.Handler.Cluster.OwnsSlices(index.Name, index.MaxSlice, s.Handler.Host) + index.Slices = s.Handler.Cluster.OwnsSlices(index.Name, index.MaxSlice, s.Handler.URI.HostPort()) } return &ns, nil @@ -107,6 +111,14 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error { return nil } // Host returns the hostname of the running server. func (s *Server) Host() string { return MustParseURLHost(s.URL) } +func (s *Server) HostURI() *pilosa.URI { + uri, err := pilosa.NewURIFromAddress(s.URL) + if err != nil { + panic(err) + } + return uri +} + // MustParseURLHost parses rawurl and returns the hostname. Panic on error. func MustParseURLHost(rawurl string) string { u, err := url.Parse(rawurl) diff --git a/uri.go b/uri.go new file mode 100644 index 000000000..de6e4bae6 --- /dev/null +++ b/uri.go @@ -0,0 +1,190 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa + +import ( + "errors" + "fmt" + "regexp" + "strconv" + "strings" +) + +var schemeRegexp = regexp.MustCompile("^[+a-z]+$") +var hostRegexp = regexp.MustCompile("^[0-9a-z.-]+$|^\\[[:0-9a-fA-F]+\\]$") +var addressRegexp = regexp.MustCompile("^(([+a-z]+):\\/\\/)?([0-9a-z.-]+|\\[[:0-9a-fA-F]+\\])?(:([0-9]+))?$") + +// URI represents a Pilosa URI. +// A Pilosa URI consists of three parts: +// 1) Scheme: Protocol of the URI. Default: http. +// 2) Host: Hostname or IP URI. Default: localhost. IPv6 addresses should be written in brackets, e.g., `[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]`. +// 3) Port: Port of the URI. Default: 10101. +// +// All parts of the URI are optional. The following are equivalent: +// http://localhost:10101 +// http://localhost +// http://:10101 +// localhost:10101 +// localhost +// :10101 +type URI struct { + scheme string + host string + port uint16 +} + +// DefaultURI creates and returns the default URI. +func DefaultURI() *URI { + return &URI{ + scheme: "http", + host: "localhost", + port: 10101, + } +} + +// NewURIFromHostPort returns a URI with specified host and port. +func NewURIFromHostPort(host string, port uint16) (*URI, error) { + uri := DefaultURI() + err := uri.SetHost(host) + if err != nil { + return nil, err + } + uri.SetPort(port) + return uri, nil +} + +// NewURIFromAddress parses the passed address and returns a URI. +func NewURIFromAddress(address string) (*URI, error) { + uri, err := parseAddress(address) + if err != nil { + return nil, err + } + return uri, err +} + +// Scheme returns the scheme of this URI. +func (u *URI) Scheme() string { + return u.scheme +} + +// SetScheme sets the scheme of this URI. +func (u *URI) SetScheme(scheme string) error { + m := schemeRegexp.FindStringSubmatch(scheme) + if m == nil { + return errors.New("invalid scheme") + } + u.scheme = scheme + return nil +} + +// Host returns the host of this URI. +func (u *URI) Host() string { + return u.host +} + +// SetHost sets the host of this URI. +func (u *URI) SetHost(host string) error { + m := hostRegexp.FindStringSubmatch(host) + if m == nil { + return errors.New("invalid host") + } + u.host = host + return nil +} + +// Port returns the port of this URI. +func (u *URI) Port() uint16 { + return u.port +} + +// SetPort sets the port of this URI. +func (u *URI) SetPort(port uint16) { + u.port = port +} + +// HostPort returns `Host:Port` +func (u *URI) HostPort() string { + // XXX: The following is just to make TestHandler_Status; remove it + if u == nil { + return "" + } + s := fmt.Sprintf("%s:%d", u.host, u.port) + return s +} + +// Normalize returns the address in a form usable by a HTTP client. +func (u *URI) Normalize() string { + scheme := u.scheme + index := strings.Index(scheme, "+") + if index >= 0 { + scheme = scheme[:index] + } + return fmt.Sprintf("%s://%s:%d", scheme, u.host, u.port) +} + +// Equals returns true if the checked URI is equivalent to this URI. +func (u URI) Equals(other *URI) bool { + if other == nil { + return false + } + return u.scheme == other.scheme && + u.host == other.host && + u.port == other.port +} + +// The following methods are required to implement pflag Value interface. + +// Set sets the time quantum value. +func (u *URI) Set(value string) error { + uri, err := NewURIFromAddress(value) + if err != nil { + return err + } + *u = *uri + return nil +} + +// Type returns the type of a time quantum value. +func (u URI) Type() string { + return "URI" +} + +func parseAddress(address string) (uri *URI, err error) { + m := addressRegexp.FindStringSubmatch(address) + if m == nil { + return nil, errors.New("invalid address") + } + scheme := "http" + if m[2] != "" { + scheme = m[2] + } + host := "localhost" + if m[3] != "" { + host = m[3] + } + var port = 10101 + if m[5] != "" { + port, err = strconv.Atoi(m[5]) + if err != nil { + return nil, errors.New("converting port string to int") + } + } + uri = &URI{ + scheme: scheme, + host: host, + port: uint16(port), + } + return uri, nil +} diff --git a/uri_test.go b/uri_test.go new file mode 100644 index 000000000..33ccd35f9 --- /dev/null +++ b/uri_test.go @@ -0,0 +1,193 @@ +// Copyright 2017 Pilosa Corp. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions +// are met: +// +// 1. Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its +// contributors may be used to endorse or promote products derived +// from this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND +// CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, +// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR +// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, +// BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH +// DAMAGE. + +package pilosa + +import "testing" + +func TestDefaultURI(t *testing.T) { + uri := DefaultURI() + compare(t, uri, "http", "localhost", 10101) +} + +func TestURIWithHostPort(t *testing.T) { + uri, err := NewURIFromHostPort("index1.pilosa.com", 3333) + if err != nil { + t.Fatal(err) + } + compare(t, uri, "http", "index1.pilosa.com", 3333) +} + +func TestURIWithInvalidHostPort(t *testing.T) { + _, err := NewURIFromHostPort("index?.pilosa.com", 3333) + if err == nil { + t.Fatalf("should have failed") + } +} + +func TestNewURIFromAddress(t *testing.T) { + for _, item := range validFixture() { + uri, err := NewURIFromAddress(item.address) + if err != nil { + t.Fatalf("Can't parse address: %s, %s", item.address, err) + } + compare(t, uri, item.scheme, item.host, item.port) + } +} + +func TestNewURIFromAddressInvalidAddress(t *testing.T) { + for _, addr := range invalidFixture() { + _, err := NewURIFromAddress(addr) + if err == nil { + t.Fatalf("Invalid address should return an error: %s", addr) + } + } +} + +func TestNormalizedAddress(t *testing.T) { + uri, err := NewURIFromAddress("http+protobuf://big-data.pilosa.com:6888") + if err != nil { + t.Fatalf("Can't parse address") + } + if uri.Normalize() != "http://big-data.pilosa.com:6888" { + t.Fatalf("Normalized address is not normal") + } +} + +func TestEquals(t *testing.T) { + uri1 := DefaultURI() + if uri1.Equals(nil) { + t.Fatalf("URI should not be equal to nil") + } + if !uri1.Equals(DefaultURI()) { + t.Fatalf("URI should be equal to another URI with the same scheme, host and port") + } +} + +func TestSetScheme(t *testing.T) { + uri := DefaultURI() + target := "fun" + err := uri.SetScheme(target) + if err != nil { + t.Fatal(err) + } + if uri.Scheme() != target { + t.Fatalf("%s != %s", uri.Scheme(), target) + } +} + +func TestSetHost(t *testing.T) { + uri := DefaultURI() + target := "10.20.30.40" + err := uri.SetHost(target) + if err != nil { + t.Fatal(err) + } + if uri.Host() != target { + t.Fatalf("%s != %s", uri.host, target) + } +} + +func TestSetPort(t *testing.T) { + uri := DefaultURI() + target := uint16(9999) + uri.SetPort(target) + if uri.Port() != target { + t.Fatalf("%d != %d", uri.port, target) + } +} + +func TestSetInvalidScheme(t *testing.T) { + uri := DefaultURI() + err := uri.SetScheme("?invalid") + if err == nil { + t.Fatalf("Should have failed") + } +} + +func TestSetInvalidHost(t *testing.T) { + uri := DefaultURI() + err := uri.SetHost("index?.pilosa.com") + if err == nil { + t.Fatalf("Should have failed") + } +} + +func TestHostPort(t *testing.T) { + uri, err := NewURIFromHostPort("i.pilosa.com", 15001) + if err != nil { + t.Fatal(err) + } + target := "i.pilosa.com:15001" + if uri.HostPort() != target { + t.Fatalf("%s != %s", uri.HostPort(), target) + } +} + +func compare(t *testing.T, uri *URI, scheme string, host string, port uint16) { + if uri.Scheme() != scheme { + t.Fatalf("Scheme does not match: %s != %s", uri.scheme, scheme) + } + if uri.Host() != host { + t.Fatalf("Host does not match: %s != %s", uri.host, host) + } + if uri.Port() != port { + t.Fatalf("Port does not match: %d != %d", uri.port, port) + } +} + +type uriItem struct { + address string + scheme string + host string + port uint16 +} + +func validFixture() []uriItem { + var test = []uriItem{ + {"http+protobuf://index1.pilosa.com:3333", "http+protobuf", "index1.pilosa.com", 3333}, + {"index1.pilosa.com:3333", "http", "index1.pilosa.com", 3333}, + {"https://index1.pilosa.com", "https", "index1.pilosa.com", 10101}, + {"index1.pilosa.com", "http", "index1.pilosa.com", 10101}, + {"https://:3333", "https", "localhost", 3333}, + {":3333", "http", "localhost", 3333}, + {"[::1]", "http", "[::1]", 10101}, + {"[::1]:3333", "http", "[::1]", 3333}, + {"[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]:3333", "http", "[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]", 3333}, + {"https://[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]:3333", "https", "[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]", 3333}, + } + return test +} + +func invalidFixture() []string { + return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80"} +}