From 7f41c0256ca49ed607608eb01237fda80c359bb9 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Sun, 29 Apr 2018 16:40:32 -0500 Subject: [PATCH 1/7] avoid creating a slice of nil timestamps on Import() --- frame.go | 6 +++++- index.go | 8 +++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/frame.go b/frame.go index 0c87210bf..3de3f3613 100644 --- a/frame.go +++ b/frame.go @@ -834,7 +834,11 @@ func (f *Frame) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro // Split import data by fragment. dataByFragment := make(map[importKey]importData) for i := range rowIDs { - rowID, columnID, timestamp := rowIDs[i], columnIDs[i], timestamps[i] + rowID, columnID := rowIDs[i], columnIDs[i] + var timestamp *time.Time + if len(timestamps) > i { + timestamp = timestamps[i] + } var standard, inverse []string if timestamp == nil { diff --git a/index.go b/index.go index 00ee98f11..dc676c304 100644 --- a/index.go +++ b/index.go @@ -642,7 +642,8 @@ func (i *Index) openInputDefinitions() error { // InputBits Process the []Bit though the Frame import process func (i *Index) InputBits(frame string, bits []*Bit) error { var rowIDs, columnIDs []uint64 - timestamps := make([]*time.Time, len(bits)) + var timestamps []*time.Time + f := i.Frame(frame) if f == nil { return fmt.Errorf("Frame not found: %s", frame) @@ -657,6 +658,11 @@ func (i *Index) InputBits(frame string, bits []*Bit) error { // Convert timestamps to time.Time. if bit.Timestamp > 0 { + // Don't create a full timestamps slice unless + // at least one bit contains a timestamp. + if len(timestamps) == 0 { + timestamps = make([]*time.Time, len(bits)) + } t := time.Unix(bit.Timestamp, 0) timestamps[i] = &t } From 15f997c1379c0c3290fb7c6904a046a97633889f Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Mon, 30 Apr 2018 16:45:31 +0300 Subject: [PATCH 2/7] Added /info endpoint. Fixes #1232 --- api.go | 11 +++++++++++ handler.go | 8 ++++++++ handler_test.go | 15 +++++++++++++++ 3 files changed, 34 insertions(+) diff --git a/api.go b/api.go index 700b21d25..1f33a6cde 100644 --- a/api.go +++ b/api.go @@ -1121,6 +1121,17 @@ func (api *API) Version() string { return strings.TrimPrefix(Version, "v") } +// Info returns information about this server instance +func (api *API) Info() ServerInfo { + return ServerInfo{ + SliceWidth: SliceWidth, + } +} + +type ServerInfo struct { + SliceWidth uint64 `json:"sliceWidth"` +} + type apiMethod int // API validation constants. diff --git a/handler.go b/handler.go index 7bf11af1b..871ba885a 100644 --- a/handler.go +++ b/handler.go @@ -126,6 +126,7 @@ func NewRouter(handler *Handler) *mux.Router { router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET") router.HandleFunc("/slices/max", handler.handleGetSlicesMax).Methods("GET") // TODO: deprecate, but it's being used by the client (for backups) router.HandleFunc("/status", handler.handleGetStatus).Methods("GET") + router.HandleFunc("/info", handler.handleGetInfo).Methods("GET") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET") router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST") @@ -252,6 +253,13 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) { } } +func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) { + info := h.API.Info() + if err := json.NewEncoder(w).Encode(info); err != nil { + h.Logger.Printf("write info response error: %s", err) + } +} + type getSchemaResponse struct { Indexes []*IndexInfo `json:"indexes"` } diff --git a/handler_test.go b/handler_test.go index 4187b3303..963c0cccc 100644 --- a/handler_test.go +++ b/handler_test.go @@ -19,6 +19,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "io/ioutil" "net/http" @@ -154,6 +155,20 @@ func TestHandler_Status(t *testing.T) { } } +func TestHandler_Info(t *testing.T) { + s := test.NewServer() + defer s.Close() + h := test.NewHandler() + + w := httptest.NewRecorder() + h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/info", nil)) + if w.Code != http.StatusOK { + t.Fatalf("unexpected status code: %d", w.Code) + } else if body := w.Body.String(); body != fmt.Sprintf("{\"sliceWidth\":%d}\n", SliceWidth) { + t.Fatalf("unexpected body: %s", body) + } +} + // Ensure the handler can abort a cluster resize. func TestHandler_ClusterResizeAbort(t *testing.T) { From b3f529cb1b08d8455b7a023c10e6b4db1a3de9ab Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Sun, 6 May 2018 19:44:36 -0700 Subject: [PATCH 3/7] 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 4/7] 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 5/7] 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 6/7] 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 7/7] 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) }