op.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  1. // Copyright 2016 The etcd Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package clientv3
  15. import pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  16. type opType int
  17. const (
  18. // A default Op has opType 0, which is invalid.
  19. tRange opType = iota + 1
  20. tPut
  21. tDeleteRange
  22. )
  23. var (
  24. noPrefixEnd = []byte{0}
  25. )
  26. // Op represents an Operation that kv can execute.
  27. type Op struct {
  28. t opType
  29. key []byte
  30. end []byte
  31. // for range
  32. limit int64
  33. sort *SortOption
  34. serializable bool
  35. keysOnly bool
  36. countOnly bool
  37. minModRev int64
  38. maxModRev int64
  39. minCreateRev int64
  40. maxCreateRev int64
  41. // for range, watch
  42. rev int64
  43. // for watch, put, delete
  44. prevKV bool
  45. // for put
  46. ignoreValue bool
  47. // progressNotify is for progress updates.
  48. progressNotify bool
  49. // createdNotify is for created event
  50. createdNotify bool
  51. // filters for watchers
  52. filterPut bool
  53. filterDelete bool
  54. // for put
  55. val []byte
  56. leaseID LeaseID
  57. }
  58. func (op Op) toRangeRequest() *pb.RangeRequest {
  59. if op.t != tRange {
  60. panic("op.t != tRange")
  61. }
  62. r := &pb.RangeRequest{
  63. Key: op.key,
  64. RangeEnd: op.end,
  65. Limit: op.limit,
  66. Revision: op.rev,
  67. Serializable: op.serializable,
  68. KeysOnly: op.keysOnly,
  69. CountOnly: op.countOnly,
  70. MinModRevision: op.minModRev,
  71. MaxModRevision: op.maxModRev,
  72. MinCreateRevision: op.minCreateRev,
  73. MaxCreateRevision: op.maxCreateRev,
  74. }
  75. if op.sort != nil {
  76. r.SortOrder = pb.RangeRequest_SortOrder(op.sort.Order)
  77. r.SortTarget = pb.RangeRequest_SortTarget(op.sort.Target)
  78. }
  79. return r
  80. }
  81. func (op Op) toRequestOp() *pb.RequestOp {
  82. switch op.t {
  83. case tRange:
  84. return &pb.RequestOp{Request: &pb.RequestOp_RequestRange{RequestRange: op.toRangeRequest()}}
  85. case tPut:
  86. r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV, IgnoreValue: op.ignoreValue}
  87. return &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: r}}
  88. case tDeleteRange:
  89. r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
  90. return &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{RequestDeleteRange: r}}
  91. default:
  92. panic("Unknown Op")
  93. }
  94. }
  95. func (op Op) isWrite() bool {
  96. return op.t != tRange
  97. }
  98. func OpGet(key string, opts ...OpOption) Op {
  99. ret := Op{t: tRange, key: []byte(key)}
  100. ret.applyOpts(opts)
  101. return ret
  102. }
  103. func OpDelete(key string, opts ...OpOption) Op {
  104. ret := Op{t: tDeleteRange, key: []byte(key)}
  105. ret.applyOpts(opts)
  106. switch {
  107. case ret.leaseID != 0:
  108. panic("unexpected lease in delete")
  109. case ret.limit != 0:
  110. panic("unexpected limit in delete")
  111. case ret.rev != 0:
  112. panic("unexpected revision in delete")
  113. case ret.sort != nil:
  114. panic("unexpected sort in delete")
  115. case ret.serializable:
  116. panic("unexpected serializable in delete")
  117. case ret.countOnly:
  118. panic("unexpected countOnly in delete")
  119. case ret.minModRev != 0, ret.maxModRev != 0:
  120. panic("unexpected mod revision filter in delete")
  121. case ret.minCreateRev != 0, ret.maxCreateRev != 0:
  122. panic("unexpected create revision filter in delete")
  123. case ret.filterDelete, ret.filterPut:
  124. panic("unexpected filter in delete")
  125. case ret.createdNotify:
  126. panic("unexpected createdNotify in delete")
  127. }
  128. return ret
  129. }
  130. func OpPut(key, val string, opts ...OpOption) Op {
  131. ret := Op{t: tPut, key: []byte(key), val: []byte(val)}
  132. ret.applyOpts(opts)
  133. switch {
  134. case ret.end != nil:
  135. panic("unexpected range in put")
  136. case ret.limit != 0:
  137. panic("unexpected limit in put")
  138. case ret.rev != 0:
  139. panic("unexpected revision in put")
  140. case ret.sort != nil:
  141. panic("unexpected sort in put")
  142. case ret.serializable:
  143. panic("unexpected serializable in put")
  144. case ret.countOnly:
  145. panic("unexpected countOnly in put")
  146. case ret.minModRev != 0, ret.maxModRev != 0:
  147. panic("unexpected mod revision filter in put")
  148. case ret.minCreateRev != 0, ret.maxCreateRev != 0:
  149. panic("unexpected create revision filter in put")
  150. case ret.filterDelete, ret.filterPut:
  151. panic("unexpected filter in put")
  152. case ret.createdNotify:
  153. panic("unexpected createdNotify in put")
  154. }
  155. return ret
  156. }
  157. func opWatch(key string, opts ...OpOption) Op {
  158. ret := Op{t: tRange, key: []byte(key)}
  159. ret.applyOpts(opts)
  160. switch {
  161. case ret.leaseID != 0:
  162. panic("unexpected lease in watch")
  163. case ret.limit != 0:
  164. panic("unexpected limit in watch")
  165. case ret.sort != nil:
  166. panic("unexpected sort in watch")
  167. case ret.serializable:
  168. panic("unexpected serializable in watch")
  169. case ret.countOnly:
  170. panic("unexpected countOnly in watch")
  171. case ret.minModRev != 0, ret.maxModRev != 0:
  172. panic("unexpected mod revision filter in watch")
  173. case ret.minCreateRev != 0, ret.maxCreateRev != 0:
  174. panic("unexpected create revision filter in watch")
  175. }
  176. return ret
  177. }
  178. func (op *Op) applyOpts(opts []OpOption) {
  179. for _, opt := range opts {
  180. opt(op)
  181. }
  182. }
  183. // OpOption configures Operations like Get, Put, Delete.
  184. type OpOption func(*Op)
  185. // WithLease attaches a lease ID to a key in 'Put' request.
  186. func WithLease(leaseID LeaseID) OpOption {
  187. return func(op *Op) { op.leaseID = leaseID }
  188. }
  189. // WithLimit limits the number of results to return from 'Get' request.
  190. func WithLimit(n int64) OpOption { return func(op *Op) { op.limit = n } }
  191. // WithRev specifies the store revision for 'Get' request.
  192. // Or the start revision of 'Watch' request.
  193. func WithRev(rev int64) OpOption { return func(op *Op) { op.rev = rev } }
  194. // WithSort specifies the ordering in 'Get' request. It requires
  195. // 'WithRange' and/or 'WithPrefix' to be specified too.
  196. // 'target' specifies the target to sort by: key, version, revisions, value.
  197. // 'order' can be either 'SortNone', 'SortAscend', 'SortDescend'.
  198. func WithSort(target SortTarget, order SortOrder) OpOption {
  199. return func(op *Op) {
  200. if target == SortByKey && order == SortAscend {
  201. // If order != SortNone, server fetches the entire key-space,
  202. // and then applies the sort and limit, if provided.
  203. // Since current mvcc.Range implementation returns results
  204. // sorted by keys in lexicographically ascending order,
  205. // client should ignore SortOrder if the target is SortByKey.
  206. order = SortNone
  207. }
  208. op.sort = &SortOption{target, order}
  209. }
  210. }
  211. // GetPrefixRangeEnd gets the range end of the prefix.
  212. // 'Get(foo, WithPrefix())' is equal to 'Get(foo, WithRange(GetPrefixRangeEnd(foo))'.
  213. func GetPrefixRangeEnd(prefix string) string {
  214. return string(getPrefix([]byte(prefix)))
  215. }
  216. func getPrefix(key []byte) []byte {
  217. end := make([]byte, len(key))
  218. copy(end, key)
  219. for i := len(end) - 1; i >= 0; i-- {
  220. if end[i] < 0xff {
  221. end[i] = end[i] + 1
  222. end = end[:i+1]
  223. return end
  224. }
  225. }
  226. // next prefix does not exist (e.g., 0xffff);
  227. // default to WithFromKey policy
  228. return noPrefixEnd
  229. }
  230. // WithPrefix enables 'Get', 'Delete', or 'Watch' requests to operate
  231. // on the keys with matching prefix. For example, 'Get(foo, WithPrefix())'
  232. // can return 'foo1', 'foo2', and so on.
  233. func WithPrefix() OpOption {
  234. return func(op *Op) {
  235. op.end = getPrefix(op.key)
  236. }
  237. }
  238. // WithRange specifies the range of 'Get', 'Delete', 'Watch' requests.
  239. // For example, 'Get' requests with 'WithRange(end)' returns
  240. // the keys in the range [key, end).
  241. // endKey must be lexicographically greater than start key.
  242. func WithRange(endKey string) OpOption {
  243. return func(op *Op) { op.end = []byte(endKey) }
  244. }
  245. // WithFromKey specifies the range of 'Get', 'Delete', 'Watch' requests
  246. // to be equal or greater than the key in the argument.
  247. func WithFromKey() OpOption { return WithRange("\x00") }
  248. // WithSerializable makes 'Get' request serializable. By default,
  249. // it's linearizable. Serializable requests are better for lower latency
  250. // requirement.
  251. func WithSerializable() OpOption {
  252. return func(op *Op) { op.serializable = true }
  253. }
  254. // WithKeysOnly makes the 'Get' request return only the keys and the corresponding
  255. // values will be omitted.
  256. func WithKeysOnly() OpOption {
  257. return func(op *Op) { op.keysOnly = true }
  258. }
  259. // WithCountOnly makes the 'Get' request return only the count of keys.
  260. func WithCountOnly() OpOption {
  261. return func(op *Op) { op.countOnly = true }
  262. }
  263. // WithMinModRev filters out keys for Get with modification revisions less than the given revision.
  264. func WithMinModRev(rev int64) OpOption { return func(op *Op) { op.minModRev = rev } }
  265. // WithMaxModRev filters out keys for Get with modification revisions greater than the given revision.
  266. func WithMaxModRev(rev int64) OpOption { return func(op *Op) { op.maxModRev = rev } }
  267. // WithMinCreateRev filters out keys for Get with creation revisions less than the given revision.
  268. func WithMinCreateRev(rev int64) OpOption { return func(op *Op) { op.minCreateRev = rev } }
  269. // WithMaxCreateRev filters out keys for Get with creation revisions greater than the given revision.
  270. func WithMaxCreateRev(rev int64) OpOption { return func(op *Op) { op.maxCreateRev = rev } }
  271. // WithFirstCreate gets the key with the oldest creation revision in the request range.
  272. func WithFirstCreate() []OpOption { return withTop(SortByCreateRevision, SortAscend) }
  273. // WithLastCreate gets the key with the latest creation revision in the request range.
  274. func WithLastCreate() []OpOption { return withTop(SortByCreateRevision, SortDescend) }
  275. // WithFirstKey gets the lexically first key in the request range.
  276. func WithFirstKey() []OpOption { return withTop(SortByKey, SortAscend) }
  277. // WithLastKey gets the lexically last key in the request range.
  278. func WithLastKey() []OpOption { return withTop(SortByKey, SortDescend) }
  279. // WithFirstRev gets the key with the oldest modification revision in the request range.
  280. func WithFirstRev() []OpOption { return withTop(SortByModRevision, SortAscend) }
  281. // WithLastRev gets the key with the latest modification revision in the request range.
  282. func WithLastRev() []OpOption { return withTop(SortByModRevision, SortDescend) }
  283. // withTop gets the first key over the get's prefix given a sort order
  284. func withTop(target SortTarget, order SortOrder) []OpOption {
  285. return []OpOption{WithPrefix(), WithSort(target, order), WithLimit(1)}
  286. }
  287. // WithProgressNotify makes watch server send periodic progress updates
  288. // every 10 minutes when there is no incoming events.
  289. // Progress updates have zero events in WatchResponse.
  290. func WithProgressNotify() OpOption {
  291. return func(op *Op) {
  292. op.progressNotify = true
  293. }
  294. }
  295. // WithCreatedNotify makes watch server sends the created event.
  296. func WithCreatedNotify() OpOption {
  297. return func(op *Op) {
  298. op.createdNotify = true
  299. }
  300. }
  301. // WithFilterPut discards PUT events from the watcher.
  302. func WithFilterPut() OpOption {
  303. return func(op *Op) { op.filterPut = true }
  304. }
  305. // WithFilterDelete discards DELETE events from the watcher.
  306. func WithFilterDelete() OpOption {
  307. return func(op *Op) { op.filterDelete = true }
  308. }
  309. // WithPrevKV gets the previous key-value pair before the event happens. If the previous KV is already compacted,
  310. // nothing will be returned.
  311. func WithPrevKV() OpOption {
  312. return func(op *Op) {
  313. op.prevKV = true
  314. }
  315. }
  316. // WithIgnoreValue updates the key using its current value.
  317. // Empty value should be passed when ignore_value is set.
  318. // Returns an error if the key does not exist.
  319. func WithIgnoreValue() OpOption {
  320. return func(op *Op) {
  321. op.ignoreValue = true
  322. }
  323. }
  324. // LeaseOp represents an Operation that lease can execute.
  325. type LeaseOp struct {
  326. id LeaseID
  327. // for TimeToLive
  328. attachedKeys bool
  329. }
  330. // LeaseOption configures lease operations.
  331. type LeaseOption func(*LeaseOp)
  332. func (op *LeaseOp) applyOpts(opts []LeaseOption) {
  333. for _, opt := range opts {
  334. opt(op)
  335. }
  336. }
  337. // WithAttachedKeys requests lease timetolive API to return
  338. // attached keys of given lease ID.
  339. func WithAttachedKeys() LeaseOption {
  340. return func(op *LeaseOp) { op.attachedKeys = true }
  341. }
  342. func toLeaseTimeToLiveRequest(id LeaseID, opts ...LeaseOption) *pb.LeaseTimeToLiveRequest {
  343. ret := &LeaseOp{id: id}
  344. ret.applyOpts(opts)
  345. return &pb.LeaseTimeToLiveRequest{ID: int64(id), Keys: ret.attachedKeys}
  346. }