scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 00:55:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/master'

Browse source
This commit is contained in:
Todd Gruben 2018-05-08 09:35:58 -05:00
commit e50b2aee24
11 changed files with 70 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
api.go
View file

@ -1118,6 +1118,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.

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
}

6
frame.go
View file

@ -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 {

8
handler.go
View file

@ -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"`
}

15
handler_test.go
View file

@ -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) {

8
index.go
View file

@ -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
}

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.