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

Merge pull request #1253 from jaffee/fixup-internal-client

Browse source 
Fixup internal client
This commit is contained in:
Matthew Jaffee 2018-05-07 12:43:12 -07:00 • committed by GitHub
commit 134a88d91d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 24 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
client.go
View file

@ -88,7 +88,7 @@ func (c *InternalHTTPClient) MaxInverseSliceByIndex(ctx context.Context) (map[st
// maxSliceByIndex returns the number of slices on a server by index.
func (c *InternalHTTPClient) maxSliceByIndex(ctx context.Context, inverse bool) (map[string]uint64, error) {
// Execute request against the host.
u := uriPathToURL(c.clientURI(ctx), "/slices/max")
u := uriPathToURL(c.defaultURI, "/slices/max")
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
@ -223,8 +223,13 @@ func (c *InternalHTTPClient) FragmentNodes(ctx context.Context, index string, sl
return a, nil
}
// ExecuteQuery executes query against index on the server.
func (c *InternalHTTPClient) ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
// Query executes query against the index.
func (c *InternalHTTPClient) Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
return c.QueryNode(ctx, c.defaultURI, index, queryRequest)
}
// QueryNode executes query against the index, sending the request to the node specified.
func (c *InternalHTTPClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
if index == "" {
return nil, ErrIndexRequired
} else if queryRequest.Query == "" {
@ -238,7 +243,7 @@ func (c *InternalHTTPClient) ExecuteQuery(ctx context.Context, index string, que
}
// Create HTTP request.
u := c.clientURI(ctx).Path(fmt.Sprintf("/index/%s/query", index))
u := uri.Path(fmt.Sprintf("/index/%s/query", index))
req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
if err != nil {
return nil, err
@ -1098,13 +1103,13 @@ func (c *InternalHTTPClient) RowAttrDiff(ctx context.Context, index, frame strin
}
// SendMessage posts a message synchronously.
func (c *InternalHTTPClient) SendMessage(ctx context.Context, pb proto.Message) error {
func (c *InternalHTTPClient) SendMessage(ctx context.Context, uri *URI, pb proto.Message) error {
msg, err := MarshalMessage(pb)
if err != nil {
return fmt.Errorf("marshaling message: %v", err)
}
u := uriPathToURL(ctx.Value("uri").(*URI), "/cluster/message")
u := uriPathToURL(uri, "/cluster/message")
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(msg))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -1132,40 +1137,6 @@ func (c *InternalHTTPClient) SendMessage(ctx context.Context, pb proto.Message)
return nil
}
func (c *InternalHTTPClient) clientURI(ctx context.Context) *URI {
clientURI := c.defaultURI
if contextURI, ok := ctx.Value("uri").(*URI); ok {
clientURI = contextURI
} else if contextURI, ok := ctx.Value("uri").(URI); ok {
clientURI = &contextURI
}
return clientURI
}
func (c *InternalHTTPClient) NodeID(uri *URI) (string, error) {
u := uriPathToURL(uri, "/id")
req, err := http.NewRequest("GET", u.String(), nil)
resp, err := c.HTTPClient.Do(req)
if err != nil {
return "", fmt.Errorf("executing http request: %v", err)
}
defer resp.Body.Close()
// Read body.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return "", fmt.Errorf("reading response body: %v", err)
}
// Return error if status is not OK.
switch resp.StatusCode {
case http.StatusOK: // ok
default:
return "", fmt.Errorf("unexpected response status code: %d: %s", resp.StatusCode, body)
}
return string(body), nil
}
// Bit represents the location of a single bit.
type Bit struct {
RowID uint64
@ -1344,7 +1315,8 @@ type InternalClient interface {
Schema(ctx context.Context) ([]*IndexInfo, error)
CreateIndex(ctx context.Context, index string, opt IndexOptions) error
FragmentNodes(ctx context.Context, index string, slice uint64) ([]*Node, error)
ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
Import(ctx context.Context, index, frame string, slice uint64, bits []Bit) error
ImportK(ctx context.Context, index, frame string, bits []Bit) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
@ -1361,6 +1333,5 @@ type InternalClient interface {
BlockData(ctx context.Context, index, frame, view string, slice uint64, block int) ([]uint64, []uint64, error)
ColumnAttrDiff(ctx context.Context, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
RowAttrDiff(ctx context.Context, index, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
SendMessage(ctx context.Context, pb proto.Message) error
NodeID(uri *URI) (string, error)
SendMessage(ctx context.Context, uri *URI, pb proto.Message) error
}

8
client_test.go
View file

@ -150,7 +150,7 @@ func TestClient_MultiNode(t *testing.T) {
Query: fmt.Sprintf(`TopN(frame="%s", n=%d)`, "f", topN),
Remote: false,
}
result, err := client[0].ExecuteQuery(context.Background(), "i", queryRequest)
result, err := client[0].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
@ -168,7 +168,7 @@ 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", queryRequest)
result, err = client[0].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
@ -188,11 +188,11 @@ func TestClient_MultiNode(t *testing.T) {
t.Fatalf("Invalid TopN result set: %s", spew.Sdump(result))
}
result1, err := client[1].ExecuteQuery(context.Background(), "i", queryRequest)
result1, err := client[1].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
result2, err := client[2].ExecuteQuery(context.Background(), "i", queryRequest)
result2, err := client[2].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}

2
ctl/bench.go
View file

@ -95,7 +95,7 @@ func (cmd *BenchCommand) runSetBit(ctx context.Context, client pilosa.InternalCl
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 {
if _, err := client.Query(ctx, cmd.Index, queryRequest); err != nil {
return err
}
}

3
executor.go
View file

@ -1490,8 +1490,7 @@ func (e *Executor) remoteExec(ctx context.Context, node *Node, index string, q *
Remote: true,
}
ctx = context.WithValue(ctx, "uri", node.URI)
pb, err := e.client.ExecuteQuery(ctx, index, pbreq)
pb, err := e.client.QueryNode(ctx, &node.URI, index, pbreq)
if err != nil {
return nil, err
}

2
fragment.go
View file

@ -1924,7 +1924,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
Query: buffers[k].String(),
Remote: true,
}
_, err := clients[i].ExecuteQuery(context.Background(), f.Index(), queryRequest)
_, err := clients[i].Query(context.Background(), f.Index(), queryRequest)
if err != nil {
return err
}

7
server.go
View file

@ -536,15 +536,15 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
func (s *Server) SendSync(pb proto.Message) error {
var eg errgroup.Group
for _, node := range s.Cluster.Nodes {
node := node
s.logger.Printf("SendSync to: %s", node.URI)
// Don't forward the message to ourselves.
if s.URI == node.URI {
continue
}
ctx := context.WithValue(context.Background(), "uri", &node.URI)
eg.Go(func() error {
return s.defaultClient.SendMessage(ctx, pb)
return s.defaultClient.SendMessage(context.Background(), &node.URI, pb)
})
}
@ -559,8 +559,7 @@ func (s *Server) SendAsync(pb proto.Message) error {
// SendTo represents an implementation of Broadcaster.
func (s *Server) SendTo(to *Node, pb proto.Message) error {
s.logger.Printf("SendTo: %s", to.URI)
ctx := context.WithValue(context.Background(), "uri", &to.URI)
return s.defaultClient.SendMessage(ctx, pb)
return s.defaultClient.SendMessage(context.Background(), &to.URI, pb)
}
// Server implements StatusHandler.