From 1bedfd6585d15840ad1eea31a435c88fea01310e Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Wed, 4 Oct 2017 08:20:00 +0300 Subject: [PATCH] use URI instead of host --- client.go | 188 ++++++++++++++++++------------------------ client_test.go | 16 ++-- cluster.go | 3 +- config.go | 2 +- executor.go | 2 +- handler.go | 21 +++-- holder.go | 10 +-- pilosa_test.go | 40 +-------- server.go | 44 +++++----- server/server.go | 7 +- server/server_test.go | 28 +++---- test/handler.go | 18 +++- uri.go | 9 +- 13 files changed, 176 insertions(+), 212 deletions(-) diff --git a/client.go b/client.go index 3d94599a3..ba03b8b0b 100644 --- a/client.go +++ b/client.go @@ -37,7 +37,7 @@ import ( // Client represents a client to the Pilosa cluster. type Client struct { - host string + host *URI // The client to use for HTTP communication. // Defaults to the http.DefaultClient. @@ -50,14 +50,26 @@ func NewClient(host string) (*Client, error) { return nil, ErrHostRequired } + uri, err := NewURIFromAddress(host) + if err != nil { + return nil, err + } + + return NewClientFromURI(uri) +} + +func NewClientFromURI(uri *URI) (*Client, error) { + if uri == nil { + return nil, ErrHostRequired + } return &Client{ - host: host, + host: uri, HTTPClient: http.DefaultClient, }, 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 +84,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 +117,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 +154,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 +191,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 +237,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 +274,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 +370,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 +458,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 +528,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 +668,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 +751,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 +797,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 +833,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 +864,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 +900,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 +953,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 +990,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 +1030,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 +1199,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..9fedb01ab 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.Host = 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 @@ -207,7 +207,7 @@ func TestClient_Import(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -258,7 +258,7 @@ func TestClient_ImportInverseEnabled(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -307,7 +307,7 @@ func TestClient_ImportValue(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -345,7 +345,7 @@ func TestClient_BackupRestore(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -410,7 +410,7 @@ func TestClient_BackupInverseView(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -447,7 +447,7 @@ func TestClient_BackupInvalidView(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = s.HostURI() s.Handler.Cluster = test.NewCluster(1) s.Handler.Cluster.Nodes[0].Host = s.Host() s.Handler.Holder = hldr.Holder @@ -476,7 +476,7 @@ func TestClient_FragmentBlocks(t *testing.T) { s := test.NewServer() defer s.Close() - s.Handler.Host = s.Host() + s.Handler.Host = 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..397c6f309 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"` } diff --git a/config.go b/config.go index 18ac93074..b2af5e5d4 100644 --- a/config.go +++ b/config.go @@ -120,7 +120,7 @@ func (c *Config) Validate() error { if err != nil { return err } - if !foundItem(c.Cluster.Hosts, bindWithDefaults.ListenAddress()) { + if !foundItem(c.Cluster.Hosts, bindWithDefaults.HostPort()) { return ErrConfigHostsMissing } } diff --git a/executor.go b/executor.go index ae960a5e1..5380b3f47 100644 --- a/executor.go +++ b/executor.go @@ -1338,7 +1338,7 @@ 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", + Scheme: node.Scheme, Host: node.Host, Path: fmt.Sprintf("/index/%s/query", index), }).String(), bytes.NewReader(buf)) diff --git a/handler.go b/handler.go index d8463862a..293a05b93 100644 --- a/handler.go +++ b/handler.go @@ -56,7 +56,7 @@ type Handler struct { StatusHandler StatusHandler // Local hostname & cluster configuration. - Host string + Host *URI Cluster *Cluster Router *mux.Router @@ -1166,7 +1166,7 @@ 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) { + if !h.Cluster.OwnsFragment(h.Host.HostPort(), req.Index, req.Slice) { mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.Index, req.Slice) http.Error(w, mesg, http.StatusPreconditionFailed) return @@ -1236,7 +1236,7 @@ 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) { + if !h.Cluster.OwnsFragment(h.Host.HostPort(), req.Index, req.Slice) { mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.Index, req.Slice) http.Error(w, mesg, http.StatusPreconditionFailed) return @@ -1302,7 +1302,7 @@ 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) { + if !h.Cluster.OwnsFragment(h.Host.HostPort(), index, slice) { mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, index, slice) http.Error(w, mesg, http.StatusPreconditionFailed) return @@ -1490,16 +1490,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) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return @@ -1529,7 +1534,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.Host.HostPort(), indexName, slice) { continue } diff --git a/holder.go b/holder.go index 0cf5f6901..e6655a336 100644 --- a/holder.go +++ b/holder.go @@ -417,7 +417,7 @@ func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.Lstd type HolderSyncer struct { Holder *Holder - Host string + Host *URI Cluster *Cluster // Signals that the sync should stop. @@ -467,7 +467,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.Host.HostPort(), di.Name, slice) { continue } @@ -503,7 +503,7 @@ func (s *HolderSyncer) syncIndex(index string) error { } // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host.HostPort()) { client, err := NewClient(node.Host) if err != nil { return err @@ -548,7 +548,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error { } // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host.HostPort()) { client, err := NewClient(node.Host) if err != nil { return err @@ -603,7 +603,7 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err // Sync fragments together. fs := FragmentSyncer{ Fragment: frag, - Host: s.Host, + Host: s.Host.HostPort(), Cluster: s.Cluster, Closing: s.Closing, } diff --git a/pilosa_test.go b/pilosa_test.go index f6ad8154d..b4d9cc2e2 100644 --- a/pilosa_test.go +++ b/pilosa_test.go @@ -1,7 +1,6 @@ package pilosa_test import ( - "reflect" "strings" "testing" @@ -81,41 +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 { - uri, err := pilosa.NewURIFromAddress(test.addr) - if err != nil { - if !strings.Contains(err.Error(), test.err) { - t.Errorf("expected error: %v, but got: %v", test.err, err) - } - } else { - actual := uri.ListenAddress() - 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 @@ -134,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) @@ -143,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 bd0a8666d..c17331618 100644 --- a/server.go +++ b/server.go @@ -60,8 +60,7 @@ type Server struct { // Cluster configuration. // Host is replaced with actual host after opening if port is ":0". Network string - Host string - Scheme string + Host *URI Cluster *Cluster // Background monitoring intervals. @@ -108,7 +107,7 @@ func (s *Server) Open() error { var err error // If bind URI has the https scheme, enable TLS - if s.Scheme == "https" { + if s.Host.Scheme() == "https" { if s.TLS.CertificatePath == "" { return errors.New("certificate path is required for TLS sockets") } @@ -120,28 +119,33 @@ func (s *Server) Open() error { return err } config := tls.Config{Certificates: []tls.Certificate{cert}} - ln, err = tls.Listen("tcp", s.Host, &config) + ln, err = tls.Listen("tcp", s.Host.HostPort(), &config) if err != nil { return err } - } else if s.Scheme == "http" { + } else if s.Host.Scheme() == "http" { // Open HTTP listener to determine port (if specified as :0). - ln, err = net.Listen(s.Network, s.Host) + ln, err = net.Listen(s.Network, s.Host.HostPort()) if err != nil { return fmt.Errorf("net.Listen: %v", err) } } else { - return fmt.Errorf("unsupported scheme: %s", s.Scheme) + return fmt.Errorf("unsupported scheme: %s", s.Host.Scheme()) } s.ln = ln - // Determine hostname based on listening port. - // s.Host = net.JoinHostPort(uri.Host(), strconv.Itoa(s.ln.Addr().(*net.TCPAddr).Port)) + if s.Host.Port() == 0 { + // If the port is 0, it is set automatically. + // Find out automatically set port and update the host. + s.Host.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.Host.Scheme(), Host: s.Host.HostPort()}, + } } for i, n := range s.Cluster.Nodes { @@ -168,7 +172,7 @@ func (s *Server) Open() error { // Create executor for executing queries. e := NewExecutor() e.Holder = s.Holder - e.Host = s.Host + e.Host = s.Host.HostPort() e.Cluster = s.Cluster e.MaxWritesPerRequest = s.MaxWritesPerRequest @@ -283,8 +287,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.Host.HostPort() != node.Host { + maxSlices, _ := 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 @@ -386,14 +390,14 @@ func (s *Server) LocalStatus() (proto.Message, error) { } ns := internal.NodeStatus{ - Host: s.Host, + Host: s.Host.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.Host.HostPort()) } return &ns, nil @@ -406,7 +410,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.Host.HostPort()) node.SetStatus(ns.(*internal.NodeStatus)) // Update NodeState for all nodes. @@ -416,7 +420,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.Host.HostPort() { nodeState = NodeStateUp } node := s.Cluster.NodeByHost(host) @@ -463,11 +467,11 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { return nil } -func checkMaxSlices(hostport string) (map[string]uint64, error) { +func 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) diff --git a/server/server.go b/server/server.go index 54447dc35..aa6430ec8 100644 --- a/server/server.go +++ b/server/server.go @@ -99,7 +99,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.Host.Normalize()) return nil } @@ -142,8 +142,7 @@ func (m *Command) SetupServer() error { if err != nil { return err } - m.Server.Host = bindWithDefaults.ListenAddress() - m.Server.Scheme = bindWithDefaults.Scheme() + m.Server.Host = bindWithDefaults // Set internal port (string). gossipPortStr := pilosa.DefaultGossipPort @@ -164,7 +163,7 @@ func (m *Command) SetupServer() error { // get the host portion of addr to use for binding gossipHost := bindWithDefaults.Host() - gossipNodeSet := gossip.NewGossipNodeSet(bindWithDefaults.ListenAddress(), gossipHost, gossipPort, gossipSeed, m.Server) + gossipNodeSet := gossip.NewGossipNodeSet(bindWithDefaults.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..070704132 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.Host.HostPort()) 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}, + {Host: m0.Server.Host.HostPort()}, + {Host: m1.Server.Host.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.Host.HostPort()) 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.Host.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.Host.HostPort()}, + {Host: m1.Server.Host.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.Host.HostPort()) if err != nil { - gossipHost = m0.Server.Host + gossipHost = m0.Server.Host.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.Host.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.Host.HostPort()) if err != nil { - gossipHost = m1.Server.Host + gossipHost = m1.Server.Host.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.Host.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.Host.HostPort()) if err != nil { panic(err) } diff --git a/test/handler.go b/test/handler.go index 70ac9e4e0..3924efeb7 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.Host = 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.Host.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.Host.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 index 0c14b6594..9b29fa87b 100644 --- a/uri.go +++ b/uri.go @@ -82,6 +82,11 @@ func (u *URI) Port() uint16 { return u.port } +// SetPort updates the port +func (u *URI) SetPort(port uint16) { + u.port = port +} + // Normalize returns the address in a form usable by a HTTP client. func (u *URI) Normalize() string { scheme := u.scheme @@ -92,8 +97,8 @@ func (u *URI) Normalize() string { return fmt.Sprintf("%s://%s:%d", scheme, u.host, u.port) } -// ListenAddress returns the address suitable for passing to `net.Listener.Listen` -func (u *URI) ListenAddress() string { +// HostPort returns the address suitable for passing to `net.Listener.Listen` +func (u *URI) HostPort() string { return fmt.Sprintf("%s:%d", u.host, u.port) }