2016-05-11 04:41:47 +00:00
|
|
|
package consul
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/armon/go-metrics"
|
2016-05-13 00:38:25 +00:00
|
|
|
"github.com/hashicorp/consul/acl"
|
2017-07-06 10:34:00 +00:00
|
|
|
"github.com/hashicorp/consul/agent/structs"
|
2016-05-11 04:41:47 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// Txn endpoint is used to perform multi-object atomic transactions.
|
|
|
|
type Txn struct {
|
|
|
|
srv *Server
|
|
|
|
}
|
|
|
|
|
2016-05-13 00:38:25 +00:00
|
|
|
// preCheck is used to verify the incoming operations before any further
|
|
|
|
// processing takes place. This checks things like ACLs.
|
2018-10-19 16:04:07 +00:00
|
|
|
func (t *Txn) preCheck(authorizer acl.Authorizer, ops structs.TxnOps) structs.TxnErrors {
|
2016-05-13 00:38:25 +00:00
|
|
|
var errors structs.TxnErrors
|
|
|
|
|
|
|
|
// Perform the pre-apply checks for any KV operations.
|
|
|
|
for i, op := range ops {
|
2018-12-03 07:11:48 +00:00
|
|
|
switch {
|
|
|
|
case op.KV != nil:
|
2018-10-19 16:04:07 +00:00
|
|
|
ok, err := kvsPreApply(t.srv, authorizer, op.KV.Verb, &op.KV.DirEnt)
|
2016-05-13 00:38:25 +00:00
|
|
|
if err != nil {
|
2017-03-23 23:05:35 +00:00
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
2016-05-13 00:38:25 +00:00
|
|
|
} else if !ok {
|
|
|
|
err = fmt.Errorf("failed to lock key %q due to lock delay", op.KV.DirEnt.Key)
|
2017-03-23 23:05:35 +00:00
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
2016-05-13 00:38:25 +00:00
|
|
|
}
|
2018-12-12 10:29:54 +00:00
|
|
|
case op.Node != nil:
|
|
|
|
node := op.Node.Node
|
|
|
|
if err := nodePreApply(node.Node, string(node.ID)); err != nil {
|
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check that the token has permissions for the given operation.
|
|
|
|
if err := vetNodeTxnOp(op.Node, authorizer); err != nil {
|
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
|
|
|
}
|
|
|
|
case op.Service != nil:
|
|
|
|
service := &op.Service.Service
|
|
|
|
if err := servicePreApply(service, authorizer); err != nil {
|
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check that the token has permissions for the given operation.
|
|
|
|
if err := vetServiceTxnOp(op.Service, authorizer); err != nil {
|
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
|
|
|
}
|
2018-12-03 07:11:48 +00:00
|
|
|
case op.Check != nil:
|
|
|
|
checkPreApply(&op.Check.Check)
|
2018-12-12 10:29:54 +00:00
|
|
|
|
|
|
|
// Check that the token has permissions for the given operation.
|
|
|
|
if err := vetCheckTxnOp(op.Check, authorizer); err != nil {
|
|
|
|
errors = append(errors, &structs.TxnError{
|
|
|
|
OpIndex: i,
|
|
|
|
What: err.Error(),
|
|
|
|
})
|
|
|
|
}
|
2016-05-13 00:38:25 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return errors
|
|
|
|
}
|
|
|
|
|
2016-05-11 04:41:47 +00:00
|
|
|
// Apply is used to apply multiple operations in a single, atomic transaction.
|
2016-05-11 08:35:27 +00:00
|
|
|
func (t *Txn) Apply(args *structs.TxnRequest, reply *structs.TxnResponse) error {
|
2016-05-11 04:41:47 +00:00
|
|
|
if done, err := t.srv.forward("Txn.Apply", args, args, reply); done {
|
|
|
|
return err
|
|
|
|
}
|
2017-10-04 23:43:27 +00:00
|
|
|
defer metrics.MeasureSince([]string{"txn", "apply"}, time.Now())
|
2016-05-11 04:41:47 +00:00
|
|
|
|
2016-05-13 00:38:25 +00:00
|
|
|
// Run the pre-checks before we send the transaction into Raft.
|
2018-10-19 16:04:07 +00:00
|
|
|
authorizer, err := t.srv.ResolveToken(args.Token)
|
2016-05-11 04:41:47 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-10-19 16:04:07 +00:00
|
|
|
reply.Errors = t.preCheck(authorizer, args.Ops)
|
2016-05-11 04:41:47 +00:00
|
|
|
if len(reply.Errors) > 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Apply the update.
|
2016-05-11 08:35:27 +00:00
|
|
|
resp, err := t.srv.raftApply(structs.TxnRequestType, args)
|
2016-05-11 04:41:47 +00:00
|
|
|
if err != nil {
|
2016-05-11 08:35:27 +00:00
|
|
|
t.srv.logger.Printf("[ERR] consul.txn: Apply failed: %v", err)
|
2016-05-11 04:41:47 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
if respErr, ok := resp.(error); ok {
|
|
|
|
return respErr
|
|
|
|
}
|
|
|
|
|
|
|
|
// Convert the return type. This should be a cheap copy since we are
|
|
|
|
// just taking the two slices.
|
2016-05-11 08:35:27 +00:00
|
|
|
if txnResp, ok := resp.(structs.TxnResponse); ok {
|
2018-10-19 16:04:07 +00:00
|
|
|
if authorizer != nil {
|
|
|
|
txnResp.Results = FilterTxnResults(authorizer, txnResp.Results)
|
2016-05-13 22:58:55 +00:00
|
|
|
}
|
2016-05-11 08:35:27 +00:00
|
|
|
*reply = txnResp
|
2016-05-11 04:41:47 +00:00
|
|
|
} else {
|
|
|
|
return fmt.Errorf("unexpected return type %T", resp)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
2016-05-13 00:38:25 +00:00
|
|
|
|
|
|
|
// Read is used to perform a read-only transaction that doesn't modify the state
|
|
|
|
// store. This is much more scaleable since it doesn't go through Raft and
|
|
|
|
// supports staleness, so this should be preferred if you're just performing
|
|
|
|
// reads.
|
|
|
|
func (t *Txn) Read(args *structs.TxnReadRequest, reply *structs.TxnReadResponse) error {
|
|
|
|
if done, err := t.srv.forward("Txn.Read", args, args, reply); done {
|
|
|
|
return err
|
|
|
|
}
|
2017-10-04 23:43:27 +00:00
|
|
|
defer metrics.MeasureSince([]string{"txn", "read"}, time.Now())
|
2016-05-13 00:38:25 +00:00
|
|
|
|
|
|
|
// We have to do this ourselves since we are not doing a blocking RPC.
|
|
|
|
t.srv.setQueryMeta(&reply.QueryMeta)
|
|
|
|
if args.RequireConsistent {
|
|
|
|
if err := t.srv.consistentRead(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Run the pre-checks before we perform the read.
|
2018-10-19 16:04:07 +00:00
|
|
|
authorizer, err := t.srv.ResolveToken(args.Token)
|
2016-05-13 00:38:25 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-10-19 16:04:07 +00:00
|
|
|
reply.Errors = t.preCheck(authorizer, args.Ops)
|
2016-05-13 00:38:25 +00:00
|
|
|
if len(reply.Errors) > 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Run the read transaction.
|
|
|
|
state := t.srv.fsm.State()
|
|
|
|
reply.Results, reply.Errors = state.TxnRO(args.Ops)
|
2018-10-19 16:04:07 +00:00
|
|
|
if authorizer != nil {
|
|
|
|
reply.Results = FilterTxnResults(authorizer, reply.Results)
|
2016-05-13 22:58:55 +00:00
|
|
|
}
|
2016-05-13 00:38:25 +00:00
|
|
|
return nil
|
|
|
|
}
|