File: txn.go

package info (click to toggle)
consul 1.5.2+dfsg1-12
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 41,656 kB
  • sloc: sh: 2,887; makefile: 127; ruby: 75; cpp: 58; xml: 36
file content (230 lines) | stat: -rw-r--r-- 6,334 bytes parent folder | download | duplicates (4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
package api

import (
	"bytes"
	"fmt"
	"io"
	"net/http"
)

// Txn is used to manipulate the Txn API
type Txn struct {
	c *Client
}

// Txn is used to return a handle to the K/V apis
func (c *Client) Txn() *Txn {
	return &Txn{c}
}

// TxnOp is the internal format we send to Consul. Currently only K/V and
// check operations are supported.
type TxnOp struct {
	KV      *KVTxnOp
	Node    *NodeTxnOp
	Service *ServiceTxnOp
	Check   *CheckTxnOp
}

// TxnOps is a list of transaction operations.
type TxnOps []*TxnOp

// TxnResult is the internal format we receive from Consul.
type TxnResult struct {
	KV      *KVPair
	Node    *Node
	Service *CatalogService
	Check   *HealthCheck
}

// TxnResults is a list of TxnResult objects.
type TxnResults []*TxnResult

// TxnError is used to return information about an operation in a transaction.
type TxnError struct {
	OpIndex int
	What    string
}

// TxnErrors is a list of TxnError objects.
type TxnErrors []*TxnError

// TxnResponse is the internal format we receive from Consul.
type TxnResponse struct {
	Results TxnResults
	Errors  TxnErrors
}

// KVOp constants give possible operations available in a transaction.
type KVOp string

const (
	KVSet            KVOp = "set"
	KVDelete         KVOp = "delete"
	KVDeleteCAS      KVOp = "delete-cas"
	KVDeleteTree     KVOp = "delete-tree"
	KVCAS            KVOp = "cas"
	KVLock           KVOp = "lock"
	KVUnlock         KVOp = "unlock"
	KVGet            KVOp = "get"
	KVGetTree        KVOp = "get-tree"
	KVCheckSession   KVOp = "check-session"
	KVCheckIndex     KVOp = "check-index"
	KVCheckNotExists KVOp = "check-not-exists"
)

// KVTxnOp defines a single operation inside a transaction.
type KVTxnOp struct {
	Verb    KVOp
	Key     string
	Value   []byte
	Flags   uint64
	Index   uint64
	Session string
}

// KVTxnOps defines a set of operations to be performed inside a single
// transaction.
type KVTxnOps []*KVTxnOp

// KVTxnResponse has the outcome of a transaction.
type KVTxnResponse struct {
	Results []*KVPair
	Errors  TxnErrors
}

// NodeOp constants give possible operations available in a transaction.
type NodeOp string

const (
	NodeGet       NodeOp = "get"
	NodeSet       NodeOp = "set"
	NodeCAS       NodeOp = "cas"
	NodeDelete    NodeOp = "delete"
	NodeDeleteCAS NodeOp = "delete-cas"
)

// NodeTxnOp defines a single operation inside a transaction.
type NodeTxnOp struct {
	Verb NodeOp
	Node Node
}

// ServiceOp constants give possible operations available in a transaction.
type ServiceOp string

const (
	ServiceGet       ServiceOp = "get"
	ServiceSet       ServiceOp = "set"
	ServiceCAS       ServiceOp = "cas"
	ServiceDelete    ServiceOp = "delete"
	ServiceDeleteCAS ServiceOp = "delete-cas"
)

// ServiceTxnOp defines a single operation inside a transaction.
type ServiceTxnOp struct {
	Verb    ServiceOp
	Node    string
	Service AgentService
}

// CheckOp constants give possible operations available in a transaction.
type CheckOp string

const (
	CheckGet       CheckOp = "get"
	CheckSet       CheckOp = "set"
	CheckCAS       CheckOp = "cas"
	CheckDelete    CheckOp = "delete"
	CheckDeleteCAS CheckOp = "delete-cas"
)

// CheckTxnOp defines a single operation inside a transaction.
type CheckTxnOp struct {
	Verb  CheckOp
	Check HealthCheck
}

// Txn is used to apply multiple Consul operations in a single, atomic transaction.
//
// Note that Go will perform the required base64 encoding on the values
// automatically because the type is a byte slice. Transactions are defined as a
// list of operations to perform, using the different fields in the TxnOp structure
// to define operations. If any operation fails, none of the changes are applied
// to the state store.
//
// Even though this is generally a write operation, we take a QueryOptions input
// and return a QueryMeta output. If the transaction contains only read ops, then
// Consul will fast-path it to a different endpoint internally which supports
// consistency controls, but not blocking. If there are write operations then
// the request will always be routed through raft and any consistency settings
// will be ignored.
//
// Here's an example:
//
//	   ops := KVTxnOps{
//		   &KVTxnOp{
//			   Verb:    KVLock,
//			   Key:     "test/lock",
//			   Session: "adf4238a-882b-9ddc-4a9d-5b6758e4159e",
//			   Value:   []byte("hello"),
//		   },
//		   &KVTxnOp{
//			   Verb:    KVGet,
//			   Key:     "another/key",
//		   },
//		   &CheckTxnOp{
//			   Verb:        CheckSet,
//			   HealthCheck: HealthCheck{
//				   Node:    "foo",
//				   CheckID: "redis:a",
//				   Name:    "Redis Health Check",
//				   Status:  "passing",
//			   },
//		   }
//	   }
//	   ok, response, _, err := kv.Txn(&ops, nil)
//
// If there is a problem making the transaction request then an error will be
// returned. Otherwise, the ok value will be true if the transaction succeeded
// or false if it was rolled back. The response is a structured return value which
// will have the outcome of the transaction. Its Results member will have entries
// for each operation. For KV operations, Deleted keys will have a nil entry in the
// results, and to save space, the Value of each key in the Results will be nil
// unless the operation is a KVGet. If the transaction was rolled back, the Errors
// member will have entries referencing the index of the operation that failed
// along with an error message.
func (t *Txn) Txn(txn TxnOps, q *QueryOptions) (bool, *TxnResponse, *QueryMeta, error) {
	return t.c.txn(txn, q)
}

func (c *Client) txn(txn TxnOps, q *QueryOptions) (bool, *TxnResponse, *QueryMeta, error) {
	r := c.newRequest("PUT", "/v1/txn")
	r.setQueryOptions(q)

	r.obj = txn
	rtt, resp, err := c.doRequest(r)
	if err != nil {
		return false, nil, nil, err
	}
	defer resp.Body.Close()

	qm := &QueryMeta{}
	parseQueryMeta(resp, qm)
	qm.RequestTime = rtt

	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusConflict {
		var txnResp TxnResponse
		if err := decodeBody(resp, &txnResp); err != nil {
			return false, nil, nil, err
		}

		return resp.StatusCode == http.StatusOK, &txnResp, qm, nil
	}

	var buf bytes.Buffer
	if _, err := io.Copy(&buf, resp.Body); err != nil {
		return false, nil, nil, fmt.Errorf("Failed to read response: %v", err)
	}
	return false, nil, nil, fmt.Errorf("Failed request: %s", buf.String())
}