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

Merge pull request #1785 from travisturner/cluster-resize-fix

Browse source 
Cluster resize fix
This commit is contained in:
Travis Turner 2018-12-18 09:25:52 -06:00 • committed by GitHub
commit 6784ab8ac1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 319 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
api.go
View file

@ -559,6 +559,23 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName, fieldName, viewNa
return blocks, nil
}
// FragmentData returns all data in the specified fragment.
func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName string, shard uint64) (io.WriterTo, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentData")
defer span.Finish()
if err := api.validate(apiFragmentData); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
f := api.holder.fragment(indexName, fieldName, viewName, shard)
if f == nil {
return nil, ErrFragmentNotFound
}
return f, nil
}
// Hosts returns a list of the hosts in the cluster including their ID,
// URL, and which is the coordinator.
func (api *API) Hosts(ctx context.Context) []*Node {
@ -1203,6 +1220,7 @@ const (
apiExportCSV
apiFragmentBlockData
apiFragmentBlocks
apiFragmentData
apiField
apiFieldAttrDiff
//apiHosts // not implemented
@ -1232,7 +1250,8 @@ var methodsCommon = map[apiMethod]struct{}{
}
var methodsResizing = map[apiMethod]struct{}{
apiResizeAbort: {},
apiFragmentData: {},
apiResizeAbort: {},
}
var methodsNormal = map[apiMethod]struct{}{

4
client.go
View file

@ -52,7 +52,7 @@ type InternalClient interface {
ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
SendMessage(ctx context.Context, uri *URI, msg []byte) error
RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error)
RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
}
@ -149,6 +149,6 @@ func (n nopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, fie
func (n nopInternalClient) SendMessage(ctx context.Context, uri *URI, msg []byte) error {
return nil
}
func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) {
func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error) {
return nil, nil
}

54
cluster.go
View file

@ -1218,7 +1218,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
Node: toCluster.unprotectedNodeByID(id),
Coordinator: c.unprotectedCoordinatorNode(),
Sources: sources,
Schema: &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node.
NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
ClusterStatus: c.unprotectedStatus(),
}
j.Instructions = append(j.Instructions, instr)
@ -1277,12 +1277,30 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "Cluster.followResizeInstruction")
defer span.Finish()
// Sync the schema received in the resize instruction.
// Sync the NodeStatus received in the resize instruction.
// Sync schema.
c.logger.Debugf("holder applySchema")
if err := c.holder.applySchema(instr.Schema); err != nil {
if err := c.holder.applySchema(instr.NodeStatus.Schema); err != nil {
return errors.Wrap(err, "applying schema")
}
// Sync available shards.
for _, is := range instr.NodeStatus.Indexes {
for _, fs := range is.Fields {
f := c.holder.Field(is.Name, fs.Name)
// if we don't know about a field locally, log an error because
// fields should be created and synced prior to shard creation
if f == nil {
c.logger.Printf("local field not found: %s/%s", is.Name, fs.Name)
continue
}
if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
return errors.Wrap(err, "adding remote available shards")
}
}
}
// Request each source file in ResizeSources.
for _, src := range instr.Sources {
c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
@ -1309,7 +1327,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
// Stream shard from remote node.
c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.Shard, srcURI)
rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.View, src.Shard, srcURI)
if err != nil {
// For now it is an acceptable error if the fragment is not found
// on the remote node. This occurs when a shard has been skipped and
@ -1318,7 +1336,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
// TODO: figure out a way to distinguish from "fragment not found" errors
// which are true errors and which simply mean the fragment doesn't have data.
if err == ErrFragmentNotFound {
return nil
continue
}
return errors.Wrap(err, "retrieving shard")
} else if rd == nil {
@ -1817,6 +1835,30 @@ func (c *cluster) nodeLeave(nodeID string) error {
return nil
}
func (c *cluster) nodeStatus() *NodeStatus {
ns := &NodeStatus{
Node: c.Node,
Schema: &Schema{Indexes: c.holder.Schema()},
}
var availableShards *roaring.Bitmap
for _, idx := range ns.Schema.Indexes {
is := &IndexStatus{Name: idx.Name}
for _, f := range idx.Fields {
if field := c.holder.Field(idx.Name, f.Name); field != nil {
availableShards = field.AvailableShards()
} else {
availableShards = roaring.NewBitmap()
}
is.Fields = append(is.Fields, &FieldStatus{
Name: f.Name,
AvailableShards: availableShards,
})
}
ns.Indexes = append(ns.Indexes, is)
}
return ns
}
func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
c.mu.Lock()
defer c.mu.Unlock()
@ -1918,7 +1960,7 @@ type ResizeInstruction struct {
Node *Node
Coordinator *Node
Sources []*ResizeSource
Schema *Schema
NodeStatus *NodeStatus
ClusterStatus *ClusterStatus
}

6
encoding/proto/proto.go
View file

@ -457,7 +457,7 @@ func encodeResizeInstruction(m *pilosa.ResizeInstruction) *internal.ResizeInstru
Node: encodeNode(m.Node),
Coordinator: encodeNode(m.Coordinator),
Sources: encodeResizeSources(m.Sources),
Schema: encodeSchema(m.Schema),
NodeStatus: encodeNodeStatus(m.NodeStatus),
ClusterStatus: encodeClusterStatus(m.ClusterStatus),
}
}
@ -737,8 +737,8 @@ func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeIns
decodeNode(ri.Coordinator, m.Coordinator)
m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources))
decodeResizeSources(ri.Sources, m.Sources)
m.Schema = &pilosa.Schema{}
decodeSchema(ri.Schema, m.Schema)
m.NodeStatus = &pilosa.NodeStatus{}
decodeNodeStatus(ri.NodeStatus, m.NodeStatus)
m.ClusterStatus = &pilosa.ClusterStatus{}
decodeClusterStatus(ri.ClusterStatus, m.ClusterStatus)
}

11
http/client.go
View file

@ -705,24 +705,19 @@ func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, i
return nil
}
func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
defer span.Finish()
node := &pilosa.Node{
URI: uri,
}
return c.backupShardNode(ctx, index, field, shard, node)
}
func (c *InternalClient) backupShardNode(ctx context.Context, index, field string, shard uint64, node *pilosa.Node) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.backupShardNode")
defer span.Finish()
u := nodePathToURL(node, "/fragment/data")
u := nodePathToURL(node, "/internal/fragment/data")
u.RawQuery = url.Values{
"index": {index},
"field": {field},
"view": {view},
"shard": {strconv.FormatUint(shard, 10)},
}.Encode()

23
http/handler.go
View file

@ -192,6 +192,7 @@ func (h *Handler) populateValidators() {
h.validators["PostClusterMessage"] = queryValidationSpecRequired()
h.validators["GetFragmentBlockData"] = queryValidationSpecRequired()
h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard")
h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "view", "shard")
h.validators["GetFragmentNodes"] = queryValidationSpecRequired("shard", "index")
h.validators["PostIndexAttrDiff"] = queryValidationSpecRequired()
h.validators["PostFieldAttrDiff"] = queryValidationSpecRequired()
@ -262,6 +263,7 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/cluster/message", handler.handlePostClusterMessage).Methods("POST").Name("PostClusterMessage")
router.HandleFunc("/internal/fragment/block/data", handler.handleGetFragmentBlockData).Methods("GET").Name("GetFragmentBlockData")
router.HandleFunc("/internal/fragment/blocks", handler.handleGetFragmentBlocks).Methods("GET").Name("GetFragmentBlocks")
router.HandleFunc("/internal/fragment/data", handler.handleGetFragmentData).Methods("GET").Name("GetFragmentData")
router.HandleFunc("/internal/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET").Name("GetFragmentNodes")
router.HandleFunc("/internal/index/{index}/attr/diff", handler.handlePostIndexAttrDiff).Methods("POST").Name("PostIndexAttrDiff")
router.HandleFunc("/internal/index/{index}/field/{field}/attr/diff", handler.handlePostFieldAttrDiff).Methods("POST").Name("PostFieldAttrDiff")
@ -1214,6 +1216,27 @@ type getFragmentBlocksResponse struct {
Blocks []pilosa.FragmentBlock `json:"blocks"`
}
// handleGetFragmentData handles GET /internal/fragment/data requests.
func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) {
// Read shard parameter.
q := r.URL.Query()
shard, err := strconv.ParseUint(q.Get("shard"), 10, 64)
if err != nil {
http.Error(w, "shard required", http.StatusBadRequest)
return
}
// Retrieve fragment data from holder.
f, err := h.api.FragmentData(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
// Stream fragment to response body.
if _, err := f.WriteTo(w); err != nil {
h.logger.Printf("error streaming fragment data: %s", err)
}
}
// handleGetVersion handles /version requests.
func (h *Handler) handleGetVersion(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {

313
internal/private.pb.go
View file

@ -32,7 +32,7 @@ func (m *IndexMeta) Reset() { *m = IndexMeta{} }
func (m *IndexMeta) String() string { return proto.CompactTextString(m) }
func (*IndexMeta) ProtoMessage() {}
func (*IndexMeta) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{0}
return fileDescriptor_private_8095a89af06a70de, []int{0}
}
func (m *IndexMeta) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -93,7 +93,7 @@ func (m *FieldOptions) Reset() { *m = FieldOptions{} }
func (m *FieldOptions) String() string { return proto.CompactTextString(m) }
func (*FieldOptions) ProtoMessage() {}
func (*FieldOptions) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{1}
return fileDescriptor_private_8095a89af06a70de, []int{1}
}
func (m *FieldOptions) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -189,7 +189,7 @@ func (m *ImportResponse) Reset() { *m = ImportResponse{} }
func (m *ImportResponse) String() string { return proto.CompactTextString(m) }
func (*ImportResponse) ProtoMessage() {}
func (*ImportResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{2}
return fileDescriptor_private_8095a89af06a70de, []int{2}
}
func (m *ImportResponse) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -240,7 +240,7 @@ func (m *BlockDataRequest) Reset() { *m = BlockDataRequest{} }
func (m *BlockDataRequest) String() string { return proto.CompactTextString(m) }
func (*BlockDataRequest) ProtoMessage() {}
func (*BlockDataRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{3}
return fileDescriptor_private_8095a89af06a70de, []int{3}
}
func (m *BlockDataRequest) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -316,7 +316,7 @@ func (m *BlockDataResponse) Reset() { *m = BlockDataResponse{} }
func (m *BlockDataResponse) String() string { return proto.CompactTextString(m) }
func (*BlockDataResponse) ProtoMessage() {}
func (*BlockDataResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{4}
return fileDescriptor_private_8095a89af06a70de, []int{4}
}
func (m *BlockDataResponse) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -370,7 +370,7 @@ func (m *Cache) Reset() { *m = Cache{} }
func (m *Cache) String() string { return proto.CompactTextString(m) }
func (*Cache) ProtoMessage() {}
func (*Cache) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{5}
return fileDescriptor_private_8095a89af06a70de, []int{5}
}
func (m *Cache) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -417,7 +417,7 @@ func (m *MaxShards) Reset() { *m = MaxShards{} }
func (m *MaxShards) String() string { return proto.CompactTextString(m) }
func (*MaxShards) ProtoMessage() {}
func (*MaxShards) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{6}
return fileDescriptor_private_8095a89af06a70de, []int{6}
}
func (m *MaxShards) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -466,7 +466,7 @@ func (m *CreateShardMessage) Reset() { *m = CreateShardMessage{} }
func (m *CreateShardMessage) String() string { return proto.CompactTextString(m) }
func (*CreateShardMessage) ProtoMessage() {}
func (*CreateShardMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{7}
return fileDescriptor_private_8095a89af06a70de, []int{7}
}
func (m *CreateShardMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -527,7 +527,7 @@ func (m *DeleteIndexMessage) Reset() { *m = DeleteIndexMessage{} }
func (m *DeleteIndexMessage) String() string { return proto.CompactTextString(m) }
func (*DeleteIndexMessage) ProtoMessage() {}
func (*DeleteIndexMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{8}
return fileDescriptor_private_8095a89af06a70de, []int{8}
}
func (m *DeleteIndexMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -575,7 +575,7 @@ func (m *CreateIndexMessage) Reset() { *m = CreateIndexMessage{} }
func (m *CreateIndexMessage) String() string { return proto.CompactTextString(m) }
func (*CreateIndexMessage) ProtoMessage() {}
func (*CreateIndexMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{9}
return fileDescriptor_private_8095a89af06a70de, []int{9}
}
func (m *CreateIndexMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -631,7 +631,7 @@ func (m *CreateFieldMessage) Reset() { *m = CreateFieldMessage{} }
func (m *CreateFieldMessage) String() string { return proto.CompactTextString(m) }
func (*CreateFieldMessage) ProtoMessage() {}
func (*CreateFieldMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{10}
return fileDescriptor_private_8095a89af06a70de, []int{10}
}
func (m *CreateFieldMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -693,7 +693,7 @@ func (m *DeleteFieldMessage) Reset() { *m = DeleteFieldMessage{} }
func (m *DeleteFieldMessage) String() string { return proto.CompactTextString(m) }
func (*DeleteFieldMessage) ProtoMessage() {}
func (*DeleteFieldMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{11}
return fileDescriptor_private_8095a89af06a70de, []int{11}
}
func (m *DeleteFieldMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -749,7 +749,7 @@ func (m *DeleteAvailableShardMessage) Reset() { *m = DeleteAvailableShar
func (m *DeleteAvailableShardMessage) String() string { return proto.CompactTextString(m) }
func (*DeleteAvailableShardMessage) ProtoMessage() {}
func (*DeleteAvailableShardMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{12}
return fileDescriptor_private_8095a89af06a70de, []int{12}
}
func (m *DeleteAvailableShardMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -812,7 +812,7 @@ func (m *Field) Reset() { *m = Field{} }
func (m *Field) String() string { return proto.CompactTextString(m) }
func (*Field) ProtoMessage() {}
func (*Field) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{13}
return fileDescriptor_private_8095a89af06a70de, []int{13}
}
func (m *Field) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -873,7 +873,7 @@ func (m *Schema) Reset() { *m = Schema{} }
func (m *Schema) String() string { return proto.CompactTextString(m) }
func (*Schema) ProtoMessage() {}
func (*Schema) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{14}
return fileDescriptor_private_8095a89af06a70de, []int{14}
}
func (m *Schema) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -921,7 +921,7 @@ func (m *Index) Reset() { *m = Index{} }
func (m *Index) String() string { return proto.CompactTextString(m) }
func (*Index) ProtoMessage() {}
func (*Index) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{15}
return fileDescriptor_private_8095a89af06a70de, []int{15}
}
func (m *Index) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -977,7 +977,7 @@ func (m *URI) Reset() { *m = URI{} }
func (m *URI) String() string { return proto.CompactTextString(m) }
func (*URI) ProtoMessage() {}
func (*URI) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{16}
return fileDescriptor_private_8095a89af06a70de, []int{16}
}
func (m *URI) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1041,7 +1041,7 @@ func (m *Node) Reset() { *m = Node{} }
func (m *Node) String() string { return proto.CompactTextString(m) }
func (*Node) ProtoMessage() {}
func (*Node) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{17}
return fileDescriptor_private_8095a89af06a70de, []int{17}
}
func (m *Node) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1110,7 +1110,7 @@ func (m *NodeStateMessage) Reset() { *m = NodeStateMessage{} }
func (m *NodeStateMessage) String() string { return proto.CompactTextString(m) }
func (*NodeStateMessage) ProtoMessage() {}
func (*NodeStateMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{18}
return fileDescriptor_private_8095a89af06a70de, []int{18}
}
func (m *NodeStateMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1165,7 +1165,7 @@ func (m *NodeEventMessage) Reset() { *m = NodeEventMessage{} }
func (m *NodeEventMessage) String() string { return proto.CompactTextString(m) }
func (*NodeEventMessage) ProtoMessage() {}
func (*NodeEventMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{19}
return fileDescriptor_private_8095a89af06a70de, []int{19}
}
func (m *NodeEventMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1221,7 +1221,7 @@ func (m *NodeStatus) Reset() { *m = NodeStatus{} }
func (m *NodeStatus) String() string { return proto.CompactTextString(m) }
func (*NodeStatus) ProtoMessage() {}
func (*NodeStatus) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{20}
return fileDescriptor_private_8095a89af06a70de, []int{20}
}
func (m *NodeStatus) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1283,7 +1283,7 @@ func (m *IndexStatus) Reset() { *m = IndexStatus{} }
func (m *IndexStatus) String() string { return proto.CompactTextString(m) }
func (*IndexStatus) ProtoMessage() {}
func (*IndexStatus) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{21}
return fileDescriptor_private_8095a89af06a70de, []int{21}
}
func (m *IndexStatus) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1338,7 +1338,7 @@ func (m *FieldStatus) Reset() { *m = FieldStatus{} }
func (m *FieldStatus) String() string { return proto.CompactTextString(m) }
func (*FieldStatus) ProtoMessage() {}
func (*FieldStatus) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{22}
return fileDescriptor_private_8095a89af06a70de, []int{22}
}
func (m *FieldStatus) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1394,7 +1394,7 @@ func (m *ClusterStatus) Reset() { *m = ClusterStatus{} }
func (m *ClusterStatus) String() string { return proto.CompactTextString(m) }
func (*ClusterStatus) ProtoMessage() {}
func (*ClusterStatus) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{23}
return fileDescriptor_private_8095a89af06a70de, []int{23}
}
func (m *ClusterStatus) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1458,7 +1458,7 @@ func (m *BSIGroup) Reset() { *m = BSIGroup{} }
func (m *BSIGroup) String() string { return proto.CompactTextString(m) }
func (*BSIGroup) ProtoMessage() {}
func (*BSIGroup) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{24}
return fileDescriptor_private_8095a89af06a70de, []int{24}
}
func (m *BSIGroup) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1528,7 +1528,7 @@ func (m *CreateViewMessage) Reset() { *m = CreateViewMessage{} }
func (m *CreateViewMessage) String() string { return proto.CompactTextString(m) }
func (*CreateViewMessage) ProtoMessage() {}
func (*CreateViewMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{25}
return fileDescriptor_private_8095a89af06a70de, []int{25}
}
func (m *CreateViewMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1591,7 +1591,7 @@ func (m *DeleteViewMessage) Reset() { *m = DeleteViewMessage{} }
func (m *DeleteViewMessage) String() string { return proto.CompactTextString(m) }
func (*DeleteViewMessage) ProtoMessage() {}
func (*DeleteViewMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{26}
return fileDescriptor_private_8095a89af06a70de, []int{26}
}
func (m *DeleteViewMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1646,7 +1646,7 @@ type ResizeInstruction struct {
Node *Node `protobuf:"bytes,2,opt,name=Node" json:"Node,omitempty"`
Coordinator *Node `protobuf:"bytes,3,opt,name=Coordinator" json:"Coordinator,omitempty"`
Sources []*ResizeSource `protobuf:"bytes,4,rep,name=Sources" json:"Sources,omitempty"`
Schema *Schema `protobuf:"bytes,5,opt,name=Schema" json:"Schema,omitempty"`
NodeStatus *NodeStatus `protobuf:"bytes,7,opt,name=NodeStatus" json:"NodeStatus,omitempty"`
ClusterStatus *ClusterStatus `protobuf:"bytes,6,opt,name=ClusterStatus" json:"ClusterStatus,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
@ -1657,7 +1657,7 @@ func (m *ResizeInstruction) Reset() { *m = ResizeInstruction{} }
func (m *ResizeInstruction) String() string { return proto.CompactTextString(m) }
func (*ResizeInstruction) ProtoMessage() {}
func (*ResizeInstruction) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{27}
return fileDescriptor_private_8095a89af06a70de, []int{27}
}
func (m *ResizeInstruction) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1714,9 +1714,9 @@ func (m *ResizeInstruction) GetSources() []*ResizeSource {
return nil
}
func (m *ResizeInstruction) GetSchema() *Schema {
func (m *ResizeInstruction) GetNodeStatus() *NodeStatus {
if m != nil {
return m.Schema
return m.NodeStatus
}
return nil
}
@ -1743,7 +1743,7 @@ func (m *ResizeSource) Reset() { *m = ResizeSource{} }
func (m *ResizeSource) String() string { return proto.CompactTextString(m) }
func (*ResizeSource) ProtoMessage() {}
func (*ResizeSource) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{28}
return fileDescriptor_private_8095a89af06a70de, []int{28}
}
func (m *ResizeSource) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1820,7 +1820,7 @@ func (m *ResizeInstructionComplete) Reset() { *m = ResizeInstructionComp
func (m *ResizeInstructionComplete) String() string { return proto.CompactTextString(m) }
func (*ResizeInstructionComplete) ProtoMessage() {}
func (*ResizeInstructionComplete) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{29}
return fileDescriptor_private_8095a89af06a70de, []int{29}
}
func (m *ResizeInstructionComplete) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1881,7 +1881,7 @@ func (m *SetCoordinatorMessage) Reset() { *m = SetCoordinatorMessage{} }
func (m *SetCoordinatorMessage) String() string { return proto.CompactTextString(m) }
func (*SetCoordinatorMessage) ProtoMessage() {}
func (*SetCoordinatorMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{30}
return fileDescriptor_private_8095a89af06a70de, []int{30}
}
func (m *SetCoordinatorMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1928,7 +1928,7 @@ func (m *UpdateCoordinatorMessage) Reset() { *m = UpdateCoordinatorMessa
func (m *UpdateCoordinatorMessage) String() string { return proto.CompactTextString(m) }
func (*UpdateCoordinatorMessage) ProtoMessage() {}
func (*UpdateCoordinatorMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{31}
return fileDescriptor_private_8095a89af06a70de, []int{31}
}
func (m *UpdateCoordinatorMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -1976,7 +1976,7 @@ func (m *Topology) Reset() { *m = Topology{} }
func (m *Topology) String() string { return proto.CompactTextString(m) }
func (*Topology) ProtoMessage() {}
func (*Topology) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{32}
return fileDescriptor_private_8095a89af06a70de, []int{32}
}
func (m *Topology) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2029,7 +2029,7 @@ func (m *RecalculateCaches) Reset() { *m = RecalculateCaches{} }
func (m *RecalculateCaches) String() string { return proto.CompactTextString(m) }
func (*RecalculateCaches) ProtoMessage() {}
func (*RecalculateCaches) Descriptor() ([]byte, []int) {
return fileDescriptor_private_08d4c0c27f7a355f, []int{33}
return fileDescriptor_private_8095a89af06a70de, []int{33}
}
func (m *RecalculateCaches) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -3270,21 +3270,21 @@ func (m *ResizeInstruction) MarshalTo(dAtA []byte) (int, error) {
i += n
}
}
if m.Schema != nil {
dAtA[i] = 0x2a
if m.ClusterStatus != nil {
dAtA[i] = 0x32
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Schema.Size()))
n18, err := m.Schema.MarshalTo(dAtA[i:])
i = encodeVarintPrivate(dAtA, i, uint64(m.ClusterStatus.Size()))
n18, err := m.ClusterStatus.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
i += n18
}
if m.ClusterStatus != nil {
dAtA[i] = 0x32
if m.NodeStatus != nil {
dAtA[i] = 0x3a
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.ClusterStatus.Size()))
n19, err := m.ClusterStatus.MarshalTo(dAtA[i:])
i = encodeVarintPrivate(dAtA, i, uint64(m.NodeStatus.Size()))
n19, err := m.NodeStatus.MarshalTo(dAtA[i:])
if err != nil {
return 0, err
}
@ -4169,14 +4169,14 @@ func (m *ResizeInstruction) Size() (n int) {
n += 1 + l + sovPrivate(uint64(l))
}
}
if m.Schema != nil {
l = m.Schema.Size()
n += 1 + l + sovPrivate(uint64(l))
}
if m.ClusterStatus != nil {
l = m.ClusterStatus.Size()
n += 1 + l + sovPrivate(uint64(l))
}
if m.NodeStatus != nil {
l = m.NodeStatus.Size()
n += 1 + l + sovPrivate(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -8034,39 +8034,6 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 5:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Schema", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthPrivate
}
postIndex := iNdEx + msglen
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.Schema == nil {
m.Schema = &Schema{}
}
if err := m.Schema.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 6:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ClusterStatus", wireType)
@ -8100,6 +8067,39 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 7:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field NodeStatus", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthPrivate
}
postIndex := iNdEx + msglen
if postIndex > l {
return io.ErrUnexpectedEOF
}
if m.NodeStatus == nil {
m.NodeStatus = &NodeStatus{}
}
if err := m.NodeStatus.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -8877,79 +8877,80 @@ var (
ErrIntOverflowPrivate = fmt.Errorf("proto: integer overflow")
)
func init() { proto.RegisterFile("private.proto", fileDescriptor_private_08d4c0c27f7a355f) }
func init() { proto.RegisterFile("private.proto", fileDescriptor_private_8095a89af06a70de) }
var fileDescriptor_private_08d4c0c27f7a355f = []byte{
// 1131 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xdd, 0x6e, 0x1b, 0xc5,
0x17, 0xff, 0xef, 0x87, 0x1d, 0xfb, 0xb8, 0x4e, 0x93, 0xed, 0xbf, 0x61, 0x0b, 0x28, 0x84, 0x51,
0x45, 0x43, 0x25, 0x42, 0xd5, 0xde, 0xf0, 0x55, 0xa9, 0x24, 0x0e, 0x65, 0x29, 0x09, 0x65, 0x9c,
0xe4, 0x8e, 0x8b, 0x89, 0x3d, 0x6a, 0x56, 0x59, 0xef, 0x98, 0xdd, 0xd9, 0x24, 0xee, 0x05, 0xb7,
0x20, 0xf1, 0x02, 0x88, 0x27, 0xe2, 0x92, 0x47, 0xa8, 0xc2, 0x8b, 0xa0, 0x39, 0x33, 0xb3, 0xbb,
0x76, 0x1c, 0x12, 0x05, 0xee, 0xe6, 0xfc, 0xce, 0x99, 0xf3, 0x7d, 0xce, 0xec, 0x42, 0x77, 0x9c,
0xc5, 0x27, 0x4c, 0xf2, 0x8d, 0x71, 0x26, 0xa4, 0x08, 0x5a, 0x71, 0x2a, 0x79, 0x96, 0xb2, 0x84,
0x3c, 0x87, 0x76, 0x94, 0x0e, 0xf9, 0xd9, 0x0e, 0x97, 0x2c, 0x08, 0xc0, 0x7f, 0xc1, 0x27, 0x79,
0xe8, 0xad, 0x39, 0xeb, 0x2d, 0x8a, 0xe7, 0xe0, 0x03, 0x58, 0xdc, 0xcb, 0xd8, 0xe0, 0x78, 0xfb,
0x2c, 0xce, 0x25, 0x4f, 0x07, 0x3c, 0xf4, 0x91, 0x3b, 0x83, 0x92, 0x37, 0x0e, 0xdc, 0xfa, 0x2a,
0xe6, 0xc9, 0xf0, 0xbb, 0xb1, 0x8c, 0x45, 0x9a, 0x07, 0xef, 0x42, 0x7b, 0x8b, 0x0d, 0x8e, 0xf8,
0xde, 0x64, 0xcc, 0x51, 0x63, 0x9b, 0x56, 0x40, 0xc9, 0xed, 0xc7, 0xaf, 0xb5, 0xc6, 0x2e, 0xad,
0x80, 0x60, 0x0d, 0x3a, 0x7b, 0xf1, 0x88, 0x7f, 0x5f, 0xb0, 0x54, 0x16, 0xa3, 0xb0, 0x81, 0xb7,
0xeb, 0x90, 0x72, 0x15, 0x15, 0xb7, 0x90, 0x85, 0xe7, 0x60, 0x09, 0xbc, 0x9d, 0x38, 0x0d, 0xdb,
0x6b, 0xce, 0xba, 0x47, 0xd5, 0x11, 0x11, 0x76, 0x16, 0x82, 0x41, 0xd8, 0x59, 0x19, 0x62, 0x67,
0x3a, 0xc4, 0x5d, 0xd1, 0x97, 0x2c, 0x1d, 0xb2, 0x6c, 0x78, 0x10, 0xf3, 0xd3, 0xf0, 0x96, 0x0e,
0x71, 0x1a, 0x25, 0x04, 0x16, 0xa3, 0xd1, 0x58, 0x64, 0x92, 0xf2, 0x7c, 0x2c, 0xd2, 0x1c, 0x2d,
0x6e, 0x67, 0x59, 0xe8, 0xa0, 0x13, 0xea, 0x48, 0x7e, 0x82, 0xa5, 0xcd, 0x44, 0x0c, 0x8e, 0x7b,
0x4c, 0x32, 0xca, 0x7f, 0x2c, 0x78, 0x2e, 0x83, 0xff, 0x43, 0x03, 0x73, 0x6c, 0xe4, 0x34, 0xa1,
0x50, 0xcc, 0x57, 0xe8, 0x6a, 0x14, 0x09, 0x85, 0xe2, 0x7d, 0xcc, 0x98, 0x4f, 0x35, 0xa1, 0xd0,
0xfe, 0x11, 0xcb, 0x86, 0x98, 0x29, 0x9f, 0x6a, 0x42, 0xc5, 0x82, 0xde, 0xea, 0xf4, 0xe0, 0x99,
0x44, 0xb0, 0x5c, 0xb3, 0x6f, 0xdc, 0x5c, 0x81, 0x26, 0x15, 0xa7, 0x51, 0x2f, 0x0f, 0x9d, 0x35,
0x6f, 0xdd, 0xa7, 0x86, 0xc2, 0x22, 0x88, 0xa4, 0x18, 0xa5, 0x8a, 0xe5, 0x22, 0xab, 0x02, 0xc8,
0x3d, 0x68, 0x60, 0x45, 0x54, 0x94, 0xd5, 0x5d, 0x75, 0x24, 0x3f, 0x3b, 0xd0, 0xde, 0x61, 0x67,
0xe8, 0x46, 0x1e, 0x3c, 0x85, 0x96, 0xcd, 0x13, 0x0a, 0x75, 0x1e, 0xbf, 0xbf, 0x61, 0x1b, 0x6c,
0xa3, 0x14, 0xdb, 0xb0, 0x32, 0xdb, 0xa9, 0xcc, 0x26, 0xb4, 0xbc, 0xf2, 0xf6, 0xe7, 0xd0, 0x9d,
0x62, 0x29, 0x7b, 0xc7, 0x7c, 0x62, 0xb3, 0x7a, 0xcc, 0x27, 0x2a, 0xfe, 0x13, 0x96, 0x14, 0x1c,
0x73, 0xe5, 0x53, 0x4d, 0x7c, 0xe6, 0x7e, 0xe2, 0x90, 0x03, 0x08, 0xb6, 0x32, 0xce, 0x24, 0x47,
0x23, 0x3b, 0x3c, 0xcf, 0xd9, 0x2b, 0x7e, 0x79, 0xc6, 0x75, 0x16, 0xdd, 0x7a, 0x16, 0xcb, 0x3a,
0x78, 0xb5, 0x3a, 0x90, 0x87, 0x10, 0xf4, 0x78, 0xc2, 0x25, 0x37, 0xd3, 0xf1, 0x0f, 0x7a, 0x49,
0xdf, 0xfa, 0x70, 0xb5, 0x6c, 0xf0, 0x00, 0x7c, 0x35, 0x6a, 0xe8, 0x42, 0xe7, 0xf1, 0x9d, 0x2a,
0x4f, 0xe5, 0x14, 0x52, 0x14, 0x20, 0x89, 0x55, 0x8a, 0xfe, 0x5c, 0x19, 0xd8, 0x9c, 0x56, 0x7a,
0x68, 0x4c, 0x79, 0x68, 0x6a, 0xa5, 0x32, 0x55, 0x1f, 0x53, 0x63, 0xed, 0x99, 0x0d, 0xf7, 0xa6,
0xd6, 0xc8, 0x00, 0xde, 0xd1, 0x1a, 0xbe, 0x3c, 0x61, 0x71, 0xc2, 0x0e, 0x93, 0x6b, 0x56, 0x64,
0x8e, 0xe3, 0x21, 0x2c, 0xe0, 0xdd, 0xa8, 0x67, 0xa6, 0xc0, 0x92, 0xe4, 0x07, 0x23, 0xaf, 0x5a,
0x7f, 0x97, 0x8d, 0xb8, 0xd1, 0x86, 0xe7, 0x32, 0x5e, 0xf7, 0xea, 0x78, 0x95, 0x61, 0x35, 0x2e,
0x6a, 0xd5, 0x79, 0xca, 0x30, 0x12, 0xe4, 0x09, 0x34, 0xfb, 0x83, 0x23, 0x3e, 0x62, 0xc1, 0x87,
0xb0, 0x80, 0x1e, 0xf2, 0xdc, 0x74, 0xf4, 0xed, 0x99, 0x4a, 0x51, 0xcb, 0x27, 0x3d, 0x13, 0xd9,
0x5c, 0x9f, 0x1e, 0x40, 0x13, 0xad, 0xe7, 0xa1, 0x3f, 0xab, 0x06, 0x71, 0x6a, 0xd8, 0x64, 0x1b,
0xbc, 0x7d, 0x1a, 0xa9, 0x49, 0x45, 0x0f, 0xac, 0x16, 0x43, 0x29, 0xdd, 0x5f, 0x8b, 0x5c, 0x9a,
0x3c, 0xe1, 0x59, 0x61, 0x2f, 0x45, 0x26, 0x31, 0x47, 0x5d, 0x8a, 0x67, 0x92, 0x83, 0xbf, 0x2b,
0x86, 0x3c, 0x58, 0x04, 0x37, 0xea, 0x19, 0x1d, 0x6e, 0xd4, 0x0b, 0xde, 0x43, 0xf5, 0x26, 0x35,
0xdd, 0xca, 0x89, 0x7d, 0x1a, 0x51, 0x34, 0x7c, 0x1f, 0xba, 0x51, 0xbe, 0x25, 0x44, 0x36, 0x8c,
0x53, 0x26, 0x45, 0x66, 0xde, 0x80, 0x69, 0x10, 0x27, 0x48, 0x32, 0xa9, 0x37, 0x76, 0x9b, 0x6a,
0x82, 0x3c, 0x83, 0x25, 0x65, 0x14, 0x09, 0x5b, 0xef, 0x15, 0x68, 0x2a, 0xac, 0x74, 0xc2, 0x50,
0x95, 0x06, 0xb7, 0xae, 0xe1, 0x5b, 0xad, 0x61, 0xfb, 0x84, 0xa7, 0xb2, 0xd6, 0x31, 0x48, 0xa3,
0x82, 0x2e, 0xd5, 0x44, 0x40, 0x74, 0x80, 0x26, 0x92, 0xc5, 0x2a, 0x12, 0x85, 0x52, 0xe4, 0x91,
0x5f, 0x1d, 0x00, 0xeb, 0x50, 0x91, 0x97, 0x57, 0x9c, 0xcb, 0xaf, 0x04, 0xeb, 0xb6, 0xf2, 0x66,
0x5a, 0x96, 0x2a, 0x29, 0x8d, 0x53, 0xdb, 0x19, 0x1f, 0x57, 0x9d, 0xa1, 0x4b, 0x7a, 0x77, 0xa6,
0x33, 0xb4, 0xd5, 0xaa, 0x3f, 0x5e, 0x42, 0xa7, 0x86, 0xcf, 0xed, 0x92, 0x8f, 0xca, 0x2e, 0x71,
0x67, 0x55, 0x22, 0x6e, 0x54, 0xda, 0x5e, 0x79, 0x01, 0x9d, 0x1a, 0x3c, 0x57, 0xe3, 0x3a, 0xdc,
0x9e, 0x9e, 0x43, 0xbb, 0xdf, 0x67, 0x61, 0x12, 0x43, 0x77, 0x2b, 0x29, 0x72, 0xc9, 0x33, 0xa3,
0x4e, 0x3d, 0x0a, 0x1a, 0x28, 0x8b, 0x57, 0x01, 0xf3, 0xeb, 0x17, 0xdc, 0x87, 0x86, 0x4a, 0xa3,
0x1e, 0xa7, 0x8b, 0x39, 0xd6, 0x4c, 0x72, 0x00, 0xad, 0xcd, 0x7e, 0xf4, 0x3c, 0x13, 0xc5, 0x78,
0xae, 0xd3, 0xf6, 0x4d, 0x77, 0x2f, 0xbe, 0xe9, 0xde, 0x85, 0x37, 0xdd, 0x2f, 0xdf, 0x74, 0xd2,
0x87, 0x65, 0xbd, 0x2a, 0xd5, 0x14, 0xdf, 0x64, 0xe1, 0xd8, 0x87, 0xd4, 0xab, 0x3d, 0xa4, 0x7d,
0x58, 0xd6, 0xfb, 0xec, 0xbf, 0x54, 0xfa, 0xbb, 0x0b, 0xcb, 0x94, 0xe7, 0xf1, 0x6b, 0x1e, 0xa5,
0xb9, 0xcc, 0x8a, 0x81, 0xda, 0x49, 0xea, 0xfe, 0x37, 0xe2, 0xd0, 0x64, 0xdb, 0xa3, 0x9a, 0xb8,
0x4e, 0xa7, 0x07, 0x8f, 0xa0, 0x33, 0x3b, 0xb3, 0x17, 0x45, 0xeb, 0x22, 0xc1, 0x23, 0x58, 0xe8,
0x8b, 0x22, 0x1b, 0x94, 0xed, 0x5b, 0xdb, 0x93, 0xda, 0x33, 0xcd, 0xa6, 0x56, 0xac, 0x36, 0x1a,
0x8d, 0x2b, 0x46, 0xe3, 0xe9, 0x4c, 0x2b, 0x85, 0x4d, 0xbc, 0xf0, 0x56, 0x75, 0x61, 0x8a, 0x4d,
0xa7, 0xa5, 0xc9, 0x2f, 0x0e, 0xdc, 0xaa, 0xbb, 0x70, 0xad, 0xc1, 0x2d, 0x2b, 0xe2, 0xce, 0xad,
0x88, 0x37, 0xaf, 0x22, 0x7e, 0x55, 0x91, 0xea, 0x9b, 0xa0, 0x51, 0xfb, 0x26, 0x20, 0xc7, 0x70,
0xef, 0x42, 0x99, 0xb6, 0xc4, 0x68, 0xac, 0xfa, 0xe1, 0x5f, 0x94, 0x4b, 0xad, 0xb4, 0x2c, 0x33,
0x85, 0x6a, 0x53, 0x4d, 0x90, 0x4f, 0xe1, 0x6e, 0x9f, 0xcb, 0x5a, 0x91, 0x6c, 0xb7, 0xad, 0x81,
0xb7, 0xcb, 0x4f, 0x2f, 0x09, 0x5f, 0xb1, 0xc8, 0x17, 0x10, 0xee, 0x8f, 0x87, 0x4c, 0xf2, 0x1b,
0xdd, 0xde, 0x84, 0xd6, 0x9e, 0x18, 0x8b, 0x44, 0xbc, 0x9a, 0x5c, 0x31, 0xf5, 0x21, 0x2c, 0xe8,
0xfd, 0xad, 0xd7, 0x48, 0x9b, 0x5a, 0x92, 0xdc, 0x51, 0x0d, 0x3d, 0x60, 0xc9, 0xa0, 0x48, 0x94,
0x1b, 0xea, 0x7b, 0x31, 0xdf, 0x5c, 0xfa, 0xe3, 0x7c, 0xd5, 0xf9, 0xf3, 0x7c, 0xd5, 0x79, 0x73,
0xbe, 0xea, 0xfc, 0xf6, 0xd7, 0xea, 0xff, 0x0e, 0x9b, 0xf8, 0xdf, 0xf1, 0xe4, 0xef, 0x00, 0x00,
0x00, 0xff, 0xff, 0x61, 0x80, 0xe4, 0xef, 0x88, 0x0c, 0x00, 0x00,
var fileDescriptor_private_8095a89af06a70de = []byte{
// 1139 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xdb, 0x6e, 0x1b, 0xc5,
0x1b, 0xff, 0xef, 0x21, 0x8e, 0xfd, 0x39, 0x4e, 0x93, 0x6d, 0x9b, 0xff, 0x16, 0x50, 0x08, 0xa3,
0x8a, 0x86, 0x4a, 0x84, 0xaa, 0xe5, 0x82, 0x53, 0xa5, 0x92, 0x38, 0x94, 0xa5, 0x24, 0x94, 0x71,
0x92, 0x3b, 0x2e, 0x26, 0xf6, 0xa8, 0x59, 0x65, 0xbd, 0x63, 0x76, 0x67, 0x93, 0xb8, 0x17, 0xdc,
0x82, 0xc4, 0x0b, 0xf0, 0x04, 0x3c, 0x0b, 0x97, 0x3c, 0x42, 0x15, 0x5e, 0x04, 0xcd, 0x37, 0x33,
0xbb, 0x6b, 0xc7, 0x21, 0x51, 0xe0, 0x6e, 0xbe, 0xdf, 0x77, 0x3e, 0xae, 0x0d, 0x9d, 0x51, 0x16,
0x9f, 0x30, 0xc9, 0x37, 0x46, 0x99, 0x90, 0x22, 0x68, 0xc6, 0xa9, 0xe4, 0x59, 0xca, 0x12, 0xf2,
0x1c, 0x5a, 0x51, 0x3a, 0xe0, 0x67, 0x3b, 0x5c, 0xb2, 0x20, 0x00, 0xff, 0x05, 0x1f, 0xe7, 0xa1,
0xb7, 0xe6, 0xac, 0x37, 0x29, 0xbe, 0x83, 0xf7, 0x61, 0x71, 0x2f, 0x63, 0xfd, 0xe3, 0xed, 0xb3,
0x38, 0x97, 0x3c, 0xed, 0xf3, 0xd0, 0x47, 0xee, 0x14, 0x4a, 0xde, 0x38, 0xb0, 0xf0, 0x55, 0xcc,
0x93, 0xc1, 0x77, 0x23, 0x19, 0x8b, 0x34, 0x0f, 0xde, 0x81, 0xd6, 0x16, 0xeb, 0x1f, 0xf1, 0xbd,
0xf1, 0x88, 0xa3, 0xc5, 0x16, 0xad, 0x80, 0x92, 0xdb, 0x8b, 0x5f, 0x6b, 0x8b, 0x1d, 0x5a, 0x01,
0xc1, 0x1a, 0xb4, 0xf7, 0xe2, 0x21, 0xff, 0xbe, 0x60, 0xa9, 0x2c, 0x86, 0xe1, 0x1c, 0x6a, 0xd7,
0x21, 0x15, 0x2a, 0x1a, 0x6e, 0x22, 0x0b, 0xdf, 0xc1, 0x12, 0x78, 0x3b, 0x71, 0x1a, 0xb6, 0xd6,
0x9c, 0x75, 0x8f, 0xaa, 0x27, 0x22, 0xec, 0x2c, 0x04, 0x83, 0xb0, 0xb3, 0x32, 0xc5, 0xf6, 0x64,
0x8a, 0xbb, 0xa2, 0x27, 0x59, 0x3a, 0x60, 0xd9, 0xe0, 0x20, 0xe6, 0xa7, 0xe1, 0x82, 0x4e, 0x71,
0x12, 0x25, 0x04, 0x16, 0xa3, 0xe1, 0x48, 0x64, 0x92, 0xf2, 0x7c, 0x24, 0xd2, 0x1c, 0x3d, 0x6e,
0x67, 0x59, 0xe8, 0x60, 0x10, 0xea, 0x49, 0x7e, 0x82, 0xa5, 0xcd, 0x44, 0xf4, 0x8f, 0xbb, 0x4c,
0x32, 0xca, 0x7f, 0x2c, 0x78, 0x2e, 0x83, 0x3b, 0x30, 0x87, 0x35, 0x36, 0x72, 0x9a, 0x50, 0x28,
0xd6, 0x2b, 0x74, 0x35, 0x8a, 0x84, 0x42, 0x51, 0x1f, 0x2b, 0xe6, 0x53, 0x4d, 0x28, 0xb4, 0x77,
0xc4, 0xb2, 0x01, 0x56, 0xca, 0xa7, 0x9a, 0x50, 0xb9, 0x60, 0xb4, 0xba, 0x3c, 0xf8, 0x26, 0x11,
0x2c, 0xd7, 0xfc, 0x9b, 0x30, 0x57, 0xa0, 0x41, 0xc5, 0x69, 0xd4, 0xcd, 0x43, 0x67, 0xcd, 0x5b,
0xf7, 0xa9, 0xa1, 0xb0, 0x09, 0x22, 0x29, 0x86, 0xa9, 0x62, 0xb9, 0xc8, 0xaa, 0x00, 0x72, 0x0f,
0xe6, 0xb0, 0x23, 0x2a, 0xcb, 0x4a, 0x57, 0x3d, 0xc9, 0xcf, 0x0e, 0xb4, 0x76, 0xd8, 0x19, 0x86,
0x91, 0x07, 0x4f, 0xa1, 0x69, 0xeb, 0x84, 0x42, 0xed, 0xc7, 0xef, 0x6d, 0xd8, 0x01, 0xdb, 0x28,
0xc5, 0x36, 0xac, 0xcc, 0x76, 0x2a, 0xb3, 0x31, 0x2d, 0x55, 0xde, 0xfa, 0x1c, 0x3a, 0x13, 0x2c,
0xe5, 0xef, 0x98, 0x8f, 0x6d, 0x55, 0x8f, 0xf9, 0x58, 0xe5, 0x7f, 0xc2, 0x92, 0x82, 0x63, 0xad,
0x7c, 0xaa, 0x89, 0xcf, 0xdc, 0x4f, 0x1c, 0x72, 0x00, 0xc1, 0x56, 0xc6, 0x99, 0xe4, 0xe8, 0x64,
0x87, 0xe7, 0x39, 0x7b, 0xc5, 0x2f, 0xaf, 0xb8, 0xae, 0xa2, 0x5b, 0xaf, 0x62, 0xd9, 0x07, 0xaf,
0xd6, 0x07, 0xf2, 0x10, 0x82, 0x2e, 0x4f, 0xb8, 0xe4, 0x66, 0x3b, 0xfe, 0xc1, 0x2e, 0xe9, 0xd9,
0x18, 0xae, 0x96, 0x0d, 0x1e, 0x80, 0xaf, 0x56, 0x0d, 0x43, 0x68, 0x3f, 0xbe, 0x5d, 0xd5, 0xa9,
0xdc, 0x42, 0x8a, 0x02, 0x24, 0xb1, 0x46, 0x31, 0x9e, 0x2b, 0x13, 0x9b, 0x31, 0x4a, 0x0f, 0x8d,
0x2b, 0x0f, 0x5d, 0xad, 0x54, 0xae, 0xea, 0x6b, 0x6a, 0xbc, 0x3d, 0xb3, 0xe9, 0xde, 0xd4, 0x1b,
0xe9, 0xc3, 0xdb, 0xda, 0xc2, 0x97, 0x27, 0x2c, 0x4e, 0xd8, 0x61, 0x72, 0xcd, 0x8e, 0xcc, 0x08,
0x3c, 0x84, 0x79, 0xd4, 0x8d, 0xba, 0x66, 0x0b, 0x2c, 0x49, 0x7e, 0x30, 0xf2, 0x6a, 0xf4, 0x77,
0xd9, 0x90, 0x1b, 0x6b, 0xf8, 0x2e, 0xf3, 0x75, 0xaf, 0xce, 0x57, 0x39, 0x56, 0xeb, 0xa2, 0x4e,
0x9d, 0xa7, 0x1c, 0x23, 0x41, 0x9e, 0x40, 0xa3, 0xd7, 0x3f, 0xe2, 0x43, 0x16, 0x7c, 0x00, 0xf3,
0x18, 0x21, 0xcf, 0xcd, 0x44, 0xdf, 0x9a, 0xea, 0x14, 0xb5, 0x7c, 0xd2, 0x35, 0x99, 0xcd, 0x8c,
0xe9, 0x01, 0x34, 0xd0, 0x7b, 0x1e, 0xfa, 0xd3, 0x66, 0x10, 0xa7, 0x86, 0x4d, 0xb6, 0xc1, 0xdb,
0xa7, 0x91, 0xda, 0x54, 0x8c, 0xc0, 0x5a, 0x31, 0x94, 0xb2, 0xfd, 0xb5, 0xc8, 0xa5, 0xa9, 0x13,
0xbe, 0x15, 0xf6, 0x52, 0x64, 0x12, 0x6b, 0xd4, 0xa1, 0xf8, 0x26, 0x39, 0xf8, 0xbb, 0x62, 0xc0,
0x83, 0x45, 0x70, 0xa3, 0xae, 0xb1, 0xe1, 0x46, 0xdd, 0xe0, 0x5d, 0x34, 0x6f, 0x4a, 0xd3, 0xa9,
0x82, 0xd8, 0xa7, 0x11, 0x45, 0xc7, 0xf7, 0xa1, 0x13, 0xe5, 0x5b, 0x42, 0x64, 0x83, 0x38, 0x65,
0x52, 0x64, 0xe6, 0x1b, 0x30, 0x09, 0xe2, 0x06, 0x49, 0x26, 0xf5, 0xc5, 0x6e, 0x51, 0x4d, 0x90,
0x67, 0xb0, 0xa4, 0x9c, 0x22, 0x61, 0xfb, 0xbd, 0x02, 0x0d, 0x85, 0x95, 0x41, 0x18, 0xaa, 0xb2,
0xe0, 0xd6, 0x2d, 0x7c, 0xab, 0x2d, 0x6c, 0x9f, 0xf0, 0x54, 0xd6, 0x26, 0x06, 0x69, 0x34, 0xd0,
0xa1, 0x9a, 0x08, 0x88, 0x4e, 0xd0, 0x64, 0xb2, 0x58, 0x65, 0xa2, 0x50, 0x8a, 0x3c, 0xf2, 0xab,
0x03, 0x60, 0x03, 0x2a, 0xf2, 0x52, 0xc5, 0xb9, 0x5c, 0x25, 0x58, 0xb7, 0x9d, 0x37, 0xdb, 0xb2,
0x54, 0x49, 0x69, 0x9c, 0xda, 0xc9, 0xf8, 0xa8, 0x9a, 0x0c, 0xdd, 0xd2, 0xbb, 0x53, 0x93, 0xa1,
0xbd, 0x56, 0xf3, 0xf1, 0x12, 0xda, 0x35, 0x7c, 0xe6, 0x94, 0x7c, 0x58, 0x4e, 0x89, 0x3b, 0x6d,
0x12, 0x71, 0x63, 0xd2, 0xce, 0xca, 0x0b, 0x68, 0xd7, 0xe0, 0x99, 0x16, 0xd7, 0xe1, 0xd6, 0xe4,
0x1e, 0xda, 0xfb, 0x3e, 0x0d, 0x93, 0x18, 0x3a, 0x5b, 0x49, 0x91, 0x4b, 0x9e, 0x19, 0x73, 0xea,
0xa3, 0xa0, 0x81, 0xb2, 0x79, 0x15, 0x30, 0xbb, 0x7f, 0xc1, 0x7d, 0x98, 0x53, 0x65, 0xd4, 0xeb,
0x74, 0xb1, 0xc6, 0x9a, 0x49, 0x0e, 0xa0, 0xb9, 0xd9, 0x8b, 0x9e, 0x67, 0xa2, 0x18, 0xcd, 0x0c,
0xda, 0x7e, 0xd3, 0xdd, 0x8b, 0xdf, 0x74, 0xef, 0xc2, 0x37, 0xdd, 0x2f, 0xbf, 0xe9, 0xa4, 0x07,
0xcb, 0xfa, 0x54, 0xaa, 0x2d, 0xbe, 0xc9, 0xc1, 0xb1, 0x1f, 0x52, 0xaf, 0xf6, 0x21, 0xed, 0xc1,
0xb2, 0xbe, 0x67, 0xff, 0xa5, 0xd1, 0xdf, 0x5d, 0x58, 0xa6, 0x3c, 0x8f, 0x5f, 0xf3, 0x28, 0xcd,
0x65, 0x56, 0xf4, 0xd5, 0x4d, 0x52, 0xfa, 0xdf, 0x88, 0x43, 0x53, 0x6d, 0x8f, 0x6a, 0xe2, 0x3a,
0x93, 0x1e, 0x3c, 0x82, 0xf6, 0xf4, 0xce, 0x5e, 0x14, 0xad, 0x8b, 0x04, 0x8f, 0x60, 0xbe, 0x27,
0x8a, 0xac, 0x5f, 0x8e, 0x6f, 0xed, 0x4e, 0xea, 0xc8, 0x34, 0x9b, 0x5a, 0xb1, 0xe0, 0xe9, 0xd4,
0x80, 0x84, 0x0d, 0xf4, 0xf2, 0xff, 0x4a, 0x6f, 0x82, 0x4d, 0xa7, 0xc6, 0xe9, 0xe3, 0xfa, 0x2e,
0x86, 0xf3, 0xa8, 0x7b, 0x67, 0x32, 0x42, 0xa3, 0x58, 0x93, 0x23, 0xbf, 0x38, 0xb0, 0x50, 0x0f,
0xe7, 0x5a, 0x4b, 0x5c, 0x76, 0xc7, 0x9d, 0xd9, 0x1d, 0x6f, 0x56, 0x77, 0xfc, 0xaa, 0x3b, 0xd5,
0xef, 0x83, 0xb9, 0xda, 0xef, 0x03, 0x72, 0x0c, 0xf7, 0x2e, 0xb4, 0x6c, 0x4b, 0x0c, 0x47, 0x6a,
0x36, 0xfe, 0x45, 0xeb, 0xd4, 0x79, 0xcb, 0x32, 0xd3, 0xb4, 0x16, 0xd5, 0x04, 0xf9, 0x14, 0xee,
0xf6, 0xb8, 0xac, 0x35, 0xcc, 0x4e, 0xde, 0x1a, 0x78, 0xbb, 0xfc, 0xf4, 0x92, 0xf4, 0x15, 0x8b,
0x7c, 0x01, 0xe1, 0xfe, 0x68, 0xc0, 0x24, 0xbf, 0x91, 0xf6, 0x26, 0x34, 0xf7, 0xc4, 0x48, 0x24,
0xe2, 0xd5, 0xf8, 0x8a, 0x0b, 0x10, 0xc2, 0xbc, 0xbe, 0xe5, 0xfa, 0xa4, 0xb4, 0xa8, 0x25, 0xc9,
0x6d, 0x35, 0xdc, 0x7d, 0x96, 0xf4, 0x8b, 0x44, 0x85, 0xa1, 0x7e, 0x3b, 0xe6, 0x9b, 0x4b, 0x7f,
0x9c, 0xaf, 0x3a, 0x7f, 0x9e, 0xaf, 0x3a, 0x6f, 0xce, 0x57, 0x9d, 0xdf, 0xfe, 0x5a, 0xfd, 0xdf,
0x61, 0x03, 0xff, 0x83, 0x3c, 0xf9, 0x3b, 0x00, 0x00, 0xff, 0xff, 0x1b, 0xd8, 0x6d, 0x1f, 0x94,
0x0c, 0x00, 0x00,
}

2
internal/private.proto
View file

@ -159,7 +159,7 @@ message ResizeInstruction {
Node Node = 2;
Node Coordinator = 3;
repeated ResizeSource Sources = 4;
Schema Schema = 5;
NodeStatus NodeStatus = 7;
ClusterStatus ClusterStatus = 6;
}

44
server/cluster_test.go
View file

@ -206,6 +206,16 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := `{"results":[{"attrs":{},"columns":[1,1300000]}]}` + "\n"
// Verify the data exists on the single node.
if res, err := m0.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
// Configure node1
m1 := test.NewCommandNode(false)
m1.Config.Gossip.Port = "0"
@ -221,6 +231,18 @@ func TestClusterResize_AddNode(t *testing.T) {
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
}
// Verify the data exists on both nodes.
if res, err := m0.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
if res, err := m1.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
})
t.Run("SkippedShard", func(t *testing.T) {
// Configure node0
@ -247,6 +269,16 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := `{"results":[{"attrs":{},"columns":[1,2400000]}]}` + "\n"
// Verify the data exists on the single node.
if res, err := m0.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
// Configure node1
m1 := test.NewCommandNode(false)
m1.Config.Gossip.Port = "0"
@ -262,6 +294,18 @@ func TestClusterResize_AddNode(t *testing.T) {
} else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected node1 cluster state: %s", m1.API.State())
}
// Verify the data exists on both nodes.
if res, err := m0.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
if res, err := m1.Query("i", "", `Row(f=1)`); err != nil {
t.Fatal(err)
} else if res != exp {
t.Fatalf("unexpected result: %s", res)
}
})
}

19
utils_internal_test.go
View file

@ -24,6 +24,7 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
)
// NewTestCluster returns a cluster with n nodes and uses a mod-based hasher.
@ -371,10 +372,26 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error
destCluster := t.clusterByID(instrNode.ID)
// Sync the schema received in the resize instruction.
if err := destCluster.holder.applySchema(instr.Schema); err != nil {
if err := destCluster.holder.applySchema(instr.NodeStatus.Schema); err != nil {
return err
}
// Sync available shards.
for _, is := range instr.NodeStatus.Indexes {
for _, fs := range is.Fields {
f := destCluster.holder.Field(is.Name, fs.Name)
// if we don't know about a field locally, log an error because
// fields should be created and synced prior to shard creation
if f == nil {
continue
}
if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
return errors.Wrap(err, "adding remote available shards")
}
}
}
for _, src := range instr.Sources {
srcCluster := t.clusterByID(src.Node.ID)