1package api 2 3import ( 4 "bytes" 5 "fmt" 6 "io" 7 "net/http" 8) 9 10// Txn is used to manipulate the Txn API 11type Txn struct { 12 c *Client 13} 14 15// Txn is used to return a handle to the K/V apis 16func (c *Client) Txn() *Txn { 17 return &Txn{c} 18} 19 20// TxnOp is the internal format we send to Consul. Currently only K/V and 21// check operations are supported. 22type TxnOp struct { 23 KV *KVTxnOp 24 Node *NodeTxnOp 25 Service *ServiceTxnOp 26 Check *CheckTxnOp 27} 28 29// TxnOps is a list of transaction operations. 30type TxnOps []*TxnOp 31 32// TxnResult is the internal format we receive from Consul. 33type TxnResult struct { 34 KV *KVPair 35 Node *Node 36 Service *CatalogService 37 Check *HealthCheck 38} 39 40// TxnResults is a list of TxnResult objects. 41type TxnResults []*TxnResult 42 43// TxnError is used to return information about an operation in a transaction. 44type TxnError struct { 45 OpIndex int 46 What string 47} 48 49// TxnErrors is a list of TxnError objects. 50type TxnErrors []*TxnError 51 52// TxnResponse is the internal format we receive from Consul. 53type TxnResponse struct { 54 Results TxnResults 55 Errors TxnErrors 56} 57 58// KVOp constants give possible operations available in a transaction. 59type KVOp string 60 61const ( 62 KVSet KVOp = "set" 63 KVDelete KVOp = "delete" 64 KVDeleteCAS KVOp = "delete-cas" 65 KVDeleteTree KVOp = "delete-tree" 66 KVCAS KVOp = "cas" 67 KVLock KVOp = "lock" 68 KVUnlock KVOp = "unlock" 69 KVGet KVOp = "get" 70 KVGetTree KVOp = "get-tree" 71 KVCheckSession KVOp = "check-session" 72 KVCheckIndex KVOp = "check-index" 73 KVCheckNotExists KVOp = "check-not-exists" 74) 75 76// KVTxnOp defines a single operation inside a transaction. 77type KVTxnOp struct { 78 Verb KVOp 79 Key string 80 Value []byte 81 Flags uint64 82 Index uint64 83 Session string 84 Namespace string `json:",omitempty"` 85} 86 87// KVTxnOps defines a set of operations to be performed inside a single 88// transaction. 89type KVTxnOps []*KVTxnOp 90 91// KVTxnResponse has the outcome of a transaction. 92type KVTxnResponse struct { 93 Results []*KVPair 94 Errors TxnErrors 95} 96 97// SessionOp constants give possible operations available in a transaction. 98type SessionOp string 99 100const ( 101 SessionDelete SessionOp = "delete" 102) 103 104// SessionTxnOp defines a single operation inside a transaction. 105type SessionTxnOp struct { 106 Verb SessionOp 107 Session Session 108} 109 110// NodeOp constants give possible operations available in a transaction. 111type NodeOp string 112 113const ( 114 NodeGet NodeOp = "get" 115 NodeSet NodeOp = "set" 116 NodeCAS NodeOp = "cas" 117 NodeDelete NodeOp = "delete" 118 NodeDeleteCAS NodeOp = "delete-cas" 119) 120 121// NodeTxnOp defines a single operation inside a transaction. 122type NodeTxnOp struct { 123 Verb NodeOp 124 Node Node 125} 126 127// ServiceOp constants give possible operations available in a transaction. 128type ServiceOp string 129 130const ( 131 ServiceGet ServiceOp = "get" 132 ServiceSet ServiceOp = "set" 133 ServiceCAS ServiceOp = "cas" 134 ServiceDelete ServiceOp = "delete" 135 ServiceDeleteCAS ServiceOp = "delete-cas" 136) 137 138// ServiceTxnOp defines a single operation inside a transaction. 139type ServiceTxnOp struct { 140 Verb ServiceOp 141 Node string 142 Service AgentService 143} 144 145// CheckOp constants give possible operations available in a transaction. 146type CheckOp string 147 148const ( 149 CheckGet CheckOp = "get" 150 CheckSet CheckOp = "set" 151 CheckCAS CheckOp = "cas" 152 CheckDelete CheckOp = "delete" 153 CheckDeleteCAS CheckOp = "delete-cas" 154) 155 156// CheckTxnOp defines a single operation inside a transaction. 157type CheckTxnOp struct { 158 Verb CheckOp 159 Check HealthCheck 160} 161 162// Txn is used to apply multiple Consul operations in a single, atomic transaction. 163// 164// Note that Go will perform the required base64 encoding on the values 165// automatically because the type is a byte slice. Transactions are defined as a 166// list of operations to perform, using the different fields in the TxnOp structure 167// to define operations. If any operation fails, none of the changes are applied 168// to the state store. 169// 170// Even though this is generally a write operation, we take a QueryOptions input 171// and return a QueryMeta output. If the transaction contains only read ops, then 172// Consul will fast-path it to a different endpoint internally which supports 173// consistency controls, but not blocking. If there are write operations then 174// the request will always be routed through raft and any consistency settings 175// will be ignored. 176// 177// Here's an example: 178// 179// ops := KVTxnOps{ 180// &KVTxnOp{ 181// Verb: KVLock, 182// Key: "test/lock", 183// Session: "adf4238a-882b-9ddc-4a9d-5b6758e4159e", 184// Value: []byte("hello"), 185// }, 186// &KVTxnOp{ 187// Verb: KVGet, 188// Key: "another/key", 189// }, 190// &CheckTxnOp{ 191// Verb: CheckSet, 192// HealthCheck: HealthCheck{ 193// Node: "foo", 194// CheckID: "redis:a", 195// Name: "Redis Health Check", 196// Status: "passing", 197// }, 198// } 199// } 200// ok, response, _, err := kv.Txn(&ops, nil) 201// 202// If there is a problem making the transaction request then an error will be 203// returned. Otherwise, the ok value will be true if the transaction succeeded 204// or false if it was rolled back. The response is a structured return value which 205// will have the outcome of the transaction. Its Results member will have entries 206// for each operation. For KV operations, Deleted keys will have a nil entry in the 207// results, and to save space, the Value of each key in the Results will be nil 208// unless the operation is a KVGet. If the transaction was rolled back, the Errors 209// member will have entries referencing the index of the operation that failed 210// along with an error message. 211func (t *Txn) Txn(txn TxnOps, q *QueryOptions) (bool, *TxnResponse, *QueryMeta, error) { 212 return t.c.txn(txn, q) 213} 214 215func (c *Client) txn(txn TxnOps, q *QueryOptions) (bool, *TxnResponse, *QueryMeta, error) { 216 r := c.newRequest("PUT", "/v1/txn") 217 r.setQueryOptions(q) 218 219 r.obj = txn 220 rtt, resp, err := c.doRequest(r) 221 if err != nil { 222 return false, nil, nil, err 223 } 224 defer closeResponseBody(resp) 225 226 qm := &QueryMeta{} 227 parseQueryMeta(resp, qm) 228 qm.RequestTime = rtt 229 230 if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusConflict { 231 var txnResp TxnResponse 232 if err := decodeBody(resp, &txnResp); err != nil { 233 return false, nil, nil, err 234 } 235 236 return resp.StatusCode == http.StatusOK, &txnResp, qm, nil 237 } 238 239 var buf bytes.Buffer 240 if _, err := io.Copy(&buf, resp.Body); err != nil { 241 return false, nil, nil, fmt.Errorf("Failed to read response: %v", err) 242 } 243 return false, nil, nil, fmt.Errorf("Failed request: %s", buf.String()) 244} 245