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

Merge pull request #152 from travisturner/db-specific-slices

Browse source 
Continuation of PR #143: DB-specific slice logic
This commit is contained in:
tgruben 2016-12-01 09:13:55 -06:00 • committed by GitHub
commit 7fb3ffda16
14 changed files with 232 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
client.go
View file

@ -45,8 +45,8 @@ func NewClient(host string) (*Client, error) {
// Host returns the host the client was initialized with.
func (c *Client) Host() string { return c.host }
// SliceN returns the number of slices on a server.
func (c *Client) SliceN(ctx context.Context) (uint64, error) {
// MaxSliceByDatabase returns the number of slices on a server by database.
func (c *Client) MaxSliceByDatabase(ctx context.Context) (map[string]uint64, error) {
// Execute request against the host.
u := url.URL{
Scheme: "http",
@ -57,24 +57,24 @@ func (c *Client) SliceN(ctx context.Context) (uint64, error) {
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return 0, err
return nil, err
}
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return 0, err
return nil, err
}
defer resp.Body.Close()
var rsp sliceMaxResponse
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("http: status=%d", resp.StatusCode)
return nil, fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return 0, fmt.Errorf("json decode: %s", err)
return nil, fmt.Errorf("json decode: %s", err)
}
return rsp.SliceMax, nil
return rsp.MaxSlices, nil
}
// Schema returns all database and frame schema information.
@ -362,13 +362,13 @@ func (c *Client) BackupTo(ctx context.Context, w io.Writer, db, frame string) er
tw := tar.NewWriter(w)
// Find the maximum number of slices.
sliceN, err := c.SliceN(ctx)
maxSlices, err := c.MaxSliceByDatabase(ctx)
if err != nil {
return fmt.Errorf("slice n: %s", err)
}
// Backup every slice to the tar file.
for i := uint64(0); i <= sliceN; i++ {
for i := uint64(0); i <= maxSlices[db]; i++ {
if err := c.backupSliceTo(ctx, tw, db, frame, i); err != nil {
return err
}

2
cluster.go
View file

@ -31,7 +31,7 @@ func (a Nodes) Contains(n *Node) bool {
return false
}
// ContainsHost returns true if host matches on of the node's host.
// ContainsHost returns true if host matches one of the node's host.
func (a Nodes) ContainsHost(host string) bool {
for _, n := range a {
if n.Host == host {

4
cmd/pilosactl/main.go
View file

@ -480,13 +480,13 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
}
// Determine slice count.
sliceN, err := client.SliceN(ctx)
maxSlices, err := client.MaxSliceByDatabase(ctx)
if err != nil {
return err
}
// Export each slice.
for slice := uint64(0); slice <= sliceN; slice++ {
for slice := uint64(0); slice <= maxSlices[cmd.Database]; slice++ {
logger.Printf("exporting slice: %d", slice)
if err := client.ExportCSV(ctx, cmd.Database, cmd.Frame, slice, w); err != nil {
return err

27
db.go
View file

@ -18,6 +18,9 @@ type DB struct {
// Frames by name.
frames map[string]*Frame
// Max Slice on any node in the cluster, according to this node
remoteMaxSlice uint64
// Profile attribute storage and cache
profileAttrStore *AttrStore
@ -27,9 +30,10 @@ type DB struct {
// NewDB returns a new instance of DB.
func NewDB(path, name string) *DB {
return &DB{
path: path,
name: name,
frames: make(map[string]*Frame),
path: path,
name: name,
frames: make(map[string]*Frame),
remoteMaxSlice: 0,
profileAttrStore: NewAttrStore(filepath.Join(path, "data")),
@ -112,14 +116,17 @@ func (db *DB) Close() error {
return nil
}
// SliceN returns the max slice in the database.
func (db *DB) SliceN() uint64 {
// MaxSlice returns the max slice in the database according to this node.
func (db *DB) MaxSlice() uint64 {
if db == nil {
return 0
}
db.mu.Lock()
defer db.mu.Unlock()
var max uint64
max := db.remoteMaxSlice
for _, f := range db.frames {
if slice := f.SliceN(); slice > max {
if slice := f.MaxSlice(); slice > max {
max = slice
}
}
@ -234,3 +241,9 @@ func MergeSchemas(a, b []*DBInfo) []*DBInfo {
return dbs
}
func (db *DB) SetRemoteMaxSlice(newmax uint64) {
db.mu.Lock()
defer db.mu.Unlock()
db.remoteMaxSlice = newmax
}

17
executor.go
View file

@ -53,11 +53,10 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices
// If slices aren't specified, then include all of them.
if len(slices) == 0 {
// Round up the number of slices.
sliceN := e.Index.SliceN()
sliceN += (sliceN % uint64(len(e.Cluster.Nodes))) + uint64(len(e.Cluster.Nodes))
maxSlice := e.Index.DB(db).MaxSlice()
// Generate a slices of all slices.
slices = make([]uint64, sliceN+1)
slices = make([]uint64, maxSlice+1)
for i := range slices {
slices[i] = uint64(i)
}
@ -714,7 +713,7 @@ func (e *Executor) mapReduce(ctx context.Context, db string, slices []uint64, c
// Iterate over all map responses and reduce.
var result interface{}
var sliceN int
var maxSlice int
for {
select {
case <-ctx.Done():
@ -739,8 +738,8 @@ func (e *Executor) mapReduce(ctx context.Context, db string, slices []uint64, c
result = reduceFn(result, resp.result)
// If all slices have been processed then return.
sliceN += len(resp.slices)
if sliceN >= len(slices) {
maxSlice += len(resp.slices)
if maxSlice >= len(slices) {
return result, nil
}
}
@ -798,7 +797,7 @@ func (e *Executor) mapperLocal(ctx context.Context, slices []uint64, mapFn mapFu
}
// Reduce results
var sliceN int
var maxSlice int
var result interface{}
for {
select {
@ -809,11 +808,11 @@ func (e *Executor) mapperLocal(ctx context.Context, slices []uint64, mapFn mapFu
return nil, resp.err
}
result = reduceFn(result, resp.result)
sliceN++
maxSlice++
}
// Exit once all slices are processed.
if sliceN == len(slices) {
if maxSlice == len(slices) {
return result, nil
}
}

4
executor_test.go
View file

@ -281,7 +281,7 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) {
t.Fatalf("unexpected db: %s", db)
} else if query.String() != `Bitmap(id=10, frame=f)` {
t.Fatalf("unexpected query: %s", query.String())
} else if !reflect.DeepEqual(slices, []uint64{0, 2, 4}) {
} else if !reflect.DeepEqual(slices, []uint64{0}) { //TODO: this is incorrect because the calling node doesn't know about slice 2
t.Fatalf("unexpected slices: %+v", slices)
}
@ -390,7 +390,7 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) {
s.Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
if db != `d` {
t.Fatalf("unexpected db: %s", db)
} else if !reflect.DeepEqual(slices, []uint64{0, 2, 4, 6}) {
} else if !reflect.DeepEqual(slices, []uint64{0, 2}) {
t.Fatalf("unexpected slices: %+v", slices)
}

8
frame.go
View file

@ -58,8 +58,8 @@ func (f *Frame) Path() string { return f.path }
// BitmapAttrStore returns the attribute storage.
func (f *Frame) BitmapAttrStore() *AttrStore { return f.bitmapAttrStore }
// SliceN returns the max slice in the frame.
func (f *Frame) SliceN() uint64 {
// MaxSlice returns the max slice in the frame.
func (f *Frame) MaxSlice() uint64 {
f.mu.Lock()
defer f.mu.Unlock()
@ -128,7 +128,7 @@ func (f *Frame) openFragments() error {
frag.BitmapAttrStore = f.bitmapAttrStore
f.fragments[frag.Slice()] = frag
f.stats.Count("sliceN", 1)
f.stats.Count("maxSlice", 1)
}
return nil
@ -202,7 +202,7 @@ func (f *Frame) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
// Save to lookup.
f.fragments[slice] = frag
f.stats.Count("sliceN", 1)
f.stats.Count("maxSlice", 1)
return frag, nil
}

17
handler.go
View file

@ -253,10 +253,10 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) error {
sm := h.Index.SliceN()
ms := h.Index.MaxSlices()
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
pb := &internal.SliceMaxResponse{
SliceMax: &sm,
pb := &internal.MaxSlicesResponse{
MaxSlices: ms,
}
if buf, err := proto.Marshal(pb); err != nil {
return err
@ -265,11 +265,13 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) erro
}
return nil
}
return json.NewEncoder(w).Encode(sliceMaxResponse{SliceMax: sm})
return json.NewEncoder(w).Encode(sliceMaxResponse{
MaxSlices: ms,
})
}
type sliceMaxResponse struct {
SliceMax uint64 `json:"SliceMax"`
MaxSlices map[string]uint64 `json:"MaxSlices"`
}
// handleDeleteDB handles DELETE /db request.
@ -816,14 +818,15 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
}
// Determine the maximum number of slices.
sliceN, err := client.SliceN(r.Context())
maxSlices, err := client.MaxSliceByDatabase(r.Context())
if err != nil {
http.Error(w, "cannot determine remote slice count: "+err.Error(), http.StatusInternalServerError)
return
}
// Loop over each slice and import it if this node owns it.
for slice := uint64(0); slice <= sliceN; slice++ {
//travis
for slice := uint64(0); slice <= maxSlices[db]; slice++ {
// Ignore this slice if we don't own it.
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
continue

24
handler_test.go
View file

@ -54,6 +54,30 @@ func TestHandler_Schema(t *testing.T) {
}
}
// Ensure the handler can return the maxslice map.
func TestHandler_MaxSlices(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d0", "f0", 1).MustSetBits(30, (1*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d0", "f0", 1).MustSetBits(30, (1*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d0", "f0", 3).MustSetBits(30, (3*SliceWidth)+4)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+8)
h := NewHandler()
h.Index = idx.Index
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
// Ensure the handler can accept URL arguments.
func TestHandler_Query_Args_URL(t *testing.T) {
h := NewHandler()

35
index.go
View file

@ -18,8 +18,7 @@ const DefaultCacheFlushInterval = 1 * time.Minute
// Index represents a container for fragments.
type Index struct {
mu sync.Mutex
remoteMax uint64
mu sync.Mutex
// Databases by name.
dbs map[string]*DB
@ -43,9 +42,8 @@ type Index struct {
// NewIndex returns a new instance of Index.
func NewIndex() *Index {
return &Index{
dbs: make(map[string]*DB),
remoteMax: 0,
closing: make(chan struct{}, 0),
dbs: make(map[string]*DB),
closing: make(chan struct{}, 0),
Stats: NopStatsClient,
@ -108,18 +106,13 @@ func (i *Index) Close() error {
return nil
}
// SliceN returns the highest slice across all frames.
func (i *Index) SliceN() uint64 {
i.mu.Lock()
defer i.mu.Unlock()
sliceN := i.remoteMax
for _, db := range i.dbs {
if n := db.SliceN(); n > sliceN {
sliceN = n
}
// MaxSlices returns MaxSlice map for all databases.
func (i *Index) MaxSlices() map[string]uint64 {
a := make(map[string]uint64)
for _, db := range i.DBs() {
a[db.Name()] = db.MaxSlice()
}
return sliceN
return a
}
// Schema returns schema data for all databases and frames.
@ -266,12 +259,6 @@ func (i *Index) CreateFragmentIfNotExists(db, frame string, slice uint64) (*Frag
return f.CreateFragmentIfNotExists(slice)
}
func (i *Index) SetMax(newmax uint64) {
i.mu.Lock()
defer i.mu.Unlock()
i.remoteMax = newmax
}
// monitorCacheFlush periodically flushes all fragment caches sequentially.
// This is run in a goroutine.
func (i *Index) monitorCacheFlush() {
@ -332,8 +319,6 @@ func (s *IndexSyncer) IsClosing() bool {
// SyncIndex compares the index on host with the local index and resolves differences.
func (s *IndexSyncer) SyncIndex() error {
sliceN := s.Index.SliceN()
// Iterate over schema in sorted order.
for _, di := range s.Index.Schema() {
// Verify syncer has not closed.
@ -357,7 +342,7 @@ func (s *IndexSyncer) SyncIndex() error {
return fmt.Errorf("frame sync error: db=%s, frame=%s, err=%s", di.Name, fi.Name, err)
}
for slice := uint64(0); slice <= sliceN; slice++ {
for slice := uint64(0); slice <= s.Index.DB(di.Name).MaxSlice(); slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) {
continue

9
index_test.go
View file

@ -111,7 +111,8 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
}
// Set highest slice.
idx0.SetMax(3)
idx0.DB("d").SetRemoteMaxSlice(1)
idx0.DB("y").SetRemoteMaxSlice(3)
// Set up syncer.
syncer := pilosa.IndexSyncer{
@ -119,6 +120,7 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
Host: cluster.Nodes[0].Host,
Cluster: cluster,
}
if err := syncer.SyncIndex(); err != nil {
t.Fatal(err)
}
@ -139,10 +141,13 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
}
f = idx.Fragment("d", "f0", 1)
a := f.Bitmap(9).Bits()
if !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
}
if a := f.Bitmap(9).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
}
f = idx.Fragment("y", "z", 3)
if a := f.Bitmap(10).Bits(); !reflect.DeepEqual(a, []uint64{(3 * SliceWidth) + 4, (3 * SliceWidth) + 5, (3 * SliceWidth) + 7}) {
t.Fatalf("unexpected bits(%d/y/z): %+v", i, a)

188
internal/internal.pb.go
View file

@ -23,7 +23,7 @@ It has these top-level messages:
BlockDataRequest
BlockDataResponse
Cache
SliceMaxResponse
MaxSlicesResponse
*/
package internal
@ -38,11 +38,13 @@ var _ = math.Inf
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
const _ = proto.GoGoProtoPackageIsVersion1
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
type Bitmap struct {
Bits []uint64 `protobuf:"varint,1,rep,name=Bits" json:"Bits,omitempty"`
Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs" json:"Attrs,omitempty"`
Bits []uint64 `protobuf:"varint,1,rep,name=Bits,json=bits" json:"Bits,omitempty"`
Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs,json=attrs" json:"Attrs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -66,8 +68,8 @@ func (m *Bitmap) GetAttrs() []*Attr {
}
type Pair struct {
Key *uint64 `protobuf:"varint,1,req,name=Key" json:"Key,omitempty"`
Count *uint64 `protobuf:"varint,2,req,name=Count" json:"Count,omitempty"`
Key *uint64 `protobuf:"varint,1,req,name=Key,json=key" json:"Key,omitempty"`
Count *uint64 `protobuf:"varint,2,req,name=Count,json=count" json:"Count,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -91,8 +93,8 @@ func (m *Pair) GetCount() uint64 {
}
type Bit struct {
BitmapID *uint64 `protobuf:"varint,1,req,name=BitmapID" json:"BitmapID,omitempty"`
ProfileID *uint64 `protobuf:"varint,2,req,name=ProfileID" json:"ProfileID,omitempty"`
BitmapID *uint64 `protobuf:"varint,1,req,name=BitmapID,json=bitmapID" json:"BitmapID,omitempty"`
ProfileID *uint64 `protobuf:"varint,2,req,name=ProfileID,json=profileID" json:"ProfileID,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -116,8 +118,8 @@ func (m *Bit) GetProfileID() uint64 {
}
type Profile struct {
ID *uint64 `protobuf:"varint,1,req,name=ID" json:"ID,omitempty"`
Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs" json:"Attrs,omitempty"`
ID *uint64 `protobuf:"varint,1,req,name=ID,json=iD" json:"ID,omitempty"`
Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs,json=attrs" json:"Attrs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -141,10 +143,10 @@ func (m *Profile) GetAttrs() []*Attr {
}
type Attr struct {
Key *string `protobuf:"bytes,1,req,name=Key" json:"Key,omitempty"`
StringValue *string `protobuf:"bytes,2,opt,name=StringValue" json:"StringValue,omitempty"`
UintValue *uint64 `protobuf:"varint,3,opt,name=UintValue" json:"UintValue,omitempty"`
BoolValue *bool `protobuf:"varint,4,opt,name=BoolValue" json:"BoolValue,omitempty"`
Key *string `protobuf:"bytes,1,req,name=Key,json=key" json:"Key,omitempty"`
StringValue *string `protobuf:"bytes,2,opt,name=StringValue,json=stringValue" json:"StringValue,omitempty"`
UintValue *uint64 `protobuf:"varint,3,opt,name=UintValue,json=uintValue" json:"UintValue,omitempty"`
BoolValue *bool `protobuf:"varint,4,opt,name=BoolValue,json=boolValue" json:"BoolValue,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -182,7 +184,7 @@ func (m *Attr) GetBoolValue() bool {
}
type AttrMap struct {
Attrs []*Attr `protobuf:"bytes,1,rep,name=Attrs" json:"Attrs,omitempty"`
Attrs []*Attr `protobuf:"bytes,1,rep,name=Attrs,json=attrs" json:"Attrs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -199,13 +201,13 @@ func (m *AttrMap) GetAttrs() []*Attr {
}
type QueryRequest struct {
DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"`
Query *string `protobuf:"bytes,2,req,name=Query" json:"Query,omitempty"`
Slices []uint64 `protobuf:"varint,3,rep,name=Slices" json:"Slices,omitempty"`
Profiles *bool `protobuf:"varint,4,opt,name=Profiles" json:"Profiles,omitempty"`
Timestamp *int64 `protobuf:"varint,5,opt,name=Timestamp" json:"Timestamp,omitempty"`
Quantum *uint32 `protobuf:"varint,6,opt,name=Quantum" json:"Quantum,omitempty"`
Remote *bool `protobuf:"varint,7,opt,name=Remote" json:"Remote,omitempty"`
DB *string `protobuf:"bytes,1,req,name=DB,json=dB" json:"DB,omitempty"`
Query *string `protobuf:"bytes,2,req,name=Query,json=query" json:"Query,omitempty"`
Slices []uint64 `protobuf:"varint,3,rep,name=Slices,json=slices" json:"Slices,omitempty"`
Profiles *bool `protobuf:"varint,4,opt,name=Profiles,json=profiles" json:"Profiles,omitempty"`
Timestamp *int64 `protobuf:"varint,5,opt,name=Timestamp,json=timestamp" json:"Timestamp,omitempty"`
Quantum *uint32 `protobuf:"varint,6,opt,name=Quantum,json=quantum" json:"Quantum,omitempty"`
Remote *bool `protobuf:"varint,7,opt,name=Remote,json=remote" json:"Remote,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -264,9 +266,9 @@ func (m *QueryRequest) GetRemote() bool {
}
type QueryResponse struct {
Err *string `protobuf:"bytes,1,opt,name=Err" json:"Err,omitempty"`
Results []*QueryResult `protobuf:"bytes,2,rep,name=Results" json:"Results,omitempty"`
Profiles []*Profile `protobuf:"bytes,3,rep,name=Profiles" json:"Profiles,omitempty"`
Err *string `protobuf:"bytes,1,opt,name=Err,json=err" json:"Err,omitempty"`
Results []*QueryResult `protobuf:"bytes,2,rep,name=Results,json=results" json:"Results,omitempty"`
Profiles []*Profile `protobuf:"bytes,3,rep,name=Profiles,json=profiles" json:"Profiles,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -297,10 +299,10 @@ func (m *QueryResponse) GetProfiles() []*Profile {
}
type QueryResult struct {
Bitmap *Bitmap `protobuf:"bytes,1,opt,name=Bitmap" json:"Bitmap,omitempty"`
N *uint64 `protobuf:"varint,2,opt,name=N" json:"N,omitempty"`
Pairs []*Pair `protobuf:"bytes,3,rep,name=Pairs" json:"Pairs,omitempty"`
Changed *bool `protobuf:"varint,4,opt,name=Changed" json:"Changed,omitempty"`
Bitmap *Bitmap `protobuf:"bytes,1,opt,name=Bitmap,json=bitmap" json:"Bitmap,omitempty"`
N *uint64 `protobuf:"varint,2,opt,name=N,json=n" json:"N,omitempty"`
Pairs []*Pair `protobuf:"bytes,3,rep,name=Pairs,json=pairs" json:"Pairs,omitempty"`
Changed *bool `protobuf:"varint,4,opt,name=Changed,json=changed" json:"Changed,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -338,11 +340,11 @@ func (m *QueryResult) GetChanged() bool {
}
type ImportRequest struct {
DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"`
Frame *string `protobuf:"bytes,2,req,name=Frame" json:"Frame,omitempty"`
Slice *uint64 `protobuf:"varint,3,req,name=Slice" json:"Slice,omitempty"`
BitmapIDs []uint64 `protobuf:"varint,4,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"`
ProfileIDs []uint64 `protobuf:"varint,5,rep,name=ProfileIDs" json:"ProfileIDs,omitempty"`
DB *string `protobuf:"bytes,1,req,name=DB,json=dB" json:"DB,omitempty"`
Frame *string `protobuf:"bytes,2,req,name=Frame,json=frame" json:"Frame,omitempty"`
Slice *uint64 `protobuf:"varint,3,req,name=Slice,json=slice" json:"Slice,omitempty"`
BitmapIDs []uint64 `protobuf:"varint,4,rep,name=BitmapIDs,json=bitmapIDs" json:"BitmapIDs,omitempty"`
ProfileIDs []uint64 `protobuf:"varint,5,rep,name=ProfileIDs,json=profileIDs" json:"ProfileIDs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -387,7 +389,7 @@ func (m *ImportRequest) GetProfileIDs() []uint64 {
}
type ImportResponse struct {
Err *string `protobuf:"bytes,1,opt,name=Err" json:"Err,omitempty"`
Err *string `protobuf:"bytes,1,opt,name=Err,json=err" json:"Err,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -404,10 +406,10 @@ func (m *ImportResponse) GetErr() string {
}
type BlockDataRequest struct {
DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"`
Frame *string `protobuf:"bytes,2,req,name=Frame" json:"Frame,omitempty"`
Slice *uint64 `protobuf:"varint,3,req,name=Slice" json:"Slice,omitempty"`
Block *uint64 `protobuf:"varint,4,req,name=Block" json:"Block,omitempty"`
DB *string `protobuf:"bytes,1,req,name=DB,json=dB" json:"DB,omitempty"`
Frame *string `protobuf:"bytes,2,req,name=Frame,json=frame" json:"Frame,omitempty"`
Slice *uint64 `protobuf:"varint,3,req,name=Slice,json=slice" json:"Slice,omitempty"`
Block *uint64 `protobuf:"varint,4,req,name=Block,json=block" json:"Block,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -445,8 +447,8 @@ func (m *BlockDataRequest) GetBlock() uint64 {
}
type BlockDataResponse struct {
BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"`
ProfileIDs []uint64 `protobuf:"varint,2,rep,name=ProfileIDs" json:"ProfileIDs,omitempty"`
BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs,json=bitmapIDs" json:"BitmapIDs,omitempty"`
ProfileIDs []uint64 `protobuf:"varint,2,rep,name=ProfileIDs,json=profileIDs" json:"ProfileIDs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -470,7 +472,7 @@ func (m *BlockDataResponse) GetProfileIDs() []uint64 {
}
type Cache struct {
BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"`
BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs,json=bitmapIDs" json:"BitmapIDs,omitempty"`
XXX_unrecognized []byte `json:"-"`
}
@ -486,21 +488,21 @@ func (m *Cache) GetBitmapIDs() []uint64 {
return nil
}
type SliceMaxResponse struct {
SliceMax *uint64 `protobuf:"varint,1,req,name=SliceMax" json:"SliceMax,omitempty"`
XXX_unrecognized []byte `json:"-"`
type MaxSlicesResponse struct {
MaxSlices map[string]uint64 `protobuf:"bytes,1,rep,name=MaxSlices,json=maxSlices" json:"MaxSlices,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
XXX_unrecognized []byte `json:"-"`
}
func (m *SliceMaxResponse) Reset() { *m = SliceMaxResponse{} }
func (m *SliceMaxResponse) String() string { return proto.CompactTextString(m) }
func (*SliceMaxResponse) ProtoMessage() {}
func (*SliceMaxResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{14} }
func (m *MaxSlicesResponse) Reset() { *m = MaxSlicesResponse{} }
func (m *MaxSlicesResponse) String() string { return proto.CompactTextString(m) }
func (*MaxSlicesResponse) ProtoMessage() {}
func (*MaxSlicesResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{14} }
func (m *SliceMaxResponse) GetSliceMax() uint64 {
if m != nil && m.SliceMax != nil {
return *m.SliceMax
func (m *MaxSlicesResponse) GetMaxSlices() map[string]uint64 {
if m != nil {
return m.MaxSlices
}
return 0
return nil
}
func init() {
@ -518,41 +520,53 @@ func init() {
proto.RegisterType((*BlockDataRequest)(nil), "internal.BlockDataRequest")
proto.RegisterType((*BlockDataResponse)(nil), "internal.BlockDataResponse")
proto.RegisterType((*Cache)(nil), "internal.Cache")
proto.RegisterType((*SliceMaxResponse)(nil), "internal.SliceMaxResponse")
proto.RegisterType((*MaxSlicesResponse)(nil), "internal.MaxSlicesResponse")
}
func init() { proto.RegisterFile("internal/internal.proto", fileDescriptorInternal) }
var fileDescriptorInternal = []byte{
// 499 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x94, 0x53, 0x5d, 0x6b, 0xd4, 0x40,
0x14, 0x65, 0x37, 0xd9, 0xcd, 0xee, 0x4d, 0x77, 0xdd, 0x1d, 0x11, 0x83, 0x50, 0x28, 0x51, 0xa4,
0xf8, 0x50, 0xa1, 0xe8, 0x8b, 0x6f, 0x6e, 0x57, 0xa1, 0x94, 0x96, 0x7e, 0xa8, 0xcf, 0x0e, 0xeb,
0xd8, 0x46, 0x93, 0x4c, 0x9c, 0x4c, 0xc0, 0x3e, 0xf9, 0xd7, 0x3d, 0x33, 0x99, 0xc9, 0x46, 0x58,
0x11, 0x9f, 0x92, 0x7b, 0xe6, 0x9e, 0x7b, 0xce, 0x9c, 0xdc, 0xd0, 0xe3, 0xac, 0xd4, 0x42, 0x95,
0x3c, 0x7f, 0xe9, 0x5f, 0x8e, 0x2a, 0x25, 0xb5, 0x64, 0x13, 0x5f, 0xa7, 0xaf, 0x69, 0xbc, 0xca,
0x74, 0xc1, 0x2b, 0xb6, 0x47, 0x21, 0xde, 0xea, 0x64, 0x70, 0x10, 0x1c, 0x86, 0x6c, 0x9f, 0x46,
0x6f, 0xb5, 0x56, 0x75, 0x32, 0x44, 0x19, 0x1f, 0xcf, 0x8f, 0xba, 0x09, 0x06, 0x4e, 0x53, 0x0a,
0x2f, 0x79, 0xa6, 0x58, 0x4c, 0xc1, 0x99, 0xb8, 0x07, 0x67, 0x08, 0xce, 0x8c, 0x46, 0x27, 0xb2,
0x29, 0x35, 0x38, 0x28, 0xd3, 0x17, 0x14, 0x60, 0x20, 0x5b, 0xd0, 0xa4, 0x55, 0x38, 0x5d, 0xbb,
0xbe, 0x25, 0x4d, 0x2f, 0x95, 0xfc, 0x9a, 0xe5, 0x02, 0x50, 0xdb, 0xfb, 0x8a, 0x22, 0x07, 0x31,
0xa2, 0x61, 0xd7, 0xf9, 0x0f, 0x17, 0x17, 0x14, 0x9a, 0x67, 0xdf, 0xc5, 0x94, 0x3d, 0xa4, 0xf8,
0x46, 0xab, 0xac, 0xbc, 0xfd, 0xc4, 0xf3, 0x46, 0x80, 0x39, 0x00, 0x08, 0xc9, 0x8f, 0xe0, 0xb6,
0x50, 0x00, 0xc8, 0xba, 0x58, 0x49, 0x99, 0xb7, 0x50, 0x08, 0x68, 0x92, 0x1e, 0x52, 0x64, 0xe6,
0x9d, 0x23, 0x8d, 0x4e, 0x79, 0xb0, 0x53, 0xf9, 0x17, 0xed, 0x5d, 0x35, 0x42, 0xdd, 0x5f, 0x8b,
0x1f, 0x8d, 0xa8, 0xb5, 0x31, 0xbd, 0x5e, 0x39, 0x03, 0x88, 0xc1, 0x9e, 0xd9, 0xab, 0x4d, 0xd9,
0x9c, 0xc6, 0x37, 0x79, 0xb6, 0x11, 0x35, 0x74, 0x4d, 0xb2, 0xc8, 0xc3, 0x5d, 0xb5, 0x6e, 0x65,
0x8d, 0x93, 0x0f, 0x59, 0x81, 0x31, 0xbc, 0xa8, 0x92, 0x11, 0xa0, 0x80, 0x3d, 0xa0, 0xe8, 0xaa,
0xe1, 0xa5, 0x6e, 0x8a, 0x64, 0x0c, 0x60, 0x66, 0xa6, 0x5c, 0x8b, 0x42, 0x6a, 0x91, 0x44, 0xd6,
0x6a, 0x46, 0x33, 0x67, 0xa0, 0xae, 0x64, 0x59, 0x0b, 0x93, 0xc1, 0x3b, 0xa5, 0x60, 0xc1, 0x5c,
0xf7, 0x39, 0x45, 0x38, 0x68, 0x72, 0xed, 0x93, 0x7b, 0xb4, 0xf5, 0xef, 0x69, 0x38, 0x65, 0x4f,
0x7b, 0x5e, 0x02, 0xdb, 0xb8, 0xdc, 0x36, 0xba, 0x93, 0xf4, 0x1b, 0xc5, 0x7d, 0xce, 0x81, 0xdf,
0x18, 0xab, 0x15, 0x1f, 0x2f, 0xb6, 0x0c, 0xb7, 0x49, 0x53, 0x1a, 0x5c, 0xd8, 0xdc, 0xed, 0x07,
0x34, 0x7b, 0xe2, 0xa7, 0xf7, 0x62, 0xb4, 0xeb, 0x83, 0x6b, 0x9e, 0xdc, 0xf1, 0xf2, 0x56, 0x7c,
0x71, 0x5f, 0xe0, 0x33, 0xcd, 0x4e, 0x8b, 0x4a, 0x2a, 0xfd, 0x97, 0x60, 0xdf, 0x2b, 0x5e, 0x08,
0x17, 0x2c, 0x4a, 0x1b, 0x2c, 0x66, 0xbb, 0xad, 0xf2, 0x7b, 0x66, 0x82, 0x35, 0x51, 0x83, 0xdd,
0x2d, 0x5a, 0x8d, 0x64, 0x81, 0xa5, 0xfb, 0x34, 0xf7, 0x0a, 0x3b, 0x92, 0x4b, 0xcf, 0x68, 0xb1,
0xca, 0xe5, 0xe6, 0xfb, 0x9a, 0x6b, 0xfe, 0xff, 0x1e, 0x50, 0x5a, 0x36, 0xf4, 0xcd, 0x56, 0xbf,
0xa1, 0x65, 0x6f, 0x98, 0x93, 0xfb, 0xc3, 0xe7, 0x60, 0x87, 0xcf, 0xa1, 0xf5, 0xf9, 0x04, 0x3f,
0x13, 0xdf, 0xdc, 0xed, 0xea, 0x4f, 0x9f, 0xd1, 0xc2, 0xaa, 0x9e, 0xf3, 0x9f, 0xdd, 0x58, 0xac,
0x95, 0xc7, 0xda, 0x9f, 0xe7, 0x77, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa7, 0x12, 0x50, 0x70, 0xfe,
0x03, 0x00, 0x00,
// 667 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xac, 0x54, 0x5f, 0x6b, 0xd4, 0x4e,
0x14, 0x65, 0xf2, 0x67, 0xb3, 0xb9, 0xdb, 0x2e, 0xed, 0xd0, 0xdf, 0xcf, 0x50, 0x44, 0x42, 0x50,
0x08, 0x82, 0x2d, 0xf4, 0x49, 0x44, 0x28, 0xa6, 0x5b, 0xb1, 0x48, 0xa5, 0x9d, 0xaa, 0xef, 0xb3,
0xdb, 0x69, 0x1b, 0x9a, 0x7f, 0x9d, 0x99, 0x14, 0xf7, 0xd1, 0x07, 0x1f, 0xfc, 0x0c, 0x7e, 0x11,
0x3f, 0x9e, 0xcc, 0x64, 0x26, 0xd9, 0x52, 0xac, 0x3e, 0xf8, 0x96, 0x73, 0x6e, 0xee, 0xdc, 0x73,
0xef, 0x99, 0x3b, 0xf0, 0x28, 0xaf, 0x24, 0xe3, 0x15, 0x2d, 0x76, 0xed, 0xc7, 0x4e, 0xc3, 0x6b,
0x59, 0xe3, 0xb1, 0xc5, 0x49, 0x06, 0xa3, 0x2c, 0x97, 0x25, 0x6d, 0x30, 0x06, 0x2f, 0xcb, 0xa5,
0x88, 0x50, 0xec, 0xa6, 0x1e, 0xf1, 0xe6, 0xb9, 0x14, 0xf8, 0x29, 0xf8, 0x6f, 0xa4, 0xe4, 0x22,
0x72, 0x62, 0x37, 0x9d, 0xec, 0x4d, 0x77, 0xfa, 0x73, 0x14, 0x4d, 0x7c, 0xaa, 0x82, 0xc9, 0x0e,
0x78, 0x27, 0x34, 0xe7, 0x78, 0x03, 0xdc, 0xf7, 0x6c, 0x19, 0xa1, 0xd8, 0x49, 0x3d, 0xe2, 0x5e,
0xb3, 0x25, 0xde, 0x02, 0xff, 0xa0, 0x6e, 0x2b, 0x19, 0x39, 0x9a, 0xf3, 0x17, 0x0a, 0x24, 0xfb,
0xe0, 0x66, 0xb9, 0xc4, 0xdb, 0x30, 0xee, 0x4a, 0x1f, 0xcd, 0x4c, 0xce, 0x78, 0x6e, 0x30, 0x7e,
0x0c, 0xe1, 0x09, 0xaf, 0x2f, 0xf2, 0x82, 0x1d, 0xcd, 0x4c, 0x72, 0xd8, 0x58, 0x22, 0xd9, 0x87,
0xc0, 0x44, 0xf1, 0x14, 0x9c, 0x3e, 0xdd, 0xc9, 0x67, 0x7f, 0xa9, 0xf8, 0x16, 0x3c, 0x05, 0x57,
0x15, 0x87, 0x9d, 0xe2, 0x18, 0x26, 0x67, 0x92, 0xe7, 0xd5, 0xe5, 0x67, 0x5a, 0xb4, 0x2c, 0x72,
0x62, 0x94, 0x86, 0x64, 0x22, 0x06, 0x4a, 0x49, 0xfb, 0x94, 0x57, 0xb2, 0x8b, 0xbb, 0x31, 0x52,
0xd2, 0x5a, 0x4b, 0xa8, 0x68, 0x56, 0xd7, 0x45, 0x17, 0xf5, 0x62, 0x94, 0x8e, 0x49, 0x38, 0xb7,
0x44, 0xb2, 0x0b, 0x81, 0xaa, 0x7b, 0x4c, 0x9b, 0x41, 0x28, 0x7a, 0x48, 0xe8, 0x4f, 0x04, 0x6b,
0xa7, 0x2d, 0xe3, 0x4b, 0xc2, 0x6e, 0x5a, 0x26, 0xa4, 0xea, 0x77, 0x96, 0x19, 0xc1, 0xce, 0x79,
0xa6, 0x26, 0xac, 0xe3, 0x7a, 0x48, 0x21, 0xf1, 0x6f, 0x14, 0xc0, 0xff, 0xc3, 0xe8, 0xac, 0xc8,
0x17, 0x4c, 0x44, 0xae, 0x76, 0x73, 0x24, 0x34, 0x52, 0x23, 0x37, 0x83, 0x13, 0x46, 0xdc, 0xd8,
0x4c, 0x55, 0x28, 0xe5, 0x1f, 0xf3, 0x92, 0x09, 0x49, 0xcb, 0x26, 0xf2, 0x63, 0x94, 0xba, 0x24,
0x94, 0x96, 0xc0, 0x11, 0x04, 0xa7, 0x2d, 0xad, 0x64, 0x5b, 0x46, 0xa3, 0x18, 0xa5, 0xeb, 0x24,
0xb8, 0xe9, 0xa0, 0xaa, 0x45, 0x58, 0x59, 0x4b, 0x16, 0x05, 0xfa, 0xc4, 0x11, 0xd7, 0x28, 0xf9,
0x8a, 0x60, 0xdd, 0x48, 0x17, 0x4d, 0x5d, 0x09, 0xa6, 0xa6, 0x7d, 0xc8, 0x79, 0x84, 0xf4, 0x4c,
0x5d, 0xc6, 0x39, 0xde, 0x85, 0x80, 0x30, 0xd1, 0x16, 0xd2, 0xfa, 0xf5, 0xdf, 0x30, 0x06, 0x9b,
0xdb, 0x16, 0x92, 0x04, 0xbc, 0xfb, 0x0b, 0xbf, 0x58, 0x69, 0xc0, 0xd5, 0x19, 0x9b, 0x43, 0x86,
0x89, 0x0c, 0x3d, 0x25, 0xdf, 0x10, 0x4c, 0x56, 0xce, 0xc1, 0xa9, 0xbd, 0xed, 0x5a, 0xc4, 0x64,
0x6f, 0x63, 0x48, 0xee, 0x78, 0x32, 0xea, 0xae, 0x20, 0x5e, 0x03, 0xf4, 0x41, 0xbb, 0xef, 0x11,
0x54, 0x29, 0xb3, 0xd4, 0x0d, 0xb7, 0x35, 0x57, 0xcc, 0x52, 0x34, 0xf1, 0x1b, 0x15, 0x54, 0x33,
0x3a, 0xb8, 0xa2, 0xd5, 0x25, 0x3b, 0x37, 0xc3, 0x0d, 0x16, 0x1d, 0x4c, 0xbe, 0x23, 0x58, 0x3f,
0x2a, 0x9b, 0x9a, 0xcb, 0x07, 0x7c, 0x7c, 0xcb, 0x69, 0xc9, 0xac, 0x8f, 0x17, 0x0a, 0x28, 0x56,
0xfb, 0x18, 0xb9, 0xdd, 0xfe, 0x68, 0x1b, 0xf5, 0x1d, 0x33, 0x8b, 0xa2, 0x6c, 0x54, 0x06, 0x87,
0x76, 0x73, 0x04, 0x7e, 0x02, 0xd0, 0xaf, 0x8e, 0x88, 0x7c, 0x1d, 0x86, 0x7e, 0x77, 0x44, 0x92,
0xc0, 0xd4, 0x4a, 0xf9, 0x9d, 0x2f, 0xc9, 0x39, 0x6c, 0x64, 0x45, 0xbd, 0xb8, 0x9e, 0x51, 0x49,
0xff, 0x85, 0xe2, 0x2d, 0xf0, 0xf5, 0x79, 0x91, 0xd7, 0xb1, 0x73, 0x05, 0x92, 0x53, 0xd8, 0x5c,
0xa9, 0x62, 0xc4, 0xdc, 0x69, 0x0e, 0x3d, 0xdc, 0x9c, 0x73, 0xaf, 0xb9, 0x67, 0xe0, 0x1f, 0xd0,
0xc5, 0xd5, 0x1f, 0x8e, 0x49, 0x7e, 0x20, 0xd8, 0x3c, 0xa6, 0x5f, 0xba, 0x1d, 0xe9, 0x4b, 0xbf,
0x83, 0xb0, 0x27, 0xcd, 0x5a, 0x3e, 0x1f, 0x9c, 0xbe, 0xf7, 0xff, 0xc0, 0x1c, 0x56, 0x92, 0x2f,
0x49, 0x58, 0x5a, 0xbc, 0xfd, 0x1a, 0xa6, 0x77, 0x83, 0x6a, 0xc6, 0xd7, 0xfa, 0xa5, 0x41, 0xf6,
0xa5, 0xd9, 0x02, 0xff, 0xb6, 0x7f, 0x63, 0x3c, 0xd2, 0x81, 0x57, 0xce, 0x4b, 0xf4, 0x2b, 0x00,
0x00, 0xff, 0xff, 0x1a, 0xd6, 0x56, 0x91, 0xb7, 0x05, 0x00, 0x00,
}

5
internal/internal.proto
View file

@ -1,3 +1,4 @@
syntax = "proto2";
package internal;
message Bitmap {
@ -82,6 +83,6 @@ message Cache {
repeated uint64 BitmapIDs = 1;
}
message SliceMaxResponse {
required uint64 SliceMax = 1;
message MaxSlicesResponse {
map<string, uint64> MaxSlices = 1;
}

52
server.go
View file

@ -115,7 +115,7 @@ func (s *Server) Open() error {
// Start background monitoring.
s.wg.Add(2)
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
go func() { defer s.wg.Done(); s.monitorMaxSlice() }()
go func() { defer s.wg.Done(); s.monitorMaxSlices() }()
return nil
}
@ -180,14 +180,14 @@ func (s *Server) monitorAntiEntropy() {
}
}
// monitorMaxSlice periodically pulls the highest slice from each node in the cluster.
func (s *Server) monitorMaxSlice() {
// monitorMaxSlices periodically pulls the highest slice from each node in the cluster.
func (s *Server) monitorMaxSlices() {
// Ignore if only one node in the cluster.
if len(s.Cluster.Nodes) <= 1 {
return
}
ticker := time.NewTicker(time.Second * time.Duration(s.PollingInterval))
ticker := time.NewTicker(s.PollingInterval)
defer ticker.Stop()
for {
@ -197,24 +197,34 @@ func (s *Server) monitorMaxSlice() {
case <-ticker.C:
}
oldmax := s.Index.SliceN()
newmax := oldmax
oldmaxslices := s.Index.MaxSlices()
for _, node := range s.Cluster.Nodes {
if s.Host != node.Host {
newslice, _ := checkMaxSlice(node.Host)
if newslice > newmax {
newmax = newslice
maxSlices, _ := checkMaxSlices(node.Host)
for db, newmax := range maxSlices {
// if we don't know about a db locally, create it
// so that the /schema endpoint can report it
if localdb := s.Index.DB(db); localdb != nil {
if newmax > oldmaxslices[db] {
oldmaxslices[db] = newmax
localdb.SetRemoteMaxSlice(newmax)
}
} else {
d, err := s.Index.CreateDBIfNotExists(db)
if err != nil {
s.logger().Printf("Failed to create DB locally: %s", db)
return
}
oldmaxslices[db] = newmax
d.SetRemoteMaxSlice(newmax)
}
}
}
}
if newmax > oldmax {
s.Index.SetMax(newmax)
}
}
}
func checkMaxSlice(hostport string) (uint64, error) {
func checkMaxSlices(hostport string) (map[string]uint64, error) {
// Create HTTP request.
req, err := http.NewRequest("GET", (&url.URL{
Scheme: "http",
@ -223,7 +233,7 @@ func checkMaxSlice(hostport string) (uint64, error) {
}).String(), nil)
if err != nil {
return 0, err
return nil, err
}
// Require protobuf encoding.
@ -233,28 +243,28 @@ func checkMaxSlice(hostport string) (uint64, error) {
// Send request to remote node.
resp, err := http.DefaultClient.Do(req)
if err != nil {
return 0, err
return nil, err
}
defer resp.Body.Close()
// Read response into buffer.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return 0, err
return nil, err
}
// Check status code.
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body)
return nil, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body)
}
// Decode response object.
pb := internal.SliceMaxResponse{}
pb := internal.MaxSlicesResponse{}
if err = proto.Unmarshal(body, &pb); err != nil {
return 0, err
return nil, err
}
return *pb.SliceMax, nil
return pb.MaxSlices, nil
}