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 branch 'develop' into crash-value-overwrite

Browse source
This commit is contained in:
tgruben 2018-06-21 20:10:54 -05:00 • committed by GitHub
commit 4b72489997
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 291 additions and 180 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
cluster.go
View file

@ -1010,6 +1010,9 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
func (c *Cluster) setStateAndBroadcast(state string) error {
c.SetState(state)
if c.Static {
return nil
}
// Broadcast cluster status changes to the cluster.
c.Logger.Printf("broadcasting ClusterStatus: %s", state)
return c.Broadcaster.SendSync(c.Status())
@ -1801,3 +1804,17 @@ func (c *Cluster) mergeClusterStatus(cs *internal.ClusterStatus) error {
return nil
}
func (c *Cluster) setStatic(hosts []string) error {
c.Static = true
c.Coordinator = c.Node.ID
for _, address := range hosts {
uri, err := NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
c.Nodes = append(c.Nodes, &Node{URI: *uri})
}
c.MemberSet = NewStaticMemberSet(c.Nodes)
return nil
}

32
cluster_internal_test.go
View file

@ -153,19 +153,19 @@ func TestFragSources(t *testing.T) {
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 101, nil)
_, err = field.SetBit(1, 101, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 1300000, nil)
_, err = field.SetBit(1, 1300000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 2600000, nil)
_, err = field.SetBit(1, 2600000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 3900000, nil)
_, err = field.SetBit(1, 3900000, nil)
if err != nil {
t.Fatal(err)
}
@ -552,7 +552,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -580,7 +580,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -605,14 +605,14 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
node1 := tc.Clusters[1]
@ -643,7 +643,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// write topology to data file
@ -664,12 +664,12 @@ func TestCluster_ResizeStates(t *testing.T) {
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
err := tc.addNode(false)
err := tc.addNode()
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
}
tc.addNode(false)
tc.addNode()
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
@ -687,7 +687,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// Open TestCluster.
@ -699,8 +699,8 @@ func TestCluster_ResizeStates(t *testing.T) {
if err := tc.CreateField("i", "f", FieldOptions{}); err != nil {
t.Fatal(err)
}
tc.SetBit("i", "f", "standard", 1, 101, nil)
tc.SetBit("i", "f", "standard", 1, 1300000, nil)
tc.SetBit("i", "f", 1, 101, nil)
tc.SetBit("i", "f", 1, 1300000, nil)
// Before starting the resize, get the CheckSum to use for
// comparison later.
@ -709,8 +709,8 @@ func TestCluster_ResizeStates(t *testing.T) {
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
// AddNode needs to block until the resize process has completed.
tc.addNode(false)
// addNode needs to block until the resize process has completed.
tc.addNode()
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.

1
ctl/import_test.go
View file

@ -183,6 +183,7 @@ func TestImportCommand_InvalidFile(t *testing.T) {
// MustNewHTTPRequest creates a new HTTP request. Panic on error.
func MustNewHTTPRequest(method, urlStr string, body io.Reader) *http.Request {
req, err := http.NewRequest(method, urlStr, body)
req.Header.Add("Accept", "application/json")
if err != nil {
panic(err)
}

2
diagnostics.go
View file

@ -222,7 +222,7 @@ func (d *DiagnosticsCollector) EnrichWithSchemaProperties() {
bsiFieldCount := 0
timeQuantumEnabled := false
for _, index := range d.server.Holder.Indexes() {
for _, index := range d.server.holder.Indexes() {
numSlices += index.MaxSlice() + 1
numIndexes += 1
for _, field := range index.Fields() {

16
executor.go
View file

@ -1026,17 +1026,17 @@ func (e *Executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
return false, fmt.Errorf("ClearBit col field '%v' required", columnLabel)
}
return e.executeClearBitView(ctx, index, c, f, ViewStandard, colID, rowID, opt)
return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
}
// executeClearBitView executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, opt *ExecOptions) (bool, error) {
// executeClearBitField executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.ClearBit(view, rowID, colID, nil)
val, err := f.ClearBit(rowID, colID, nil)
if err != nil {
return false, err
} else if val {
@ -1101,18 +1101,18 @@ func (e *Executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
timestamp = &t
}
return e.executeSetBitView(ctx, index, c, f, ViewStandard, colID, rowID, timestamp, opt)
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
}
// executeSetBitView executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
// executeSetBitField executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.SetBit(view, rowID, colID, timestamp)
val, err := f.SetBit(rowID, colID, timestamp)
if err != nil {
return false, err
} else if val {

22
field.go
View file

@ -654,14 +654,11 @@ func (f *Field) ViewRow(viewName string, rowID uint64) (*Row, error) {
}
// SetBit sets a bit on a view within the field.
func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -679,7 +676,7 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// If a timestamp is specified then set bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -696,14 +693,11 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// ClearBit clears a bit within the field.
func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) ClearBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -721,7 +715,7 @@ func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
}
// If a timestamp is specified then clear bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)

6
holder_test.go
View file

@ -210,7 +210,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -231,7 +231,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -257,7 +257,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if view, err := field.CreateViewIfNotExists(pilosa.ViewStandard); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := view.Fragment(0).FlushCache(); err != nil {
t.Fatal(err)

7
http/client.go
View file

@ -90,6 +90,7 @@ func (c *InternalClient) maxSliceByIndex(ctx context.Context) (map[string]uint64
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -120,6 +121,7 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -195,6 +197,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, slice
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -685,6 +688,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -777,6 +781,7 @@ func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pilosa.URI, in
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -820,6 +825,7 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
@ -859,6 +865,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, pb pr
}
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))

96
http/handler.go
View file

@ -164,11 +164,12 @@ func (h *Handler) queryArgValidator(next http.Handler) http.Handler {
func NewRouter(handler *Handler) *mux.Router {
router := mux.NewRouter()
router.HandleFunc("/", handler.handleHome).Methods("GET")
router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST")
router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
router.Handle("/debug/vars", expvar.Handler()).Methods("GET")
router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET")
router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST")
router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST")
router.HandleFunc("/slices/max", handler.handleGetSlicesMax).Methods("GET") // TODO: deprecate, but it's being used by the client
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
router.HandleFunc("/info", handler.handleGetInfo).Methods("GET")
@ -257,8 +258,28 @@ func (h *Handler) handleHome(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information.", http.StatusNotFound)
}
func checkHeaderAcceptJSON(header http.Header) bool {
v, found := header["Accept"]
sendError := false
if found {
sendError = true
for _, v := range v {
if v == "application/json" {
sendError = false
}
}
}
return sendError
}
// handleGetSchema handles GET /schema requests.
func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
schema := h.API.Schema(r.Context())
if err := json.NewEncoder(w).Encode(getSchemaResponse{
Indexes: schema,
@ -269,6 +290,10 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
// handleGetStatus handles GET /status requests.
func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
status := getStatusResponse{
State: h.API.State(),
Nodes: h.API.Hosts(r.Context()),
@ -280,6 +305,10 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
info := h.API.Info()
if err := json.NewEncoder(w).Encode(info); err != nil {
h.Logger.Printf("write info response error: %s", err)
@ -333,6 +362,10 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
// handleGetSlicesMax handles GET /schema requests.
func (h *Handler) handleGetSlicesMax(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
if err := json.NewEncoder(w).Encode(getSlicesMaxResponse{
Standard: h.API.MaxSlices(r.Context()),
}); err != nil {
@ -351,6 +384,10 @@ func (h *Handler) handleGetIndexes(w http.ResponseWriter, r *http.Request) {
// handleGetIndex handles GET /index/<indexname> requests.
func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
for _, idx := range h.API.Schema(r.Context()) {
if idx.Name == indexName {
@ -429,6 +466,10 @@ type postIndexResponse struct{}
// handleDeleteIndex handles DELETE /index request.
func (h *Handler) handleDeleteIndex(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
err := h.API.DeleteIndex(r.Context(), indexName)
if err != nil {
@ -447,6 +488,10 @@ type deleteIndexResponse struct{}
// handlePostIndex handles POST /index request.
func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
// Decode request.
@ -477,6 +522,10 @@ func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) {
// handlePostIndexAttrDiff handles POST /index/attr/diff requests.
func (h *Handler) handlePostIndexAttrDiff(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
// Decode request.
@ -514,6 +563,10 @@ type postIndexAttrDiffResponse struct {
// handlePostField handles POST /field request.
func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
@ -592,6 +645,11 @@ type postFieldResponse struct{}
// handleDeleteField handles DELETE /field request.
func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
@ -617,6 +675,10 @@ type deleteFieldResponse struct{}
// handlePostFieldAttrDiff handles POST /field/attr/diff requests.
func (h *Handler) handlePostFieldAttrDiff(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
@ -709,7 +771,7 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*pilosa.QueryRequest, er
// writeQueryResponse writes the response from the executor to w.
func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, resp *pilosa.QueryResponse) error {
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
if checkHeaderAcceptJSON(r.Header) {
return h.writeProtobufQueryResponse(w, resp)
}
return h.writeJSONQueryResponse(w, resp)
@ -872,6 +934,10 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) {
// handleGetFragmentNodes handles /fragment/nodes requests.
func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
q := r.URL.Query()
index := q.Get("index")
@ -917,6 +983,10 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ
// handleGetFragmentBlocks handles GET /fragment/blocks requests.
func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Read slice parameter.
q := r.URL.Query()
slice, err := strconv.ParseUint(q.Get("slice"), 10, 64)
@ -949,6 +1019,10 @@ type getFragmentBlocksResponse struct {
// handleGetVersion handles /version requests.
func (h *Handler) handleGetVersion(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
err := json.NewEncoder(w).Encode(struct {
Version string `json:"version"`
}{
@ -1047,6 +1121,10 @@ func errorString(err error) string {
}
func (h *Handler) handlePostClusterResizeSetCoordinator(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Decode request.
var req setCoordinatorRequest
err := json.NewDecoder(r.Body).Decode(&req)
@ -1084,6 +1162,10 @@ type setCoordinatorResponse struct {
// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request.
func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Decode request.
var req removeNodeRequest
err := json.NewDecoder(r.Body).Decode(&req)
@ -1120,6 +1202,10 @@ type removeNodeResponse struct {
// handlePostClusterResizeAbort handles POST /cluster/resize/abort request.
func (h *Handler) handlePostClusterResizeAbort(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
err := h.API.ResizeAbort()
var msg string
if err != nil {
@ -1157,6 +1243,10 @@ func (h *Handler) handleRecalculateCaches(w http.ResponseWriter, r *http.Request
}
func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Request) {
if checkHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Verify that request is only communicating over protobufs.
if r.Header.Get("Content-Type") != "application/x-protobuf" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)

28
http/handler_test.go
View file

@ -85,12 +85,12 @@ func TestHandler_Schema(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
@ -122,12 +122,12 @@ func TestHandler_Status(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
@ -750,11 +750,17 @@ func TestHandler_Index_AttrStore_Diff(t *testing.T) {
blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
// Send block checksums to determine diff.
resp, err := gohttp.Post(
req, err := gohttp.NewRequest(
"POST",
s.URL+"/index/i/attr/diff",
"application/json",
strings.NewReader(`{"blocks":`+string(test.MustMarshalJSON(blks))+`}`),
)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
client := &gohttp.Client{}
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}
@ -800,11 +806,17 @@ func TestHandler_Field_AttrStore_Diff(t *testing.T) {
blks[1].Checksum = []byte("MISMATCHED_CHECKSUM")
// Send block checksums to determine diff.
resp, err := gohttp.Post(
req, err := gohttp.NewRequest(
"POST",
s.URL+"/index/i/field/meta/attr/diff",
"application/json",
strings.NewReader(`{"blocks":`+string(test.MustMarshalJSON(blks))+`}`),
)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
client := &gohttp.Client{}
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}

154
server.go
View file

@ -52,19 +52,19 @@ type Server struct {
closing chan struct{}
// Internal
Holder *Holder
Cluster *Cluster
TranslateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
holder *Holder
Cluster *Cluster
translateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
hosts []string
clusterDisabled bool
// External
handler Handler
Broadcaster Broadcaster
BroadcastReceiver BroadcastReceiver
systemInfo SystemInfo
gcNotifier GCNotifier
NewAttrStore func(string) AttrStore
logger Logger
ln net.Listener
@ -107,8 +107,7 @@ func OptServerDataDir(dir string) ServerOption {
func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
return func(s *Server) error {
s.NewAttrStore = af
s.Holder.NewAttrStore = af
s.holder.NewAttrStore = af
return nil
}
}
@ -180,7 +179,7 @@ func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
func OptServerStatsClient(sc StatsClient) ServerOption {
return func(s *Server) error {
s.Holder.Stats = sc
s.holder.Stats = sc
return nil
}
}
@ -207,21 +206,28 @@ func OptServerURI(uri *URI) ServerOption {
}
}
// OptClusterDisabled tells the server whether to use a static cluster with the
// defined hosts. Mostly used for testing.
func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
return func(s *Server) error {
s.hosts = hosts
s.clusterDisabled = disabled
return nil
}
}
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
Cluster: NewCluster(),
Holder: NewHolder(),
Broadcaster: NopBroadcaster,
holder: NewHolder(),
BroadcastReceiver: NopBroadcastReceiver,
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
systemInfo: NewNopSystemInfo(),
gcNotifier: NopGCNotifier,
NewAttrStore: NewNopAttrStore,
antiEntropyInterval: time.Minute * 10,
metricInterval: 0,
diagnosticInterval: 0,
@ -242,19 +248,19 @@ func NewServer(opts ...ServerOption) (*Server, error) {
return nil, err
}
s.Holder.Path = path
s.Holder.Logger = s.logger
s.Holder.Stats.SetLogger(s.logger)
s.holder.Path = path
s.holder.Logger = s.logger
s.holder.Stats.SetLogger(s.logger)
s.Cluster.Path = path
s.Cluster.Logger = s.logger
s.Cluster.Holder = s.Holder
s.Cluster.Holder = s.holder
// Initialize translation database.
s.TranslateFile = NewTranslateFile()
s.TranslateFile.Path = filepath.Join(path, "keys")
s.TranslateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.TranslateFile.Open(); err != nil {
s.translateFile = NewTranslateFile()
s.translateFile.Path = filepath.Join(path, "keys")
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.translateFile.Open(); err != nil {
return nil, err
}
@ -263,6 +269,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
// Get or create NodeID.
s.NodeID = s.LoadNodeID()
// Set Cluster Node.
node := &Node{
@ -271,15 +278,23 @@ func NewServer(opts ...ServerOption) (*Server, error) {
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
}
s.Cluster.Node = node
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
if s.clusterDisabled {
err := s.Cluster.setStatic(s.hosts)
if err != nil {
return nil, errors.Wrap(err, "setting cluster static")
}
}
s.executor.Holder = s.Holder
// Append the NodeID tag to stats.
s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
s.executor.Holder = s.holder
s.executor.Node = node
s.executor.Cluster = s.Cluster
s.executor.TranslateStore = s.TranslateFile
s.executor.TranslateStore = s.translateFile
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.handler.GetAPI().Executor = s.executor
s.handler.GetAPI().TranslateStore = s.TranslateFile
s.handler.GetAPI().TranslateStore = s.translateFile
return s, nil
}
@ -287,39 +302,30 @@ func NewServer(opts ...ServerOption) (*Server, error) {
// Open opens and initializes the server.
func (s *Server) Open() error {
s.logger.Printf("open server")
// s.ln can be configured prior to Open() via s.OpenListener().
if s.ln == nil {
return errors.New("Must pass a listener option to NewServer")
return errors.New("must pass a listener option to NewServer")
}
// Log startup
err := s.Holder.logStartup()
err := s.holder.logStartup()
if err != nil {
log.Println(errors.Wrap(err, "logging startup"))
}
// Get or create NodeID.
// Append the NodeID tag to stats.
// Create default HTTP client
// Create executor for executing queries.
// Cluster settings.
s.Cluster.Broadcaster = s.Broadcaster
s.Cluster.Broadcaster = s
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
// Initialize HTTP handler.
api := s.handler.GetAPI()
api.Holder = s.Holder
api.Broadcaster = s.Broadcaster
api.Holder = s.holder
api.Broadcaster = s
api.BroadcastHandler = s
api.StatusHandler = s
api.Cluster = s.Cluster
// Initialize Holder.
s.Holder.Broadcaster = s.Broadcaster
s.holder.Broadcaster = s
// Serve handler.
go s.handler.Serve(s.ln, s.closing)
@ -335,7 +341,7 @@ func (s *Server) Open() error {
}
// Open holder.
if err := s.Holder.Open(); err != nil {
if err := s.holder.Open(); err != nil {
return fmt.Errorf("opening Holder: %v", err)
}
if err := s.Cluster.setNodeState(NodeStateReady); err != nil {
@ -370,11 +376,11 @@ func (s *Server) Close() error {
if s.Cluster != nil {
s.Cluster.close()
}
if s.Holder != nil {
s.Holder.Close()
if s.holder != nil {
s.holder.Close()
}
if s.TranslateFile != nil {
s.TranslateFile.Close()
if s.translateFile != nil {
s.translateFile.Close()
}
return nil
@ -386,7 +392,7 @@ func (s *Server) LoadNodeID() string {
if s.NodeID != "" {
return s.NodeID
}
nodeID, err := s.Holder.loadNodeID()
nodeID, err := s.holder.loadNodeID()
if err != nil {
s.logger.Printf("loading NodeID: %v", err)
return s.NodeID
@ -414,18 +420,18 @@ func (s *Server) monitorAntiEntropy() {
case <-s.closing:
return
case <-ticker.C:
s.Holder.Stats.Count("AntiEntropy", 1, 1.0)
s.holder.Stats.Count("AntiEntropy", 1, 1.0)
}
t := time.Now()
s.logger.Printf("holder sync beginning")
// Initialize syncer with local holder and remote client.
var syncer HolderSyncer
syncer.Holder = s.Holder
syncer.Holder = s.holder
syncer.Node = s.Cluster.Node
syncer.Cluster = s.Cluster
syncer.Closing = s.closing
syncer.Stats = s.Holder.Stats.WithTags("HolderSyncer")
syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
// Sync holders.
if err := syncer.SyncHolder(); err != nil {
@ -436,7 +442,7 @@ func (s *Server) monitorAntiEntropy() {
// Record successful sync in log.
s.logger.Printf("holder sync complete")
dif := time.Since(t)
s.Holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
s.holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
}
}
@ -444,23 +450,23 @@ func (s *Server) monitorAntiEntropy() {
func (s *Server) ReceiveMessage(pb proto.Message) error {
switch obj := pb.(type) {
case *internal.CreateSliceMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
idx.SetRemoteMaxSlice(obj.Slice)
case *internal.CreateIndexMessage:
opt := IndexOptions{}
_, err := s.Holder.CreateIndex(obj.Index, opt)
_, err := s.holder.CreateIndex(obj.Index, opt)
if err != nil {
return err
}
case *internal.DeleteIndexMessage:
if err := s.Holder.DeleteIndex(obj.Index); err != nil {
if err := s.holder.DeleteIndex(obj.Index); err != nil {
return err
}
case *internal.CreateFieldMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
@ -470,12 +476,12 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.DeleteFieldMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if err := idx.DeleteField(obj.Field); err != nil {
return err
}
case *internal.CreateViewMessage:
f := s.Holder.Field(obj.Index, obj.Field)
f := s.holder.Field(obj.Index, obj.Field)
if f == nil {
return fmt.Errorf("Local Field not found: %s", obj.Field)
}
@ -484,7 +490,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.DeleteViewMessage:
f := s.Holder.Field(obj.Index, obj.Field)
f := s.holder.Field(obj.Index, obj.Field)
if f == nil {
return fmt.Errorf("Local Field not found: %s", obj.Field)
}
@ -517,7 +523,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.RecalculateCaches:
s.Holder.RecalculateCaches()
s.holder.RecalculateCaches()
case *internal.NodeEventMessage:
s.Cluster.ReceiveEvent(DecodeNodeEvent(obj))
}
@ -569,14 +575,14 @@ func (s *Server) LocalStatus() (proto.Message, error) {
if s.Cluster == nil {
return nil, errors.New("Server.Cluster is nil")
}
if s.Holder == nil {
if s.holder == nil {
return nil, errors.New("Server.Holder is nil")
}
ns := internal.NodeStatus{
Node: EncodeNode(s.Cluster.Node),
MaxSlices: s.Holder.EncodeMaxSlices(),
Schema: s.Holder.EncodeSchema(),
MaxSlices: s.holder.EncodeMaxSlices(),
Schema: s.holder.EncodeSchema(),
}
return &ns, nil
@ -596,7 +602,7 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error {
go func() {
// Make sure the holder has opened.
<-s.Holder.opened
<-s.holder.opened
err := s.mergeRemoteStatus(pb.(*internal.NodeStatus))
if err != nil {
@ -614,14 +620,14 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
}
// Sync schema.
if err := s.Holder.ApplySchema(ns.Schema); err != nil {
if err := s.holder.ApplySchema(ns.Schema); err != nil {
return errors.Wrap(err, "applying schema")
}
// Sync maxSlices.
oldmaxslices := s.Holder.MaxSlices()
oldmaxslices := s.holder.MaxSlices()
for index, newMax := range ns.MaxSlices.Standard {
localIndex := s.Holder.Index(index)
localIndex := s.holder.Index(index)
// if we don't know about an index locally, log an error because
// indexes should be created and synced prior to slice creation
if localIndex == nil {
@ -709,26 +715,26 @@ func (s *Server) monitorRuntime() {
return
case <-s.gcNotifier.AfterGC():
// GC just ran.
s.Holder.Stats.Count("garbage_collection", 1, 1.0)
s.holder.Stats.Count("garbage_collection", 1, 1.0)
case <-ticker.C:
}
// Record the number of go routines.
s.Holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0)
s.holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0)
openFiles, err := countOpenFiles()
// Open File handles.
if err == nil {
s.Holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0)
s.holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0)
}
// Runtime memory metrics.
runtime.ReadMemStats(&m)
s.Holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0)
s.Holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0)
s.Holder.Stats.Gauge("StackInuse", float64(m.StackInuse), 1.0)
s.Holder.Stats.Gauge("Mallocs", float64(m.Mallocs), 1.0)
s.Holder.Stats.Gauge("Frees", float64(m.Frees), 1.0)
s.holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0)
s.holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0)
s.holder.Stats.Gauge("StackInuse", float64(m.StackInuse), 1.0)
s.holder.Stats.Gauge("Mallocs", float64(m.Mallocs), 1.0)
s.holder.Stats.Gauge("Frees", float64(m.Frees), 1.0)
}
}

32
server/server.go
View file

@ -65,10 +65,10 @@ type Command struct {
// Standard input/output
*pilosa.CmdIO
// Started will be closed once Command.Run is finished.
// Started will be closed once Command.Start is finished.
Started chan struct{}
// Done will be closed when Command.Close() is called
Done chan struct{}
// done will be closed when Command.Close() is called
done chan struct{}
// Passed to the Gossip implementation.
logOutput io.Writer
@ -83,7 +83,7 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
Started: make(chan struct{}),
Done: make(chan struct{}),
done: make(chan struct{}),
}
}
@ -125,7 +125,7 @@ func (m *Command) Wait() error {
// Second signal causes a hard shutdown.
go func() { <-c; os.Exit(1) }()
return errors.Wrap(m.Close(), "closing command")
case <-m.Done:
case <-m.done:
m.logger.Printf("Server closed externally")
return nil
}
@ -242,6 +242,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
)
return errors.Wrap(err, "new server")
@ -249,25 +250,7 @@ func (m *Command) SetupServer() error {
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {
m.Server.NodeID = m.Server.LoadNodeID()
if m.Config.Cluster.Disabled {
m.Server.Cluster.Static = true
m.Server.Cluster.Coordinator = m.Server.NodeID
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
m.Server.Cluster.Nodes = append(m.Server.Cluster.Nodes, &pilosa.Node{
URI: *uri,
})
}
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet(m.Server.Cluster.Nodes)
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
return nil
}
@ -310,7 +293,6 @@ func (m *Command) SetupNetworking() error {
}
gossipMemberSet.Logger = m.logger
m.Server.Cluster.MemberSet = gossipMemberSet
m.Server.Broadcaster = m.Server
m.Server.BroadcastReceiver = gossipMemberSet
return nil
}
@ -322,7 +304,7 @@ func (m *Command) Close() error {
if closer, ok := m.logOutput.(io.Closer); ok {
logErr = closer.Close()
}
close(m.Done)
close(m.done)
if serveErr != nil && logErr != nil {
return fmt.Errorf("closing server: '%v', closing logs: '%v'", serveErr, logErr)
} else if logErr != nil {

10
test/frame.go → test/field.go
View file

@ -18,7 +18,6 @@ import (
"io/ioutil"
"os"
"testing"
"time"
"github.com/pilosa/pilosa"
)
@ -75,15 +74,6 @@ func (f *Field) Reopen() error {
return nil
}
// MustSetBit sets a bit on the field. Panic on error.
func (f *Field) MustSetBit(view string, rowID, columnID uint64, t *time.Time) (changed bool) {
changed, err := f.SetBit(view, rowID, columnID, t)
if err != nil {
panic(err)
}
return changed
}
// Ensure field can set its cache
func TestField_SetCacheSize(t *testing.T) {
f := MustOpenField()

1
test/handler.go
View file

@ -148,6 +148,7 @@ func MustParseURLHost(rawurl string) string {
// MustNewHTTPRequest creates a new HTTP request. Panic on error.
func MustNewHTTPRequest(method, urlStr string, body io.Reader) *gohttp.Request {
req, err := gohttp.NewRequest(method, urlStr, body)
req.Header.Add("Accept", "application/json")
if err != nil {
panic(err)
}

4
test/holder.go
View file

@ -142,7 +142,7 @@ func (h *Holder) SetBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.SetBit(pilosa.ViewStandard, rowID, columnID, nil)
f.SetBit(rowID, columnID, nil)
}
// ClearBit clears a bit on the given field.
@ -152,7 +152,7 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.ClearBit(pilosa.ViewStandard, rowID, columnID, nil)
f.ClearBit(rowID, columnID, nil)
}
// MustSetBits sets columns on a row. Panic on error.

17
test/pilosa.go
View file

@ -25,7 +25,6 @@ import (
"testing"
"time"
"github.com/pilosa/pilosa/boltdb"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
@ -164,9 +163,6 @@ func (m *Main) Reopen() error {
return errors.Wrap(err, "setting up server")
}
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = m.Server.NewAttrStore
// Run new program.
if err := m.Start(); err != nil {
return err
@ -267,10 +263,15 @@ func (m *Main) RecalculateCaches() error {
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
func MustDo(method, urlStr string, body string) *httpResponse {
req, err := gohttp.NewRequest(method, urlStr, strings.NewReader(body))
if err != nil {
panic(err)
}
req, err := gohttp.NewRequest(
method,
urlStr,
strings.NewReader(body),
)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
resp, err := gohttp.DefaultClient.Do(req)
if err != nil {
panic(err)

16
test/pilosa_test.go
View file

@ -17,6 +17,7 @@ package test_test
import (
"encoding/json"
"net/http"
"strings"
"testing"
"github.com/pilosa/pilosa"
@ -32,12 +33,21 @@ func TestNewCluster(t *testing.T) {
t.Fatalf("node %d does not have the same coordinator as node 0. '%v' and '%v' respectively", i, coordi, coordinator)
}
}
req, err := http.NewRequest(
"GET",
"http://"+cluster[0].Server.Addr().String()+"/status",
strings.NewReader(""),
)
response, err := http.Get("http://" + cluster[0].Server.Addr().String() + "/status")
req.Header.Set("Accept", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("getting schema: %v", err)
t.Fatalf("sending request: %v", err)
}
dec := json.NewDecoder(response.Body)
defer resp.Body.Close()
dec := json.NewDecoder(resp.Body)
body := struct {
State string
Nodes []struct {

10
utils_internal_test.go
View file

@ -117,7 +117,7 @@ func (t *ClusterCluster) CreateField(index, field string, opt FieldOptions) erro
return nil
}
func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64, x *time.Time) error {
func (t *ClusterCluster) SetBit(index, field string, rowID, colID uint64, x *time.Time) error {
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
slice := colID / SliceWidth
@ -132,7 +132,7 @@ func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64,
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
_, err := f.SetBit(view, rowID, colID, x)
_, err := f.SetBit(rowID, colID, x)
if err != nil {
return err
}
@ -150,11 +150,11 @@ func (t *ClusterCluster) clusterByID(id string) *Cluster {
return nil
}
// AddNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode(saveTopology bool) error {
// addNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode() error {
id := len(t.Clusters)
c, err := t.addCluster(id, saveTopology)
c, err := t.addCluster(id, false)
if err != nil {
return err
}