Initial client refactoring

client.go

@@ -43,8 +43,8 @@ type ClientOptions struct {
 
 // Client represents a client to the Pilosa cluster.
 type Client struct {
-	host    *URI
-	options *ClientOptions
+	defaultURI *URI
+	options    *ClientOptions
 
 	// The client to use for HTTP communication.
 	HTTPClient *http.Client
@@ -64,10 +64,7 @@ func NewClient(host string, options *ClientOptions) (*Client, error) {
 	return NewClientFromURI(uri, options)
 }
 
-func NewClientFromURI(uri *URI, options *ClientOptions) (*Client, error) {
-	if uri == nil {
-		return nil, ErrHostRequired
-	}
+func NewClientFromURI(defaultURI *URI, options *ClientOptions) (*Client, error) {
 	if options == nil {
 		options = &ClientOptions{}
 	}
@@ -77,13 +74,13 @@ func NewClientFromURI(uri *URI, options *ClientOptions) (*Client, error) {
 	}
 	client := &http.Client{Transport: transport}
 	return &Client{
-		host:       uri,
+		defaultURI: defaultURI,
 		HTTPClient: client,
 	}, nil
 }
 
 // Host returns the host the client was initialized with.
-func (c *Client) Host() *URI { return c.host }
+func (c *Client) Host() *URI { return c.defaultURI }
 
 // MaxSliceByIndex returns the number of slices on a server by index.
 func (c *Client) MaxSliceByIndex(ctx context.Context) (map[string]uint64, error) {
@@ -98,7 +95,7 @@ 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 := uriPathToURL(c.host, "/slices/max")
+	u := uriPathToURL(c.defaultURI, "/slices/max")
 	u.RawQuery = (&url.Values{
 		"inverse": {strconv.FormatBool(inverse)},
 	}).Encode()
@@ -131,7 +128,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 := uriPathToURL(c.host, "/schema")
+	u := uriPathToURL(c.defaultURI, "/schema")
 
 	// Build request.
 	req, err := http.NewRequest("GET", u.String(), nil)
@@ -168,7 +165,7 @@ func (c *Client) CreateIndex(ctx context.Context, index string, opt IndexOptions
 	}
 
 	// Create URL & HTTP request.
-	u := uriPathToURL(c.host, fmt.Sprintf("/index/%s", index))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s", index))
 	req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
 	if err != nil {
 		return err
@@ -205,7 +202,7 @@ 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 := uriPathToURL(c.host, "/fragment/nodes")
+	u := uriPathToURL(c.defaultURI, "/fragment/nodes")
 	u.RawQuery = (url.Values{"index": {index}, "slice": {strconv.FormatUint(slice, 10)}}).Encode()
 
 	// Build request.
@@ -234,28 +231,30 @@ func (c *Client) FragmentNodes(ctx context.Context, index string, slice uint64)
 }
 
 // ExecuteQuery executes query against index on the server.
-func (c *Client) ExecuteQuery(ctx context.Context, index, query string, allowRedirect bool) (result interface{}, err error) {
+func (c *Client) ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
 	if index == "" {
 		return nil, ErrIndexRequired
-	} else if query == "" {
+	} else if queryRequest.Query == "" {
 		return nil, ErrQueryRequired
 	}
 
-	// Encode query request.
-	buf, err := proto.Marshal(&internal.QueryRequest{
-		Query:  query,
-		Remote: !allowRedirect,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("marshal: %s", err)
-	}
-
-	// Create URL & HTTP request.
-	u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/query", index))
-	req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
+	// Encode request object.
+	buf, err := proto.Marshal(queryRequest)
 	if err != nil {
 		return nil, err
 	}
+
+	// Create HTTP request.
+	clientURI := c.defaultURI
+	if contextURI, ok := ctx.Value("uri").(*URI); ok {
+		clientURI = contextURI
+	}
+	u := clientURI.Path(fmt.Sprintf("/index/%s/query", index))
+	req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
+	if err != nil {
+		return nil, err
+	}
+
 	req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
 	req.Header.Set("Content-Type", "application/x-protobuf")
 	req.Header.Set("Accept", "application/x-protobuf")
@@ -276,8 +275,8 @@ func (c *Client) ExecuteQuery(ctx context.Context, index, query string, allowRed
 		return nil, errors.New(string(body))
 	}
 
-	var qresp internal.QueryResponse
-	if err := proto.Unmarshal(body, &qresp); err != nil {
+	qresp := &internal.QueryResponse{}
+	if err := proto.Unmarshal(body, qresp); err != nil {
 		return nil, fmt.Errorf("unmarshal response: %s", err)
 	} else if s := qresp.Err; s != "" {
 		return nil, errors.New(s)
@@ -288,7 +287,7 @@ 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 := uriPathToURL(c.host, "/query")
+	u := uriPathToURL(c.defaultURI, "/query")
 	u.RawQuery = url.Values{"index": {index}}.Encode()
 
 	req, err := http.NewRequest("POST", u.String(), bytes.NewReader([]byte(query)))
@@ -811,7 +810,7 @@ func (c *Client) CreateFrame(ctx context.Context, index, frame string, opt Frame
 	}
 
 	// Create URL & HTTP request.
-	u := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s", index, frame))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s", index, frame))
 	req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
 	if err != nil {
 		return err
@@ -847,7 +846,7 @@ 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 := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/restore", index, frame))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s/restore", index, frame))
 	u.RawQuery = url.Values{
 		"host": {host},
 	}.Encode()
@@ -878,7 +877,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 := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/views", index, frame))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s/views", index, frame))
 	req, err := http.NewRequest("GET", u.String(), nil)
 	if err != nil {
 		return nil, err
@@ -914,7 +913,7 @@ 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 := uriPathToURL(c.host, "/fragment/blocks")
+	u := uriPathToURL(c.defaultURI, "/fragment/blocks")
 	u.RawQuery = url.Values{
 		"index": {index},
 		"frame": {frame},
@@ -967,7 +966,7 @@ func (c *Client) BlockData(ctx context.Context, index, frame, view string, slice
 		return nil, nil, err
 	}
 
-	u := uriPathToURL(c.host, "/fragment/block/data")
+	u := uriPathToURL(c.defaultURI, "/fragment/block/data")
 	req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf))
 	if err != nil {
 		return nil, nil, err
@@ -1004,7 +1003,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 := uriPathToURL(c.host, fmt.Sprintf("/index/%s/attr/diff", index))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/attr/diff", index))
 
 	// Encode request.
 	buf, err := json.Marshal(postIndexAttrDiffRequest{Blocks: blks})
@@ -1044,7 +1043,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 := uriPathToURL(c.host, fmt.Sprintf("/index/%s/frame/%s/attr/diff", index, frame))
+	u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s/attr/diff", index, frame))
 
 	// Encode request.
 	buf, err := json.Marshal(postFrameAttrDiffRequest{Blocks: blks})
@@ -1229,3 +1228,7 @@ func nodePathToURL(node *Node, path string) url.URL {
 		Path:   path,
 	}
 }
+
+type InternalClient interface {
+	ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
+}

client_test.go

@@ -54,7 +54,10 @@ 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(nil)
+		e, err := pilosa.NewExecutor(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
 		e.Holder = hldr[0].Holder
 		e.Scheme = cluster.Nodes[0].Scheme
 		e.Host = cluster.Nodes[0].Host
@@ -62,7 +65,10 @@ func TestClient_MultiNode(t *testing.T) {
 		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(nil)
+		e, err := pilosa.NewExecutor(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
 		e.Holder = hldr[1].Holder
 		e.Scheme = cluster.Nodes[1].Scheme
 		e.Host = cluster.Nodes[1].Host
@@ -70,7 +76,10 @@ func TestClient_MultiNode(t *testing.T) {
 		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(nil)
+		e, err := pilosa.NewExecutor(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
 		e.Holder = hldr[2].Holder
 		e.Scheme = cluster.Nodes[2].Scheme
 		e.Host = cluster.Nodes[2].Host
@@ -140,15 +149,17 @@ func TestClient_MultiNode(t *testing.T) {
 	client[2] = test.MustNewClient(s[2].Host())
 
 	topN := 4
-	q := fmt.Sprintf(`TopN(frame="%s", n=%d)`, "f", topN)
-
-	result, err := client[0].ExecuteQuery(context.Background(), "i", q, true)
+	queryRequest := &internal.QueryRequest{
+		Query:  fmt.Sprintf(`TopN(frame="%s", n=%d)`, "f", topN),
+		Remote: false,
+	}
+	result, err := client[0].ExecuteQuery(context.Background(), "i", queryRequest)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Check the results before every node has the correct max slice value.
-	pairs := result.(internal.QueryResponse).Results[0].Pairs
+	pairs := result.Results[0].Pairs
 	for _, pair := range pairs {
 		if pair.Key == 22 && pair.Count != 3 {
 			t.Fatalf("Invalid Cluster wide MaxSlice prevents accurate calculation of %s", pair)
@@ -160,13 +171,13 @@ func TestClient_MultiNode(t *testing.T) {
 	hldr[1].Index("i").SetRemoteMaxSlice(maxSlice)
 	hldr[2].Index("i").SetRemoteMaxSlice(maxSlice)
 
-	result, err = client[0].ExecuteQuery(context.Background(), "i", q, true)
+	result, err = client[0].ExecuteQuery(context.Background(), "i", queryRequest)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Test must return exactly N results.
-	if len(result.(internal.QueryResponse).Results[0].Pairs) != topN {
+	if len(result.Results[0].Pairs) != topN {
 		t.Fatalf("unexpected number of TopN results: %s", spew.Sdump(result))
 	}
 	p := []*internal.Pair{
@@ -176,15 +187,15 @@ func TestClient_MultiNode(t *testing.T) {
 		{Key: 99, Count: 7}}
 
 	// Valdidate the Top 4 result counts.
-	if !reflect.DeepEqual(result.(internal.QueryResponse).Results[0].Pairs, p) {
+	if !reflect.DeepEqual(result.Results[0].Pairs, p) {
 		t.Fatalf("Invalid TopN result set: %s", spew.Sdump(result))
 	}
 
-	result1, err := client[1].ExecuteQuery(context.Background(), "i", q, true)
+	result1, err := client[1].ExecuteQuery(context.Background(), "i", queryRequest)
 	if err != nil {
 		t.Fatal(err)
 	}
-	result2, err := client[2].ExecuteQuery(context.Background(), "i", q, true)
+	result2, err := client[2].ExecuteQuery(context.Background(), "i", queryRequest)
 	if err != nil {
 		t.Fatal(err)
 	}

ctl/bench.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/internal"
 )
 
 // BenchCommand represents a command for benchmarking index operations.
@@ -89,9 +90,11 @@ func (cmd *BenchCommand) runSetBit(ctx context.Context, client *pilosa.Client) e
 		rowID := rand.Intn(maxRowID)
 		columnID := rand.Intn(maxColumnID)
 
-		q := fmt.Sprintf(`SetBit(id=%d, frame="%s", columnID=%d)`, rowID, cmd.Frame, columnID)
-
-		if _, err := client.ExecuteQuery(ctx, cmd.Index, q, true); err != nil {
+		queryRequest := &internal.QueryRequest{
+			Query:  fmt.Sprintf(`SetBit(id=%d, frame="%s", columnID=%d)`, rowID, cmd.Frame, columnID),
+			Remote: false,
+		}
+		if _, err := client.ExecuteQuery(ctx, cmd.Index, queryRequest); err != nil {
 			return err
 		}
 	}

executor.go

@@ -15,16 +15,12 @@
 package pilosa
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
-	"io/ioutil"
-	"net/http"
 	"sort"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
 	"github.com/pilosa/pilosa/internal"
 	"github.com/pilosa/pilosa/pql"
 )
@@ -47,26 +43,25 @@ type Executor struct {
 	Host    string
 	Cluster *Cluster
 
-	// Client used for remote HTTP requests.
-	HTTPClient *http.Client
+	// Client used for remote requests.
+	client InternalClient
 
 	// Maximum number of SetBit() or ClearBit() commands per request.
 	MaxWritesPerRequest int
 }
 
 // NewExecutor returns a new instance of Executor.
-func NewExecutor(clientOptions *ClientOptions) *Executor {
+func NewExecutor(clientOptions *ClientOptions) (*Executor, error) {
 	if clientOptions == nil {
 		clientOptions = &ClientOptions{}
 	}
-	transport := &http.Transport{}
-	if clientOptions.TLS != nil {
-		transport.TLSClientConfig = clientOptions.TLS
+	client, err := NewClientFromURI(nil, clientOptions)
+	if err != nil {
+		return nil, err
 	}
-	client := &http.Client{Transport: transport}
 	return &Executor{
-		HTTPClient: client,
-	}
+		client: client,
+	}, nil
 }
 
 // Execute executes a PQL query.
@@ -1377,48 +1372,17 @@ func (e *Executor) exec(ctx context.Context, node *Node, index string, q *pql.Qu
 		Slices: slices,
 		Remote: true,
 	}
-	buf, err := proto.Marshal(pbreq)
+	uri, err := NewURIFromAddress(node.Host)
 	if err != nil {
 		return nil, err
 	}
-
-	// Create HTTP request.
-	u := nodePathToURL(node, fmt.Sprintf("/index/%s/query", index))
-	u.Scheme = e.Scheme
-	req, err := http.NewRequest("POST", (&u).String(), bytes.NewReader(buf))
+	uri.SetScheme(node.Scheme)
+	ctx = context.WithValue(ctx, "uri", uri)
+	pb, err := e.client.ExecuteQuery(ctx, index, pbreq)
 	if err != nil {
 		return nil, err
 	}
 
-	// Require protobuf encoding.
-	req.Header.Set("Accept", "application/x-protobuf")
-	req.Header.Set("Content-Type", "application/x-protobuf")
-	req.Header.Set("User-Agent", "pilosa/"+Version)
-
-	// Send request to remote node.
-	resp, err := e.HTTPClient.Do(req)
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-
-	// Read response into buffer.
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return nil, err
-	}
-
-	// Check status code.
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("invalid status Executor.exec: code=%d, err=%s, req: %v", resp.StatusCode, body, req)
-	}
-
-	// Decode response object.
-	var pb internal.QueryResponse
-	if err := proto.Unmarshal(body, &pb); err != nil {
-		return nil, err
-	}
-
 	// Return an error, if specified on response.
 	if err := decodeError(pb.Err); err != nil {
 		return nil, err

fragment.go

@@ -1848,7 +1848,11 @@ func (s *FragmentSyncer) syncBlock(id int) error {
 		}
 
 		// Execute query.
-		_, err := clients[i].ExecuteQuery(context.Background(), f.Index(), buf.String(), false)
+		queryRequest := &internal.QueryRequest{
+			Query:  buf.String(),
+			Remote: true,
+		}
+		_, err := clients[i].ExecuteQuery(context.Background(), f.Index(), queryRequest)
 		if err != nil {
 			return err
 		}

holder_test.go

@@ -314,7 +314,10 @@ 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(nil)
+		e, err := pilosa.NewExecutor(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
 		e.Holder = hldr1.Holder
 		e.Scheme = cluster.Nodes[1].Scheme
 		e.Host = cluster.Nodes[1].Host

server.go

@@ -164,7 +164,10 @@ func (s *Server) Open() error {
 	s.createDefaultClient()
 
 	// Create executor for executing queries.
-	e := NewExecutor(&ClientOptions{TLS: s.TLS})
+	e, err := NewExecutor(&ClientOptions{TLS: s.TLS})
+	if err != nil {
+		return err
+	}
 	e.Holder = s.Holder
 	e.Scheme = s.URI.Scheme()
 	e.Host = s.URI.HostPort()

test/executor.go

@@ -15,7 +15,8 @@ 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(nil)}
+	executor, _ := pilosa.NewExecutor(nil)
+	e := &Executor{Executor: executor}
 	e.Holder = holder
 	e.Cluster = cluster
 	e.Scheme = cluster.Nodes[0].Scheme

uri.go

@@ -144,6 +144,11 @@ func (u URI) Equals(other *URI) bool {
 		u.port == other.port
 }
 
+// Path returns URI with path
+func (u *URI) Path(path string) string {
+	return fmt.Sprintf("%s%s", u.Normalize(), path)
+}
+
 // The following methods are required to implement pflag Value interface.
 
 // Set sets the time quantum value.

uri_test.go

@@ -83,6 +83,17 @@ func TestNormalizedAddress(t *testing.T) {
 	}
 }
 
+func TestURIPath(t *testing.T) {
+	uri, err := NewURIFromAddress("http+protobuf://big-data.pilosa.com:6888")
+	if err != nil {
+		t.Fatal(err)
+	}
+	target := "http://big-data.pilosa.com:6888/index/foo"
+	if uri.Path("/index/foo") != target {
+		t.Fatalf("%s != %s", uri.Path("/index/foo"), target)
+	}
+}
+
 func TestEquals(t *testing.T) {
 	uri1 := DefaultURI()
 	if uri1.Equals(nil) {