diff --git a/api.go b/api.go index 7ced231ae..7c2da4893 100644 --- a/api.go +++ b/api.go @@ -1585,6 +1585,34 @@ func (api *API) PrimaryReplicaNodeURL() url.URL { return node.URI.URL() } +func (api *API) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool, remote bool) (*Transaction, error) { + if err := api.validate(apiStartTransaction); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.server.StartTransaction(ctx, id, timeout, exclusive, remote) +} + +func (api *API) FinishTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { + if err := api.validate(apiFinishTransaction); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.server.FinishTransaction(ctx, id, remote) +} + +func (api *API) Transactions(ctx context.Context) (map[string]*Transaction, error) { + if err := api.validate(apiTransactions); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.server.Transactions(ctx) +} + +func (api *API) GetTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { + if err := api.validate(apiGetTransaction); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.server.GetTransaction(ctx, id, remote) +} + type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` Memory uint64 `json:"memory"` @@ -1632,6 +1660,10 @@ const ( //apiVersion // not implemented apiViews apiApplySchema + apiStartTransaction + apiFinishTransaction + apiTransactions + apiGetTransaction ) var methodsCommon = map[apiMethod]struct{}{ @@ -1667,4 +1699,8 @@ var methodsNormal = map[apiMethod]struct{}{ apiShardNodes: {}, apiViews: {}, apiApplySchema: {}, + apiStartTransaction: {}, + apiFinishTransaction: {}, + apiTransactions: {}, + apiGetTransaction: {}, } diff --git a/broadcast.go b/broadcast.go index 6f2245992..37d2bb39d 100644 --- a/broadcast.go +++ b/broadcast.go @@ -69,6 +69,7 @@ const ( messageTypeRecalculateCaches messageTypeNodeEvent messageTypeNodeStatus + messageTypeTransaction ) // MarshalInternalMessage serializes the pilosa message and adds pilosa internal @@ -116,6 +117,8 @@ func getMessage(typ byte) Message { return &NodeEvent{} case messageTypeNodeStatus: return &NodeStatus{} + case messageTypeTransaction: + return &TransactionMessage{} default: panic(fmt.Sprintf("unknown message type %d", typ)) } @@ -155,6 +158,8 @@ func getMessageType(m Message) byte { return messageTypeNodeEvent case *NodeStatus: return messageTypeNodeStatus + case *TransactionMessage: + return messageTypeTransaction default: panic(fmt.Sprintf("don't have type for message %#v", m)) } diff --git a/client.go b/client.go index a1b808a77..01579b1e8 100644 --- a/client.go +++ b/client.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "io" + "time" ) // Bit represents the intersection of a row and a column. It can be specified by @@ -73,6 +74,11 @@ type InternalClient interface { RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error + + StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) + FinishTransaction(ctx context.Context, id string) (*Transaction, error) + Transactions(ctx context.Context) (map[string]*Transaction, error) + GetTransaction(ctx context.Context, id string) (*Transaction, error) } //=============== @@ -204,3 +210,16 @@ func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, fiel func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) { return nil, nil } + +func (n nopInternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) { + return nil, nil +} +func (n nopInternalClient) FinishTransaction(ctx context.Context, id string) (*Transaction, error) { + return nil, nil +} +func (n nopInternalClient) Transactions(ctx context.Context) (map[string]*Transaction, error) { + return nil, nil +} +func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (*Transaction, error) { + return nil, nil +} diff --git a/cluster.go b/cluster.go index af0b43e77..ae7c2106c 100644 --- a/cluster.go +++ b/cluster.go @@ -2589,3 +2589,15 @@ type FieldStatus struct { // RecalculateCaches is an internal message for recalculating all caches // within a holder. type RecalculateCaches struct{} + +// Transaction Actions +const ( + TRANSACTION_START = "start" + TRANSACTION_FINISH = "finish" + TRANSACTION_VALIDATE = "validate" +) + +type TransactionMessage struct { + Transaction *Transaction + Action string +} diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 3c72e1d16..678b2d04c 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -17,6 +17,7 @@ package proto import ( "fmt" "sort" + "time" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2" @@ -290,6 +291,14 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error { } decodeTranslateIDsResponse(msg, mt) return nil + case *pilosa.TransactionMessage: + msg := &internal.TransactionMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling TransactionMessage") + } + decodeTransactionMessage(msg, mt) + return nil default: panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m)) } @@ -359,6 +368,8 @@ func encodeToProto(m pilosa.Message) proto.Message { return encodeTranslateIDsRequest(mt) case *pilosa.TranslateIDsResponse: return encodeTranslateIDsResponse(mt) + case *pilosa.TransactionMessage: + return encodeTransactionMessage(mt) } return nil } @@ -829,6 +840,38 @@ func encodeTranslateIDsResponse(response *pilosa.TranslateIDsResponse) *internal } } +func encodeTransactionMessage(msg *pilosa.TransactionMessage) *internal.TransactionMessage { + return &internal.TransactionMessage{ + Action: msg.Action, + Transaction: encodeTransaction(msg.Transaction), + } +} + +func encodeTransaction(trns *pilosa.Transaction) *internal.Transaction { + if trns == nil { + return nil + } + return &internal.Transaction{ + ID: trns.ID, + Active: trns.Active, + Exclusive: trns.Exclusive, + Timeout: int64(trns.Timeout), + Deadline: encodeTransactionDeadline(trns.Deadline), + Stats: encodeTransactionStats(trns.Stats), + } +} + +func encodeTransactionDeadline(deadline time.Time) int64 { + if deadline.Year() > 2262 || deadline.Year() < 1678 { + return 0 + } + return deadline.UnixNano() +} + +func encodeTransactionStats(stats pilosa.TransactionStats) *internal.TransactionStats { + return &internal.TransactionStats{} +} + func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeInstruction) { m.JobID = ri.JobID m.Node = &pilosa.Node{} @@ -1198,6 +1241,27 @@ func decodeTranslateIDsResponse(pb *internal.TranslateIDsResponse, m *pilosa.Tra m.Keys = pb.Keys } +func decodeTransactionMessage(pb *internal.TransactionMessage, m *pilosa.TransactionMessage) { + m.Action = pb.Action + if pb.Transaction == nil { + m.Transaction = nil + return + } else if m.Transaction == nil { + m.Transaction = &pilosa.Transaction{} + } + decodeTransaction(pb.Transaction, m.Transaction) +} + +func decodeTransaction(pb *internal.Transaction, trns *pilosa.Transaction) { + + trns.ID = pb.ID + trns.Active = pb.Active + trns.Exclusive = pb.Exclusive + trns.Timeout = time.Duration(pb.Timeout) + trns.Deadline = time.Unix(0, pb.Deadline) + // TODO m.Stats... once it has anything +} + // QueryResult types. const ( queryResultTypeNil uint32 = iota diff --git a/holder.go b/holder.go index 51a89826b..ad52ee082 100644 --- a/holder.go +++ b/holder.go @@ -85,6 +85,12 @@ type Holder struct { OpenTranslateStore OpenTranslateStoreFunc OpenTranslateReader OpenTranslateReaderFunc + // Func to open whatever implementation of transaction store we're using. + OpenTransactionStore OpenTransactionStoreFunc + + // transactionManager + transactionManager *TransactionManager + translationSyncer translationSyncer // Queue of fields (having a foreign index) which have @@ -98,6 +104,22 @@ type Holder struct { opening bool } +func (h *Holder) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) { + return h.transactionManager.Start(ctx, id, timeout, exclusive) +} + +func (h *Holder) FinishTransaction(ctx context.Context, id string) (*Transaction, error) { + return h.transactionManager.Finish(ctx, id) +} + +func (h *Holder) Transactions(ctx context.Context) (map[string]*Transaction, error) { + return h.transactionManager.List(ctx) +} + +func (h *Holder) GetTransaction(ctx context.Context, id string) (*Transaction, error) { + return h.transactionManager.Get(ctx, id) +} + // lockedChan looks a little ridiculous admittedly, but exists for good reason. // The channel within is used (for example) to signal to other goroutines when // the Holder has finished opening (via closing the channel). However, it is @@ -142,6 +164,8 @@ func NewHolder(partitionN int) *Holder { OpenTranslateStore: OpenInMemTranslateStore, + OpenTransactionStore: OpenInMemTransactionStore, + translationSyncer: NopTranslationSyncer, Logger: logger.NopLogger, @@ -170,6 +194,13 @@ func (h *Holder) Open() error { return ErrCannotOpenV1TranslateFile } + tstore, err := h.OpenTransactionStore(h.Path) + if err != nil { + return errors.Wrap(err, "opening transaction store") + } + h.transactionManager = NewTransactionManager(tstore) + h.transactionManager.Log = h.Logger + // Open path to read all index directories. f, err := os.Open(h.Path) if err != nil { diff --git a/http/client.go b/http/client.go index a8a5106d6..2b112e9c5 100644 --- a/http/client.go +++ b/http/client.go @@ -26,6 +26,7 @@ import ( "net/url" "sort" "strconv" + "time" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/encoding/proto" @@ -392,6 +393,13 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits // Get the coordinator node; all bits are sent to the // primary translate store (i.e. coordinator). + // TODO... is that right^^? + // RESPONSE: It looks like in ctl/import.go, we could change the + // logic in ImportCommand.importBits() to only use ImportK + // when useRowKeys = true. It's no longer necessary to + // send column key translations to the coordinator (although + // it should still work). As far as I know, the only thing + // that uses ImportK is the pilosa import sub-command. nodes, err := c.Nodes(ctx) if err != nil { return fmt.Errorf("getting nodes: %s", err) @@ -1227,11 +1235,170 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI, return tkresp.Keys, nil } +func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.Transaction, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions") + defer span.Finish() + + u := uriPathToURL(c.defaultURI, "/transactions") + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + return nil, errors.Wrap(err, "creating transactions request") + } + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer func() { + _, _ = io.Copy(ioutil.Discard, resp.Body) + _ = resp.Body.Close() + }() + trnsMap := make(map[string]*pilosa.Transaction) + err = json.NewDecoder(resp.Body).Decode(&trnsMap) + return trnsMap, errors.Wrap(err, "json decoding") +} + +func (c *InternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*pilosa.Transaction, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.StartTransaction") + defer span.Finish() + buf, err := json.Marshal(&pilosa.Transaction{ + ID: id, + Timeout: timeout, + Exclusive: exclusive, + }) + if err != nil { + return nil, errors.Wrap(err, "marshalling payload") + } + // We're using the defaultURI here because this is only used by + // tests, and we want to test requests against all hosts. A robust + // client implementation would ensure that these requests go to + // the coordinator. + u := uriPathToURL(c.defaultURI, "/transaction/"+id) + req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf)) + if err != nil { + return nil, errors.Wrap(err, "creating post transaction request") + } + req.Header.Set("Content-Length", strconv.Itoa(len(buf))) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer func() { + _, _ = io.Copy(ioutil.Discard, resp.Body) + _ = resp.Body.Close() + }() + tr := &TransactionResponse{} + err = json.NewDecoder(resp.Body).Decode(tr) + if err != nil { + return nil, errors.Wrap(err, "decoding response") + } + if resp.StatusCode == 409 { + err = pilosa.ErrTransactionExclusive + } else if tr.Error != "" { + err = errors.New(tr.Error) + } + return tr.Transaction, err +} + +func (c *InternalClient) FinishTransaction(ctx context.Context, id string) (*pilosa.Transaction, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FinishTransaction") + defer span.Finish() + + u := uriPathToURL(c.defaultURI, "/transaction/"+id+"/finish") + req, err := http.NewRequest("POST", u.String(), nil) + if err != nil { + return nil, errors.Wrap(err, "creating finish transaction request") + } + + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer func() { + _, _ = io.Copy(ioutil.Discard, resp.Body) + _ = resp.Body.Close() + }() + tr := &TransactionResponse{} + err = json.NewDecoder(resp.Body).Decode(tr) + if err != nil { + return nil, errors.Wrap(err, "decoding response") + } + + if tr.Error != "" { + err = errors.New(tr.Error) + } + return tr.Transaction, err +} + +func (c *InternalClient) GetTransaction(ctx context.Context, id string) (*pilosa.Transaction, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.GetTransaction") + defer span.Finish() + + // We're using the defaultURI here because this is only used by + // tests, and we want to test requests against all hosts. A robust + // client implementation would ensure that these requests go to + // the coordinator. + u := uriPathToURL(c.defaultURI, "/transaction/"+id) + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + return nil, errors.Wrap(err, "creating get transaction request") + } + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + resp, err := c.executeRequest(req.WithContext(ctx), giveRawResponse(true)) + if err != nil { + return nil, errors.Wrap(err, "executing request") + } + defer func() { + _, _ = io.Copy(ioutil.Discard, resp.Body) + _ = resp.Body.Close() + }() + tr := &TransactionResponse{} + err = json.NewDecoder(resp.Body).Decode(tr) + if err != nil { + return nil, errors.Wrap(err, "decoding response") + } + + if tr.Error != "" { + err = errors.New(tr.Error) + } + return tr.Transaction, err +} + +type executeOpts struct { + // giveRawResponse instructs executeRequest not to process the + // respStatusCode and try to extract errors or whatever. + giveRawResponse bool +} + +type executeRequestOption func(*executeOpts) + +func giveRawResponse(b bool) executeRequestOption { + return func(eo *executeOpts) { + eo.giveRawResponse = b + } +} + // executeRequest executes the given request and checks the Response. For // responses with non-2XX status, the body is read and closed, and an error is // returned. If the error is nil, the caller must ensure that the response body // is closed. -func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, error) { +func (c *InternalClient) executeRequest(req *http.Request, opts ...executeRequestOption) (*http.Response, error) { + eo := &executeOpts{} + for _, opt := range opts { + opt(eo) + } + tracing.GlobalTracer.InjectHTTPHeaders(req) req.Close = false resp, err := c.httpClient.Do(req) @@ -1241,6 +1408,9 @@ func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, erro } return nil, errors.Wrap(err, "getting response") } + if eo.giveRawResponse { + return resp, nil + } if resp.StatusCode < 200 || resp.StatusCode >= 300 { defer resp.Body.Close() buf, err := ioutil.ReadAll(resp.Body) diff --git a/http/client_test.go b/http/client_test.go index 9d8f61aca..fb5568658 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -23,6 +23,7 @@ import ( gohttp "net/http" "reflect" "strconv" + "strings" "testing" "time" @@ -32,6 +33,7 @@ import ( "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "github.com/pkg/errors" ) // Test distributed TopN Row count across 3 nodes. @@ -1242,6 +1244,163 @@ func TestClient_CreateDecimalField(t *testing.T) { } } +func TestClientTransactions(t *testing.T) { + c := test.MustRunCluster(t, 3) + defer c.Close() + + client0 := MustNewClient(c[0].URL(), http.GetHTTPClient(nil)) + client1 := MustNewClient(c[1].URL(), http.GetHTTPClient(nil)) + + // can create, list, get, and finish a transaction + var expDeadline time.Time + if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err != nil { + t.Fatalf("error starting transaction: %v", err) + } else { + expDeadline = time.Now().Add(time.Minute) + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + + if trnsMap, err := client0.Transactions(context.Background()); err != nil { + t.Errorf("listing transactions: %v", err) + } else { + if len(trnsMap) != 1 { + t.Errorf("unexpected trnsMap: %+v", trnsMap) + } + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trnsMap["blah"]) + } + + if trns, err := client0.GetTransaction(context.Background(), "blah"); err != nil { + t.Fatalf("error getting transaction: %v", err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + + if trns, err := client0.FinishTransaction(context.Background(), "blah"); err != nil { + t.Fatalf("error finishing transaction: %v", err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + + // can create exclusive transaction + if trns, err := client0.StartTransaction(context.Background(), "blahe", time.Minute, true); err != nil { + t.Fatalf("error starting transaction: %v", err) + } else { + expDeadline = time.Now().Add(time.Minute) + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline}, + trns) + } + + // cannot start new transaction - correct error and exclusive transaction are returned + if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); errors.Cause(err) != pilosa.ErrTransactionExclusive { + t.Fatalf("shouldn't be able to start transaction while an exclusive is running, but got: %+v, %v", trns, err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline}, + trns) + } + + // finish exclusive transaction + if trns, err := client0.FinishTransaction(context.Background(), "blahe"); err != nil { + t.Fatalf("error finishing transaction: %v", err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: true, Exclusive: true, Deadline: expDeadline}, + trns) + } + + // start new transaction + if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err != nil { + t.Fatalf("error starting transaction: %v", err) + } else { + expDeadline = time.Now().Add(time.Minute) + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + + // try to start same transaction + if trns, err := client0.StartTransaction(context.Background(), "blah", time.Minute, false); err == nil || + !strings.Contains(err.Error(), pilosa.ErrTransactionExists.Error()) { + t.Fatalf("expected ErrTransactionExists, but got: %v", err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blah", Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + + // start an exclusive transaction which can't go active + if trns, err := client0.StartTransaction(context.Background(), "blahe", time.Minute, true); err != nil { + t.Fatalf("error starting transaction: %v", err) + } else { + expDeadline = time.Now().Add(time.Minute) + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: false, Exclusive: true, Deadline: expDeadline}, + trns) + } + + // finish exclusive transaction that never went active + if trns, err := client0.FinishTransaction(context.Background(), "blahe"); err != nil { + t.Fatalf("error finishing transaction: %v", err) + } else { + test.CompareTransactions(t, + &pilosa.Transaction{ID: "blahe", Timeout: time.Minute, Active: false, Exclusive: true, Deadline: expDeadline}, + trns) + } + + // finish non-existent transaction + if trns, err := client0.FinishTransaction(context.Background(), "zzz"); err == nil || + !strings.Contains(err.Error(), pilosa.ErrTransactionNotFound.Error()) { + t.Fatalf("unexpected error finishing nonexistent transaction: %v", err) + } else { + test.CompareTransactions(t, + nil, + trns) + } + + // get non-existent transaction + if trns, err := client0.GetTransaction(context.Background(), "xxx"); err == nil || + !strings.Contains(err.Error(), pilosa.ErrTransactionNotFound.Error()) { + t.Fatalf("unexpected error getting nonexistent transaction: %v", err) + } else { + test.CompareTransactions(t, + nil, + trns) + } + + // non-coordinator + if trns, err := client1.StartTransaction(context.Background(), "blah", time.Minute, false); err == nil || + !strings.Contains(err.Error(), pilosa.ErrNodeNotCoordinator.Error()) { + t.Fatalf("unexpected error starting on non-coordinator: %v", err) + } else { + test.CompareTransactions(t, + nil, + trns) + } + + // start transaction with blank id + if trns, err := client0.StartTransaction(context.Background(), "", time.Minute, false); err != nil { + t.Fatalf("error starting transaction: %v", err) + } else { + expDeadline = time.Now().Add(time.Minute) + if len(trns.ID) != 36 { + t.Errorf("expected generated UUID, but got '%s'", trns.ID) + } + test.CompareTransactions(t, + &pilosa.Transaction{ID: trns.ID, Timeout: time.Minute, Active: true, Deadline: expDeadline}, + trns) + } + +} + // Client represents a test wrapper for pilosa.Client. type Client struct { *http.InternalClient diff --git a/http/handler.go b/http/handler.go index cecaa666e..c46f3914f 100644 --- a/http/handler.go +++ b/http/handler.go @@ -204,6 +204,10 @@ func (h *Handler) populateValidators() { h.validators["PostFieldAttrDiff"] = queryValidationSpecRequired() h.validators["GetNodes"] = queryValidationSpecRequired() h.validators["GetShardMax"] = queryValidationSpecRequired() + h.validators["GetTransactions"] = queryValidationSpecRequired() + h.validators["GetTransaction"] = queryValidationSpecRequired() + h.validators["PostTransaction"] = queryValidationSpecRequired() + h.validators["PostFinishTransaction"] = queryValidationSpecRequired() } type contextKeyQuery int @@ -336,6 +340,12 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET").Name("GetSchema") router.HandleFunc("/schema", handler.handlePostSchema).Methods("POST").Name("PostSchema") router.HandleFunc("/status", handler.handleGetStatus).Methods("GET").Name("GetStatus") + router.HandleFunc("/transaction", handler.handlePostTransaction).Methods("POST").Name("PostTransaction") + router.HandleFunc("/transaction/", handler.handlePostTransaction).Methods("POST").Name("PostTransaction") + router.HandleFunc("/transaction/{id}", handler.handleGetTransaction).Methods("GET").Name("GetTransaction") + router.HandleFunc("/transaction/{id}", handler.handlePostTransaction).Methods("POST").Name("PostTransaction") + router.HandleFunc("/transaction/{id}/finish", handler.handlePostFinishTransaction).Methods("POST").Name("PostFinishTransaction") + router.HandleFunc("/transactions", handler.handleGetTransactions).Methods("GET").Name("GetTransactions") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion") // /internal endpoints are for internal use only; they may change at any time. @@ -1031,6 +1041,102 @@ func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) { resp.write(w, err) } +func (h *Handler) handleGetTransactions(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + trnsMap, err := h.api.Transactions(r.Context()) + if err != nil { + switch errors.Cause(err) { + case pilosa.ErrNodeNotCoordinator: + http.Error(w, err.Error(), http.StatusBadRequest) + default: + http.Error(w, "problem getting transactions: "+err.Error(), http.StatusInternalServerError) + } + return + } + + if err := json.NewEncoder(w).Encode(trnsMap); err != nil { + h.logger.Printf("encoding GetTransactions response: %s", err) + } +} + +type TransactionResponse struct { + Transaction *pilosa.Transaction `json:"transaction,omitempty"` + Error string `json:"error,omitempty"` +} + +func (h *Handler) doTransactionResponse(w http.ResponseWriter, err error, trns *pilosa.Transaction) { + if err != nil { + switch errors.Cause(err) { + case pilosa.ErrNodeNotCoordinator, pilosa.ErrTransactionExists: + w.WriteHeader(http.StatusBadRequest) + case pilosa.ErrTransactionExclusive: + w.WriteHeader(http.StatusConflict) + case pilosa.ErrTransactionNotFound: + w.WriteHeader(http.StatusNotFound) + default: + w.WriteHeader(http.StatusInternalServerError) + } + } + + var errString string + if err != nil { + errString = err.Error() + } + err = json.NewEncoder(w).Encode( + TransactionResponse{Error: errString, Transaction: trns}) + if err != nil { + h.logger.Printf("encoding transaction response: %v", err) + } + +} + +func (h *Handler) handleGetTransaction(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + id := mux.Vars(r)["id"] + trns, err := h.api.GetTransaction(r.Context(), id, false) + h.doTransactionResponse(w, err, trns) +} + +func (h *Handler) handlePostTransaction(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + reqTrns := &pilosa.Transaction{} + if err := json.NewDecoder(r.Body).Decode(reqTrns); err != nil || reqTrns.Timeout == 0 { + if err == nil { + http.Error(w, "timeout is required and cannot be 0", http.StatusBadRequest) + } else { + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + id, ok := mux.Vars(r)["id"] + if !ok { + id = reqTrns.ID + } + trns, err := h.api.StartTransaction(r.Context(), id, reqTrns.Timeout, reqTrns.Exclusive, false) + + h.doTransactionResponse(w, err, trns) +} + +func (h *Handler) handlePostFinishTransaction(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + id := mux.Vars(r)["id"] + trns, err := h.api.FinishTransaction(r.Context(), id, false) + h.doTransactionResponse(w, err, trns) +} + // handleDeleteRemoteAvailableShard handles DELETE /field/{field}/available-shards/{shardID} request. func (h *Handler) handleDeleteRemoteAvailableShard(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { diff --git a/http/handler_test.go b/http/handler_test.go index d275d450e..53c7d2da2 100644 --- a/http/handler_test.go +++ b/http/handler_test.go @@ -15,11 +15,13 @@ package http_test import ( + "encoding/json" "net" "testing" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/http" + "github.com/pilosa/pilosa/v2/test" ) func TestHandlerOptions(t *testing.T) { @@ -40,3 +42,39 @@ func TestHandlerOptions(t *testing.T) { t.Fatalf("expected error making handler without options, got nil") } } + +func TestMarshalUnmarshalTransactionResponse(t *testing.T) { + tests := []struct { + name string + tr *http.TransactionResponse + }{ + { + name: "nil transaction", + tr: &http.TransactionResponse{}, + }, + { + name: "empty transaction", + tr: &http.TransactionResponse{Transaction: &pilosa.Transaction{}}, + }, + } + + for _, tst := range tests { + t.Run(tst.name, func(t *testing.T) { + data, err := json.Marshal(tst.tr) + if err != nil { + t.Fatalf("marshaling: %v", err) + } + + mytr := &http.TransactionResponse{} + err = json.Unmarshal(data, mytr) + if err != nil { + t.Fatalf("unmarshalling: %v", err) + } + + if mytr.Error != tst.tr.Error { + t.Errorf("errors mismatch:exp/got \n%v\n%v", tst.tr.Error, mytr.Error) + } + test.CompareTransactions(t, tst.tr.Transaction, mytr.Transaction) + }) + } +} diff --git a/internal/private.pb.go b/internal/private.pb.go index c78b91ff5..a1d0870bd 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -2187,6 +2187,187 @@ func (m *RecalculateCaches) XXX_DiscardUnknown() { var xxx_messageInfo_RecalculateCaches proto.InternalMessageInfo +type TransactionMessage struct { + Action string `protobuf:"bytes,1,opt,name=Action,proto3" json:"Action,omitempty"` + Transaction *Transaction `protobuf:"bytes,2,opt,name=Transaction,proto3" json:"Transaction,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *TransactionMessage) Reset() { *m = TransactionMessage{} } +func (m *TransactionMessage) String() string { return proto.CompactTextString(m) } +func (*TransactionMessage) ProtoMessage() {} +func (*TransactionMessage) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{35} +} +func (m *TransactionMessage) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *TransactionMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_TransactionMessage.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *TransactionMessage) XXX_Merge(src proto.Message) { + xxx_messageInfo_TransactionMessage.Merge(m, src) +} +func (m *TransactionMessage) XXX_Size() int { + return m.Size() +} +func (m *TransactionMessage) XXX_DiscardUnknown() { + xxx_messageInfo_TransactionMessage.DiscardUnknown(m) +} + +var xxx_messageInfo_TransactionMessage proto.InternalMessageInfo + +func (m *TransactionMessage) GetAction() string { + if m != nil { + return m.Action + } + return "" +} + +func (m *TransactionMessage) GetTransaction() *Transaction { + if m != nil { + return m.Transaction + } + return nil +} + +type Transaction struct { + ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"` + Active bool `protobuf:"varint,2,opt,name=Active,proto3" json:"Active,omitempty"` + Exclusive bool `protobuf:"varint,3,opt,name=Exclusive,proto3" json:"Exclusive,omitempty"` + Timeout int64 `protobuf:"varint,4,opt,name=Timeout,proto3" json:"Timeout,omitempty"` + Deadline int64 `protobuf:"varint,5,opt,name=Deadline,proto3" json:"Deadline,omitempty"` + Stats *TransactionStats `protobuf:"bytes,6,opt,name=Stats,proto3" json:"Stats,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Transaction) Reset() { *m = Transaction{} } +func (m *Transaction) String() string { return proto.CompactTextString(m) } +func (*Transaction) ProtoMessage() {} +func (*Transaction) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{36} +} +func (m *Transaction) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Transaction) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Transaction.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *Transaction) XXX_Merge(src proto.Message) { + xxx_messageInfo_Transaction.Merge(m, src) +} +func (m *Transaction) XXX_Size() int { + return m.Size() +} +func (m *Transaction) XXX_DiscardUnknown() { + xxx_messageInfo_Transaction.DiscardUnknown(m) +} + +var xxx_messageInfo_Transaction proto.InternalMessageInfo + +func (m *Transaction) GetID() string { + if m != nil { + return m.ID + } + return "" +} + +func (m *Transaction) GetActive() bool { + if m != nil { + return m.Active + } + return false +} + +func (m *Transaction) GetExclusive() bool { + if m != nil { + return m.Exclusive + } + return false +} + +func (m *Transaction) GetTimeout() int64 { + if m != nil { + return m.Timeout + } + return 0 +} + +func (m *Transaction) GetDeadline() int64 { + if m != nil { + return m.Deadline + } + return 0 +} + +func (m *Transaction) GetStats() *TransactionStats { + if m != nil { + return m.Stats + } + return nil +} + +type TransactionStats struct { + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *TransactionStats) Reset() { *m = TransactionStats{} } +func (m *TransactionStats) String() string { return proto.CompactTextString(m) } +func (*TransactionStats) ProtoMessage() {} +func (*TransactionStats) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{37} +} +func (m *TransactionStats) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *TransactionStats) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_TransactionStats.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *TransactionStats) XXX_Merge(src proto.Message) { + xxx_messageInfo_TransactionStats.Merge(m, src) +} +func (m *TransactionStats) XXX_Size() int { + return m.Size() +} +func (m *TransactionStats) XXX_DiscardUnknown() { + xxx_messageInfo_TransactionStats.DiscardUnknown(m) +} + +var xxx_messageInfo_TransactionStats proto.InternalMessageInfo + func init() { proto.RegisterType((*IndexMeta)(nil), "internal.IndexMeta") proto.RegisterType((*FieldOptions)(nil), "internal.FieldOptions") @@ -2224,94 +2405,103 @@ func init() { proto.RegisterType((*UpdateCoordinatorMessage)(nil), "internal.UpdateCoordinatorMessage") proto.RegisterType((*Topology)(nil), "internal.Topology") proto.RegisterType((*RecalculateCaches)(nil), "internal.RecalculateCaches") + proto.RegisterType((*TransactionMessage)(nil), "internal.TransactionMessage") + proto.RegisterType((*Transaction)(nil), "internal.Transaction") + proto.RegisterType((*TransactionStats)(nil), "internal.TransactionStats") } func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) } var fileDescriptor_d2a91b51c7bdc125 = []byte{ - // 1298 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, 0x96, 0xa5, 0x96, 0xe5, 0xc8, 0x93, 0xc4, 0xd9, 0xe4, 0x4f, 0x19, 0x31, - 0xa4, 0x88, 0x48, 0x15, 0x26, 0x95, 0x70, 0xe0, 0x2b, 0x55, 0xc1, 0x96, 0x13, 0x44, 0xb0, 0x93, - 0x8c, 0x9c, 0xdc, 0x38, 0x8c, 0x57, 0x53, 0xf1, 0x96, 0x57, 0xbb, 0x62, 0x77, 0xd6, 0x91, 0x73, - 0xe0, 0x0a, 0x55, 0xbc, 0x00, 0x47, 0x1e, 0x87, 0x23, 0x8f, 0x40, 0x85, 0xa7, 0xe0, 0x46, 0x4d, - 0xcf, 0xcc, 0x7e, 0x28, 0x16, 0x0e, 0x0e, 0xb7, 0xe9, 0x5f, 0xf7, 0xf4, 0x77, 0xf7, 0xec, 0x42, - 0x67, 0x9a, 0x04, 0xc7, 0x5c, 0x8a, 0xcd, 0x69, 0x12, 0xcb, 0x98, 0x34, 0x83, 0x48, 0x8a, 0x24, - 0xe2, 0xe1, 0xb5, 0x95, 0x69, 0x76, 0x10, 0x06, 0xbe, 0xc6, 0xe9, 0x03, 0x68, 0x0d, 0xa3, 0xb1, - 0x98, 0xed, 0x0a, 0xc9, 0x09, 0x81, 0xfa, 0x43, 0x71, 0x92, 0x7a, 0x6e, 0xcf, 0xe9, 0x37, 0x19, - 0x9e, 0xc9, 0x07, 0xb0, 0xba, 0x9f, 0x70, 0xff, 0x68, 0x67, 0x16, 0xa4, 0x52, 0x44, 0xbe, 0xf0, - 0xea, 0xc8, 0x9d, 0x43, 0xe9, 0xaf, 0x2e, 0xac, 0xdc, 0x0f, 0x44, 0x38, 0x7e, 0x34, 0x95, 0x41, - 0x1c, 0xa5, 0x4a, 0xd9, 0xfe, 0xc9, 0x54, 0x78, 0xcd, 0x9e, 0xd3, 0x6f, 0x31, 0x3c, 0x93, 0x77, - 0xa0, 0xb5, 0xcd, 0xfd, 0x43, 0x81, 0x0c, 0x17, 0x19, 0x05, 0x90, 0x73, 0x47, 0xc1, 0x4b, 0x6d, - 0xa5, 0xc3, 0x0a, 0x80, 0xf4, 0xa0, 0xbd, 0x1f, 0x4c, 0xc4, 0x93, 0x8c, 0x47, 0x32, 0x9b, 0x78, - 0x4b, 0x78, 0xbb, 0x0c, 0x91, 0x75, 0x68, 0x3c, 0x0a, 0xc7, 0xbb, 0x41, 0xe4, 0xb5, 0x7a, 0x4e, - 0xdf, 0x65, 0x86, 0xb2, 0x38, 0x9f, 0x79, 0x50, 0xe0, 0x7c, 0x96, 0x87, 0xdb, 0xae, 0x86, 0xbb, - 0x17, 0x8f, 0x24, 0x8f, 0xc6, 0x3c, 0x19, 0x3f, 0x0b, 0xc4, 0x0b, 0x6f, 0x45, 0x87, 0x5b, 0x45, - 0xd5, 0xdd, 0x2d, 0x9e, 0x0a, 0xaf, 0x83, 0x1a, 0xf1, 0x4c, 0xae, 0x41, 0x73, 0x2b, 0x90, 0x03, - 0x31, 0x95, 0x87, 0xde, 0x6a, 0xcf, 0xe9, 0xd7, 0x59, 0x4e, 0x93, 0x4b, 0xb0, 0x34, 0xf2, 0x79, - 0x28, 0xbc, 0x0b, 0x78, 0x41, 0x13, 0x84, 0xc2, 0xca, 0xfd, 0x38, 0x11, 0xc1, 0xf3, 0x08, 0x8b, - 0xe0, 0x75, 0x31, 0xa8, 0x0a, 0x46, 0xde, 0x07, 0x57, 0x85, 0xb4, 0xd6, 0x73, 0xfa, 0xed, 0xdb, - 0x6b, 0x9b, 0xb6, 0x8e, 0x9b, 0x03, 0xe1, 0x07, 0x13, 0x1e, 0x32, 0xc5, 0x45, 0x21, 0x3e, 0xf3, - 0xc8, 0x62, 0x21, 0x3e, 0xa3, 0x14, 0x56, 0x87, 0x93, 0x69, 0x9c, 0x48, 0x26, 0xd2, 0x69, 0x1c, - 0xa5, 0x82, 0x74, 0xc1, 0xdd, 0x49, 0x12, 0xcf, 0x41, 0xb3, 0xea, 0x48, 0x7f, 0x80, 0xee, 0x56, - 0x18, 0xfb, 0x47, 0x03, 0x2e, 0x39, 0x13, 0xdf, 0x67, 0x22, 0x95, 0xca, 0x77, 0xed, 0x9e, 0x96, - 0xd3, 0x84, 0x42, 0xb1, 0xde, 0x5e, 0x4d, 0xa3, 0x48, 0xa8, 0xbc, 0x60, 0xd6, 0x74, 0x79, 0xf0, - 0x8c, 0xb1, 0x1f, 0xf2, 0x64, 0x8c, 0x35, 0xad, 0x33, 0x4d, 0x28, 0x14, 0x2d, 0x61, 0x1f, 0xd4, - 0x99, 0x26, 0xe8, 0x10, 0xd6, 0x4a, 0xf6, 0x8d, 0x9b, 0xeb, 0xd0, 0x60, 0xf1, 0x8b, 0xe1, 0x20, - 0xf5, 0x9c, 0x9e, 0xdb, 0xaf, 0x33, 0x43, 0x61, 0xc3, 0xc4, 0x61, 0x36, 0x89, 0x14, 0xab, 0x86, - 0xac, 0x02, 0xa0, 0x57, 0x61, 0x09, 0xbb, 0x47, 0x45, 0x59, 0xdc, 0x55, 0x47, 0xfa, 0xa3, 0x03, - 0xad, 0x5d, 0x3e, 0x43, 0x47, 0x52, 0x72, 0x17, 0x9a, 0xb6, 0xb6, 0x28, 0xd4, 0xbe, 0xfd, 0x5e, - 0x91, 0xc1, 0x5c, 0x6c, 0xd3, 0xca, 0xec, 0x44, 0x32, 0x39, 0x61, 0xf9, 0x95, 0x6b, 0x5f, 0x40, - 0xa7, 0xc2, 0x52, 0xf6, 0x8e, 0xc4, 0x89, 0xcd, 0xea, 0x91, 0x38, 0x51, 0xb1, 0x1e, 0xf3, 0x30, - 0x13, 0x98, 0xab, 0x3a, 0xd3, 0xc4, 0xe7, 0xb5, 0x4f, 0x1d, 0xfa, 0x0c, 0xc8, 0x76, 0x22, 0xb8, - 0x14, 0x68, 0x64, 0x57, 0xa4, 0x29, 0x7f, 0x2e, 0xce, 0xca, 0xb8, 0x5b, 0xce, 0x78, 0x9e, 0xdd, - 0x5a, 0x29, 0xbb, 0xf4, 0x26, 0x90, 0x81, 0x08, 0x85, 0x14, 0x66, 0xba, 0xff, 0x41, 0x2f, 0x1d, - 0x59, 0x1f, 0xce, 0x96, 0x25, 0x37, 0xa0, 0xae, 0x56, 0x05, 0x1a, 0x6b, 0xdf, 0xbe, 0x58, 0xe4, - 0x29, 0xdf, 0x22, 0x0c, 0x05, 0x68, 0x68, 0x95, 0xa2, 0x97, 0x6f, 0x18, 0x58, 0xa5, 0x95, 0x6e, - 0x1a, 0x53, 0x2e, 0x9a, 0x5a, 0x2f, 0x4c, 0x95, 0xd7, 0x8c, 0xb1, 0x76, 0xcf, 0x86, 0x7b, 0x5e, - 0x6b, 0xd4, 0x87, 0xff, 0x6b, 0x0d, 0x5f, 0x1d, 0xf3, 0x20, 0xe4, 0x07, 0xe1, 0xbf, 0xaa, 0x48, - 0xc5, 0x71, 0x0f, 0x96, 0xf1, 0xee, 0x70, 0x60, 0x7a, 0xdb, 0x92, 0xf4, 0x3b, 0x28, 0xc6, 0x64, - 0x8f, 0x4f, 0x84, 0xd1, 0x86, 0xe7, 0x3c, 0xde, 0xda, 0xd9, 0xf1, 0x2a, 0xc3, 0x6a, 0xb4, 0xd4, - 0xaa, 0x76, 0x95, 0x61, 0x24, 0xe8, 0x1d, 0x68, 0x8c, 0xfc, 0x43, 0x31, 0xe1, 0xe4, 0x43, 0x58, - 0x46, 0x0f, 0x45, 0x6a, 0x3a, 0xfa, 0xc2, 0x5c, 0xa5, 0x98, 0xe5, 0xd3, 0xd4, 0x44, 0x76, 0xaa, - 0x4f, 0x1f, 0xc1, 0xb2, 0x31, 0x8c, 0x13, 0xbd, 0xa0, 0xe2, 0x56, 0x86, 0xdc, 0x80, 0x06, 0x3a, - 0x9b, 0x7a, 0xf5, 0x79, 0xab, 0x88, 0x33, 0xc3, 0xa6, 0x3b, 0xe0, 0x3e, 0x65, 0x43, 0x35, 0xd8, - 0xe8, 0xb0, 0x35, 0x6a, 0x28, 0xe5, 0xca, 0xd7, 0x71, 0x2a, 0x4d, 0x5a, 0xf1, 0xac, 0xb0, 0xc7, - 0x71, 0x22, 0x31, 0xa5, 0x1d, 0x86, 0x67, 0x9a, 0x42, 0x7d, 0x2f, 0x1e, 0x0b, 0xb2, 0x0a, 0xb5, - 0xe1, 0xc0, 0xe8, 0xa8, 0x0d, 0x07, 0xe4, 0x5d, 0x54, 0x6f, 0x32, 0xd9, 0x29, 0x9c, 0x78, 0xca, - 0x86, 0x0c, 0x0d, 0x5f, 0x87, 0xce, 0x30, 0xdd, 0x8e, 0xe3, 0x64, 0x1c, 0x44, 0x5c, 0xc6, 0x89, - 0x79, 0xf2, 0xaa, 0x20, 0x8e, 0x96, 0xe4, 0x52, 0x3f, 0x46, 0x2d, 0xa6, 0x09, 0x7a, 0x0f, 0xba, - 0xca, 0x28, 0x12, 0xb6, 0x3d, 0xd6, 0xa1, 0xa1, 0xb0, 0xdc, 0x09, 0x43, 0x15, 0x1a, 0x6a, 0x65, - 0x0d, 0xdf, 0x6a, 0x0d, 0x3b, 0xc7, 0x22, 0x92, 0xa5, 0x06, 0x43, 0x1a, 0x15, 0x74, 0x98, 0x26, - 0x08, 0xd5, 0x01, 0x9a, 0x48, 0x56, 0x8b, 0x48, 0x14, 0xca, 0x90, 0x47, 0x7f, 0x76, 0x00, 0xac, - 0x43, 0x59, 0x9a, 0x5f, 0x71, 0x16, 0x5f, 0x21, 0x7d, 0xdb, 0x28, 0x66, 0xb8, 0xba, 0x85, 0x94, - 0xc6, 0x99, 0x6d, 0xa4, 0x8f, 0x8b, 0x46, 0xd2, 0x25, 0xbd, 0x3c, 0xd7, 0x00, 0xda, 0x6a, 0xd1, - 0x4e, 0x8f, 0xa1, 0x5d, 0xc2, 0x17, 0x34, 0x95, 0xed, 0x92, 0xda, 0xbc, 0x4a, 0xc4, 0x8d, 0x4a, - 0xdb, 0x2b, 0x0f, 0xa1, 0x5d, 0x82, 0x4f, 0xd5, 0xd8, 0x87, 0x0b, 0xd5, 0xb1, 0xb5, 0xcf, 0xc1, - 0x3c, 0x4c, 0x03, 0xe8, 0x6c, 0x87, 0x59, 0x2a, 0x45, 0x62, 0xd4, 0xa9, 0x37, 0x44, 0x03, 0x79, - 0xf1, 0x0a, 0xe0, 0xf4, 0xfa, 0x91, 0xeb, 0xb0, 0xa4, 0xd2, 0xa8, 0xa7, 0xef, 0xf5, 0x1c, 0x6b, - 0x26, 0x7d, 0x06, 0xcd, 0xad, 0xd1, 0xf0, 0x41, 0x12, 0x67, 0xd3, 0x53, 0x9d, 0xb6, 0x1f, 0x48, - 0xb5, 0xd2, 0x07, 0x52, 0x57, 0x3f, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0xcb, 0xde, 0xd5, 0x2f, 0x7b, - 0xdd, 0x20, 0x5c, 0xad, 0xeb, 0x35, 0xbd, 0x59, 0xd5, 0xd0, 0x9f, 0x67, 0x3f, 0xd9, 0x37, 0xda, - 0x2d, 0xde, 0x68, 0xa5, 0x54, 0xaf, 0xbf, 0xff, 0x52, 0xe9, 0x5f, 0x35, 0x58, 0x63, 0x22, 0x0d, - 0x5e, 0x8a, 0x61, 0x94, 0xca, 0x24, 0xf3, 0xd5, 0x96, 0x50, 0xf7, 0xbf, 0x89, 0x0f, 0x4c, 0xb6, - 0x5d, 0xa6, 0x89, 0x37, 0xe9, 0x74, 0x72, 0x0b, 0xda, 0xf3, 0x33, 0xfb, 0xba, 0x68, 0x59, 0x84, - 0xdc, 0x82, 0xe5, 0x51, 0x9c, 0x25, 0x7e, 0xde, 0xbe, 0xa5, 0xb5, 0xaa, 0x3d, 0xd3, 0x6c, 0x66, - 0xc5, 0xc8, 0x13, 0x20, 0xfb, 0x09, 0x8f, 0xd2, 0x90, 0x2b, 0x67, 0xed, 0xe5, 0xe6, 0xfc, 0x67, - 0x41, 0x49, 0xa6, 0xa2, 0xe7, 0x94, 0xcb, 0xe4, 0x93, 0xf2, 0x7c, 0x7a, 0xcb, 0xe8, 0xf5, 0xa5, - 0xaa, 0xd7, 0xa6, 0xe5, 0xcb, 0x73, 0x7c, 0x77, 0xae, 0x53, 0xbd, 0x06, 0x5e, 0xbc, 0x52, 0x5c, - 0xac, 0xb0, 0x59, 0x55, 0x9a, 0xfe, 0xe4, 0xc0, 0x4a, 0xd9, 0xb3, 0x37, 0xda, 0x0b, 0x79, 0xc1, - 0x6b, 0x67, 0x7f, 0x77, 0xd8, 0x82, 0xd7, 0x4f, 0xfb, 0xd2, 0x5b, 0x2a, 0x7f, 0x8b, 0x64, 0x70, - 0x65, 0x41, 0xba, 0xde, 0xc2, 0xa9, 0x1e, 0xb4, 0x1f, 0xf3, 0x44, 0x06, 0x4a, 0xa5, 0x79, 0x68, - 0x97, 0x58, 0x19, 0xa2, 0x47, 0x70, 0xf5, 0xb5, 0xe6, 0xdb, 0x8e, 0x27, 0x53, 0xd5, 0xe5, 0x6f, - 0xd1, 0x84, 0x6a, 0x51, 0x27, 0x89, 0x69, 0xbf, 0x16, 0xd3, 0x04, 0xfd, 0x0c, 0x2e, 0x8f, 0x84, - 0x2c, 0xb5, 0x9e, 0x9d, 0xa1, 0x1e, 0xb8, 0x7b, 0xe2, 0xc5, 0x82, 0x00, 0x15, 0x8b, 0x7e, 0x09, - 0xde, 0xd3, 0xe9, 0x98, 0x4b, 0x71, 0xae, 0xdb, 0x5b, 0xd0, 0xdc, 0x8f, 0xa7, 0x71, 0x18, 0x3f, - 0x3f, 0x39, 0x63, 0x97, 0x79, 0xb0, 0xac, 0x5f, 0x25, 0xbd, 0x1c, 0x5b, 0xcc, 0x92, 0xf4, 0xa2, - 0x1a, 0x53, 0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0xba, 0xd5, 0xfd, 0xed, 0xd5, - 0x86, 0xf3, 0xfb, 0xab, 0x0d, 0xe7, 0x8f, 0x57, 0x1b, 0xce, 0x2f, 0x7f, 0x6e, 0xfc, 0xef, 0xa0, - 0x81, 0xbf, 0x8c, 0x77, 0xfe, 0x0e, 0x00, 0x00, 0xff, 0xff, 0xa0, 0xbb, 0xce, 0xd0, 0x5b, 0x0e, - 0x00, 0x00, + // 1395 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x17, 0xcd, 0x72, 0xdb, 0x44, + 0x18, 0x59, 0x8e, 0x63, 0x7f, 0x8e, 0x53, 0x67, 0xdb, 0xa6, 0x6a, 0x60, 0x82, 0x59, 0x3a, 0xd4, + 0x74, 0x86, 0xd0, 0x69, 0x99, 0xe1, 0xb7, 0x33, 0x6d, 0xe2, 0xb4, 0x98, 0x92, 0xb4, 0x5d, 0xa7, + 0xbd, 0x71, 0xd8, 0xc8, 0x3b, 0x8d, 0x26, 0xb2, 0x64, 0xa4, 0x55, 0xea, 0xf4, 0xc0, 0x15, 0x66, + 0x78, 0x01, 0x8e, 0xbc, 0x07, 0x2f, 0xc0, 0x91, 0x47, 0x60, 0xca, 0x53, 0x70, 0x63, 0xf6, 0xdb, + 0x5d, 0x49, 0x76, 0x1c, 0x52, 0x52, 0x6e, 0xfb, 0xfd, 0xff, 0x7f, 0x9f, 0x04, 0xad, 0x71, 0x12, + 0x1c, 0x71, 0x29, 0x36, 0xc6, 0x49, 0x2c, 0x63, 0x52, 0x0f, 0x22, 0x29, 0x92, 0x88, 0x87, 0x6b, + 0x4b, 0xe3, 0x6c, 0x3f, 0x0c, 0x7c, 0x8d, 0xa7, 0x0f, 0xa0, 0xd1, 0x8f, 0x86, 0x62, 0xb2, 0x23, + 0x24, 0x27, 0x04, 0xaa, 0x0f, 0xc5, 0x71, 0xea, 0xb9, 0x1d, 0xa7, 0x5b, 0x67, 0xf8, 0x26, 0x1f, + 0xc0, 0xf2, 0x5e, 0xc2, 0xfd, 0xc3, 0xed, 0x49, 0x90, 0x4a, 0x11, 0xf9, 0xc2, 0xab, 0x22, 0x75, + 0x06, 0x4b, 0x7f, 0x75, 0x61, 0xe9, 0x7e, 0x20, 0xc2, 0xe1, 0xa3, 0xb1, 0x0c, 0xe2, 0x28, 0x55, + 0xca, 0xf6, 0x8e, 0xc7, 0xc2, 0xab, 0x77, 0x9c, 0x6e, 0x83, 0xe1, 0x9b, 0xbc, 0x03, 0x8d, 0x2d, + 0xee, 0x1f, 0x08, 0x24, 0xb8, 0x48, 0x28, 0x10, 0x39, 0x75, 0x10, 0xbc, 0xd4, 0x56, 0x5a, 0xac, + 0x40, 0x90, 0x0e, 0x34, 0xf7, 0x82, 0x91, 0x78, 0x92, 0xf1, 0x48, 0x66, 0x23, 0x6f, 0x01, 0xa5, + 0xcb, 0x28, 0xb2, 0x0a, 0xb5, 0x47, 0xe1, 0x70, 0x27, 0x88, 0xbc, 0x46, 0xc7, 0xe9, 0xba, 0xcc, + 0x40, 0x16, 0xcf, 0x27, 0x1e, 0x14, 0x78, 0x3e, 0xc9, 0xc3, 0x6d, 0x4e, 0x87, 0xbb, 0x1b, 0x0f, + 0x24, 0x8f, 0x86, 0x3c, 0x19, 0x3e, 0x0b, 0xc4, 0x0b, 0x6f, 0x49, 0x87, 0x3b, 0x8d, 0x55, 0xb2, + 0x9b, 0x3c, 0x15, 0x5e, 0x0b, 0x35, 0xe2, 0x9b, 0xac, 0x41, 0x7d, 0x33, 0x90, 0x3d, 0x31, 0x96, + 0x07, 0xde, 0x72, 0xc7, 0xe9, 0x56, 0x59, 0x0e, 0x93, 0x4b, 0xb0, 0x30, 0xf0, 0x79, 0x28, 0xbc, + 0x0b, 0x28, 0xa0, 0x01, 0x42, 0x61, 0xe9, 0x7e, 0x9c, 0x88, 0xe0, 0x79, 0x84, 0x45, 0xf0, 0xda, + 0x18, 0xd4, 0x14, 0x8e, 0xbc, 0x0f, 0xae, 0x0a, 0x69, 0xa5, 0xe3, 0x74, 0x9b, 0xb7, 0x56, 0x36, + 0x6c, 0x1d, 0x37, 0x7a, 0xc2, 0x0f, 0x46, 0x3c, 0x64, 0x8a, 0x8a, 0x4c, 0x7c, 0xe2, 0x91, 0xd3, + 0x99, 0xf8, 0x84, 0x52, 0x58, 0xee, 0x8f, 0xc6, 0x71, 0x22, 0x99, 0x48, 0xc7, 0x71, 0x94, 0x0a, + 0xd2, 0x06, 0x77, 0x3b, 0x49, 0x3c, 0x07, 0xcd, 0xaa, 0x27, 0xfd, 0x01, 0xda, 0x9b, 0x61, 0xec, + 0x1f, 0xf6, 0xb8, 0xe4, 0x4c, 0x7c, 0x9f, 0x89, 0x54, 0x2a, 0xdf, 0xb5, 0x7b, 0x9a, 0x4f, 0x03, + 0x0a, 0x8b, 0xf5, 0xf6, 0x2a, 0x1a, 0x8b, 0x80, 0xca, 0x0b, 0x66, 0x4d, 0x97, 0x07, 0xdf, 0x18, + 0xfb, 0x01, 0x4f, 0x86, 0x58, 0xd3, 0x2a, 0xd3, 0x80, 0xc2, 0xa2, 0x25, 0xec, 0x83, 0x2a, 0xd3, + 0x00, 0xed, 0xc3, 0x4a, 0xc9, 0xbe, 0x71, 0x73, 0x15, 0x6a, 0x2c, 0x7e, 0xd1, 0xef, 0xa5, 0x9e, + 0xd3, 0x71, 0xbb, 0x55, 0x66, 0x20, 0x6c, 0x98, 0x38, 0xcc, 0x46, 0x91, 0x22, 0x55, 0x90, 0x54, + 0x20, 0xe8, 0x55, 0x58, 0xc0, 0xee, 0x51, 0x51, 0x16, 0xb2, 0xea, 0x49, 0x7f, 0x74, 0xa0, 0xb1, + 0xc3, 0x27, 0xe8, 0x48, 0x4a, 0xee, 0x40, 0xdd, 0xd6, 0x16, 0x99, 0x9a, 0xb7, 0xde, 0x2b, 0x32, + 0x98, 0xb3, 0x6d, 0x58, 0x9e, 0xed, 0x48, 0x26, 0xc7, 0x2c, 0x17, 0x59, 0xfb, 0x12, 0x5a, 0x53, + 0x24, 0x65, 0xef, 0x50, 0x1c, 0xdb, 0xac, 0x1e, 0x8a, 0x63, 0x15, 0xeb, 0x11, 0x0f, 0x33, 0x81, + 0xb9, 0xaa, 0x32, 0x0d, 0x7c, 0x51, 0xf9, 0xcc, 0xa1, 0xcf, 0x80, 0x6c, 0x25, 0x82, 0x4b, 0x81, + 0x46, 0x76, 0x44, 0x9a, 0xf2, 0xe7, 0xe2, 0xac, 0x8c, 0xbb, 0xe5, 0x8c, 0xe7, 0xd9, 0xad, 0x94, + 0xb2, 0x4b, 0x6f, 0x00, 0xe9, 0x89, 0x50, 0x48, 0x61, 0xa6, 0xfb, 0x5f, 0xf4, 0xd2, 0x81, 0xf5, + 0xe1, 0x6c, 0x5e, 0x72, 0x1d, 0xaa, 0x6a, 0x55, 0xa0, 0xb1, 0xe6, 0xad, 0x8b, 0x45, 0x9e, 0xf2, + 0x2d, 0xc2, 0x90, 0x81, 0x86, 0x56, 0x29, 0x7a, 0xf9, 0x9a, 0x81, 0x4d, 0xb5, 0xd2, 0x0d, 0x63, + 0xca, 0x45, 0x53, 0xab, 0x85, 0xa9, 0xf2, 0x9a, 0x31, 0xd6, 0xee, 0xda, 0x70, 0xcf, 0x6b, 0x8d, + 0xfa, 0xf0, 0xb6, 0xd6, 0x70, 0xef, 0x88, 0x07, 0x21, 0xdf, 0x0f, 0xff, 0x53, 0x45, 0xa6, 0x1c, + 0xf7, 0x60, 0x11, 0x65, 0xfb, 0x3d, 0xd3, 0xdb, 0x16, 0xa4, 0xdf, 0x41, 0x31, 0x26, 0xbb, 0x7c, + 0x24, 0x8c, 0x36, 0x7c, 0xe7, 0xf1, 0x56, 0xce, 0x8e, 0x57, 0x19, 0x56, 0xa3, 0xa5, 0x56, 0xb5, + 0xab, 0x0c, 0x23, 0x40, 0x6f, 0x43, 0x6d, 0xe0, 0x1f, 0x88, 0x11, 0x27, 0x1f, 0xc2, 0x22, 0x7a, + 0x28, 0x52, 0xd3, 0xd1, 0x17, 0x66, 0x2a, 0xc5, 0x2c, 0x9d, 0xa6, 0x26, 0xb2, 0xb9, 0x3e, 0x7d, + 0x04, 0x8b, 0xc6, 0x30, 0x4e, 0xf4, 0x29, 0x15, 0xb7, 0x3c, 0xe4, 0x3a, 0xd4, 0xd0, 0xd9, 0xd4, + 0xab, 0xce, 0x5a, 0x45, 0x3c, 0x33, 0x64, 0xba, 0x0d, 0xee, 0x53, 0xd6, 0x57, 0x83, 0x8d, 0x0e, + 0x5b, 0xa3, 0x06, 0x52, 0xae, 0x7c, 0x1d, 0xa7, 0xd2, 0xa4, 0x15, 0xdf, 0x0a, 0xf7, 0x38, 0x4e, + 0x24, 0xa6, 0xb4, 0xc5, 0xf0, 0x4d, 0x53, 0xa8, 0xee, 0xc6, 0x43, 0x41, 0x96, 0xa1, 0xd2, 0xef, + 0x19, 0x1d, 0x95, 0x7e, 0x8f, 0xbc, 0x8b, 0xea, 0x4d, 0x26, 0x5b, 0x85, 0x13, 0x4f, 0x59, 0x9f, + 0xa1, 0xe1, 0x6b, 0xd0, 0xea, 0xa7, 0x5b, 0x71, 0x9c, 0x0c, 0x83, 0x88, 0xcb, 0x38, 0x31, 0x27, + 0x6f, 0x1a, 0x89, 0xa3, 0x25, 0xb9, 0xd4, 0xc7, 0xa8, 0xc1, 0x34, 0x40, 0xef, 0x42, 0x5b, 0x19, + 0x45, 0xc0, 0xb6, 0xc7, 0x2a, 0xd4, 0x14, 0x2e, 0x77, 0xc2, 0x40, 0x85, 0x86, 0x4a, 0x59, 0xc3, + 0xb7, 0x5a, 0xc3, 0xf6, 0x91, 0x88, 0x64, 0xa9, 0xc1, 0x10, 0x46, 0x05, 0x2d, 0xa6, 0x01, 0x42, + 0x75, 0x80, 0x26, 0x92, 0xe5, 0x22, 0x12, 0x85, 0x65, 0x48, 0xa3, 0x3f, 0x3b, 0x00, 0xd6, 0xa1, + 0x2c, 0xcd, 0x45, 0x9c, 0xd3, 0x45, 0x48, 0xd7, 0x36, 0x8a, 0x19, 0xae, 0x76, 0xc1, 0xa5, 0xf1, + 0xcc, 0x36, 0xd2, 0xc7, 0x45, 0x23, 0xe9, 0x92, 0x5e, 0x9e, 0x69, 0x00, 0x6d, 0xb5, 0x68, 0xa7, + 0xc7, 0xd0, 0x2c, 0xe1, 0x4f, 0x69, 0x2a, 0xdb, 0x25, 0x95, 0x59, 0x95, 0x88, 0x37, 0x2a, 0x6d, + 0xaf, 0x3c, 0x84, 0x66, 0x09, 0x3d, 0x57, 0x63, 0x17, 0x2e, 0x4c, 0x8f, 0xad, 0x3d, 0x07, 0xb3, + 0x68, 0x1a, 0x40, 0x6b, 0x2b, 0xcc, 0x52, 0x29, 0x12, 0xa3, 0x4e, 0xdd, 0x10, 0x8d, 0xc8, 0x8b, + 0x57, 0x20, 0xe6, 0xd7, 0x8f, 0x5c, 0x83, 0x05, 0x95, 0x46, 0x3d, 0x7d, 0x27, 0x73, 0xac, 0x89, + 0xf4, 0x19, 0xd4, 0x37, 0x07, 0xfd, 0x07, 0x49, 0x9c, 0x8d, 0xe7, 0x3a, 0x6d, 0x3f, 0x90, 0x2a, + 0xa5, 0x0f, 0xa4, 0xb6, 0x3e, 0xf6, 0x2e, 0x7e, 0x24, 0xe0, 0x65, 0x6f, 0xeb, 0xcb, 0x5e, 0x35, + 0x18, 0xae, 0xd6, 0xf5, 0x8a, 0xde, 0xac, 0x6a, 0xe8, 0xcf, 0xb3, 0x9f, 0xec, 0x8d, 0x76, 0x8b, + 0x1b, 0xad, 0x94, 0xea, 0xf5, 0xf7, 0x7f, 0x2a, 0xfd, 0xbb, 0x02, 0x2b, 0x4c, 0xa4, 0xc1, 0x4b, + 0xd1, 0x8f, 0x52, 0x99, 0x64, 0xbe, 0xda, 0x12, 0x4a, 0xfe, 0x9b, 0x78, 0xdf, 0x64, 0xdb, 0x65, + 0x1a, 0x78, 0x9d, 0x4e, 0x27, 0x37, 0xa1, 0x39, 0x3b, 0xb3, 0x27, 0x59, 0xcb, 0x2c, 0xe4, 0x26, + 0x2c, 0x0e, 0xe2, 0x2c, 0xf1, 0xf3, 0xf6, 0x2d, 0xad, 0x55, 0xed, 0x99, 0x26, 0x33, 0xcb, 0x46, + 0x9e, 0x00, 0xd9, 0x4b, 0x78, 0x94, 0x86, 0x5c, 0x39, 0x6b, 0x85, 0xeb, 0xb3, 0x9f, 0x05, 0x25, + 0x9e, 0x29, 0x3d, 0x73, 0x84, 0xc9, 0x27, 0xe5, 0xf9, 0xf4, 0x16, 0xd1, 0xeb, 0x4b, 0xd3, 0x5e, + 0x9b, 0x96, 0x2f, 0xcf, 0xf1, 0x9d, 0x99, 0x4e, 0xf5, 0x6a, 0x28, 0x78, 0xa5, 0x10, 0x9c, 0x22, + 0xb3, 0x69, 0x6e, 0xfa, 0x93, 0x03, 0x4b, 0x65, 0xcf, 0x5e, 0x6b, 0x2f, 0xe4, 0x05, 0xaf, 0x9c, + 0xfd, 0xdd, 0x61, 0x0b, 0x5e, 0x9d, 0xf7, 0xa5, 0xb7, 0x50, 0xfe, 0x16, 0xc9, 0xe0, 0xca, 0x29, + 0xe9, 0x7a, 0x03, 0xa7, 0x3a, 0xd0, 0x7c, 0xcc, 0x13, 0x19, 0x28, 0x95, 0xe6, 0xd0, 0x2e, 0xb0, + 0x32, 0x8a, 0x1e, 0xc2, 0xd5, 0x13, 0xcd, 0xb7, 0x15, 0x8f, 0xc6, 0xaa, 0xcb, 0xdf, 0xa0, 0x09, + 0xd5, 0xa2, 0x4e, 0x12, 0xd3, 0x7e, 0x0d, 0xa6, 0x01, 0xfa, 0x39, 0x5c, 0x1e, 0x08, 0x59, 0x6a, + 0x3d, 0x3b, 0x43, 0x1d, 0x70, 0x77, 0xc5, 0x8b, 0x53, 0x02, 0x54, 0x24, 0xfa, 0x15, 0x78, 0x4f, + 0xc7, 0x43, 0x2e, 0xc5, 0xb9, 0xa4, 0x37, 0xa1, 0xbe, 0x17, 0x8f, 0xe3, 0x30, 0x7e, 0x7e, 0x7c, + 0xc6, 0x2e, 0xf3, 0x60, 0x51, 0x5f, 0x25, 0xbd, 0x1c, 0x1b, 0xcc, 0x82, 0xf4, 0xa2, 0x1a, 0x53, + 0x9f, 0x87, 0x7e, 0x16, 0x2a, 0x37, 0xd4, 0x47, 0x73, 0x4a, 0x85, 0x19, 0x04, 0x8e, 0x89, 0x2b, + 0x1d, 0xba, 0x7b, 0x88, 0xb0, 0x87, 0x4e, 0x43, 0xe4, 0x53, 0x68, 0x96, 0xb8, 0x4d, 0x02, 0x2f, + 0xcf, 0xcc, 0x8b, 0x26, 0xb2, 0x32, 0x27, 0xfd, 0xcd, 0x99, 0x92, 0x3c, 0x71, 0xca, 0x8d, 0xc1, + 0x23, 0x5d, 0x94, 0x3a, 0x33, 0x90, 0x8a, 0x75, 0x7b, 0xe2, 0x87, 0x59, 0xaa, 0x48, 0xfa, 0x7a, + 0x17, 0x08, 0x15, 0xab, 0xfa, 0x33, 0x8c, 0x33, 0x69, 0x36, 0xa7, 0x05, 0xd5, 0x4f, 0x5a, 0x4f, + 0xf0, 0x61, 0x18, 0x44, 0x02, 0xbb, 0xd4, 0x65, 0x39, 0x4c, 0x6e, 0xea, 0x6d, 0x6f, 0x47, 0x6d, + 0x6d, 0xae, 0xfb, 0xc8, 0xa1, 0x2f, 0x41, 0x4a, 0x09, 0xb4, 0x67, 0x49, 0x9b, 0xed, 0xdf, 0x5f, + 0xad, 0x3b, 0x7f, 0xbc, 0x5a, 0x77, 0xfe, 0x7c, 0xb5, 0xee, 0xfc, 0xf2, 0xd7, 0xfa, 0x5b, 0xfb, + 0x35, 0xfc, 0xd7, 0xbe, 0xfd, 0x4f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x2f, 0x76, 0xf5, 0x59, 0x94, + 0x0f, 0x00, 0x00, } func (m *IndexMeta) Marshal() (dAtA []byte, err error) { @@ -4146,6 +4336,155 @@ func (m *RecalculateCaches) MarshalToSizedBuffer(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *TransactionMessage) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *TransactionMessage) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *TransactionMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + if m.Transaction != nil { + { + size, err := m.Transaction.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x12 + } + if len(m.Action) > 0 { + i -= len(m.Action) + copy(dAtA[i:], m.Action) + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Action))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *Transaction) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *Transaction) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Transaction) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + if m.Stats != nil { + { + size, err := m.Stats.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x32 + } + if m.Deadline != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.Deadline)) + i-- + dAtA[i] = 0x28 + } + if m.Timeout != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.Timeout)) + i-- + dAtA[i] = 0x20 + } + if m.Exclusive { + i-- + if m.Exclusive { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x18 + } + if m.Active { + i-- + if m.Active { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x10 + } + if len(m.ID) > 0 { + i -= len(m.ID) + copy(dAtA[i:], m.ID) + i = encodeVarintPrivate(dAtA, i, uint64(len(m.ID))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *TransactionStats) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *TransactionStats) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *TransactionStats) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + return len(dAtA) - i, nil +} + func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int { offset -= sovPrivate(v) base := offset @@ -4988,6 +5327,70 @@ func (m *RecalculateCaches) Size() (n int) { return n } +func (m *TransactionMessage) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Action) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Transaction != nil { + l = m.Transaction.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + +func (m *Transaction) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.ID) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.Active { + n += 2 + } + if m.Exclusive { + n += 2 + } + if m.Timeout != 0 { + n += 1 + sovPrivate(uint64(m.Timeout)) + } + if m.Deadline != 0 { + n += 1 + sovPrivate(uint64(m.Deadline)) + } + if m.Stats != nil { + l = m.Stats.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + +func (m *TransactionStats) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + func sovPrivate(x uint64) (n int) { return (math_bits.Len64(x|1) + 6) / 7 } @@ -10137,6 +10540,382 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error { } return nil } +func (m *TransactionMessage) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: TransactionMessage: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: TransactionMessage: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Action", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Action = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Transaction", 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 < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Transaction == nil { + m.Transaction = &Transaction{} + } + if err := m.Transaction.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Transaction) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: Transaction: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Transaction: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ID = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Active", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.Active = bool(v != 0) + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Exclusive", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.Exclusive = bool(v != 0) + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Timeout", wireType) + } + m.Timeout = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Timeout |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Deadline", wireType) + } + m.Deadline = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Deadline |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 6: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Stats", 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 < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Stats == nil { + m.Stats = &TransactionStats{} + } + if err := m.Stats.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *TransactionStats) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: TransactionStats: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: TransactionStats: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := skipPrivate(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthPrivate + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} func skipPrivate(dAtA []byte) (n int, err error) { l := len(dAtA) iNdEx := 0 diff --git a/internal/private.proto b/internal/private.proto index 989428756..804c3f2cd 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -207,3 +207,19 @@ message Topology { } message RecalculateCaches {} + +message TransactionMessage { + string Action = 1; + Transaction Transaction = 2; +} + +message Transaction { + string ID = 1; + bool Active = 2; + bool Exclusive = 3; + int64 Timeout = 4; + int64 Deadline = 5; + TransactionStats Stats = 6; +} + +message TransactionStats {} \ No newline at end of file diff --git a/server.go b/server.go index 936e08756..7e0690c30 100644 --- a/server.go +++ b/server.go @@ -28,6 +28,8 @@ import ( "time" "github.com/molecula/ext" + uuid "github.com/satori/go.uuid" + // extensions pulls in some extensions depending on build tags _ "github.com/pilosa/pilosa/v2/extensions" "github.com/pilosa/pilosa/v2/logger" @@ -794,11 +796,42 @@ func (s *Server) receiveMessage(m Message) error { } case *NodeStatus: s.handleRemoteStatus(obj) + case *TransactionMessage: + err := s.handleTransactionMessage(obj) + if err != nil { + return errors.Wrapf(err, "handling transaction message: %v", obj) + } } return nil } +func (s *Server) handleTransactionMessage(tm *TransactionMessage) error { + mtrns := tm.Transaction // message transaction + ctx := context.Background() + switch tm.Action { + case TRANSACTION_START: + _, err := s.StartTransaction(ctx, mtrns.ID, mtrns.Timeout, mtrns.Exclusive, true) + if err != nil { + return errors.Wrap(err, "starting transaction locally") + } + case TRANSACTION_FINISH: + _, err := s.FinishTransaction(ctx, mtrns.ID, true) + if err != nil { + return errors.Wrap(err, "finishing transaction locally") + } + case TRANSACTION_VALIDATE: + trns, err := s.GetTransaction(ctx, mtrns.ID, true) + if err != nil { + return errors.Wrap(err, "getting local transaction to validate") + } + return CompareTransactions(mtrns, trns) + default: + return errors.Errorf("unknown transaction action: '%s'", tm.Action) + } + return nil +} + // SendSync represents an implementation of Broadcaster. func (s *Server) SendSync(m Message) error { var eg errgroup.Group @@ -991,6 +1024,127 @@ func (s *Server) monitorRuntime() { } } +func (srv *Server) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool, remote bool) (*Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return nil, ErrNodeNotCoordinator + } + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return nil, errors.New("got a remote start call to coordinator or single node cluster... shouldn't ever happen") + } + + if remote { + return srv.holder.StartTransaction(ctx, id, timeout, exclusive) + } + + // empty string id should generate an id + if id == "" { + id = uuid.NewV4().String() + } + trns, err := srv.holder.StartTransaction(ctx, id, timeout, exclusive) + if err != nil { + return trns, errors.Wrap(err, "starting transaction") + } + err = srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_START, + Transaction: trns, + }) + if err != nil { + // try to clean up, but ignore errors + _, errLocal := srv.holder.FinishTransaction(ctx, id) + errBroadcast := srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_FINISH, + Transaction: trns, + }, + ) + if errLocal != nil || errBroadcast != nil { + srv.logger.Printf("error(s) while trying to clean up transaction which failed to start, local: %v, broadcast: %v", + errLocal, + errBroadcast, + ) + } + return trns, errors.Wrap(err, "broadcasting transaction start") + } + return trns, nil +} + +func (srv *Server) FinishTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return nil, ErrNodeNotCoordinator + } + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return nil, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen") + } + + if remote { + return srv.holder.FinishTransaction(ctx, id) + } + trns, err := srv.holder.FinishTransaction(ctx, id) + if err != nil { + return trns, errors.Wrap(err, "finishing transaction") + } + err = srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_FINISH, + Transaction: trns, + }, + ) + if err != nil { + srv.logger.Printf("error broadcasting transaction finish: %v", err) + // TODO retry? + } + return trns, nil +} + +func (srv *Server) Transactions(ctx context.Context) (map[string]*Transaction, error) { + node := srv.node() + if !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return nil, ErrNodeNotCoordinator + } + + return srv.holder.Transactions(ctx) +} + +func (srv *Server) GetTransaction(ctx context.Context, id string, remote bool) (*Transaction, error) { + node := srv.node() + if !remote && !node.IsCoordinator && len(srv.cluster.Nodes()) > 1 { + return nil, ErrNodeNotCoordinator + } + + if remote && (node.IsCoordinator || len(srv.cluster.Nodes()) == 1) { + return nil, errors.New("got a remote finish call to coordinator or single node cluster... shouldn't ever happen") + } + + trns, err := srv.holder.GetTransaction(ctx, id) + if err != nil { + return nil, errors.Wrap(err, "getting transaction") + } + + // The way a client would find out that the exclusive transaction + // it requested is active is by polling the GetTransaction + // endpoint. Therefore, returning an active, exclusive + // transaction, from here is what truly makes the transaction + // "live". Before doing so, we want to make sure all nodes + // agree. (in case other nodes have activity on this transaction + // we're not aware of) + if !remote && trns.Exclusive && trns.Active { + err := srv.SendSync( + &TransactionMessage{ + Action: TRANSACTION_VALIDATE, + Transaction: trns, + }, + ) + if err != nil { + return nil, errors.Wrap(err, "contacting remote hosts") + } + return trns, nil + } + return trns, nil +} + // countOpenFiles on operating systems that support lsof. func countOpenFiles() (int, error) { switch runtime.GOOS { diff --git a/server/server.go b/server/server.go index 87fc1c361..478e5cfe9 100644 --- a/server/server.go +++ b/server/server.go @@ -87,6 +87,7 @@ type Command struct { listenURI *pilosa.URI tlsConfig *tls.Config closeTimeout time.Duration + noSleep bool serverOptions []pilosa.ServerOption } @@ -114,6 +115,17 @@ func OptCommandConfig(config *Config) CommandOption { } } +// OptCommandNoSleep disables the 5 second sleep for non-coordinator +// nodes on startup. See https://github.com/molecula/pilosa/issues/266 +// This option should only be used by tests, and expect it to be +// deprecated. +func OptCommandNoSleep() CommandOption { + return func(c *Command) error { + c.noSleep = true + return nil + } +} + // NewCommand returns a new instance of Main. func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command { c := &Command{ @@ -149,7 +161,7 @@ func (m *Command) Start() (err error) { if !m.API.Node().IsCoordinator { // hack to give coordinator a head start // TODO https://github.com/molecula/pilosa/issues/266 - if len(m.Config.Gossip.Seeds) > 0 { + if len(m.Config.Gossip.Seeds) > 0 && !m.noSleep { time.Sleep(5 * time.Second) } } diff --git a/server/server_test.go b/server/server_test.go index 7809ee466..d7cb9452e 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -36,6 +36,7 @@ import ( "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -374,6 +375,133 @@ func TestConcurrentFieldCreation(t *testing.T) { } } +func TestTransactionsAPI(t *testing.T) { + cluster := test.MustRunCluster(t, 3) + defer cluster.Close() + + api0 := cluster[0].API + api1 := cluster[1].API + ctx := context.Background() + //api2 := cluster[2].API + + // can fetch empty transactions + if trnsMap, err := api0.Transactions(ctx); err != nil { + t.Fatalf("getting transactions: %v", err) + } else if len(trnsMap) != 0 { + t.Fatalf("unexpectedly has transactions: %v", trnsMap) + } + + // can't fetch transactions from non-coordinator + if _, err := api1.Transactions(ctx); err != pilosa.ErrNodeNotCoordinator { + t.Errorf("api1 should return ErrNodeNotCoordinator when asked for transactions but got: %v", err) + } + + // can start transaction + if trns, err := api0.StartTransaction(ctx, "a", time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + test.CompareTransactions(t, &pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can retrieve transaction from other nodes with remote=true + if trns, err := api1.GetTransaction(ctx, "a", true); err != nil { + t.Errorf("couldn't fetch transaction from other node with remote=true: %v", err) + } else { + test.CompareTransactions(t, &pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can start transaction with blank id and get uuid back + id := "" + if trns, err := api0.StartTransaction(ctx, id, time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + id = trns.ID + if len(id) != 36 { // UUID + t.Errorf("unexpected generated ID: %s", id) + } + test.CompareTransactions(t, &pilosa.Transaction{ID: id, Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can't finish transaction on non-coordinator + if _, err := api1.FinishTransaction(ctx, id, false); err != pilosa.ErrNodeNotCoordinator { + t.Errorf("unexpected error is not ErrNodeNotCoordinator: %v", err) + } + + // can finish transaction + if _, err := api0.FinishTransaction(ctx, id, false); err != nil { + t.Errorf("couldn't finish transaction: %v", err) + } + + // can finish previous transaction + if _, err := api0.FinishTransaction(ctx, "a", false); err != nil { + t.Errorf("couldn't finish transaction a: %v", err) + } + + // can start exclusive transaction + if te, err := api0.StartTransaction(ctx, "exc", time.Minute, true, false); err != nil { + t.Errorf("couldn't start exclusive transaction: %v", err) + } else if !te.Active { + t.Errorf("expected exclusive transaction to be active: %+v", te) + } + + // can finish exclusive transaction + if _, err := api0.FinishTransaction(ctx, "exc", false); err != nil { + t.Errorf("couldn't finish exclusive transaction: %v", err) + } + + // can start transaction (with same name as previous finished transaction) + if trns, err := api0.StartTransaction(ctx, "a", time.Minute, false, false); err != nil { + t.Errorf("couldn't start transaction: %v", err) + } else { + test.CompareTransactions(t, &pilosa.Transaction{ID: "a", Active: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // can start exclusive transaction and is not immediately active + if te, err := api0.StartTransaction(ctx, "exc", time.Minute, true, false); err != nil { + t.Errorf("couldn't start exclusive transaction: %v", err) + } else if te.Active { + t.Errorf("expected exclusive transaction to be inactive: %+v", te) + } + + // can finish non-exclusive transaction + if _, err := api0.FinishTransaction(ctx, "a", false); err != nil { + t.Errorf("couldn't finish transaction a: %v", err) + } + + // can poll exclusive transaction and is active + var excTrns *pilosa.Transaction + if trns, err := api0.GetTransaction(ctx, "exc", false); err != nil { + t.Errorf("couldn't poll exclusive transaction: %v", err) + } else { + excTrns = &pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)} + test.CompareTransactions(t, excTrns, trns) + } + + // can't start another exclusive transaction + if trns, err := api0.StartTransaction(ctx, "exc2", time.Minute, true, false); errors.Cause(err) != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error: %v", err) + } else { + // returned transaction should be the exclusive one which is blocking this one + test.CompareTransactions(t, excTrns, trns) + } + + // can't keep the second exclusive name but make it nonexclusive and start a transaction + if trns, err := api0.StartTransaction(ctx, "exc2", time.Minute, false, false); errors.Cause(err) != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error: %v", err) + } else { + test.CompareTransactions(t, excTrns, trns) + } + + // transaction is active on other nodes with remote=true + if trns, err := api1.GetTransaction(ctx, "exc", true); err != nil { + t.Errorf("couldn't poll exclusive transaction: %v", err) + } else { + test.CompareTransactions(t, &pilosa.Transaction{ID: "exc", Active: true, Exclusive: true, Timeout: time.Minute, Deadline: time.Now().Add(time.Minute)}, trns) + } + + // LATER, test deadline extension on non-coordinator blocks active, exclusive transaction being returned +} + func TestMain_RecalculateHashes(t *testing.T) { const clusterSize = 5 cluster := test.MustRunCluster(t, clusterSize) @@ -1058,4 +1186,5 @@ Set("h", adec=100.22) } else if !strings.Contains(result.Body, `"count":1`) { t.Fatalf("expected count 1, but got: '%s'", result.Body) } + } diff --git a/test/pilosa.go b/test/pilosa.go index 8179c04e0..6903a2010 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -65,6 +65,7 @@ func newCommand(opts ...server.CommandOption) *Command { // does not fail on 32-bit systems. opts = append([]server.CommandOption{ server.OptCommandCloseTimeout(time.Millisecond * 2), + server.OptCommandNoSleep(), }, opts...) m := &Command{commandOptions: opts} m.Command = server.NewCommand(bytes.NewReader(nil), ioutil.Discard, ioutil.Discard, opts...) @@ -409,6 +410,7 @@ func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) { if err != nil { return nil, errors.Wrap(err, "new cluster") } + if err = cluster.Start(); err != nil { return nil, errors.Wrap(err, "starting cluster") } diff --git a/test/transaction.go b/test/transaction.go new file mode 100644 index 000000000..90addbeae --- /dev/null +++ b/test/transaction.go @@ -0,0 +1,46 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package test + +import ( + "testing" + "time" + + "github.com/pilosa/pilosa/v2" +) + +const deadlineSkew = time.Millisecond * 10 + +// CompareTransactions errors describing how the +// transactions differ (if at all). The deadlines need only be close +// (within deadlineSkew). +func CompareTransactions(t *testing.T, trns1, trns2 *pilosa.Transaction) { + t.Helper() + if err := pilosa.CompareTransactions(trns1, trns2); err != nil { + t.Errorf("%v", err) + } + if trns1 == nil || trns2 == nil { + return + } + + diff := trns1.Deadline.Sub(trns2.Deadline) + + if diff > deadlineSkew || diff < -deadlineSkew { + t.Errorf("Deadlines differ by %v:\n%+v\n%+v", diff, trns1, trns2) + } + if trns1.Stats != trns2.Stats { + t.Errorf("Stats differ:\n%+v\n%+v", trns1, trns2) + } +} diff --git a/transaction.go b/transaction.go new file mode 100644 index 000000000..1e142ccb2 --- /dev/null +++ b/transaction.go @@ -0,0 +1,461 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/pilosa/pilosa/v2/logger" + "github.com/pkg/errors" +) + +// Transaction contains information related to a block of work that +// needs to be tracked and spans multiple API calls. +type Transaction struct { + // ID is an arbitrary string identifier. All transactions must have a unique ID. + ID string `json:"id"` + + // Active notes whether an exclusive transaction is active, or + // still pending (if other active transactions exist). All + // non-exclusive transactions are always active. + Active bool `json:"active"` + + // Exclusive is set to true for transactions which can only become active when no other + // transactions exist. + Exclusive bool `json:"exclusive"` + + // Timeout is the minimum idle time for which this transaction should continue to exist. + Timeout time.Duration `json:"timeout"` + + // Deadline is calculated from Timeout. TODO reset deadline each time there is activity on the transaction. (we can't do this until there is some method of associating a request/call with a transaction) + // time there is activity on the transaction. + Deadline time.Time `json:"deadline"` + + // Stats track statistics for the transaction. Not yet used. + Stats TransactionStats `json:"stats"` +} + +type TransactionStats struct{} + +// TransactionManager enforces the rules for transactions on a single +// node. It is goroutine-safe. It should be created by a call to +// NewTransactionManager where it takes a TransactionStore. If logging +// is desired, Log should be set before an instance of +// TransactionManager is used. +type TransactionManager struct { + mu sync.RWMutex + + Log logger.Logger + + store TransactionStore + + checkingDeadlines bool +} + +// NewTransactionManager creates a new TransactionManager with the +// given store, and starts a deadline-checker in a goroutine. +func NewTransactionManager(store TransactionStore) *TransactionManager { + tm := &TransactionManager{ + Log: logger.NopLogger, + store: store, + checkingDeadlines: true, + } + // start deadline checker in case we've just started up, but there is already state in the store. + go tm.deadlineChecker() + return tm +} + +// Start starts a new transaction with the given parameters. If an +// exclusive transaction is pending or in progress, +// ErrTransactionExclusive is returned. If a transaction with the same +// id already exists, that transaction is returned along with +// ErrTransactionExists. If there is no error, the created transaction +// is returned—this is primarily so that the caller can discover if an +// exclusive transaction has been made immediately active or if they +// need to poll. +func (tm *TransactionManager) Start(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) { + tm.mu.Lock() + defer tm.mu.Unlock() + + trnsMap, err := tm.store.List() + if err != nil { + return nil, errors.Wrap(err, "listing transactions in Start") + } + + // check for an exclusive transaction + for _, trns := range trnsMap { + if trns.Exclusive { + // if someone wants a transaction, and we're not able to + // give it to them, we want to be checking deadlines. + // TODO: it would be nice if we could identify whether + // this trns has expired, and if so, automatically remove + // it and continue without returning ErrTransactionExclusive + // on this iteration of the loop. One way we could do that is + // to call tm.checkDeadlines() here (note that we'd have to + // have an unprotectedCheckDeadlines()), and then after that + // check if trns still exists in tm.store. If not, continue. + tm.startDeadlineChecker() + return trns, ErrTransactionExclusive + } + } + if trns, ok := trnsMap[id]; ok { + return trns, ErrTransactionExists + } + + // set new transaction to active if it is not exclusive or if + // there are no other transactions. + active := !exclusive || (len(trnsMap) == 0) + + // set deadline according to timeout + deadline := time.Now().Add(timeout) + trns := &Transaction{ + ID: id, + Active: active, + Exclusive: exclusive, + Timeout: timeout, + Deadline: deadline, + } + if err = tm.store.Put(trns); err != nil { + return nil, errors.Wrap(err, "adding to store") + } + + // we won't check deadlines unless there's actually a + // transaction pending + if exclusive && !active { + tm.startDeadlineChecker() + } + + return trns, nil +} + +// Finish completes and removes a transaction, returning the completed +// transaction (so that the caller can e.g. view the Stats) +func (tm *TransactionManager) Finish(ctx context.Context, id string) (*Transaction, error) { + tm.mu.Lock() + defer tm.mu.Unlock() + return tm.finish(id) +} + +// finish is the unprotected implementation of Finish +func (tm *TransactionManager) finish(id string) (*Transaction, error) { + trns, err := tm.store.Remove(id) + if err != nil { + return nil, err + } + + // After removing, check to see if we need to activate an exclusive transaction + trnsMap, err := tm.store.List() + if err != nil { + tm.log().Printf("error listing transactions in Finish: %v", err) + return trns, nil + } + + if len(trnsMap) == 1 { + for _, etrans := range trnsMap { + if etrans.Exclusive { + if etrans.Active { // sanity check + panic("we just removed a transaction, and the sole remaining exclusive transaction was already active") + } + etrans.Active = true + etrans.Deadline = time.Now().Add(etrans.Timeout) + if err := tm.store.Put(etrans); err != nil { + tm.log().Printf("activating exclusive transaction after finishing last transaction: %v", err) + return trns, nil + } + } + } + } + return trns, nil +} + +// Get retrieves the transaction with the given ID. Returns ErrTransactionNotFound +// if there isn't one. +func (tm *TransactionManager) Get(ctx context.Context, id string) (*Transaction, error) { + tm.mu.RLock() + defer tm.mu.RUnlock() + + return tm.store.Get(id) +} + +// List returns map of all transactions by their ID. It is a copy and +// so may be retained and modified by the caller. +func (tm *TransactionManager) List(ctx context.Context) (map[string]*Transaction, error) { + tm.mu.RLock() + defer tm.mu.RUnlock() + return tm.store.List() +} + +// ResetDeadline updates the deadline for the transaction with the +// given ID to be equal to the current time plus the transaction's +// timeout. +func (tm *TransactionManager) ResetDeadline(ctx context.Context, id string) (*Transaction, error) { + tm.mu.Lock() + defer tm.mu.Unlock() + trns, err := tm.store.Get(id) + if err != nil { + return nil, errors.Wrap(err, "getting transaction") + } + + trns.Deadline = time.Now().Add(trns.Timeout) + + err = tm.store.Put(trns) + return trns, errors.Wrap(err, "storing transaction with new timeout") +} + +// startDeadlineChecker may only be called while tm.mu is held. +func (tm *TransactionManager) startDeadlineChecker() { + if !tm.checkingDeadlines { + tm.checkingDeadlines = true + go tm.deadlineChecker() + } +} + +// deadlineChecker loops continuously checking for expired +// deadlines. It stops when there are no upcoming deadlines. +func (tm *TransactionManager) deadlineChecker() { + interval := tm.checkDeadlines() + for interval != 0 { + time.Sleep(interval) + interval = tm.checkDeadlines() + } + tm.mu.Lock() + tm.checkingDeadlines = false + tm.mu.Unlock() +} + +// checkDeadlines finishes transactions which are past their +// deadlines. It returns the duration until the next deadline. If +// there are no exclusive transactions, it does nothing and returns 0 +// as a signal to stop checking. +func (tm *TransactionManager) checkDeadlines() time.Duration { + tm.mu.Lock() + defer tm.mu.Unlock() + + trnsMap, err := tm.store.List() + if err != nil { + tm.log().Printf("transaction deadline checker couldn't list transactions: %v", err) + return 0 + } + + hasExclusive := false + for _, trns := range trnsMap { + if trns.Exclusive { + hasExclusive = true + break + } + } + if !hasExclusive { + return 0 // no need to expire things if nothing is waiting + } + + now := time.Now() + // track the time interval to next deadline + nextInterval := time.Duration(0) + for id, trns := range trnsMap { + if !trns.Active { + continue + } + if !now.Before(trns.Deadline) { + trnsF, err := tm.finish(id) + if err != nil { + tm.log().Printf("error finishing expired transaction '%s': %+v: %v", id, trnsF, err) + } else { + tm.log().Printf("cleared expired transaction: %+v", trnsF) + } + } else { + interval := trns.Deadline.Sub(now) + if nextInterval == 0 || interval < nextInterval { + nextInterval = interval + } + } + } + return nextInterval +} + +func (tm *TransactionManager) log() logger.Logger { + if tm.Log != nil { + return tm.Log + } + return logger.NopLogger +} + +// TransactionStore declares the functionality which a store for +// Pilosa transactions must implement. +type TransactionStore interface { + // Put stores a new transaction or replaces an existing transaction with the given one. + Put(trns *Transaction) error + // Get retrieves the transaction at id or returns ErrTransactionNotFound if there isn't one. + Get(id string) (*Transaction, error) + // List returns a map of all transactions by ID. The map must be safe to modify by the caller. + List() (map[string]*Transaction, error) + // Remove deletes the transaction from the store. It must return ErrTransactionNotFound if there isn't one. + Remove(id string) (*Transaction, error) +} + +type OpenTransactionStoreFunc func(path string) (TransactionStore, error) + +func OpenInMemTransactionStore(path string) (TransactionStore, error) { + return NewInMemTransactionStore(), nil +} + +// InMemTransactionStore does not persist transaction data and is only +// useful for testing. +type InMemTransactionStore struct { + mu sync.RWMutex + tmap map[string]*Transaction +} + +func NewInMemTransactionStore() *InMemTransactionStore { + return &InMemTransactionStore{ + tmap: make(map[string]*Transaction), + } +} + +func (s *InMemTransactionStore) Put(trns *Transaction) error { + s.mu.Lock() + defer s.mu.Unlock() + + s.tmap[trns.ID] = trns.Copy() + return nil +} + +func (s *InMemTransactionStore) Get(id string) (*Transaction, error) { + s.mu.RLock() + defer s.mu.RUnlock() + + if trns, ok := s.tmap[id]; ok { + return trns.Copy(), nil + } + return nil, ErrTransactionNotFound +} + +func (s *InMemTransactionStore) List() (map[string]*Transaction, error) { + cp := make(map[string]*Transaction) + for id, trns := range s.tmap { + cp[id] = trns.Copy() + } + return cp, nil +} + +func (s *InMemTransactionStore) Remove(id string) (*Transaction, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if trns, ok := s.tmap[id]; ok { + delete(s.tmap, id) + return trns.Copy(), nil + } + return nil, ErrTransactionNotFound +} + +type Error string + +func (e Error) Error() string { return string(e) } + +const ErrTransactionNotFound = Error("transaction not found") +const ErrTransactionExclusive = Error("there is an exclusive transaction, try later") +const ErrTransactionExists = Error("transaction with the given id already exists") + +func CompareTransactions(t1, t2 *Transaction) error { + if t1 == nil && t2 == nil { + return nil + } + if t1 == nil || t2 == nil { + return errors.Errorf("transactions are not equal: %+v %+v", t1, t2) + } + if t1.ID != t2.ID { + return errors.Errorf("transaction IDs not equal: %+v %+v", t1, t2) + } + if t1.Active != t2.Active { + return errors.Errorf("transaction Actives not equal: %+v %+v", t1, t2) + } + if t1.Exclusive != t2.Exclusive { + return errors.Errorf("transaction Exclusives not equal: %+v %+v", t1, t2) + } + if t1.Timeout != t2.Timeout { + return errors.Errorf("transaction Timeouts not equal: %+v %+v", t1, t2) + } + // don't care about Deadline or Stats + return nil +} + +func (trns *Transaction) UnmarshalJSON(b []byte) error { + tmp := &struct { + ID string `json:"id"` + Active bool `json:"active"` + Exclusive bool `json:"exclusive"` + Timeout interface{} `json:"timeout"` + Deadline string `json:"deadline"` + }{} + err := json.Unmarshal(b, tmp) + if err != nil { + return err + } + trns.ID = tmp.ID + trns.Active = tmp.Active + trns.Exclusive = tmp.Exclusive + switch tm := tmp.Timeout.(type) { + case string: + dur, err := time.ParseDuration(tm) + if err != nil { + return errors.Wrapf(err, "timeout as string must be a valid duration got: '%s'", tm) + } + trns.Timeout = dur + case float64: + // interpret as number of seconds + seconds := int64(tm) + nsec := (tm - float64(seconds)) * 1e9 + trns.Timeout = time.Duration(seconds*1e9 + int64(nsec)) + case nil: + break + default: + return errors.New("timeout must be float64 or string") + } + + if tmp.Deadline != "" { + trns.Deadline, err = time.Parse(time.RFC3339Nano, tmp.Deadline) + } + return errors.Wrap(err, "parsing deadline") +} + +func (trns *Transaction) MarshalJSON() ([]byte, error) { + return json.Marshal(&struct { + ID string `json:"id"` + Active bool `json:"active"` + Exclusive bool `json:"exclusive"` + Timeout string `json:"timeout"` + Deadline string `json:"deadline"` + }{ + ID: trns.ID, + Active: trns.Active, + Exclusive: trns.Exclusive, + Timeout: trns.Timeout.String(), + Deadline: trns.Deadline.In(time.UTC).Format(time.RFC3339Nano), + }) +} + +func (trns *Transaction) Copy() *Transaction { + return &Transaction{ + ID: trns.ID, + Active: trns.Active, + Exclusive: trns.Exclusive, + Timeout: trns.Timeout, + Deadline: trns.Deadline, + Stats: trns.Stats, + } +} diff --git a/transaction.md b/transaction.md new file mode 100644 index 000000000..f8f7f5723 --- /dev/null +++ b/transaction.md @@ -0,0 +1,256 @@ +# Initial Transaction Support + +This is not full-featured transaction support with commit and rollback +for now; this is a placeholder intended to allow us to solve shorter-term +problems. + +The primary purpose of this is to allow an exclusive transaction to +block new ingest activity from starting, while permitting existing ingest +operations to complete, even if a single ingest requires multiple operations. +This allows users with cooperating ingest operations to ensure a stable state +for the data on disk before triggering snapshots or other writes. + +## Overview: What transactions are + +A transaction reflects an ongoing set of related operations that may be +occurring in multiple or distinct messages. There is no support for +rolling back a failed transaction. Transactions can coexist, and there's +nothing controlling simultaneous access to fields. + +However, a transaction can be exclusive. An exclusive transaction cannot +start until other transactions complete, but no non-exclusive transaction +can start while an exclusive transaction is waiting. + +Transactions are holder-wide, not index-specific. Transactions are also +presumably cluster-wide. + +### API Details + +The base transaction endpoints are `/transactions`, for listing or creating +transactions, and `/transaction/[id]`, for listing, creating, finishing, or +cancelling a transaction. + +A POST to `/transactions` attempts to create a transaction, assigning it an +arbitrary ID that is not the ID of any existing transaction. A `GET` from +`/transactions` lists existing transactions. + +A POST to `/transaction/[id]` tries to create a transaction with the given +ID, failing if it can't for any reason, including the reason "this ID is +already in use". A GET from `/transaction/[id]` retrieves information about +the transaction. + +When creating a transaction, you may specify an options object: + + ``` + { + "exclusive": true, // default is false + "timeout": 300 // in seconds, default is 300 + } + ``` + +For an exclusive transaction, you may also specify the optional parameter +"pause-snapshots" as a boolean. A `true` value indicates that the snapshot +queue should be paused once this transaction becomes active. *Note that pausing +the snapshot queue can cause some write operations to block indefinitely.* +If a transaction requests that the snapshot queue be paused, it will not +report itself "active" until the snapshot queue has completed any outstanding +snapshots and paused itself. The full sequence of events, then, is: + +* Stop allowing new transactions to start. +* Wait for transactions to complete. +* Pause snapshot queue. +* Wait for snapshot queue to report that it's successfully paused. +* Transition to active state. + +Exclusive transactions which pause the snapshot queue should not write to +the database; this is used as a way to block activity so backups can be made. + +When requesting information about a transaction, you get back an object: + + ``` + { + "active": true, + "timeout": 300, // timeout time in seconds + "stats": { + "idle": 0, // time in seconds since last activity + "queries": 3, // queries submitted in this transaction + "errors": 0 // errors produced by queries + } + } + ``` + +To mark a transaction as complete, you POST to `/transaction/[id]/finish`, and +get back the same information you'd have gotten from a GET for that transaction. +The finish request may block if any existing queries are running as part of +that transaction, but immediately prevents any new queries from starting for +that transaction. + +Queries can be associated with a transaction by including +`X-Pilosa-Transaction: [id]` in their request headers. A transaction's idle +timer is reset by any query against it, even a query which doesn't write +anything. + +When an exclusive transaction is created, it does not necessarily start out +in the `active` state. It immediately blocks the starting of new non-exclusive +transactions, but does not transition to an `active` state until existing +transactions complete. During this time, a GET to it should return: + + ``` + { + "active": false, + "blocked-by": [ "id" ] + } + ``` + +where blocked-by is a list of the IDs of any transactions blocking the +transition. + +If multiple exclusive transactions are requested, they become active +sequentially in the order the requests came in, and the snapshot queue and +other transactions are not permitted to resume until the exclusive transactions +all complete. + + + +### Implementation Notes + +All requests go through coordinator. + +When creating a new transaction, we'll create it on every node in the +cluster and persist it to disk. + +Only the coordinator will accept requests to start a transaction. + +Timeouts only expire when there has been *no activity* on a transaction for the timeout duration. +Any activity on the transaction may extend the deadline (unimplemented). + +When finishing a transaction, we'll finish it on the coordinator and +then broadcast the finish to the cluster before returning to the +client. + +When getting an exclusive transaction, if the transaction is active, +we'll make sure that all nodes agree before returning it. + + +Coordinator forwards all requests to every other node so they can stay +in sync. If the coordinator doesn't hear back from a node, the request +fails. The coordinator only reaches out to active nodes, so if the +cluster is in DEGRADED, things can still continue. + +If a node is down and comes back up, it needs to synchronize its state +with the coordinator (unimplemented). + +There is a separate TransactionManager and TransactionStore + +The store is just responsible for persisting info about +transactions. The manager handles all the logic (at the node level). +Logic related to cluster and remote vs local node is handled by the +Server. The Holder contains the TransactionManager, and the Server +contains the logic for how to handle external vs intra cluster +requests (remote=true). + +There is intra-cluster messaging for transactions which is handled +with the new TransactionMessage and goes through the usual +SendMessage/Broadcaster stuff. + +There is also external API which is handled by the HTTP handler and +goes through API (and is passed directly to Server). (unimplemented) + + +#### TODO + +- [x] implement api layer and cluster logic, startup, etc. +- [ ] add new cluster state to explicitly reject certain requests during exclusive transaction? +- [x] implement HTTP layer +- [ ] implement transaction id in header +- [x] propagate context +- [ ] implement and use persistent transaction store rather than inmem. + - [ ] implement some method for syncing transaction stores +- [ ] update go-pilosa/gpexp to actually USE transactions + - [ ] update IDK to use updated go-pilosa +- [ ] external testing with e.g. curl +- [ ] validate incoming transaction IDs + +- ID validation. No slashes, no non-URL safe chars + +#### Testing TransactionManager +- there should never be more than one Exclusive transaction +- if the Exclusive transaction is active, there should be no other transactions + + +### Documentation + +Before performing a backup, you must request an exclusive "transaction" with the cluster. Do this via and HTTP POST to the coordinator node at path: + +`/transaction` OR `/transaction/{id}` if you wish to specify a custom ID (any alphanum+dash). Otherwise a UUID will be generated and returned in the response. + +Use headers: + +``` +Accept: application/json +Content-Type: application/json +``` + +And body like: + +``` +{ + "timeout": "10m", + "exclusive": true +} +``` + +You MUST specify a timeout. You may choose any timeout you like, +though it's better to err on the longer side of how long you expect +the backup to take. You explicitly finish the transaction once you're +done, so the timeout exists solely for cleanup in the case of +failures. + +This will return a JSON "transaction response" object. +``` +{ +"transaction": { + "id":"5e572d95-4204-40cd-804c-92976b68dc9b", + "active":true, + "exclusive":false, + "timeout":"1m0s", + "deadline":"2020-04-17T21:54:18.69359-05:00" + }, +"error":"some message" +} +``` + +The `error` field MAY not be present if there is no error. + +You MUST check whether `active` is true. If not, you must poll the transaction endpoint with a GET request and your ID until it is true. This looks like: + +GET `/transaction/5e572d95-4204-40cd-804c-92976b68dc9b` + +with headers: + +``` +Accept: application/json +``` + +and also returns a "transaction response" object. + +Once an "active", "exclusive" transaction is returned, proceed with your backup. + +Once the backup is complete, finish the transaction with + +POST `/transaction/{id}/finish` + +with headers: + +``` +Accept: application/json +``` + +Finishing the transaction removes it from the transaction store +completely. A 200 response indicates that this was completed +successfully. The "finish" request will also return a Transaction +response object which contains the transaction as it looked at the +time of its removal. Notably, if the transaction was active, it will +contain `active: true` though it does not exist any more and cannot be +used. + diff --git a/transaction_test.go b/transaction_test.go new file mode 100644 index 000000000..171d1bbab --- /dev/null +++ b/transaction_test.go @@ -0,0 +1,328 @@ +// Copyright 2020 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pilosa_test + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/test" +) + +// TestTransactionManager currently uses an in memory transaction +// store, but tests a variety of timeouts, and therefore could be +// sensitive to slowness in the implementation. Especially if a store +// were used that actually wrote things to disk. +func TestTransactionManager(t *testing.T) { + store := pilosa.NewInMemTransactionStore() + + tm := pilosa.NewTransactionManager(store) + tm.Log = test.NewBufferLogger() + ctx := context.Background() + + // can add a non-exclusive transaction + trns1 := mustStart(t, tm, "a", time.Microsecond, false) + test.CompareTransactions(t, &pilosa.Transaction{ID: "a", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns1) + + // can have two non exclusive transactions + trns2 := mustStart(t, tm, "b", time.Microsecond, false) + test.CompareTransactions(t, &pilosa.Transaction{ID: "b", Active: true, Timeout: time.Microsecond, Deadline: time.Now()}, trns2) + + // trying to start a transaction with same name errors and returns previous transaction + t3, err := tm.Start(ctx, "a", time.Second, true) + if err != pilosa.ErrTransactionExists { + t.Errorf("expected transaction exists, but got: '%v'", err) + } + test.CompareTransactions(t, trns1, t3) + + // can get an existing transaction + trns2_2 := mustGet(t, tm, "b") + test.CompareTransactions(t, trns2, trns2_2) + + // can list all transactions + trnsMap := mustList(t, tm) + if len(trnsMap) != 2 { + t.Errorf("unexpected number of transactions in map: %d", len(trnsMap)) + } + test.CompareTransactions(t, trnsMap["a"], trns1) + test.CompareTransactions(t, trnsMap["b"], trns2) + + // can submit an exclusive transaction + trnsE := mustStart(t, tm, "ce", time.Millisecond*5, true) + test.CompareTransactions(t, &pilosa.Transaction{ID: "ce", Active: false, Exclusive: true, Timeout: time.Millisecond * 5, Deadline: time.Now().Add(time.Millisecond * 5)}, trnsE) + + // can't start new transactions while an exclusive transaction is pending + if _, err := tm.Start(ctx, "d", time.Millisecond, false); err != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err) + } + + // can't start new exclusive transactions while an exclusive transaction is pending + if _, err := tm.Start(ctx, "ee", time.Millisecond, true); err != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err) + } + + // exclusive transaction becomes active after deadlines expire + for i := 0; true; i++ { + time.Sleep(time.Microsecond) + trnsE, err := tm.Get(ctx, "ce") + if err != nil { + t.Errorf("error retrieving exclusive transaction: %v", err) + } + if trnsE.Active { + break + } + if i > 100 { + t.Fatalf("exclusive transaction never became active: %+v", trnsE) + } + } + + // can't start new transactions while an exclusive transaction is active + if _, err := tm.Start(ctx, "f", time.Millisecond, false); err != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err) + } + + // can't start new exclusive transactions while an exclusive transaction is active + if _, err := tm.Start(ctx, "ge", time.Millisecond, true); err != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err) + } + + // exclusive transaction gets expired after other transactions have attempted to start + for i := 0; true; i++ { + time.Sleep(time.Millisecond * 2) + trnsE, err := tm.Get(ctx, "ce") + if err == nil { + if i > 10 { + t.Fatalf("exclusive transaction didn't expire: %+v", trnsE) + } + } else if err != pilosa.ErrTransactionNotFound { + t.Errorf("unexpected error fetching transaction while waiting for expiration: %v", err) + } else { + break // transaction was not found, therefore it expired and we can happily continue + } + } + + // can start a new exclusive transaction and it's immediately active + trnsHE := mustStart(t, tm, "he", time.Hour, true) + test.CompareTransactions(t, &pilosa.Transaction{ID: "he", Active: true, Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsHE) + + // can't start new transactions while an exclusive transaction is active + if _, err := tm.Start(ctx, "i", time.Millisecond, false); err != pilosa.ErrTransactionExclusive { + t.Errorf("unexpected error starting transaction while an exclusive transaction exists: %v", err) + } + + // can finish an active exclusive transaction + trnsHE_finish := mustFinish(t, tm, "he") + test.CompareTransactions(t, trnsHE, trnsHE_finish) + + // can start normal transaction after finishing exclusive transaction + trnsJ := mustStart(t, tm, "j", time.Hour, false) + test.CompareTransactions(t, &pilosa.Transaction{ID: "j", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsJ) + + // can finish normal transaction + trnsJ_finish := mustFinish(t, tm, "j") + test.CompareTransactions(t, trnsJ, trnsJ_finish) + + // can start normal transaction after finishing normal transaction + trnsK := mustStart(t, tm, "k", time.Hour, false) + test.CompareTransactions(t, &pilosa.Transaction{ID: "k", Active: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsK) + + // can start new exclusive transaction, but not immediately active + trnsLE := mustStart(t, tm, "le", time.Hour, true) + test.CompareTransactions(t, &pilosa.Transaction{ID: "le", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsLE) + + // finishing k should activate le + trnsK_finish := mustFinish(t, tm, "k") + test.CompareTransactions(t, trnsK, trnsK_finish) + trnsLE_active := mustGet(t, tm, "le") + trnsLE.Active = true + test.CompareTransactions(t, trnsLE, trnsLE_active) + + mustFinish(t, tm, "le") + + // can start normal transaction to test deadline reset + trnsM := mustStart(t, tm, "m", time.Millisecond*4, false) + test.CompareTransactions(t, &pilosa.Transaction{ID: "m", Active: true, Timeout: time.Millisecond * 4, Deadline: time.Now().Add(time.Millisecond * 4)}, trnsM) + + // start new exclusive transaction to trigger deadline check + trnsNE := mustStart(t, tm, "ne", time.Hour, true) + test.CompareTransactions(t, &pilosa.Transaction{ID: "ne", Exclusive: true, Timeout: time.Hour, Deadline: time.Now().Add(time.Hour)}, trnsNE) + + // sleep for most of the deadline + time.Sleep(time.Millisecond * 3) + + // reset deadline + trnsM_reset, err := tm.ResetDeadline(ctx, "m") + if err != nil { + t.Errorf("resetting deadline: %v", err) + } + trnsM.Deadline = time.Now().Add(time.Millisecond * 4) + test.CompareTransactions(t, trnsM, trnsM_reset) + + // sleep until past the original deadline + time.Sleep(time.Millisecond * 2) + + // verify that trnsM still exists + trnsM_again := mustGet(t, tm, "m") + test.CompareTransactions(t, trnsM, trnsM_again) + +} + +func mustStart(t *testing.T, tm *pilosa.TransactionManager, id string, timeout time.Duration, exclusive bool) *pilosa.Transaction { + t.Helper() + trns, err := tm.Start(context.Background(), id, timeout, exclusive) + if err != nil { + t.Errorf("starting transaction: %v", err) + } + return trns +} + +func mustFinish(t *testing.T, tm *pilosa.TransactionManager, id string) *pilosa.Transaction { + t.Helper() + trns, err := tm.Finish(context.Background(), id) + if err != nil { + t.Errorf("finishing transaction: %v", err) + } + return trns +} + +func mustGet(t *testing.T, tm *pilosa.TransactionManager, id string) *pilosa.Transaction { + t.Helper() + trns, err := tm.Get(context.Background(), id) + if err != nil { + t.Errorf("getting transaction %s: %v", id, err) + } + return trns +} + +func mustList(t *testing.T, tm *pilosa.TransactionManager) map[string]*pilosa.Transaction { + t.Helper() + trnsMap, err := tm.List(context.Background()) + if err != nil { + t.Errorf("getting transaction list: %v", err) + } + return trnsMap +} + +func TestInMemTransactionStore(t *testing.T) { + ims := pilosa.NewInMemTransactionStore() + + err := ims.Put(&pilosa.Transaction{ID: "blah", Timeout: time.Second}) + if err != nil { + t.Fatalf("adding blah: %v", err) + } + + trns, err := ims.Get("blah") + if err != nil { + t.Fatalf("getting blah: %v", err) + } + if trns.ID != "blah" || trns.Timeout != time.Second { + t.Fatalf("unexpected transaction for blah: %+v", t) + } + + _, err = ims.Get("nope") + if err != pilosa.ErrTransactionNotFound { + t.Fatalf("unexpected error: %v", err) + } + + l, err := ims.List() + if err != nil { + t.Fatalf("listing transactions: %v", err) + } + if len(l) != 1 { + t.Errorf("unexpected number of transactions: %d", len(l)) + } + if l["blah"].ID != "blah" || l["blah"].Timeout != time.Second { + t.Errorf("unexpected transaction at blah: %+v", l["blah"]) + } + +} + +func TestMarshalUnmarshalTransaction(t *testing.T) { + tests := []struct { + name string + transaction *pilosa.Transaction + }{ + { + name: "empty", + transaction: &pilosa.Transaction{}, + }, + { + name: "basic", + transaction: &pilosa.Transaction{ + ID: "blah", + Active: true, + Exclusive: true, + Timeout: time.Minute, + Deadline: time.Now(), + }, + }, + } + + for _, tst := range tests { + t.Run(tst.name, func(t *testing.T) { + bytes, err := json.Marshal(tst.transaction) + if err != nil { + t.Errorf("marshalling: %v", err) + } + + nt := &pilosa.Transaction{} + err = json.Unmarshal(bytes, nt) + if err != nil { + t.Fatalf("unmarshalling: %v", err) + } + + test.CompareTransactions(t, tst.transaction, nt) + }) + } +} + +func TestUnmarshalTransaction(t *testing.T) { + tests := []struct { + name string + transactionJSON string + exp *pilosa.Transaction + }{ + { + name: "empty", + transactionJSON: `{}`, + exp: &pilosa.Transaction{}, + }, + { + name: "basicPost", + transactionJSON: `{"id": "blah", "exclusive": false, "timeout": "1m"}`, + exp: &pilosa.Transaction{ID: "blah", Timeout: time.Minute}, + }, + { + name: "basicPostFloatTimeout", + transactionJSON: `{"id": "blah", "exclusive": false, "timeout": 10.5}`, + exp: &pilosa.Transaction{ID: "blah", Timeout: time.Second*10 + time.Second/2}, + }, + } + + for _, tst := range tests { + t.Run(tst.name, func(t *testing.T) { + nt := &pilosa.Transaction{} + err := json.Unmarshal([]byte(tst.transactionJSON), nt) + if err != nil { + t.Fatalf("unmarshalling: %v", err) + } + + test.CompareTransactions(t, tst.exp, nt) + }) + } +}