From b3f529cb1b08d8455b7a023c10e6b4db1a3de9ab Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sun, 6 May 2018 19:44:36 -0700 Subject: [PATCH 1/5] remove unused NodeID method on InternalClient --- client.go | 25 ------------------------- 1 file changed, 25 deletions(-) diff --git a/client.go b/client.go index c2e88c77b..eee1a3a9c 100644 --- a/client.go +++ b/client.go @@ -1142,30 +1142,6 @@ func (c *InternalHTTPClient) clientURI(ctx context.Context) *URI { 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 @@ -1362,5 +1338,4 @@ type InternalClient interface { 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) } From 0d3df71e37b3a34cd31c1ff6c7723c5ac7e4d586 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sun, 6 May 2018 20:12:23 -0700 Subject: [PATCH 2/5] add URI argument to InternalClient.SendMessage passing values through context is error prone and usually bad practice. --- client.go | 6 +++--- server.go | 7 +++---- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/client.go b/client.go index eee1a3a9c..ecfae5801 100644 --- a/client.go +++ b/client.go @@ -1098,13 +1098,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) @@ -1337,5 +1337,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 + SendMessage(ctx context.Context, uri *URI, pb proto.Message) error } diff --git a/server.go b/server.go index 7617d7e69..52999b1c5 100644 --- a/server.go +++ b/server.go @@ -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. From 2738c922864edabfc3237511eb0364640f1cf10d Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sun, 6 May 2018 20:30:58 -0700 Subject: [PATCH 3/5] stop passing uri via context to InternalClient.ExecuteQuery --- client.go | 14 ++++++++++---- client_test.go | 8 ++++---- ctl/bench.go | 2 +- executor.go | 3 +-- fragment.go | 2 +- 5 files changed, 17 insertions(+), 12 deletions(-) diff --git a/client.go b/client.go index ecfae5801..8950e8406 100644 --- a/client.go +++ b/client.go @@ -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) { +// QueryNode 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 @@ -1320,7 +1325,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 diff --git a/client_test.go b/client_test.go index be3505acd..88dc3accb 100644 --- a/client_test.go +++ b/client_test.go @@ -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) } diff --git a/ctl/bench.go b/ctl/bench.go index 9e37fb704..b9743bb5e 100644 --- a/ctl/bench.go +++ b/ctl/bench.go @@ -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 } } diff --git a/executor.go b/executor.go index 62bc0e8c8..59e40511d 100644 --- a/executor.go +++ b/executor.go @@ -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 } diff --git a/fragment.go b/fragment.go index 448ba3202..7168f4443 100644 --- a/fragment.go +++ b/fragment.go @@ -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 } From 420631e748925d8f3542fc734ab6b5f809629d83 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sun, 6 May 2018 20:36:52 -0700 Subject: [PATCH 4/5] remove last vestiges of passing URI via context this chould be safe as context.WithValue doesn't seem to appear anywhere else in Pilosa --- client.go | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/client.go b/client.go index 8950e8406..314f7d4e2 100644 --- a/client.go +++ b/client.go @@ -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) @@ -1137,16 +1137,6 @@ func (c *InternalHTTPClient) SendMessage(ctx context.Context, uri *URI, pb proto 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 -} - // Bit represents the location of a single bit. type Bit struct { RowID uint64 From b2eb8f02ee2395cfe6a5f3712ca2ea91bc0e7ee1 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 7 May 2018 12:42:39 -0700 Subject: [PATCH 5/5] fix QueryNode comment --- client.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client.go b/client.go index 314f7d4e2..5997cf4ef 100644 --- a/client.go +++ b/client.go @@ -223,7 +223,7 @@ func (c *InternalHTTPClient) FragmentNodes(ctx context.Context, index string, sl return a, nil } -// QueryNode executes query against the index. +// 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) }