From 210c7239ab5001b6a5cd109395769feeed1e783d Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 17 Apr 2020 16:30:54 -0500 Subject: [PATCH] add HTTP handlers and client for transactions --- client.go | 19 ++++++ http/client.go | 161 +++++++++++++++++++++++++++++++++++++++++++- http/client_test.go | 146 +++++++++++++++++++++++++++++++++++++++ http/handler.go | 109 ++++++++++++++++++++++++++++++ transaction.go | 73 +++++++++++++++++--- transaction.md | 2 + transaction_test.go | 74 ++++++++++++++++++++ 7 files changed, 574 insertions(+), 10 deletions(-) diff --git a/client.go b/client.go index a1b808a77..1fb893ee8 100644 --- a/client.go +++ b/client.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "io" + "time" ) // Bit represents the intersection of a row and a column. It can be specified by @@ -73,6 +74,11 @@ type InternalClient interface { RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error + + StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error) + FinishTransaction(ctx context.Context, id string) (Transaction, error) + Transactions(ctx context.Context) (map[string]Transaction, error) + GetTransaction(ctx context.Context, id string) (Transaction, error) } //=============== @@ -204,3 +210,16 @@ func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, fiel func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) { return nil, nil } + +func (n nopInternalClient) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (Transaction, error) { + return Transaction{}, nil +} +func (n nopInternalClient) FinishTransaction(ctx context.Context, id string) (Transaction, error) { + return Transaction{}, 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 Transaction{}, nil +} diff --git a/http/client.go b/http/client.go index a8a5106d6..e3bee63c6 100644 --- a/http/client.go +++ b/http/client.go @@ -26,6 +26,7 @@ import ( "net/url" "sort" "strconv" + "time" "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/encoding/proto" @@ -392,6 +393,7 @@ 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^^? nodes, err := c.Nodes(ctx) if err != nil { return fmt.Errorf("getting nodes: %s", err) @@ -1227,11 +1229,165 @@ 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() + + trnsMap := make(map[string]pilosa.Transaction) + + u := uriPathToURL(c.defaultURI, "/transactions") + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + return trnsMap, 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 trnsMap, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + tmpTrnsMap := make(map[string]*pilosa.Transaction) + err = json.NewDecoder(resp.Body).Decode(&tmpTrnsMap) + + for id, trnsp := range tmpTrnsMap { + trnsMap[id] = *trnsp + } + + 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() + tr := &TransactionResponse{Transaction: &pilosa.Transaction{}} + buf, err := json.Marshal(&pilosa.Transaction{ + ID: id, + Timeout: timeout, + Exclusive: exclusive, + }) + if err != nil { + return pilosa.Transaction{}, 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 pilosa.Transaction{}, 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 pilosa.Transaction{}, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + err = json.NewDecoder(resp.Body).Decode(&tr) + if err != nil { + return pilosa.Transaction{}, 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 pilosa.Transaction{}, 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 pilosa.Transaction{}, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + tr := &TransactionResponse{Transaction: &pilosa.Transaction{}} + err = json.NewDecoder(resp.Body).Decode(&tr) + if err != nil { + return pilosa.Transaction{}, 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 pilosa.Transaction{}, 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 pilosa.Transaction{}, errors.Wrap(err, "executing request") + } + defer resp.Body.Close() + tr := &TransactionResponse{Transaction: &pilosa.Transaction{}} + err = json.NewDecoder(resp.Body).Decode(&tr) + if err != nil { + return pilosa.Transaction{}, 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 +1397,9 @@ func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, erro } return nil, errors.Wrap(err, "getting response") } + if eo.giveRawResponse { + return resp, nil + } if resp.StatusCode < 200 || resp.StatusCode >= 300 { defer resp.Body.Close() buf, err := ioutil.ReadAll(resp.Body) diff --git a/http/client_test.go b/http/client_test.go index 9d8f61aca..eb76e43ce 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -23,6 +23,7 @@ import ( gohttp "net/http" "reflect" "strconv" + "strings" "testing" "time" @@ -32,6 +33,7 @@ import ( "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "github.com/pkg/errors" ) // Test distributed TopN Row count across 3 nodes. @@ -1242,6 +1244,150 @@ 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, + pilosa.Transaction{}, + 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, + pilosa.Transaction{}, + 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 { + expDeadline = time.Now().Add(time.Minute) + test.CompareTransactions(t, + pilosa.Transaction{}, + trns) + } +} + // Client represents a test wrapper for pilosa.Client. type Client struct { *http.InternalClient diff --git a/http/handler.go b/http/handler.go index cecaa666e..b5e307bf0 100644 --- a/http/handler.go +++ b/http/handler.go @@ -204,6 +204,10 @@ func (h *Handler) populateValidators() { h.validators["PostFieldAttrDiff"] = queryValidationSpecRequired() h.validators["GetNodes"] = queryValidationSpecRequired() h.validators["GetShardMax"] = queryValidationSpecRequired() + h.validators["GetTransactions"] = queryValidationSpecRequired() + h.validators["GetTransaction"] = queryValidationSpecRequired() + h.validators["PostTransaction"] = queryValidationSpecRequired() + h.validators["PostFinishTransaction"] = queryValidationSpecRequired() } type contextKeyQuery int @@ -337,6 +341,10 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/schema", handler.handlePostSchema).Methods("POST").Name("PostSchema") router.HandleFunc("/status", handler.handleGetStatus).Methods("GET").Name("GetStatus") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion") + router.HandleFunc("/transactions", handler.handleGetTransactions).Methods("GET").Name("GetTransactions") + 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") // /internal endpoints are for internal use only; they may change at any time. // DO NOT rely on these for external applications! @@ -1031,6 +1039,107 @@ 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() + 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 + } + + // JSON marshalling bullshit. Maybe we should just use + // *Transaction everywhere. + tmapP := make(map[string]*pilosa.Transaction) + for id, trns := range trnsMap { + trns := trns + tmapP[id] = &trns + } + + if err := json.NewEncoder(w).Encode(tmapP); 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(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 := mux.Vars(r)["id"] + trns, err := h.api.StartTransaction(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(id, false) + h.doTransactionResponse(w, err, trns) +} + // handleDeleteRemoteAvailableShard handles DELETE /field/{field}/available-shards/{shardID} request. func (h *Handler) handleDeleteRemoteAvailableShard(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { diff --git a/transaction.go b/transaction.go index ccfad9ed9..ca053fc9c 100644 --- a/transaction.go +++ b/transaction.go @@ -1,6 +1,7 @@ package pilosa import ( + "encoding/json" "sync" "time" @@ -12,25 +13,25 @@ import ( // 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 + 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 + Active bool `json:"active"` // Exclusive is set on transactions which can only become active when no other transactions exist. - Exclusive bool + Exclusive bool `json:"exclusive"` // Timeout is the minimum idle time for which this transaction should continue to exist. - Timeout time.Duration + Timeout time.Duration `json:"timeout"` // Deadline is calculated from Timeout, and should be reset each // time there is activity on the transaction. - Deadline time.Time + Deadline time.Time `json:"deadline"` // Stats track statistics for the transaction. Not yet used. - Stats TransactionStats + Stats TransactionStats `json:"stats"` } type TransactionStats struct{} @@ -86,7 +87,7 @@ func (tm *TransactionManager) Start(id string, timeout time.Duration, exclusive // if someone wants a transaction, and we're not able to // give it to them, we want to be checking deadlines. tm.startDeadlineChecker() - return Transaction{}, ErrTransactionExclusive + return trns, ErrTransactionExclusive } } if trns, ok := trnsMap[id]; ok { @@ -354,9 +355,8 @@ type Error string func (e Error) Error() string { return string(e) } const ErrTransactionNotFound = Error("transaction not found") -const ErrTransactionExclusive = Error("there is already an exclusive transaction") +const ErrTransactionExclusive = Error("there is an exclusive transaction, try later") const ErrTransactionExists = Error("transaction with the given id already exists") -const ErrTransactionInactive = Error("cannot finish an inactive transaction") func CompareTransactions(t1, t2 Transaction) error { if t1.ID != t2.ID { @@ -374,3 +374,58 @@ func CompareTransactions(t1, t2 Transaction) error { // 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.Format(time.RFC3339Nano), + }) +} diff --git a/transaction.md b/transaction.md index d938b4a00..109dbfd9d 100644 --- a/transaction.md +++ b/transaction.md @@ -164,6 +164,8 @@ goes through API (and is passed directly to Server). (unimplemented) - [ ] implement and use persistent transaction store rather than inmem. - [ ] update go-pilosa/gpexp to actually USE transactions - [ ] update IDK to use updated go-pilosa + +- ID validation. No slashes, no non-URL safe chars #### Testing TransactionManager - there should never be more than one Exclusive transaction diff --git a/transaction_test.go b/transaction_test.go index 6621dcda3..917eaf982 100644 --- a/transaction_test.go +++ b/transaction_test.go @@ -1,6 +1,7 @@ package pilosa_test import ( + "encoding/json" "testing" "time" @@ -234,3 +235,76 @@ func TestInMemTransactionStore(t *testing.T) { } } + +func TestMarshalUnmarshalTransaction(t *testing.T) { + tests := []struct { + name string + transaction pilosa.Transaction + }{ + { + name: "empty", + }, + { + 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: `{}`, + }, + { + 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) + }) + } +}