From 3d270f45d2df9e0affe48044ca99427ab4bceb4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Wed, 27 May 2020 21:56:11 +0200 Subject: [PATCH] Add (in memory) ETag to index and fields --- api.go | 38 +++- api_test.go | 25 ++- cluster.go | 11 +- encoding/proto/proto.go | 12 ++ field.go | 10 + gossip/gossip.go | 10 +- holder.go | 51 +++++- http/handler.go | 272 ++++++++++++++++++---------- index.go | 23 ++- internal/private.pb.go | 392 +++++++++++++++++++++++++++++++--------- internal/private.proto | 6 + pilosa.go | 5 + server.go | 13 +- server/handler_test.go | 60 ++++++ 14 files changed, 706 insertions(+), 222 deletions(-) diff --git a/api.go b/api.go index 210569f6a..43dba98f0 100644 --- a/api.go +++ b/api.go @@ -181,10 +181,15 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index if err != nil { return nil, errors.Wrap(err, "creating index") } + index.mu.Lock() + index.etag = newETag() + index.mu.Unlock() + // Send the create index message to all nodes. err = api.server.SendSync( &CreateIndexMessage{ Index: indexName, + ETag: index.ETag(), Meta: &options, }) if err != nil { @@ -269,14 +274,17 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str if err != nil { return nil, errors.Wrap(err, "creating field") } + field.mu.Lock() + field.etag = newETag() + field.mu.Unlock() // Send the create field message to all nodes. - err = api.server.SendSync( - &CreateFieldMessage{ - Index: indexName, - Field: fieldName, - Meta: &fo, - }) + err = api.server.SendSync(&CreateFieldMessage{ + Index: indexName, + Field: fieldName, + ETag: field.ETag(), + Meta: &fo, + }) if err != nil { api.server.logger.Printf("problem sending CreateField message: %s", err) return nil, errors.Wrap(err, "sending CreateField message") @@ -803,6 +811,18 @@ func (api *API) ApplySchema(ctx context.Context, s *Schema, remote bool) error { return errors.Wrap(err, "validating api method") } + // set etags for indexes and fields (if empty), and then apply schema. + for _, index := range s.Indexes { + if index.ETag == 0 { + index.ETag = newETag() + } + for _, field := range index.Fields { + if field.ETag == 0 { + field.ETag = newETag() + } + } + } + if !remote { nodes := api.cluster.Nodes() for i, node := range nodes { @@ -813,7 +833,11 @@ func (api *API) ApplySchema(ctx context.Context, s *Schema, remote bool) error { } } - return api.holder.applySchema(s) + if err := api.holder.applySchema(s); err != nil { + return errors.Wrap(err, "applying schema") + } + + return nil } // Views returns the views in the given field. diff --git a/api_test.go b/api_test.go index b7f0fc06f..c11ac4f8a 100644 --- a/api_test.go +++ b/api_test.go @@ -186,17 +186,24 @@ func TestAPI_Import(t *testing.T) { t.Run("RowIDColumnKey", func(t *testing.T) { ctx := context.Background() - index := "rick" - field := "f" + indexName := "rick" + fieldName := "f" - _, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}) + index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}) if err != nil { t.Fatalf("creating index: %v", err) } - _, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) + if index.ETag() == 0 { + t.Fatal("index etag is empty") + } + + field, err := m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) if err != nil { t.Fatalf("creating field: %v", err) } + if field.ETag() == 0 { + t.Fatal("field etag is empty") + } rowID := uint64(1) timestamp := int64(0) @@ -215,8 +222,8 @@ func TestAPI_Import(t *testing.T) { // Import data with keys to the coordinator (node0) and verify that it gets // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) req := &pilosa.ImportRequest{ - Index: index, - Field: field, + Index: indexName, + Field: fieldName, Shard: 0, RowIDs: rowIDs, ColumnKeys: colKeys, @@ -226,10 +233,10 @@ func TestAPI_Import(t *testing.T) { t.Fatal(err) } - pql := fmt.Sprintf("Row(%s=%d)", field, rowID) + pql := fmt.Sprintf("Row(%s=%d)", fieldName, rowID) // Query node0. - if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil { t.Fatal(err) } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { t.Fatalf("unexpected column keys: %#v", keys) @@ -237,7 +244,7 @@ func TestAPI_Import(t *testing.T) { // Query node1. if err := test.RetryUntil(5*time.Second, func() error { - if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil { + if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil { return err } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { return fmt.Errorf("unexpected column keys: %#v", keys) diff --git a/cluster.go b/cluster.go index e979710d8..72ce09b17 100644 --- a/cluster.go +++ b/cluster.go @@ -1272,9 +1272,7 @@ func (c *cluster) listenForJoins() { // Then we want to set the cluster state to NORMAL and resume processing of joiningLeavingNodes events. // We use a bool `setNormal` to indicate when at least one node has joined. var setNormal bool - for { - // Handle all pending joins before changing state back to NORMAL. select { case nodeAction := <-c.joiningLeavingNodes: @@ -2146,7 +2144,7 @@ func (c *cluster) nodeStatus() *NodeStatus { } var availableShards *roaring.Bitmap for _, idx := range ns.Schema.Indexes { - is := &IndexStatus{Name: idx.Name} + is := &IndexStatus{Name: idx.Name, ETag: idx.ETag} for _, f := range idx.Fields { if field := c.holder.Field(idx.Name, f.Name); field != nil { availableShards = field.AvailableShards() @@ -2155,6 +2153,7 @@ func (c *cluster) nodeStatus() *NodeStatus { } is.Fields = append(is.Fields, &FieldStatus{ Name: f.Name, + ETag: f.ETag, AvailableShards: availableShards, }) } @@ -2491,7 +2490,7 @@ type translationResizeNode struct { // Schema contains information about indexes and their configuration. type Schema struct { - Indexes []*IndexInfo + Indexes []*IndexInfo `json:"indexes"` } func encodeTopology(topology *Topology) *internal.Topology { @@ -2530,6 +2529,7 @@ type CreateShardMessage struct { // CreateIndexMessage is an internal message indicating index creation. type CreateIndexMessage struct { Index string + ETag int64 Meta *IndexOptions } @@ -2542,6 +2542,7 @@ type DeleteIndexMessage struct { type CreateFieldMessage struct { Index string Field string + ETag int64 Meta *FieldOptions } @@ -2606,12 +2607,14 @@ type NodeStatus struct { // IndexStatus is an internal message representing the contents of an index. type IndexStatus struct { Name string + ETag int64 Fields []*FieldStatus } // FieldStatus is an internal message representing the contents of a field. type FieldStatus struct { Name string + ETag int64 AvailableShards *roaring.Bitmap } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index a4508f652..6eb389302 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -594,6 +594,7 @@ func (s Serializer) encodeIndexInfos(idxs []*pilosa.IndexInfo) []*internal.Index func (s Serializer) encodeIndexInfo(idx *pilosa.IndexInfo) *internal.Index { return &internal.Index{ Name: idx.Name, + ETag: idx.ETag, Options: s.encodeIndexMeta(&idx.Options), Fields: s.encodeFieldInfos(idx.Fields), } @@ -610,6 +611,7 @@ func (s Serializer) encodeFieldInfos(fs []*pilosa.FieldInfo) []*internal.Field { func (s Serializer) encodeFieldInfo(f *pilosa.FieldInfo) *internal.Field { ifield := &internal.Field{ Name: f.Name, + ETag: f.ETag, Meta: s.encodeFieldOptions(&f.Options), Views: make([]string, 0, len(f.Views)), } @@ -686,6 +688,7 @@ func (s Serializer) encodeCreateShardMessage(m *pilosa.CreateShardMessage) *inte func (s Serializer) encodeCreateIndexMessage(m *pilosa.CreateIndexMessage) *internal.CreateIndexMessage { return &internal.CreateIndexMessage{ Index: m.Index, + ETag: m.ETag, Meta: s.encodeIndexMeta(m.Meta), } } @@ -707,6 +710,7 @@ func (s Serializer) encodeCreateFieldMessage(m *pilosa.CreateFieldMessage) *inte return &internal.CreateFieldMessage{ Index: m.Index, Field: m.Field, + ETag: m.ETag, Meta: s.encodeFieldOptions(m.Meta), } } @@ -787,6 +791,7 @@ func (s Serializer) encodeNodeStatus(m *pilosa.NodeStatus) *internal.NodeStatus func (s Serializer) encodeIndexStatus(m *pilosa.IndexStatus) *internal.IndexStatus { return &internal.IndexStatus{ Name: m.Name, + ETag: m.ETag, Fields: s.encodeFieldStatuses(m.Fields), } } @@ -802,6 +807,7 @@ func (s Serializer) encodeIndexStatuses(a []*pilosa.IndexStatus) []*internal.Ind func (s Serializer) encodeFieldStatus(m *pilosa.FieldStatus) *internal.FieldStatus { return &internal.FieldStatus{ Name: m.Name, + ETag: m.ETag, AvailableShards: m.AvailableShards.Slice(), } } @@ -938,6 +944,7 @@ func (s Serializer) decodeIndexes(idxs []*internal.Index, m []*pilosa.IndexInfo) func (s Serializer) decodeIndex(idx *internal.Index, m *pilosa.IndexInfo) { m.Name = idx.Name + m.ETag = idx.ETag m.Options = pilosa.IndexOptions{} s.decodeIndexMeta(idx.Options, &m.Options) m.Fields = make([]*pilosa.FieldInfo, len(idx.Fields)) @@ -953,6 +960,7 @@ func (s Serializer) decodeFields(fs []*internal.Field, m []*pilosa.FieldInfo) { func (s Serializer) decodeField(f *internal.Field, m *pilosa.FieldInfo) { m.Name = f.Name + m.ETag = f.ETag m.Options = pilosa.FieldOptions{} s.decodeFieldOptions(f.Meta, &m.Options) m.Views = make([]*pilosa.ViewInfo, 0, len(f.Views)) @@ -1016,6 +1024,7 @@ func (s Serializer) decodeCreateShardMessage(pb *internal.CreateShardMessage, m func (s Serializer) decodeCreateIndexMessage(pb *internal.CreateIndexMessage, m *pilosa.CreateIndexMessage) { m.Index = pb.Index + m.ETag = pb.ETag m.Meta = &pilosa.IndexOptions{} s.decodeIndexMeta(pb.Meta, m.Meta) } @@ -1034,6 +1043,7 @@ func (s Serializer) decodeDeleteIndexMessage(pb *internal.DeleteIndexMessage, m func (s Serializer) decodeCreateFieldMessage(pb *internal.CreateFieldMessage, m *pilosa.CreateFieldMessage) { m.Index = pb.Index m.Field = pb.Field + m.ETag = pb.ETag m.Meta = &pilosa.FieldOptions{} s.decodeFieldOptions(pb.Meta, m.Meta) } @@ -1107,6 +1117,7 @@ func (s Serializer) decodeIndexStatuses(a []*internal.IndexStatus) []*pilosa.Ind func (s Serializer) decodeIndexStatus(pb *internal.IndexStatus, m *pilosa.IndexStatus) { m.Name = pb.Name + m.ETag = pb.ETag m.Fields = s.decodeFieldStatuses(pb.Fields) } @@ -1121,6 +1132,7 @@ func (s Serializer) decodeFieldStatuses(a []*internal.FieldStatus) []*pilosa.Fie func (s Serializer) decodeFieldStatus(pb *internal.FieldStatus, m *pilosa.FieldStatus) { m.Name = pb.Name + m.ETag = pb.ETag m.AvailableShards = roaring.NewBitmap(pb.AvailableShards...) } diff --git a/field.go b/field.go index d38aa2ffb..272717565 100644 --- a/field.go +++ b/field.go @@ -87,6 +87,7 @@ var availableShardFileFlushDuration = &protected{ // Field represents a container for views. type Field struct { mu sync.RWMutex + etag int64 path string index string name string @@ -382,6 +383,14 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) { // Name returns the name the field was initialized with. func (f *Field) Name() string { return f.name } +// ETag is an identifier for a specific version of field. +func (f *Field) ETag() int64 { + f.mu.RLock() + defer f.mu.RUnlock() + + return f.etag +} + // Index returns the index name the field was initialized with. func (f *Field) Index() string { return f.index } @@ -1972,6 +1981,7 @@ func (p fieldSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() } // FieldInfo represents schema information for a field. type FieldInfo struct { Name string `json:"name"` + ETag int64 `json:"etag,omitempty"` Options FieldOptions `json:"options"` Views []*ViewInfo `json:"views,omitempty"` } diff --git a/gossip/gossip.go b/gossip/gossip.go index 84f914975..115c4274b 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -324,16 +324,20 @@ func (g *memberSet) LocalState(join bool) []byte { Schema: &pilosa.Schema{Indexes: g.papi.Schema(context.Background())}, } for _, idx := range m.Schema.Indexes { - is := &pilosa.IndexStatus{Name: idx.Name} + is := &pilosa.IndexStatus{Name: idx.Name, ETag: idx.ETag} + for _, f := range idx.Fields { availableShards := roaring.NewBitmap() if field, _ := g.papi.Field(context.Background(), idx.Name, f.Name); field != nil { availableShards = field.AvailableShards() } - is.Fields = append(is.Fields, &pilosa.FieldStatus{ + + fs := &pilosa.FieldStatus{ Name: f.Name, + ETag: f.ETag, AvailableShards: availableShards, - }) + } + is.Fields = append(is.Fields, fs) } m.Indexes = append(m.Indexes, is) } diff --git a/holder.go b/holder.go index ad52ee082..7434a0d0c 100644 --- a/holder.go +++ b/holder.go @@ -234,7 +234,13 @@ func (h *Holder) Open() error { return errors.Wrap(err, "opening index") } - if err := index.Open(); err != nil { + if h.isCoordinator() { + index.etag = newETag() + err = index.OpenWithETag() + } else { + err = index.Open() + } + if err != nil { if err == ErrName { h.Logger.Printf("ERROR opening index: %s, err=%s", index.Name(), err) continue @@ -381,10 +387,15 @@ func (h *Holder) Schema() []*IndexInfo { for _, index := range h.Indexes() { di := &IndexInfo{ Name: index.Name(), + ETag: index.ETag(), Options: index.Options(), } for _, field := range index.Fields() { - fi := &FieldInfo{Name: field.Name(), Options: field.Options()} + fi := &FieldInfo{ + Name: field.Name(), + ETag: field.ETag(), + Options: field.Options(), + } for _, view := range field.views() { fi.Views = append(fi.Views, &ViewInfo{Name: view.name}) } @@ -404,6 +415,7 @@ func (h *Holder) limitedSchema() []*IndexInfo { for _, index := range h.Indexes() { di := &IndexInfo{ Name: index.Name(), + ETag: index.ETag(), Options: index.Options(), ShardWidth: ShardWidth, } @@ -411,7 +423,11 @@ func (h *Holder) limitedSchema() []*IndexInfo { if strings.HasPrefix(field.name, "_") { continue } - fi := &FieldInfo{Name: field.Name(), Options: field.Options()} + fi := &FieldInfo{ + Name: field.Name(), + ETag: field.ETag(), + Options: field.Options(), + } di.Fields = append(di.Fields, fi) } sort.Sort(fieldInfoSlice(di.Fields)) @@ -424,20 +440,32 @@ func (h *Holder) limitedSchema() []*IndexInfo { // applySchema applies an internal Schema to Holder. func (h *Holder) applySchema(schema *Schema) error { // Create indexes that don't exist. - for _, index := range schema.Indexes { - idx, err := h.CreateIndexIfNotExists(index.Name, index.Options) + for _, i := range schema.Indexes { + idx, err := h.CreateIndexIfNotExists(i.Name, i.Options) if err != nil { return errors.Wrap(err, "creating index") } + if i.ETag != 0 { + idx.mu.Lock() + idx.etag = i.ETag + idx.mu.Unlock() + } + // Create fields that don't exist. - for _, f := range index.Fields { - field, err := idx.createFieldIfNotExists(f.Name, &f.Options) + for _, f := range i.Fields { + fld, err := idx.createFieldIfNotExists(f.Name, &f.Options) if err != nil { return errors.Wrap(err, "creating field") } + if f.ETag != 0 { + fld.mu.Lock() + fld.etag = f.ETag + fld.mu.Unlock() + } + // Create views that don't exist. for _, v := range f.Views { - _, err := field.createViewIfNotExists(v.Name) + _, err := fld.createViewIfNotExists(v.Name) if err != nil { return errors.Wrap(err, "creating view") } @@ -652,6 +680,13 @@ func (h *Holder) recalculateCaches() { } } +func (h *Holder) isCoordinator() bool { + if s, ok := h.broadcaster.(*Server); ok { + return s.isCoordinator + } + return false +} + // setFileLimit attempts to set the open file limit to the FileLimit constant defined above. func (h *Holder) setFileLimit() { oldLimit := &syscall.Rlimit{} diff --git a/http/handler.go b/http/handler.go index 5b9eaf405..555c789c9 100644 --- a/http/handler.go +++ b/http/handler.go @@ -15,13 +15,13 @@ package http import ( + "bytes" "context" "crypto/tls" "encoding/json" "expvar" "fmt" "io" - "io/ioutil" "math" "net" "net/http" @@ -504,7 +504,7 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") schema := h.api.Schema(r.Context()) - if err := json.NewEncoder(w).Encode(map[string]interface{}{"indexes": schema}); err != nil { // TODO: use pilosa.Schema instead of map[string]interface{} here? + if err := json.NewEncoder(w).Encode(pilosa.Schema{Indexes: schema}); err != nil { h.logger.Printf("write schema response error: %s", err) } } @@ -787,8 +787,14 @@ func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) { resp.write(w, err) return } - _, err = h.api.CreateIndex(r.Context(), indexName, req.Options) - + index, err := h.api.CreateIndex(r.Context(), indexName, req.Options) + if index != nil { + w.Header().Add("ETag", strconv.FormatInt(index.ETag(), 10)) + } else if _, ok = err.(pilosa.ConflictError); ok { + if index, _ = h.api.Index(r.Context(), indexName); index != nil { + w.Header().Add("ETag", strconv.FormatInt(index.ETag(), 10)) + } + } resp.write(w, err) } @@ -925,11 +931,18 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { fos = append(fos, pilosa.OptFieldForeignIndex(*req.Options.ForeignIndex)) } - _, err = h.api.CreateField(r.Context(), indexName, fieldName, fos...) + field, err := h.api.CreateField(r.Context(), indexName, fieldName, fos...) if _, ok := err.(pilosa.BadRequestError); ok { http.Error(w, err.Error(), http.StatusBadRequest) return } + if field != nil { + w.Header().Add("ETag", strconv.FormatInt(field.ETag(), 10)) + } else if _, ok = err.(pilosa.ConflictError); ok { + if field, _ = h.api.Field(r.Context(), indexName, fieldName); field != nil { + w.Header().Add("ETag", strconv.FormatInt(field.ETag(), 10)) + } + } resp.write(w, err) } @@ -1240,7 +1253,7 @@ func (h *Handler) readQueryRequest(r *http.Request) (*pilosa.QueryRequest, error // readProtobufQueryRequest parses query parameters in protobuf from r. func (h *Handler) readProtobufQueryRequest(r *http.Request) (*pilosa.QueryRequest, error) { // Slurp the body. - body, err := ioutil.ReadAll(r.Body) + body, err := readBody(r) if err != nil { return nil, errors.Wrap(err, "reading") } @@ -1258,7 +1271,7 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*pilosa.QueryRequest, er q := r.URL.Query() // Parse query string. - buf, err := ioutil.ReadAll(r.Body) + buf, err := readBody(r) if err != nil { return nil, errors.Wrap(err, "reading") } @@ -1319,95 +1332,38 @@ func (h *Handler) writeJSONQueryResponse(w io.Writer, resp *pilosa.QueryResponse return json.NewEncoder(w).Encode(resp) } -// handlePostImport handles /import requests. -func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { - // Verify that request is only communicating over protobufs. +func validateProtobufHeader(r *http.Request) (error string, code int) { if r.Header.Get("Content-Type") != "application/x-protobuf" { - http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) - return - } else if r.Header.Get("Accept") != "application/x-protobuf" { - http.Error(w, "Not acceptable", http.StatusNotAcceptable) - return + return "Unsupported media type", http.StatusUnsupportedMediaType } - indexName := mux.Vars(r)["index"] - fieldName := mux.Vars(r)["field"] - - // If the clear flag is true, treat the import as clear bits. - q := r.URL.Query() - doClear := q.Get("clear") == "true" - doIgnoreKeyCheck := q.Get("ignoreKeyCheck") == "true" - - opts := []pilosa.ImportOption{ - pilosa.OptImportOptionsClear(doClear), - pilosa.OptImportOptionsIgnoreKeyCheck(doIgnoreKeyCheck), + if r.Header.Get("Accept") != "application/x-protobuf" { + return "Not acceptable", http.StatusNotAcceptable } + return +} - // Get index and field type to determine how to handle the - // import data. - field, err := h.api.Field(r.Context(), indexName, fieldName) - if err != nil { - switch errors.Cause(err) { - case pilosa.ErrIndexNotFound: - fallthrough - case pilosa.ErrFieldNotFound: - http.Error(w, err.Error(), http.StatusNotFound) - default: - http.Error(w, err.Error(), http.StatusInternalServerError) - } - return +func validateETagHeader(r *http.Request, index *pilosa.Index, field *pilosa.Field) (error string, code int) { + etags := strings.Split(r.Header.Get("If-Match"), ",") + netags := len(etags) + for i := 0; i < netags && i < 2; i++ { + etags[i] = strings.TrimLeft(strings.TrimSpace(etags[i]), "W/") } - - // Read entire body. - body, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return + if netags == 1 && etags[0] != "" && etags[0] != "*" { + return "Precondition Failed", http.StatusPreconditionFailed } - - // Unmarshal request based on field type. - if field.Type() == pilosa.FieldTypeInt || field.Type() == pilosa.FieldTypeDecimal { - // Field type: Int - // Marshal into request object. - req := &pilosa.ImportValueRequest{} - if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return + if netags > 1 { + indexETag := etags[0] + if indexETag != "" && strconv.FormatInt(index.ETag(), 10) != indexETag { + return "Precondition Failed", http.StatusPreconditionFailed } - if err := h.api.ImportValue(r.Context(), req, opts...); err != nil { - switch errors.Cause(err) { - case pilosa.ErrClusterDoesNotOwnShard: - http.Error(w, err.Error(), http.StatusPreconditionFailed) - default: - http.Error(w, err.Error(), http.StatusInternalServerError) - } - return - } - } else { - // Field type: set, time, mutex - // Marshal into request object. - req := &pilosa.ImportRequest{} - if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - if err := h.api.Import(r.Context(), req, opts...); err != nil { - switch errors.Cause(err) { - case pilosa.ErrClusterDoesNotOwnShard: - http.Error(w, err.Error(), http.StatusPreconditionFailed) - default: - http.Error(w, err.Error(), http.StatusInternalServerError) - } - return + fieldETag := etags[1] + if fieldETag != "" && strconv.FormatInt(field.ETag(), 10) != fieldETag { + return "Precondition Failed", http.StatusPreconditionFailed } } - // Write response. - _, err = w.Write(importOk) - if err != nil { - h.logger.Printf("writing import response: %v", err) - } + return } // handleGetExport handles /export requests. @@ -1885,6 +1841,101 @@ func GetHTTPClient(t *tls.Config) *http.Client { return &http.Client{Transport: transport} } +// handlePostImport handles /import requests. +func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { + // Verify that request is only communicating over protobufs. + if error, code := validateProtobufHeader(r); error != "" { + http.Error(w, error, code) + return + } + + // Get index and field type to determine how to handle the + // import data. + indexName := mux.Vars(r)["index"] + index, err := h.api.Index(r.Context(), indexName) + if err != nil { + if errors.Cause(err) == pilosa.ErrIndexNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + fieldName := mux.Vars(r)["field"] + field := index.Field(fieldName) + if field == nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + + // Verify if request matches etag + if error, code := validateETagHeader(r, index, field); error != "" { + http.Error(w, error, code) + return + } + + // If the clear flag is true, treat the import as clear bits. + q := r.URL.Query() + doClear := q.Get("clear") == "true" + doIgnoreKeyCheck := q.Get("ignoreKeyCheck") == "true" + + opts := []pilosa.ImportOption{ + pilosa.OptImportOptionsClear(doClear), + pilosa.OptImportOptionsIgnoreKeyCheck(doIgnoreKeyCheck), + } + + // Read entire body. + body, err := readBody(r) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + // Unmarshal request based on field type. + if field.Type() == pilosa.FieldTypeInt || field.Type() == pilosa.FieldTypeDecimal { + // Field type: Int + // Marshal into request object. + req := &pilosa.ImportValueRequest{} + if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := h.api.ImportValue(r.Context(), req, opts...); err != nil { + switch errors.Cause(err) { + case pilosa.ErrClusterDoesNotOwnShard: + http.Error(w, err.Error(), http.StatusPreconditionFailed) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + } else { + // Field type: set, time, mutex + // Marshal into request object. + req := &pilosa.ImportRequest{} + if err := proto.DefaultSerializer.Unmarshal(body, req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := h.api.Import(r.Context(), req, opts...); err != nil { + switch errors.Cause(err) { + case pilosa.ErrClusterDoesNotOwnShard: + http.Error(w, err.Error(), http.StatusPreconditionFailed) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + } + + // Write response. + _, err = w.Write(importOk) + if err != nil { + h.logger.Printf("writing import response: %v", err) + } +} + // handlePostImportColumnAttrs func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Request) { // Verify that request is only communicating over protobufs. @@ -1898,7 +1949,7 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req opts := []pilosa.ImportOption{} - body, err := ioutil.ReadAll(r.Body) + body, err := readBody(r) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -1922,18 +1973,38 @@ func (h *Handler) handlePostImportColumnAttrs(w http.ResponseWriter, r *http.Req } } -// handlPostRoaringImport +// handlePostImportRoaring func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request) { // 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) - return - } else if r.Header.Get("Accept") != "application/x-protobuf" { - http.Error(w, "Not acceptable", http.StatusNotAcceptable) + if error, code := validateProtobufHeader(r); error != "" { + http.Error(w, error, code) return } + + // Get index and field type to determine how to handle the + // import data. indexName := mux.Vars(r)["index"] + index, err := h.api.Index(r.Context(), indexName) + if err != nil { + if errors.Cause(err) == pilosa.ErrIndexNotFound { + http.Error(w, err.Error(), http.StatusNotFound) + } else { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } fieldName := mux.Vars(r)["field"] + field := index.Field(fieldName) + if field == nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + + // Verify if request matches etag + if error, code := validateETagHeader(r, index, field); error != "" { + http.Error(w, error, code) + return + } q := r.URL.Query() remoteStr := q.Get("remote") @@ -1946,7 +2017,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request // Read entire body. span, _ := tracing.StartSpanFromContext(ctx, "ioutil.ReadAll-Body") - body, err := ioutil.ReadAll(r.Body) + body, err := readBody(r) span.LogKV("bodySize", len(body)) span.Finish() if err != nil { @@ -2040,3 +2111,18 @@ func (h *Handler) handlePostTranslateIDs(w http.ResponseWriter, r *http.Request) h.logger.Printf("writing translate keys response: %v", err) } } + +// Read entire request body. +func readBody(r *http.Request) ([]byte, error) { + var contentLength int64 = bytes.MinRead + if r.ContentLength > 0 { + contentLength = r.ContentLength + } + + buf := bytes.NewBuffer(make([]byte, 0, 1+contentLength)) + if _, err := buf.ReadFrom(r.Body); err != nil { + return nil, err + } + + return buf.Bytes(), nil +} diff --git a/index.go b/index.go index a5f143bfc..aaf61e70f 100644 --- a/index.go +++ b/index.go @@ -37,6 +37,7 @@ import ( // Index represents a container for fields. type Index struct { mu sync.RWMutex + etag int64 path string name string keys bool // use string keys @@ -103,6 +104,13 @@ func NewIndex(path, name string, partitionN int) (*Index, error) { }, nil } +// ETag is an identifier for a specific version of an index. +func (i *Index) ETag() int64 { + i.mu.RLock() + defer i.mu.RUnlock() + return i.etag +} + // Name returns name of the index. func (i *Index) Name() string { return i.name } @@ -140,7 +148,12 @@ func (i *Index) options() IndexOptions { } // Open opens and initializes the index. -func (i *Index) Open() (err error) { +func (i *Index) Open() error { return i.open(false) } + +// OpenWithETag opens and initializes the index and set a new ETag for fields. +func (i *Index) OpenWithETag() error { return i.open(true) } + +func (i *Index) open(withETag bool) (err error) { // Ensure the path exists. i.logger.Debugf("ensure index path exists: %s", i.path) if err := os.MkdirAll(i.path, 0777); err != nil { @@ -154,7 +167,7 @@ func (i *Index) Open() (err error) { } i.logger.Debugf("open fields for index: %s", i.name) - if err := i.openFields(); err != nil { + if err := i.openFields(withETag); err != nil { return errors.Wrap(err, "opening fields") } @@ -197,7 +210,7 @@ func (i *Index) Open() (err error) { var indexQueue = make(chan struct{}, 8) // openFields opens and initializes the fields inside the index. -func (i *Index) openFields() error { +func (i *Index) openFields(withETag bool) error { f, err := os.Open(i.path) if err != nil { return errors.Wrap(err, "opening directory") @@ -229,6 +242,9 @@ fileLoop: i.logger.Debugf("open field: %s", fi.Name()) mu.Lock() fld, err := i.newField(i.fieldPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + if withETag { + fld.etag = newETag() + } mu.Unlock() if err != nil { return errors.Wrapf(ErrName, "'%s'", fi.Name()) @@ -559,6 +575,7 @@ func (p indexSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() } // IndexInfo represents schema information for an index. type IndexInfo struct { Name string `json:"name"` + ETag int64 `json:"etag,omitempty"` Options IndexOptions `json:"options"` Fields []*FieldInfo `json:"fields"` ShardWidth uint64 `json:"shardWidth"` diff --git a/internal/private.pb.go b/internal/private.pb.go index 38eb7af3d..694935fb0 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -616,6 +616,7 @@ func (m *DeleteIndexMessage) GetIndex() string { type CreateIndexMessage struct { Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"` Meta *IndexMeta `protobuf:"bytes,2,opt,name=Meta,proto3" json:"Meta,omitempty"` + ETag int64 `protobuf:"varint,3,opt,name=ETag,proto3" json:"ETag,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -668,10 +669,18 @@ func (m *CreateIndexMessage) GetMeta() *IndexMeta { return nil } +func (m *CreateIndexMessage) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + type CreateFieldMessage struct { Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"` Field string `protobuf:"bytes,2,opt,name=Field,proto3" json:"Field,omitempty"` Meta *FieldOptions `protobuf:"bytes,3,opt,name=Meta,proto3" json:"Meta,omitempty"` + ETag int64 `protobuf:"varint,4,opt,name=ETag,proto3" json:"ETag,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -731,6 +740,13 @@ func (m *CreateFieldMessage) GetMeta() *FieldOptions { return nil } +func (m *CreateFieldMessage) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + type DeleteFieldMessage struct { Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"` Field string `protobuf:"bytes,2,opt,name=Field,proto3" json:"Field,omitempty"` @@ -853,6 +869,7 @@ type Field struct { Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` Meta *FieldOptions `protobuf:"bytes,2,opt,name=Meta,proto3" json:"Meta,omitempty"` Views []string `protobuf:"bytes,3,rep,name=Views,proto3" json:"Views,omitempty"` + ETag int64 `protobuf:"varint,4,opt,name=ETag,proto3" json:"ETag,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -912,6 +929,13 @@ func (m *Field) GetViews() []string { return nil } +func (m *Field) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + type Schema struct { Indexes []*Index `protobuf:"bytes,1,rep,name=Indexes,proto3" json:"Indexes,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` @@ -961,6 +985,7 @@ func (m *Schema) GetIndexes() []*Index { type Index struct { Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` + ETag int64 `protobuf:"varint,2,opt,name=ETag,proto3" json:"ETag,omitempty"` Options *IndexMeta `protobuf:"bytes,5,opt,name=Options,proto3" json:"Options,omitempty"` Fields []*Field `protobuf:"bytes,4,rep,name=Fields,proto3" json:"Fields,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` @@ -1008,6 +1033,13 @@ func (m *Index) GetName() string { return "" } +func (m *Index) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + func (m *Index) GetOptions() *IndexMeta { if m != nil { return m.Options @@ -1340,6 +1372,7 @@ func (m *NodeStatus) GetIndexes() []*IndexStatus { type IndexStatus struct { Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` Fields []*FieldStatus `protobuf:"bytes,2,rep,name=Fields,proto3" json:"Fields,omitempty"` + ETag int64 `protobuf:"varint,3,opt,name=ETag,proto3" json:"ETag,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -1392,9 +1425,17 @@ func (m *IndexStatus) GetFields() []*FieldStatus { return nil } +func (m *IndexStatus) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + type FieldStatus struct { Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` AvailableShards []uint64 `protobuf:"varint,2,rep,packed,name=AvailableShards,proto3" json:"AvailableShards,omitempty"` + ETag int64 `protobuf:"varint,3,opt,name=ETag,proto3" json:"ETag,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -1447,6 +1488,13 @@ func (m *FieldStatus) GetAvailableShards() []uint64 { return nil } +func (m *FieldStatus) GetETag() int64 { + if m != nil { + return m.ETag + } + return 0 +} + type ClusterStatus struct { ClusterID string `protobuf:"bytes,1,opt,name=ClusterID,proto3" json:"ClusterID,omitempty"` State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"` @@ -2421,96 +2469,98 @@ func init() { func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) } var fileDescriptor_d2a91b51c7bdc125 = []byte{ - // 1418 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xcd, 0x72, 0x1b, 0xc5, - 0x13, 0xff, 0xaf, 0x56, 0xb6, 0xa4, 0x96, 0xe5, 0xc8, 0x93, 0xc4, 0xd9, 0xf8, 0x4f, 0x19, 0x31, - 0xa4, 0x88, 0x48, 0x15, 0x26, 0x95, 0x50, 0xc5, 0x67, 0xaa, 0x12, 0x5b, 0x4e, 0x10, 0xc1, 0x8e, - 0x33, 0x72, 0x72, 0xe3, 0x30, 0x5e, 0x4d, 0xc5, 0x5b, 0x5e, 0xed, 0x8a, 0xdd, 0x59, 0x47, 0xce, - 0x81, 0x2b, 0x54, 0xf1, 0x02, 0x1c, 0x38, 0xf0, 0x1e, 0xbc, 0x00, 0x47, 0x1e, 0x81, 0x0a, 0x4f, - 0xc1, 0x8d, 0x9a, 0x9e, 0x99, 0xdd, 0x95, 0x2c, 0xe3, 0x90, 0x70, 0xdb, 0xfe, 0xf5, 0x77, 0x4f, - 0x77, 0xcf, 0x2c, 0xb4, 0xc6, 0x49, 0x70, 0xcc, 0xa5, 0xd8, 0x18, 0x27, 0xb1, 0x8c, 0x49, 0x3d, - 0x88, 0xa4, 0x48, 0x22, 0x1e, 0xae, 0x2d, 0x8d, 0xb3, 0x83, 0x30, 0xf0, 0x35, 0x4e, 0x1f, 0x40, - 0xa3, 0x1f, 0x0d, 0xc5, 0x64, 0x47, 0x48, 0x4e, 0x08, 0x54, 0x1f, 0x8a, 0x93, 0xd4, 0x73, 0x3b, - 0x4e, 0xb7, 0xce, 0xf0, 0x9b, 0xbc, 0x07, 0xcb, 0xfb, 0x09, 0xf7, 0x8f, 0xb6, 0x27, 0x41, 0x2a, - 0x45, 0xe4, 0x0b, 0xaf, 0x8a, 0xdc, 0x19, 0x94, 0xfe, 0xe2, 0xc2, 0xd2, 0xfd, 0x40, 0x84, 0xc3, - 0x47, 0x63, 0x19, 0xc4, 0x51, 0xaa, 0x8c, 0xed, 0x9f, 0x8c, 0x85, 0x57, 0xef, 0x38, 0xdd, 0x06, - 0xc3, 0x6f, 0xf2, 0x16, 0x34, 0xb6, 0xb8, 0x7f, 0x28, 0x90, 0xe1, 0x22, 0xa3, 0x00, 0x72, 0xee, - 0x20, 0x78, 0xa1, 0xbd, 0xb4, 0x58, 0x01, 0x90, 0x0e, 0x34, 0xf7, 0x83, 0x91, 0x78, 0x9c, 0xf1, - 0x48, 0x66, 0x23, 0x6f, 0x01, 0xb5, 0xcb, 0x10, 0x59, 0x85, 0xc5, 0x47, 0xe1, 0x70, 0x27, 0x88, - 0xbc, 0x46, 0xc7, 0xe9, 0xba, 0xcc, 0x50, 0x16, 0xe7, 0x13, 0x0f, 0x0a, 0x9c, 0x4f, 0xf2, 0x74, - 0x9b, 0xd3, 0xe9, 0xee, 0xc6, 0x03, 0xc9, 0xa3, 0x21, 0x4f, 0x86, 0x4f, 0x03, 0xf1, 0xdc, 0x5b, - 0xd2, 0xe9, 0x4e, 0xa3, 0x4a, 0x77, 0x93, 0xa7, 0xc2, 0x6b, 0xa1, 0x45, 0xfc, 0x26, 0x6b, 0x50, - 0xdf, 0x0c, 0x64, 0x4f, 0x8c, 0xe5, 0xa1, 0xb7, 0xdc, 0x71, 0xba, 0x55, 0x96, 0xd3, 0xe4, 0x12, - 0x2c, 0x0c, 0x7c, 0x1e, 0x0a, 0xef, 0x02, 0x2a, 0x68, 0x82, 0x50, 0x58, 0xba, 0x1f, 0x27, 0x22, - 0x78, 0x16, 0xe1, 0x21, 0x78, 0x6d, 0x4c, 0x6a, 0x0a, 0x23, 0xef, 0x82, 0xab, 0x52, 0x5a, 0xe9, - 0x38, 0xdd, 0xe6, 0xad, 0x95, 0x0d, 0x7b, 0x8e, 0x1b, 0x3d, 0xe1, 0x07, 0x23, 0x1e, 0x32, 0xc5, - 0x45, 0x21, 0x3e, 0xf1, 0xc8, 0xd9, 0x42, 0x7c, 0x42, 0x29, 0x2c, 0xf7, 0x47, 0xe3, 0x38, 0x91, - 0x4c, 0xa4, 0xe3, 0x38, 0x4a, 0x05, 0x69, 0x83, 0xbb, 0x9d, 0x24, 0x9e, 0x83, 0x6e, 0xd5, 0x27, - 0xfd, 0x0e, 0xda, 0x9b, 0x61, 0xec, 0x1f, 0xf5, 0xb8, 0xe4, 0x4c, 0x7c, 0x9b, 0x89, 0x54, 0xaa, - 0xd8, 0x75, 0x78, 0x5a, 0x4e, 0x13, 0x0a, 0xc5, 0xf3, 0xf6, 0x2a, 0x1a, 0x45, 0x42, 0xd5, 0x05, - 0xab, 0xa6, 0x8f, 0x07, 0xbf, 0x31, 0xf7, 0x43, 0x9e, 0x0c, 0xf1, 0x4c, 0xab, 0x4c, 0x13, 0x0a, - 0x45, 0x4f, 0xd8, 0x07, 0x55, 0xa6, 0x09, 0xda, 0x87, 0x95, 0x92, 0x7f, 0x13, 0xe6, 0x2a, 0x2c, - 0xb2, 0xf8, 0x79, 0xbf, 0x97, 0x7a, 0x4e, 0xc7, 0xed, 0x56, 0x99, 0xa1, 0xb0, 0x61, 0xe2, 0x30, - 0x1b, 0x45, 0x8a, 0x55, 0x41, 0x56, 0x01, 0xd0, 0xab, 0xb0, 0x80, 0xdd, 0xa3, 0xb2, 0x2c, 0x74, - 0xd5, 0x27, 0xfd, 0xde, 0x81, 0xc6, 0x0e, 0x9f, 0x60, 0x20, 0x29, 0xb9, 0x03, 0x75, 0x7b, 0xb6, - 0x28, 0xd4, 0xbc, 0xf5, 0x4e, 0x51, 0xc1, 0x5c, 0x6c, 0xc3, 0xca, 0x6c, 0x47, 0x32, 0x39, 0x61, - 0xb9, 0xca, 0xda, 0xe7, 0xd0, 0x9a, 0x62, 0x29, 0x7f, 0x47, 0xe2, 0xc4, 0x56, 0xf5, 0x48, 0x9c, - 0xa8, 0x5c, 0x8f, 0x79, 0x98, 0x09, 0xac, 0x55, 0x95, 0x69, 0xe2, 0xb3, 0xca, 0x27, 0x0e, 0x7d, - 0x0a, 0x64, 0x2b, 0x11, 0x5c, 0x0a, 0x74, 0xb2, 0x23, 0xd2, 0x94, 0x3f, 0x13, 0xe7, 0x55, 0xdc, - 0x2d, 0x57, 0x3c, 0xaf, 0x6e, 0xa5, 0x54, 0x5d, 0x7a, 0x03, 0x48, 0x4f, 0x84, 0x42, 0x0a, 0x33, - 0xdd, 0xff, 0x60, 0x97, 0x0e, 0x6c, 0x0c, 0xe7, 0xcb, 0x92, 0xeb, 0x50, 0x55, 0xab, 0x02, 0x9d, - 0x35, 0x6f, 0x5d, 0x2c, 0xea, 0x94, 0x6f, 0x11, 0x86, 0x02, 0x34, 0xb4, 0x46, 0x31, 0xca, 0x57, - 0x4c, 0x6c, 0xaa, 0x95, 0x6e, 0x18, 0x57, 0x2e, 0xba, 0x5a, 0x2d, 0x5c, 0x95, 0xd7, 0x8c, 0xf1, - 0x76, 0xd7, 0xa6, 0xfb, 0xba, 0xde, 0xa8, 0x0f, 0xff, 0xd7, 0x16, 0xee, 0x1d, 0xf3, 0x20, 0xe4, - 0x07, 0xe1, 0xbf, 0x3a, 0x91, 0xa9, 0xc0, 0x3d, 0xa8, 0xa1, 0x6e, 0xbf, 0x67, 0x7a, 0xdb, 0x92, - 0xf4, 0x1b, 0x28, 0xc6, 0x64, 0x97, 0x8f, 0x84, 0xb1, 0x86, 0xdf, 0x79, 0xbe, 0x95, 0xf3, 0xf3, - 0x55, 0x8e, 0xd5, 0x68, 0xa9, 0x55, 0xed, 0x2a, 0xc7, 0x48, 0xd0, 0xdb, 0xb0, 0x38, 0xf0, 0x0f, - 0xc5, 0x88, 0x93, 0xf7, 0xa1, 0x86, 0x11, 0x8a, 0xd4, 0x74, 0xf4, 0x85, 0x99, 0x93, 0x62, 0x96, - 0x4f, 0x53, 0x93, 0xd9, 0xdc, 0x98, 0x3e, 0x80, 0x9a, 0x71, 0x8c, 0x13, 0x7d, 0xc6, 0x89, 0x5b, - 0x19, 0x72, 0x1d, 0x16, 0x31, 0xd8, 0xd4, 0xab, 0xce, 0x7a, 0x45, 0x9c, 0x19, 0x36, 0xdd, 0x06, - 0xf7, 0x09, 0xeb, 0xab, 0xc1, 0xc6, 0x80, 0xad, 0x53, 0x43, 0xa9, 0x50, 0xbe, 0x8c, 0x53, 0x69, - 0xca, 0x8a, 0xdf, 0x0a, 0xdb, 0x8b, 0x13, 0x89, 0x25, 0x6d, 0x31, 0xfc, 0xa6, 0x3f, 0x3b, 0x50, - 0xdd, 0x8d, 0x87, 0x82, 0x2c, 0x43, 0xa5, 0xdf, 0x33, 0x46, 0x2a, 0xfd, 0x1e, 0x79, 0x1b, 0xed, - 0x9b, 0x52, 0xb6, 0x8a, 0x28, 0x9e, 0xb0, 0x3e, 0x43, 0xcf, 0xd7, 0xa0, 0xd5, 0x4f, 0xb7, 0xe2, - 0x38, 0x19, 0x06, 0x11, 0x97, 0x71, 0x62, 0xee, 0xbc, 0x69, 0x10, 0x67, 0x4b, 0x72, 0xa9, 0x6f, - 0xa3, 0x06, 0xd3, 0x04, 0xb9, 0x0e, 0xb5, 0x07, 0x6c, 0x6f, 0x4b, 0x39, 0x58, 0x98, 0xe7, 0xc0, - 0x72, 0xe9, 0x5d, 0x68, 0xab, 0xe8, 0x50, 0xcb, 0x36, 0xd2, 0x2a, 0x2c, 0x2a, 0x2c, 0x8f, 0xd6, - 0x50, 0x85, 0xab, 0x4a, 0xc9, 0x15, 0xfd, 0x5a, 0x5b, 0xd8, 0x3e, 0x16, 0x91, 0x2c, 0xb5, 0x22, - 0xd2, 0x68, 0xa0, 0xc5, 0x34, 0x41, 0xa8, 0xae, 0x84, 0x49, 0x79, 0xb9, 0x88, 0x48, 0xa1, 0x0c, - 0x79, 0xf4, 0x47, 0x07, 0xc0, 0x06, 0x94, 0xa5, 0xb9, 0x8a, 0x73, 0xb6, 0x0a, 0xe9, 0xda, 0x96, - 0x32, 0x63, 0xd8, 0x2e, 0xa4, 0x34, 0xce, 0x6c, 0xcb, 0x7d, 0x58, 0xb4, 0x9c, 0x3e, 0xfc, 0xcb, - 0x33, 0xad, 0xa2, 0xbd, 0x16, 0x8d, 0xb7, 0x07, 0xcd, 0x12, 0x7e, 0x46, 0xfb, 0xd9, 0x7e, 0xaa, - 0xcc, 0x9a, 0x44, 0xdc, 0x98, 0xb4, 0x5d, 0xf5, 0x10, 0x9a, 0x25, 0x78, 0xae, 0xc5, 0x2e, 0x5c, - 0x98, 0x1e, 0x70, 0x7b, 0x71, 0xcc, 0xc2, 0x34, 0x80, 0xd6, 0x56, 0x98, 0xa5, 0x52, 0x24, 0xc6, - 0x9c, 0xba, 0x6d, 0x34, 0x90, 0x1f, 0x5e, 0x01, 0xcc, 0x3f, 0x3f, 0x72, 0x0d, 0x16, 0x54, 0x19, - 0xf5, 0x9c, 0x9e, 0xae, 0xb1, 0x66, 0xd2, 0xa7, 0x50, 0xdf, 0x1c, 0xf4, 0x1f, 0x24, 0x71, 0x36, - 0x9e, 0x1b, 0xb4, 0x7d, 0x4a, 0x55, 0x4a, 0x4f, 0xa9, 0xb6, 0x7e, 0x16, 0xb8, 0xf8, 0x9c, 0xc0, - 0x37, 0x40, 0x5b, 0xbf, 0x01, 0xaa, 0x06, 0xe1, 0x6a, 0xb1, 0xaf, 0xe8, 0x1d, 0xac, 0xd6, 0xc3, - 0xeb, 0x6c, 0x32, 0x7b, 0x9b, 0xbb, 0xc5, 0x6d, 0xae, 0x8c, 0xea, 0x45, 0xf9, 0x5f, 0x1a, 0xfd, - 0xab, 0x02, 0x2b, 0x4c, 0xa4, 0xc1, 0x0b, 0xd1, 0x8f, 0x52, 0x99, 0x64, 0xbe, 0xda, 0x27, 0x4a, - 0xff, 0xab, 0xf8, 0xc0, 0x54, 0xdb, 0x65, 0x9a, 0x78, 0x95, 0x4e, 0x27, 0x37, 0xa1, 0x39, 0x3b, - 0xdc, 0xa7, 0x45, 0xcb, 0x22, 0xe4, 0x26, 0xd4, 0x06, 0x71, 0x96, 0xf8, 0x79, 0xfb, 0x96, 0x16, - 0xb0, 0x8e, 0x4c, 0xb3, 0x99, 0x15, 0x23, 0x8f, 0x81, 0xec, 0x27, 0x3c, 0x4a, 0x43, 0xae, 0x82, - 0xb5, 0xca, 0xf5, 0xd9, 0x07, 0x44, 0x49, 0x66, 0xca, 0xce, 0x1c, 0x65, 0xf2, 0x51, 0x79, 0x3e, - 0xbd, 0x1a, 0x46, 0x7d, 0x69, 0x3a, 0x6a, 0xd3, 0xf2, 0xe5, 0x39, 0xbe, 0x33, 0xd3, 0xa9, 0xde, - 0x22, 0x2a, 0x5e, 0x29, 0x14, 0xa7, 0xd8, 0x6c, 0x5a, 0x9a, 0xfe, 0xe0, 0xc0, 0x52, 0x39, 0xb2, - 0x57, 0xda, 0x0b, 0xf9, 0x81, 0x57, 0xce, 0x7f, 0xa1, 0xd8, 0x03, 0xaf, 0xce, 0x7b, 0x13, 0x2e, - 0x94, 0x5f, 0x2d, 0x19, 0x5c, 0x39, 0xa3, 0x5c, 0x6f, 0x10, 0x54, 0x07, 0x9a, 0x7b, 0x3c, 0x91, - 0x81, 0x32, 0x69, 0xae, 0xe4, 0x05, 0x56, 0x86, 0xe8, 0x11, 0x5c, 0x3d, 0xd5, 0x7c, 0x5b, 0xf1, - 0x68, 0xac, 0xba, 0xfc, 0x0d, 0x9a, 0x50, 0x2d, 0xea, 0x24, 0x31, 0xed, 0xd7, 0x60, 0x9a, 0xa0, - 0x9f, 0xc2, 0xe5, 0x81, 0x90, 0xa5, 0xd6, 0xb3, 0x33, 0xd4, 0x01, 0x77, 0x57, 0x3c, 0x3f, 0x23, - 0x41, 0xc5, 0xa2, 0x5f, 0x80, 0xf7, 0x64, 0x3c, 0xe4, 0x52, 0xbc, 0x96, 0xf6, 0x26, 0xd4, 0xf7, - 0xe3, 0x71, 0x1c, 0xc6, 0xcf, 0x4e, 0xce, 0xd9, 0x65, 0x1e, 0xd4, 0xf4, 0xad, 0xa4, 0x97, 0x63, - 0x83, 0x59, 0x92, 0x5e, 0x54, 0x63, 0xea, 0xf3, 0xd0, 0xcf, 0x42, 0x15, 0x86, 0x7a, 0x5e, 0xa7, - 0x54, 0x98, 0x41, 0xe0, 0x58, 0xb8, 0xd2, 0x45, 0x77, 0x0f, 0x01, 0x7b, 0xd1, 0x69, 0x8a, 0x7c, - 0x0c, 0xcd, 0x92, 0xb4, 0x29, 0xe0, 0xe5, 0x99, 0x79, 0xd1, 0x4c, 0x56, 0x96, 0xa4, 0xbf, 0x3a, - 0x53, 0x9a, 0xa7, 0xee, 0x7c, 0xe3, 0xf0, 0x58, 0x1f, 0x4a, 0x9d, 0x19, 0x4a, 0xe5, 0xba, 0x3d, - 0xf1, 0xc3, 0x2c, 0x55, 0x2c, 0x7d, 0xcd, 0x17, 0x80, 0xca, 0x55, 0xfd, 0x43, 0xc6, 0x99, 0x34, - 0x9b, 0xd3, 0x92, 0xea, 0x77, 0xae, 0x27, 0xf8, 0x30, 0x0c, 0x22, 0x81, 0x5d, 0xea, 0xb2, 0x9c, - 0x26, 0x37, 0xf5, 0xb6, 0xb7, 0xa3, 0xb6, 0x36, 0x37, 0x7c, 0x94, 0xd0, 0x37, 0x41, 0x4a, 0x09, - 0xb4, 0x67, 0x59, 0x9b, 0xed, 0xdf, 0x5e, 0xae, 0x3b, 0xbf, 0xbf, 0x5c, 0x77, 0xfe, 0x78, 0xb9, - 0xee, 0xfc, 0xf4, 0xe7, 0xfa, 0xff, 0x0e, 0x16, 0xf1, 0xaf, 0xfc, 0xf6, 0xdf, 0x01, 0x00, 0x00, - 0xff, 0xff, 0x8a, 0x80, 0x2a, 0x36, 0xbe, 0x0f, 0x00, 0x00, + // 1448 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xdd, 0x6e, 0x1b, 0xc5, + 0x17, 0xff, 0xaf, 0xd7, 0x4e, 0xec, 0xe3, 0x38, 0x75, 0xa6, 0x6d, 0xba, 0xcd, 0x1f, 0x05, 0x33, + 0x54, 0xd4, 0x54, 0x6a, 0xa8, 0x5a, 0x24, 0x3e, 0x2b, 0xb5, 0x89, 0xdd, 0x62, 0x20, 0x69, 0x3b, + 0x4e, 0x7b, 0x8b, 0x26, 0xf6, 0x28, 0x59, 0x65, 0xbd, 0xeb, 0xee, 0xce, 0xa6, 0x4e, 0x2f, 0x10, + 0x77, 0x20, 0xf1, 0x02, 0x5c, 0x70, 0xc1, 0x7b, 0xf0, 0x02, 0x5c, 0xf2, 0x08, 0xa8, 0x3c, 0x05, + 0x77, 0x68, 0xce, 0xcc, 0xec, 0xae, 0x1d, 0x87, 0x84, 0x96, 0xbb, 0x39, 0x5f, 0x73, 0x7e, 0xe7, + 0x73, 0xc7, 0x86, 0xc6, 0x38, 0xf6, 0x8f, 0xb8, 0x14, 0x1b, 0xe3, 0x38, 0x92, 0x11, 0xa9, 0xfa, + 0xa1, 0x14, 0x71, 0xc8, 0x83, 0xb5, 0xa5, 0x71, 0xba, 0x17, 0xf8, 0x03, 0xcd, 0xa7, 0x0f, 0xa1, + 0xd6, 0x0b, 0x87, 0x62, 0xb2, 0x2d, 0x24, 0x27, 0x04, 0xca, 0x5f, 0x89, 0xe3, 0xc4, 0x73, 0x5b, + 0x4e, 0xbb, 0xca, 0xf0, 0x4c, 0xde, 0x83, 0xe5, 0xdd, 0x98, 0x0f, 0x0e, 0xbb, 0x13, 0x3f, 0x91, + 0x22, 0x1c, 0x08, 0xaf, 0x8c, 0xd2, 0x19, 0x2e, 0xfd, 0xc5, 0x85, 0xa5, 0x07, 0xbe, 0x08, 0x86, + 0x8f, 0xc6, 0xd2, 0x8f, 0xc2, 0x44, 0x5d, 0xb6, 0x7b, 0x3c, 0x16, 0x5e, 0xb5, 0xe5, 0xb4, 0x6b, + 0x0c, 0xcf, 0xe4, 0x2d, 0xa8, 0x6d, 0xf1, 0xc1, 0x81, 0x40, 0x81, 0x8b, 0x82, 0x9c, 0x91, 0x49, + 0xfb, 0xfe, 0x4b, 0xed, 0xa5, 0xc1, 0x72, 0x06, 0x69, 0x41, 0x7d, 0xd7, 0x1f, 0x89, 0x27, 0x29, + 0x0f, 0x65, 0x3a, 0xf2, 0x2a, 0x68, 0x5d, 0x64, 0x91, 0x55, 0x58, 0x78, 0x14, 0x0c, 0xb7, 0xfd, + 0xd0, 0xab, 0xb5, 0x9c, 0xb6, 0xcb, 0x0c, 0x65, 0xf9, 0x7c, 0xe2, 0x41, 0xce, 0xe7, 0x93, 0x2c, + 0xdc, 0xfa, 0x74, 0xb8, 0x3b, 0x51, 0x5f, 0xf2, 0x70, 0xc8, 0xe3, 0xe1, 0x33, 0x5f, 0xbc, 0xf0, + 0x96, 0x74, 0xb8, 0xd3, 0x5c, 0x65, 0xbb, 0xc9, 0x13, 0xe1, 0x35, 0xf0, 0x46, 0x3c, 0x93, 0x35, + 0xa8, 0x6e, 0xfa, 0xb2, 0x23, 0xc6, 0xf2, 0xc0, 0x5b, 0x6e, 0x39, 0xed, 0x32, 0xcb, 0x68, 0x72, + 0x09, 0x2a, 0xfd, 0x01, 0x0f, 0x84, 0x77, 0x01, 0x0d, 0x34, 0x41, 0x28, 0x2c, 0x3d, 0x88, 0x62, + 0xe1, 0xef, 0x87, 0x58, 0x04, 0xaf, 0x89, 0x41, 0x4d, 0xf1, 0xc8, 0xbb, 0xe0, 0xaa, 0x90, 0x56, + 0x5a, 0x4e, 0xbb, 0x7e, 0x7b, 0x65, 0xc3, 0xd6, 0x71, 0xa3, 0x23, 0x06, 0xfe, 0x88, 0x07, 0x4c, + 0x49, 0x51, 0x89, 0x4f, 0x3c, 0x72, 0xba, 0x12, 0x9f, 0x50, 0x0a, 0xcb, 0xbd, 0xd1, 0x38, 0x8a, + 0x25, 0x13, 0xc9, 0x38, 0x0a, 0x13, 0x41, 0x9a, 0xe0, 0x76, 0xe3, 0xd8, 0x73, 0xd0, 0xad, 0x3a, + 0xd2, 0x6f, 0xa1, 0xb9, 0x19, 0x44, 0x83, 0xc3, 0x0e, 0x97, 0x9c, 0x89, 0xe7, 0xa9, 0x48, 0xa4, + 0xc2, 0xae, 0xe1, 0x69, 0x3d, 0x4d, 0x28, 0x2e, 0xd6, 0xdb, 0x2b, 0x69, 0x2e, 0x12, 0x2a, 0x2f, + 0x98, 0x35, 0x5d, 0x1e, 0x3c, 0x63, 0xec, 0x07, 0x3c, 0x1e, 0x62, 0x4d, 0xcb, 0x4c, 0x13, 0x8a, + 0x8b, 0x9e, 0xb0, 0x0f, 0xca, 0x4c, 0x13, 0xb4, 0x07, 0x2b, 0x05, 0xff, 0x06, 0xe6, 0x2a, 0x2c, + 0xb0, 0xe8, 0x45, 0xaf, 0x93, 0x78, 0x4e, 0xcb, 0x6d, 0x97, 0x99, 0xa1, 0xb0, 0x61, 0xa2, 0x20, + 0x1d, 0x85, 0x4a, 0x54, 0x42, 0x51, 0xce, 0xa0, 0x57, 0xa1, 0x82, 0xdd, 0xa3, 0xa2, 0xcc, 0x6d, + 0xd5, 0x91, 0x7e, 0xef, 0x40, 0x6d, 0x9b, 0x4f, 0x10, 0x48, 0x42, 0xee, 0x42, 0xd5, 0xd6, 0x16, + 0x95, 0xea, 0xb7, 0xdf, 0xc9, 0x33, 0x98, 0xa9, 0x6d, 0x58, 0x9d, 0x6e, 0x28, 0xe3, 0x63, 0x96, + 0x99, 0xac, 0x7d, 0x06, 0x8d, 0x29, 0x91, 0xf2, 0x77, 0x28, 0x8e, 0x6d, 0x56, 0x0f, 0xc5, 0xb1, + 0x8a, 0xf5, 0x88, 0x07, 0xa9, 0xc0, 0x5c, 0x95, 0x99, 0x26, 0x3e, 0x2d, 0x7d, 0xec, 0xd0, 0x67, + 0x40, 0xb6, 0x62, 0xc1, 0xa5, 0x40, 0x27, 0xdb, 0x22, 0x49, 0xf8, 0xbe, 0x38, 0x2b, 0xe3, 0x6e, + 0x31, 0xe3, 0x59, 0x76, 0x4b, 0x85, 0xec, 0xd2, 0x1b, 0x40, 0x3a, 0x22, 0x10, 0x52, 0x98, 0xe9, + 0xfe, 0x87, 0x7b, 0xe9, 0xbe, 0xc5, 0x70, 0xb6, 0x2e, 0xb9, 0x0e, 0x65, 0xb5, 0x2a, 0xd0, 0x59, + 0xfd, 0xf6, 0xc5, 0x3c, 0x4f, 0xd9, 0x16, 0x61, 0x65, 0xbb, 0x4b, 0xba, 0xbb, 0x7c, 0x1f, 0xb1, + 0xba, 0x0c, 0xcf, 0xf4, 0x3b, 0xc7, 0x7a, 0x42, 0xe8, 0xe7, 0x8c, 0x76, 0xaa, 0xbf, 0x6e, 0x18, + 0xff, 0x2e, 0xfa, 0x5f, 0xcd, 0xfd, 0x17, 0x77, 0xcf, 0x0c, 0x84, 0x72, 0x01, 0xc2, 0x3d, 0x9b, + 0x97, 0xd7, 0x45, 0x40, 0x07, 0xf0, 0x7f, 0x7d, 0xc3, 0xfd, 0x23, 0xee, 0x07, 0x7c, 0x2f, 0xf8, + 0x57, 0xa5, 0x9b, 0x0a, 0xc6, 0x83, 0x45, 0xb4, 0xed, 0x75, 0xcc, 0x10, 0x58, 0x92, 0x3e, 0x87, + 0x7c, 0x9e, 0x76, 0xf8, 0x48, 0x98, 0xdb, 0xf0, 0x9c, 0xe5, 0xa0, 0x74, 0x8e, 0x1c, 0x5c, 0x82, + 0x8a, 0x9a, 0x41, 0xb5, 0xd3, 0x5d, 0xe5, 0x18, 0x89, 0xb9, 0x99, 0xb9, 0x03, 0x0b, 0xfd, 0xc1, + 0x81, 0x18, 0x71, 0xf2, 0x3e, 0x2c, 0x22, 0x6a, 0x91, 0x98, 0x71, 0xb8, 0x30, 0x53, 0x66, 0x66, + 0xe5, 0x6a, 0x90, 0x4c, 0x84, 0xf3, 0x80, 0x5a, 0x37, 0xa5, 0xdc, 0x0d, 0xb9, 0x09, 0x8b, 0x06, + 0x21, 0xee, 0x88, 0x53, 0x7a, 0xc8, 0xea, 0x90, 0xeb, 0xb0, 0x80, 0x51, 0x25, 0x5e, 0x79, 0x16, + 0x0a, 0xf2, 0x99, 0x11, 0xd3, 0x2e, 0xb8, 0x4f, 0x59, 0x4f, 0xad, 0x0a, 0x8c, 0xc2, 0x02, 0x31, + 0x94, 0x82, 0xf2, 0x45, 0x94, 0x48, 0x93, 0x7f, 0x3c, 0x2b, 0xde, 0xe3, 0x28, 0x96, 0x98, 0xfb, + 0x06, 0xc3, 0x33, 0xfd, 0xd9, 0x81, 0xf2, 0x4e, 0x34, 0x14, 0x64, 0x19, 0x4a, 0xbd, 0x8e, 0xb9, + 0xa4, 0xd4, 0xeb, 0x90, 0xb7, 0xf1, 0x7e, 0x93, 0xf3, 0x46, 0x8e, 0xe2, 0x29, 0xeb, 0x31, 0xf4, + 0x7c, 0x0d, 0x1a, 0xbd, 0x64, 0x2b, 0x8a, 0xe2, 0xa1, 0x1f, 0x72, 0x19, 0xc5, 0xe6, 0x2b, 0x3a, + 0xcd, 0xc4, 0x69, 0x95, 0x5c, 0xea, 0xef, 0x5b, 0x8d, 0x69, 0x82, 0x5c, 0x87, 0xc5, 0x87, 0xec, + 0xf1, 0x96, 0x72, 0x50, 0x99, 0xe7, 0xc0, 0x4a, 0xe9, 0x3d, 0x68, 0x2a, 0x74, 0x68, 0x65, 0x3b, + 0x6e, 0x15, 0x16, 0x14, 0x2f, 0x43, 0x6b, 0xa8, 0xdc, 0x55, 0xa9, 0xe0, 0x8a, 0x7e, 0xad, 0x6f, + 0xe8, 0x1e, 0x89, 0x50, 0x16, 0x7a, 0x16, 0x69, 0xbc, 0xa0, 0xc1, 0x34, 0x41, 0xa8, 0xce, 0x84, + 0x09, 0x79, 0x39, 0x47, 0xa4, 0xb8, 0x0c, 0x65, 0xf4, 0x47, 0x07, 0xc0, 0x02, 0x4a, 0x93, 0xcc, + 0xc4, 0x39, 0xdd, 0x84, 0xb4, 0x6d, 0x9f, 0x99, 0x19, 0x6e, 0xe6, 0x5a, 0x9a, 0xcf, 0x6c, 0x1f, + 0x7e, 0x90, 0xf7, 0xa1, 0x2e, 0xfe, 0xe5, 0x99, 0x56, 0xd1, 0x5e, 0xf3, 0x6e, 0x1c, 0x42, 0xbd, + 0xc0, 0x9f, 0xdb, 0x92, 0x37, 0xb3, 0x7e, 0x2a, 0xcd, 0x5e, 0x89, 0x7c, 0x73, 0xa5, 0x51, 0x9a, + 0xbb, 0xc5, 0xbe, 0x81, 0x7a, 0x41, 0x75, 0xae, 0x97, 0x36, 0x5c, 0x98, 0xde, 0x0e, 0xf6, 0xf3, + 0x34, 0xcb, 0x9e, 0xeb, 0xc0, 0x87, 0xc6, 0x56, 0x90, 0x26, 0x52, 0xc4, 0xc6, 0x85, 0xfa, 0xce, + 0x69, 0x46, 0x56, 0xe4, 0x9c, 0x31, 0xbf, 0xce, 0xe4, 0x1a, 0x54, 0x54, 0xba, 0xf5, 0xe0, 0x9f, + 0xac, 0x85, 0x16, 0xd2, 0x67, 0x50, 0xdd, 0xec, 0xf7, 0x1e, 0xc6, 0x51, 0x3a, 0x3e, 0x6d, 0x82, + 0xf1, 0xad, 0x56, 0x2a, 0x3c, 0xe2, 0x9a, 0xfa, 0x41, 0xa2, 0x11, 0xe3, 0xeb, 0xa3, 0xa9, 0x5f, + 0x1f, 0x65, 0xc3, 0xe1, 0x13, 0xda, 0x87, 0x15, 0xbd, 0xe8, 0xd5, 0xbe, 0x79, 0x9d, 0xd5, 0x68, + 0xdf, 0x11, 0x6e, 0xfe, 0x8e, 0x50, 0x97, 0xea, 0xcd, 0xfb, 0x5f, 0x5e, 0xfa, 0x57, 0x09, 0x56, + 0x98, 0x48, 0xfc, 0x97, 0xa2, 0x17, 0x26, 0x32, 0x4e, 0x07, 0x6a, 0xef, 0x28, 0xfb, 0x2f, 0xa3, + 0x3d, 0x93, 0x6d, 0x97, 0x69, 0xe2, 0x3c, 0x13, 0x41, 0x6e, 0x41, 0x7d, 0x76, 0x09, 0x9c, 0x54, + 0x2d, 0xaa, 0x90, 0x5b, 0xb0, 0xd8, 0x8f, 0xd2, 0x78, 0x90, 0xb5, 0x79, 0x61, 0xa3, 0x6b, 0x64, + 0x5a, 0xcc, 0xac, 0x1a, 0x79, 0x02, 0x64, 0x37, 0xe6, 0x61, 0x12, 0x70, 0x05, 0xd6, 0x1a, 0x57, + 0x67, 0x9f, 0x2e, 0x05, 0x9d, 0xa9, 0x7b, 0xe6, 0x18, 0x93, 0x0f, 0x8b, 0x73, 0xec, 0x2d, 0x22, + 0xea, 0x4b, 0xd3, 0xa8, 0xcd, 0x68, 0x14, 0xe7, 0xfd, 0xee, 0x4c, 0xa7, 0x7a, 0x0b, 0x68, 0x78, + 0x25, 0x37, 0x9c, 0x12, 0xb3, 0x69, 0x6d, 0xfa, 0x83, 0x03, 0x4b, 0x45, 0x64, 0xe7, 0xda, 0x1f, + 0x59, 0xc1, 0x4b, 0x67, 0xbf, 0x8d, 0x6c, 0xc1, 0xcb, 0xf3, 0x5e, 0xa3, 0x95, 0xe2, 0x7b, 0x29, + 0x85, 0x2b, 0xa7, 0xa4, 0xeb, 0x0d, 0x40, 0xb5, 0xa0, 0xfe, 0x98, 0xc7, 0xd2, 0x57, 0x57, 0x9a, + 0x6f, 0x7c, 0x85, 0x15, 0x59, 0xf4, 0x10, 0xae, 0x9e, 0x68, 0xbe, 0xad, 0x68, 0x34, 0x56, 0x5d, + 0xfe, 0x06, 0x4d, 0xa8, 0x16, 0x7a, 0x1c, 0x9b, 0xf6, 0xab, 0x31, 0x4d, 0xd0, 0x4f, 0xe0, 0x72, + 0x5f, 0xc8, 0x42, 0xeb, 0xd9, 0x19, 0x6a, 0x81, 0xbb, 0x23, 0x5e, 0x9c, 0x12, 0xa0, 0x12, 0xd1, + 0xcf, 0xc1, 0x7b, 0x3a, 0x1e, 0x72, 0x29, 0x5e, 0xcb, 0x7a, 0x13, 0xaa, 0xbb, 0xd1, 0x38, 0x0a, + 0xa2, 0xfd, 0xe3, 0x33, 0x76, 0x99, 0x07, 0x8b, 0xfa, 0xeb, 0xa5, 0x17, 0x66, 0x8d, 0x59, 0x92, + 0x5e, 0x54, 0x63, 0x3a, 0xe0, 0xc1, 0x20, 0x0d, 0x14, 0x0c, 0xf5, 0xb0, 0x4f, 0xa8, 0x30, 0x83, + 0xc0, 0x31, 0x71, 0x85, 0x0f, 0xe2, 0x7d, 0x64, 0xd8, 0x0f, 0xa2, 0xa6, 0xc8, 0x47, 0x50, 0x2f, + 0x68, 0x9b, 0x04, 0x5e, 0x9e, 0x99, 0x17, 0x2d, 0x64, 0x45, 0x4d, 0xfa, 0xab, 0x33, 0x65, 0x79, + 0xe2, 0x6d, 0x60, 0x1c, 0x1e, 0xe9, 0xa2, 0x54, 0x99, 0xa1, 0x54, 0xac, 0xdd, 0xc9, 0x20, 0x48, + 0x13, 0x25, 0xd2, 0xcf, 0x81, 0x9c, 0xa1, 0x62, 0x55, 0xbf, 0x5e, 0xa3, 0x54, 0x9a, 0xcd, 0x69, + 0x49, 0xf5, 0x43, 0xb2, 0x23, 0xf8, 0x30, 0xf0, 0x43, 0x81, 0x5d, 0xea, 0xb2, 0x8c, 0x26, 0xb7, + 0xf4, 0xb6, 0xb7, 0xa3, 0xb6, 0x36, 0x17, 0x3e, 0x6a, 0xe8, 0x2f, 0x41, 0x42, 0x09, 0x34, 0x67, + 0x45, 0x9b, 0xcd, 0xdf, 0x5e, 0xad, 0x3b, 0xbf, 0xbf, 0x5a, 0x77, 0xfe, 0x78, 0xb5, 0xee, 0xfc, + 0xf4, 0xe7, 0xfa, 0xff, 0xf6, 0x16, 0xf0, 0xff, 0x80, 0x3b, 0x7f, 0x07, 0x00, 0x00, 0xff, 0xff, + 0x4f, 0xdc, 0x5a, 0x3d, 0x38, 0x10, 0x00, 0x00, } func (m *IndexMeta) Marshal() (dAtA []byte, err error) { @@ -3043,6 +3093,11 @@ func (m *CreateIndexMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x18 + } if m.Meta != nil { { size, err := m.Meta.MarshalToSizedBuffer(dAtA[:i]) @@ -3089,6 +3144,11 @@ func (m *CreateFieldMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x20 + } if m.Meta != nil { { size, err := m.Meta.MarshalToSizedBuffer(dAtA[:i]) @@ -3229,6 +3289,11 @@ func (m *Field) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x20 + } if len(m.Views) > 0 { for iNdEx := len(m.Views) - 1; iNdEx >= 0; iNdEx-- { i -= len(m.Views[iNdEx]) @@ -3351,6 +3416,11 @@ func (m *Index) MarshalToSizedBuffer(dAtA []byte) (int, error) { dAtA[i] = 0x22 } } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x10 + } if len(m.Name) > 0 { i -= len(m.Name) copy(dAtA[i:], m.Name) @@ -3656,6 +3726,11 @@ func (m *IndexStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x18 + } if len(m.Fields) > 0 { for iNdEx := len(m.Fields) - 1; iNdEx >= 0; iNdEx-- { { @@ -3704,6 +3779,11 @@ func (m *FieldStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.ETag != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.ETag)) + i-- + dAtA[i] = 0x18 + } if len(m.AvailableShards) > 0 { dAtA19 := make([]byte, len(m.AvailableShards)*10) var j18 int @@ -4759,6 +4839,9 @@ func (m *CreateIndexMessage) Size() (n int) { l = m.Meta.Size() n += 1 + l + sovPrivate(uint64(l)) } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -4783,6 +4866,9 @@ func (m *CreateFieldMessage) Size() (n int) { l = m.Meta.Size() n += 1 + l + sovPrivate(uint64(l)) } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -4852,6 +4938,9 @@ func (m *Field) Size() (n int) { n += 1 + l + sovPrivate(uint64(l)) } } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -4886,6 +4975,9 @@ func (m *Index) Size() (n int) { if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if len(m.Fields) > 0 { for _, e := range m.Fields { l = e.Size() @@ -5037,6 +5129,9 @@ func (m *IndexStatus) Size() (n int) { n += 1 + l + sovPrivate(uint64(l)) } } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -5060,6 +5155,9 @@ func (m *FieldStatus) Size() (n int) { } n += 1 + sovPrivate(uint64(l)) + l } + if m.ETag != 0 { + n += 1 + sovPrivate(uint64(m.ETag)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -7021,6 +7119,25 @@ func (m *CreateIndexMessage) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -7175,6 +7292,25 @@ func (m *CreateFieldMessage) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -7584,6 +7720,25 @@ func (m *Field) Unmarshal(dAtA []byte) error { } m.Views = append(m.Views, string(dAtA[iNdEx:postIndex])) iNdEx = postIndex + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -7758,6 +7913,25 @@ func (m *Index) Unmarshal(dAtA []byte) error { } m.Name = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } case 4: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field Fields", wireType) @@ -8682,6 +8856,25 @@ func (m *IndexStatus) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -8844,6 +9037,25 @@ func (m *FieldStatus) Unmarshal(dAtA []byte) error { } else { return fmt.Errorf("proto: wrong wireType = %d for field AvailableShards", wireType) } + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field ETag", wireType) + } + m.ETag = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.ETag |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) diff --git a/internal/private.proto b/internal/private.proto index 849fff04a..502d3b7bd 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -64,12 +64,14 @@ message DeleteIndexMessage { message CreateIndexMessage { string Index = 1; IndexMeta Meta = 2; + int64 ETag = 3; } message CreateFieldMessage { string Index = 1; string Field = 2; FieldOptions Meta = 3; + int64 ETag = 4; } message DeleteFieldMessage { @@ -87,6 +89,7 @@ message Field { string Name = 1; FieldOptions Meta = 2; repeated string Views = 3; + int64 ETag = 4; } message Schema { @@ -95,6 +98,7 @@ message Schema { message Index { string Name = 1; + int64 ETag = 2; IndexMeta Options = 5; repeated Field Fields = 4; } @@ -132,11 +136,13 @@ message NodeStatus { message IndexStatus { string Name = 1; repeated FieldStatus Fields = 2; + int64 ETag = 3; } message FieldStatus { string Name = 1; repeated uint64 AvailableShards = 2; + int64 ETag = 3; } message ClusterStatus { diff --git a/pilosa.go b/pilosa.go index d7eb2f3c8..e49af5406 100644 --- a/pilosa.go +++ b/pilosa.go @@ -17,6 +17,7 @@ package pilosa import ( "encoding/json" "regexp" + "time" "github.com/pkg/errors" ) @@ -191,6 +192,10 @@ func stringSlicesAreEqual(a, b []string) bool { return true } +func newETag() int64 { + return time.Now().UTC().UnixNano() +} + // AddressWithDefaults converts addr into a valid address, // using defaults when necessary. func AddressWithDefaults(addr string) (*URI, error) { diff --git a/server.go b/server.go index fd07ec0b2..586d14e73 100644 --- a/server.go +++ b/server.go @@ -356,14 +356,11 @@ func NewServer(opts ...ServerOption) (*Server, error) { } s.executor = newExecutor(executorOpts...) - // s.holder.translateFile.logger = s.logger - path, err := expandDirName(s.dataDir) if err != nil { return nil, err } s.holder.Path = path - // s.holder.translateFile.Path = filepath.Join(path, ".keys") s.holder.Logger = s.logger s.holder.Stats.SetLogger(s.logger) @@ -716,10 +713,13 @@ func (s *Server) receiveMessage(m Message) error { } case *CreateIndexMessage: opt := obj.Meta - _, err := s.holder.CreateIndex(obj.Index, *opt) + idx, err := s.holder.CreateIndex(obj.Index, *opt) if err != nil { return err } + idx.mu.Lock() + idx.etag = obj.ETag + idx.mu.Unlock() case *DeleteIndexMessage: if err := s.holder.DeleteIndex(obj.Index); err != nil { return err @@ -730,10 +730,13 @@ func (s *Server) receiveMessage(m Message) error { return fmt.Errorf("local index not found: %s", obj.Index) } opt := obj.Meta - _, err := idx.createFieldIfNotExists(obj.Field, opt) + fld, err := idx.createFieldIfNotExists(obj.Field, opt) if err != nil { return err } + fld.mu.Lock() + fld.etag = obj.ETag + fld.mu.Unlock() case *DeleteFieldMessage: idx := s.holder.Index(obj.Index) if err := idx.DeleteField(obj.Field); err != nil { diff --git a/server/handler_test.go b/server/handler_test.go index b01b15c59..9cd25f50e 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -203,6 +203,53 @@ func TestHandler_Endpoints(t *testing.T) { } }) + t.Run("Import", func(t *testing.T) { + indexInfo := cmd.API.Schema(context.Background()) + err := cmd.API.ApplySchema(context.Background(), &pilosa.Schema{Indexes: indexInfo}, false) + if err != nil { + t.Fatalf("applying schema: %v", err) + } + + idx := indexInfo[0] + fld := indexInfo[0].Fields[0] + msg := pilosa.ImportRequest{ + Index: idx.Name, + Field: fld.Name, + Shard: 0, + } + ser := proto.Serializer{} + data, err := ser.Marshal(&msg) + if err != nil { + t.Fatal(err) + } + path := fmt.Sprintf("/index/%s/field/%s/import", idx.Name, fld.Name) + etag := fmt.Sprintf("%d, %d", idx.ETag, fld.ETag) + + httpReq := test.MustNewHTTPRequest("POST", path, bytes.NewBuffer(data)) + httpReq.Header.Set("Content-Type", "application/x-protobuf") + httpReq.Header.Set("Accept", "application/x-protobuf") + httpReq.Header.Set("If-Match", etag) + + w := httptest.NewRecorder() + h.ServeHTTP(w, httpReq) + if w.Body.String() != "" { + t.Fatalf(w.Body.String()) + } + + etag = "invalid-index-etag, invalid-field-etag" + + httpReq = test.MustNewHTTPRequest("POST", path, bytes.NewBuffer(data)) + httpReq.Header.Set("Content-Type", "application/x-protobuf") + httpReq.Header.Set("Accept", "application/x-protobuf") + httpReq.Header.Set("If-Match", etag) + + h.ServeHTTP(w, httpReq) + + if strings.TrimSpace(w.Body.String()) != "Precondition Failed" { + t.Fatal("expected: Precondition Failed, got:" + w.Body.String()) + } + }) + t.Run("ImportRoaring", func(t *testing.T) { w := httptest.NewRecorder() roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100") @@ -217,10 +264,23 @@ func TestHandler_Endpoints(t *testing.T) { if err != nil { t.Fatal(err) } + idx, err := cmd.API.Index(context.Background(), "i0") + if err != nil { + t.Fatal(err) + } + fld, err := cmd.API.Field(context.Background(), "i0", "f1") + if err != nil { + t.Fatal(err) + } + httpReq := test.MustNewHTTPRequest("POST", "/index/i0/field/f1/import-roaring/0", bytes.NewBuffer(data)) httpReq.Header.Set("Content-Type", "application/x-protobuf") httpReq.Header.Set("Accept", "application/x-protobuf") + httpReq.Header.Set("If-Match", fmt.Sprintf("%d, %d", idx.ETag(), fld.ETag())) h.ServeHTTP(w, httpReq) + if w.Body.String() != "" { + t.Fatalf("Unexpected response body: %s", w.Body.String()) + } resp, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i0", Query: "TopN(f1)"}) if err != nil { t.Fatalf("querying: %v", err)