From 48c0dbaee75bf5c9e13216add47c3e6d1cdc85b1 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Tue, 17 Oct 2017 11:35:01 +0300 Subject: [PATCH] Initial client refactoring --- client.go | 75 +++++++++++++++++++++++++----------------------- client_test.go | 35 ++++++++++++++-------- ctl/bench.go | 9 ++++-- executor.go | 60 ++++++++------------------------------ fragment.go | 6 +++- holder_test.go | 5 +++- server.go | 5 +++- test/executor.go | 3 +- uri.go | 5 ++++ uri_test.go | 11 +++++++ 10 files changed, 111 insertions(+), 103 deletions(-) diff --git a/client.go b/client.go index fbd0d3717..3babe2919 100644 --- a/client.go +++ b/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) +} diff --git a/client_test.go b/client_test.go index 70ea6c669..509039618 100644 --- a/client_test.go +++ b/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) } diff --git a/ctl/bench.go b/ctl/bench.go index 24e6e7658..2f1e7a52f 100644 --- a/ctl/bench.go +++ b/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 } } diff --git a/executor.go b/executor.go index 64a068149..8d70cb2f1 100644 --- a/executor.go +++ b/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 diff --git a/fragment.go b/fragment.go index 67f9ff000..cdc8c00ff 100644 --- a/fragment.go +++ b/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 } diff --git a/holder_test.go b/holder_test.go index 104bd1bb0..cb33045e6 100644 --- a/holder_test.go +++ b/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 diff --git a/server.go b/server.go index 773cf9025..e73354cf2 100644 --- a/server.go +++ b/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() diff --git a/test/executor.go b/test/executor.go index 73445a1cd..4ef99d351 100644 --- a/test/executor.go +++ b/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 diff --git a/uri.go b/uri.go index de6e4bae6..ee5d23d75 100644 --- a/uri.go +++ b/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. diff --git a/uri_test.go b/uri_test.go index 33ccd35f9..70ea3f273 100644 --- a/uri_test.go +++ b/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) {