scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

use URI instead of host

Browse source
This commit is contained in:
Yuce Tekol 2017-10-04 08:20:00 +03:00
parent 5760d89742
commit 1bedfd6585
No known key found for this signature in database
GPG key ID: CB59E46D2FB90573
13 changed files with 176 additions and 212 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
client.go
View file

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

16
client_test.go
View file

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

3
cluster.go
View file

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

2
config.go
View file

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

2
executor.go
View file

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

21
handler.go
View file

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

10
holder.go
View file

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

40
pilosa_test.go
View file

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

44
server.go
View file

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

7
server/server.go
View file

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

28
server/server_test.go
View file

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

18
test/handler.go
View file

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

9
uri.go
View file

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