Merge pull request #309 from molecula/transactions

Transactions
This commit is contained in:
Travis Turner 2020-04-22 15:52:42 -05:00 committed by GitHub
commit 2795a3f7e2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 2908 additions and 85 deletions

36
api.go
View file

@ -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: {},
}

View file

@ -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))
}

View file

@ -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
}

View file

@ -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
}

View file

@ -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

View file

@ -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 {

View file

@ -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)

View file

@ -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

View file

@ -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) {

View file

@ -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)
})
}
}

View file

@ -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

View file

@ -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 {}

154
server.go
View file

@ -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 {

View file

@ -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)
}
}

View file

@ -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)
}
}

View file

@ -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")
}

46
test/transaction.go Normal file
View file

@ -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)
}
}

461
transaction.go Normal file
View file

@ -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,
}
}

256
transaction.md Normal file
View file

@ -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.

328
transaction_test.go Normal file
View file

@ -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)
})
}
}