apply.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838
  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 etcdserver
  15. import (
  16. "bytes"
  17. "fmt"
  18. "sort"
  19. "time"
  20. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  21. "github.com/coreos/etcd/lease"
  22. "github.com/coreos/etcd/mvcc"
  23. "github.com/coreos/etcd/mvcc/mvccpb"
  24. "github.com/coreos/etcd/pkg/types"
  25. "github.com/gogo/protobuf/proto"
  26. "golang.org/x/net/context"
  27. )
  28. const (
  29. // noTxn is an invalid txn ID.
  30. // To apply with independent Range, Put, Delete, you can pass noTxn
  31. // to apply functions instead of a valid txn ID.
  32. noTxn = -1
  33. warnApplyDuration = 100 * time.Millisecond
  34. )
  35. type applyResult struct {
  36. resp proto.Message
  37. err error
  38. // physc signals the physical effect of the request has completed in addition
  39. // to being logically reflected by the node. Currently only used for
  40. // Compaction requests.
  41. physc <-chan struct{}
  42. }
  43. // applierV3 is the interface for processing V3 raft messages
  44. type applierV3 interface {
  45. Apply(r *pb.InternalRaftRequest) *applyResult
  46. Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error)
  47. Range(txnID int64, r *pb.RangeRequest) (*pb.RangeResponse, error)
  48. DeleteRange(txnID int64, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error)
  49. Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error)
  50. Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, error)
  51. LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error)
  52. LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error)
  53. Alarm(*pb.AlarmRequest) (*pb.AlarmResponse, error)
  54. Authenticate(r *pb.InternalAuthenticateRequest) (*pb.AuthenticateResponse, error)
  55. AuthEnable() (*pb.AuthEnableResponse, error)
  56. AuthDisable() (*pb.AuthDisableResponse, error)
  57. UserAdd(ua *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error)
  58. UserDelete(ua *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error)
  59. UserChangePassword(ua *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error)
  60. UserGrantRole(ua *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error)
  61. UserGet(ua *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error)
  62. UserRevokeRole(ua *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error)
  63. RoleAdd(ua *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error)
  64. RoleGrantPermission(ua *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error)
  65. RoleGet(ua *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error)
  66. RoleRevokePermission(ua *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error)
  67. RoleDelete(ua *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error)
  68. UserList(ua *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error)
  69. RoleList(ua *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error)
  70. }
  71. type applierV3backend struct {
  72. s *EtcdServer
  73. }
  74. func (s *EtcdServer) newApplierV3() applierV3 {
  75. return newAuthApplierV3(
  76. s.AuthStore(),
  77. newQuotaApplierV3(s, &applierV3backend{s}),
  78. )
  79. }
  80. func (a *applierV3backend) Apply(r *pb.InternalRaftRequest) *applyResult {
  81. ar := &applyResult{}
  82. // call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
  83. switch {
  84. case r.Range != nil:
  85. ar.resp, ar.err = a.s.applyV3.Range(noTxn, r.Range)
  86. case r.Put != nil:
  87. ar.resp, ar.err = a.s.applyV3.Put(noTxn, r.Put)
  88. case r.DeleteRange != nil:
  89. ar.resp, ar.err = a.s.applyV3.DeleteRange(noTxn, r.DeleteRange)
  90. case r.Txn != nil:
  91. ar.resp, ar.err = a.s.applyV3.Txn(r.Txn)
  92. case r.Compaction != nil:
  93. ar.resp, ar.physc, ar.err = a.s.applyV3.Compaction(r.Compaction)
  94. case r.LeaseGrant != nil:
  95. ar.resp, ar.err = a.s.applyV3.LeaseGrant(r.LeaseGrant)
  96. case r.LeaseRevoke != nil:
  97. ar.resp, ar.err = a.s.applyV3.LeaseRevoke(r.LeaseRevoke)
  98. case r.Alarm != nil:
  99. ar.resp, ar.err = a.s.applyV3.Alarm(r.Alarm)
  100. case r.Authenticate != nil:
  101. ar.resp, ar.err = a.s.applyV3.Authenticate(r.Authenticate)
  102. case r.AuthEnable != nil:
  103. ar.resp, ar.err = a.s.applyV3.AuthEnable()
  104. case r.AuthDisable != nil:
  105. ar.resp, ar.err = a.s.applyV3.AuthDisable()
  106. case r.AuthUserAdd != nil:
  107. ar.resp, ar.err = a.s.applyV3.UserAdd(r.AuthUserAdd)
  108. case r.AuthUserDelete != nil:
  109. ar.resp, ar.err = a.s.applyV3.UserDelete(r.AuthUserDelete)
  110. case r.AuthUserChangePassword != nil:
  111. ar.resp, ar.err = a.s.applyV3.UserChangePassword(r.AuthUserChangePassword)
  112. case r.AuthUserGrantRole != nil:
  113. ar.resp, ar.err = a.s.applyV3.UserGrantRole(r.AuthUserGrantRole)
  114. case r.AuthUserGet != nil:
  115. ar.resp, ar.err = a.s.applyV3.UserGet(r.AuthUserGet)
  116. case r.AuthUserRevokeRole != nil:
  117. ar.resp, ar.err = a.s.applyV3.UserRevokeRole(r.AuthUserRevokeRole)
  118. case r.AuthRoleAdd != nil:
  119. ar.resp, ar.err = a.s.applyV3.RoleAdd(r.AuthRoleAdd)
  120. case r.AuthRoleGrantPermission != nil:
  121. ar.resp, ar.err = a.s.applyV3.RoleGrantPermission(r.AuthRoleGrantPermission)
  122. case r.AuthRoleGet != nil:
  123. ar.resp, ar.err = a.s.applyV3.RoleGet(r.AuthRoleGet)
  124. case r.AuthRoleRevokePermission != nil:
  125. ar.resp, ar.err = a.s.applyV3.RoleRevokePermission(r.AuthRoleRevokePermission)
  126. case r.AuthRoleDelete != nil:
  127. ar.resp, ar.err = a.s.applyV3.RoleDelete(r.AuthRoleDelete)
  128. case r.AuthUserList != nil:
  129. ar.resp, ar.err = a.s.applyV3.UserList(r.AuthUserList)
  130. case r.AuthRoleList != nil:
  131. ar.resp, ar.err = a.s.applyV3.RoleList(r.AuthRoleList)
  132. default:
  133. panic("not implemented")
  134. }
  135. return ar
  136. }
  137. func (a *applierV3backend) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  138. resp := &pb.PutResponse{}
  139. resp.Header = &pb.ResponseHeader{}
  140. var (
  141. rev int64
  142. err error
  143. )
  144. var rr *mvcc.RangeResult
  145. if p.PrevKv {
  146. if txnID != noTxn {
  147. rr, err = a.s.KV().TxnRange(txnID, p.Key, nil, mvcc.RangeOptions{})
  148. if err != nil {
  149. return nil, err
  150. }
  151. } else {
  152. rr, err = a.s.KV().Range(p.Key, nil, mvcc.RangeOptions{})
  153. if err != nil {
  154. return nil, err
  155. }
  156. }
  157. }
  158. if txnID != noTxn {
  159. rev, err = a.s.KV().TxnPut(txnID, p.Key, p.Value, lease.LeaseID(p.Lease))
  160. if err != nil {
  161. return nil, err
  162. }
  163. } else {
  164. leaseID := lease.LeaseID(p.Lease)
  165. if leaseID != lease.NoLease {
  166. if l := a.s.lessor.Lookup(leaseID); l == nil {
  167. return nil, lease.ErrLeaseNotFound
  168. }
  169. }
  170. rev = a.s.KV().Put(p.Key, p.Value, leaseID)
  171. }
  172. resp.Header.Revision = rev
  173. if rr != nil && len(rr.KVs) != 0 {
  174. resp.PrevKv = &rr.KVs[0]
  175. }
  176. return resp, nil
  177. }
  178. func (a *applierV3backend) DeleteRange(txnID int64, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
  179. resp := &pb.DeleteRangeResponse{}
  180. resp.Header = &pb.ResponseHeader{}
  181. var (
  182. n int64
  183. rev int64
  184. err error
  185. )
  186. if isGteRange(dr.RangeEnd) {
  187. dr.RangeEnd = []byte{}
  188. }
  189. var rr *mvcc.RangeResult
  190. if dr.PrevKv {
  191. if txnID != noTxn {
  192. rr, err = a.s.KV().TxnRange(txnID, dr.Key, dr.RangeEnd, mvcc.RangeOptions{})
  193. if err != nil {
  194. return nil, err
  195. }
  196. } else {
  197. rr, err = a.s.KV().Range(dr.Key, dr.RangeEnd, mvcc.RangeOptions{})
  198. if err != nil {
  199. return nil, err
  200. }
  201. }
  202. }
  203. if txnID != noTxn {
  204. n, rev, err = a.s.KV().TxnDeleteRange(txnID, dr.Key, dr.RangeEnd)
  205. if err != nil {
  206. return nil, err
  207. }
  208. } else {
  209. n, rev = a.s.KV().DeleteRange(dr.Key, dr.RangeEnd)
  210. }
  211. resp.Deleted = n
  212. if rr != nil {
  213. for i := range rr.KVs {
  214. resp.PrevKvs = append(resp.PrevKvs, &rr.KVs[i])
  215. }
  216. }
  217. resp.Header.Revision = rev
  218. return resp, nil
  219. }
  220. func (a *applierV3backend) Range(txnID int64, r *pb.RangeRequest) (*pb.RangeResponse, error) {
  221. resp := &pb.RangeResponse{}
  222. resp.Header = &pb.ResponseHeader{}
  223. var (
  224. rr *mvcc.RangeResult
  225. err error
  226. )
  227. if isGteRange(r.RangeEnd) {
  228. r.RangeEnd = []byte{}
  229. }
  230. limit := r.Limit
  231. if r.SortOrder != pb.RangeRequest_NONE ||
  232. r.MinModRevision != 0 || r.MaxModRevision != 0 ||
  233. r.MinCreateRevision != 0 || r.MaxCreateRevision != 0 {
  234. // fetch everything; sort and truncate afterwards
  235. limit = 0
  236. }
  237. if limit > 0 {
  238. // fetch one extra for 'more' flag
  239. limit = limit + 1
  240. }
  241. ro := mvcc.RangeOptions{
  242. Limit: limit,
  243. Rev: r.Revision,
  244. Count: r.CountOnly,
  245. }
  246. if txnID != noTxn {
  247. rr, err = a.s.KV().TxnRange(txnID, r.Key, r.RangeEnd, ro)
  248. if err != nil {
  249. return nil, err
  250. }
  251. } else {
  252. rr, err = a.s.KV().Range(r.Key, r.RangeEnd, ro)
  253. if err != nil {
  254. return nil, err
  255. }
  256. }
  257. if r.MaxModRevision != 0 {
  258. f := func(kv *mvccpb.KeyValue) bool { return kv.ModRevision > r.MaxModRevision }
  259. pruneKVs(rr, f)
  260. }
  261. if r.MinModRevision != 0 {
  262. f := func(kv *mvccpb.KeyValue) bool { return kv.ModRevision < r.MinModRevision }
  263. pruneKVs(rr, f)
  264. }
  265. if r.MaxCreateRevision != 0 {
  266. f := func(kv *mvccpb.KeyValue) bool { return kv.CreateRevision > r.MaxCreateRevision }
  267. pruneKVs(rr, f)
  268. }
  269. if r.MinCreateRevision != 0 {
  270. f := func(kv *mvccpb.KeyValue) bool { return kv.CreateRevision < r.MinCreateRevision }
  271. pruneKVs(rr, f)
  272. }
  273. sortOrder := r.SortOrder
  274. if r.SortTarget != pb.RangeRequest_KEY && sortOrder == pb.RangeRequest_NONE {
  275. // Since current mvcc.Range implementation returns results
  276. // sorted by keys in lexiographically ascending order,
  277. // sort ASCEND by default only when target is not 'KEY'
  278. sortOrder = pb.RangeRequest_ASCEND
  279. }
  280. if sortOrder != pb.RangeRequest_NONE {
  281. var sorter sort.Interface
  282. switch {
  283. case r.SortTarget == pb.RangeRequest_KEY:
  284. sorter = &kvSortByKey{&kvSort{rr.KVs}}
  285. case r.SortTarget == pb.RangeRequest_VERSION:
  286. sorter = &kvSortByVersion{&kvSort{rr.KVs}}
  287. case r.SortTarget == pb.RangeRequest_CREATE:
  288. sorter = &kvSortByCreate{&kvSort{rr.KVs}}
  289. case r.SortTarget == pb.RangeRequest_MOD:
  290. sorter = &kvSortByMod{&kvSort{rr.KVs}}
  291. case r.SortTarget == pb.RangeRequest_VALUE:
  292. sorter = &kvSortByValue{&kvSort{rr.KVs}}
  293. }
  294. switch {
  295. case sortOrder == pb.RangeRequest_ASCEND:
  296. sort.Sort(sorter)
  297. case sortOrder == pb.RangeRequest_DESCEND:
  298. sort.Sort(sort.Reverse(sorter))
  299. }
  300. }
  301. if r.Limit > 0 && len(rr.KVs) > int(r.Limit) {
  302. rr.KVs = rr.KVs[:r.Limit]
  303. resp.More = true
  304. }
  305. resp.Header.Revision = rr.Rev
  306. resp.Count = int64(rr.Count)
  307. for i := range rr.KVs {
  308. if r.KeysOnly {
  309. rr.KVs[i].Value = nil
  310. }
  311. resp.Kvs = append(resp.Kvs, &rr.KVs[i])
  312. }
  313. return resp, nil
  314. }
  315. func (a *applierV3backend) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
  316. ok := true
  317. for _, c := range rt.Compare {
  318. if _, ok = a.applyCompare(c); !ok {
  319. break
  320. }
  321. }
  322. var reqs []*pb.RequestOp
  323. if ok {
  324. reqs = rt.Success
  325. } else {
  326. reqs = rt.Failure
  327. }
  328. if err := a.checkRequestLeases(reqs); err != nil {
  329. return nil, err
  330. }
  331. if err := a.checkRequestRange(reqs); err != nil {
  332. return nil, err
  333. }
  334. // When executing the operations of txn, we need to hold the txn lock.
  335. // So the reader will not see any intermediate results.
  336. txnID := a.s.KV().TxnBegin()
  337. resps := make([]*pb.ResponseOp, len(reqs))
  338. for i := range reqs {
  339. resps[i] = a.applyUnion(txnID, reqs[i])
  340. }
  341. err := a.s.KV().TxnEnd(txnID)
  342. if err != nil {
  343. panic(fmt.Sprint("unexpected error when closing txn", txnID))
  344. }
  345. txnResp := &pb.TxnResponse{}
  346. txnResp.Header = &pb.ResponseHeader{}
  347. txnResp.Header.Revision = a.s.KV().Rev()
  348. txnResp.Responses = resps
  349. txnResp.Succeeded = ok
  350. return txnResp, nil
  351. }
  352. // applyCompare applies the compare request.
  353. // It returns the revision at which the comparison happens. If the comparison
  354. // succeeds, the it returns true. Otherwise it returns false.
  355. func (a *applierV3backend) applyCompare(c *pb.Compare) (int64, bool) {
  356. rr, err := a.s.KV().Range(c.Key, nil, mvcc.RangeOptions{})
  357. rev := rr.Rev
  358. if err != nil {
  359. if err == mvcc.ErrTxnIDMismatch {
  360. panic("unexpected txn ID mismatch error")
  361. }
  362. return rev, false
  363. }
  364. var ckv mvccpb.KeyValue
  365. if len(rr.KVs) != 0 {
  366. ckv = rr.KVs[0]
  367. } else {
  368. // Use the zero value of ckv normally. However...
  369. if c.Target == pb.Compare_VALUE {
  370. // Always fail if we're comparing a value on a key that doesn't exist.
  371. // We can treat non-existence as the empty set explicitly, such that
  372. // even a key with a value of length 0 bytes is still a real key
  373. // that was written that way
  374. return rev, false
  375. }
  376. }
  377. // -1 is less, 0 is equal, 1 is greater
  378. var result int
  379. switch c.Target {
  380. case pb.Compare_VALUE:
  381. tv, _ := c.TargetUnion.(*pb.Compare_Value)
  382. if tv != nil {
  383. result = bytes.Compare(ckv.Value, tv.Value)
  384. }
  385. case pb.Compare_CREATE:
  386. tv, _ := c.TargetUnion.(*pb.Compare_CreateRevision)
  387. if tv != nil {
  388. result = compareInt64(ckv.CreateRevision, tv.CreateRevision)
  389. }
  390. case pb.Compare_MOD:
  391. tv, _ := c.TargetUnion.(*pb.Compare_ModRevision)
  392. if tv != nil {
  393. result = compareInt64(ckv.ModRevision, tv.ModRevision)
  394. }
  395. case pb.Compare_VERSION:
  396. tv, _ := c.TargetUnion.(*pb.Compare_Version)
  397. if tv != nil {
  398. result = compareInt64(ckv.Version, tv.Version)
  399. }
  400. }
  401. switch c.Result {
  402. case pb.Compare_EQUAL:
  403. if result != 0 {
  404. return rev, false
  405. }
  406. case pb.Compare_NOT_EQUAL:
  407. if result == 0 {
  408. return rev, false
  409. }
  410. case pb.Compare_GREATER:
  411. if result != 1 {
  412. return rev, false
  413. }
  414. case pb.Compare_LESS:
  415. if result != -1 {
  416. return rev, false
  417. }
  418. }
  419. return rev, true
  420. }
  421. func (a *applierV3backend) applyUnion(txnID int64, union *pb.RequestOp) *pb.ResponseOp {
  422. switch tv := union.Request.(type) {
  423. case *pb.RequestOp_RequestRange:
  424. if tv.RequestRange != nil {
  425. resp, err := a.Range(txnID, tv.RequestRange)
  426. if err != nil {
  427. plog.Panicf("unexpected error during txn: %v", err)
  428. }
  429. return &pb.ResponseOp{Response: &pb.ResponseOp_ResponseRange{ResponseRange: resp}}
  430. }
  431. case *pb.RequestOp_RequestPut:
  432. if tv.RequestPut != nil {
  433. resp, err := a.Put(txnID, tv.RequestPut)
  434. if err != nil {
  435. plog.Panicf("unexpected error during txn: %v", err)
  436. }
  437. return &pb.ResponseOp{Response: &pb.ResponseOp_ResponsePut{ResponsePut: resp}}
  438. }
  439. case *pb.RequestOp_RequestDeleteRange:
  440. if tv.RequestDeleteRange != nil {
  441. resp, err := a.DeleteRange(txnID, tv.RequestDeleteRange)
  442. if err != nil {
  443. plog.Panicf("unexpected error during txn: %v", err)
  444. }
  445. return &pb.ResponseOp{Response: &pb.ResponseOp_ResponseDeleteRange{ResponseDeleteRange: resp}}
  446. }
  447. default:
  448. // empty union
  449. return nil
  450. }
  451. return nil
  452. }
  453. func (a *applierV3backend) Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, error) {
  454. resp := &pb.CompactionResponse{}
  455. resp.Header = &pb.ResponseHeader{}
  456. ch, err := a.s.KV().Compact(compaction.Revision)
  457. if err != nil {
  458. return nil, ch, err
  459. }
  460. // get the current revision. which key to get is not important.
  461. rr, _ := a.s.KV().Range([]byte("compaction"), nil, mvcc.RangeOptions{})
  462. resp.Header.Revision = rr.Rev
  463. return resp, ch, err
  464. }
  465. func (a *applierV3backend) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  466. l, err := a.s.lessor.Grant(lease.LeaseID(lc.ID), lc.TTL)
  467. resp := &pb.LeaseGrantResponse{}
  468. if err == nil {
  469. resp.ID = int64(l.ID)
  470. resp.TTL = l.TTL()
  471. resp.Header = &pb.ResponseHeader{Revision: a.s.KV().Rev()}
  472. }
  473. return resp, err
  474. }
  475. func (a *applierV3backend) LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) {
  476. err := a.s.lessor.Revoke(lease.LeaseID(lc.ID))
  477. return &pb.LeaseRevokeResponse{Header: &pb.ResponseHeader{Revision: a.s.KV().Rev()}}, err
  478. }
  479. func (a *applierV3backend) Alarm(ar *pb.AlarmRequest) (*pb.AlarmResponse, error) {
  480. resp := &pb.AlarmResponse{}
  481. oldCount := len(a.s.alarmStore.Get(ar.Alarm))
  482. switch ar.Action {
  483. case pb.AlarmRequest_GET:
  484. resp.Alarms = a.s.alarmStore.Get(ar.Alarm)
  485. case pb.AlarmRequest_ACTIVATE:
  486. m := a.s.alarmStore.Activate(types.ID(ar.MemberID), ar.Alarm)
  487. if m == nil {
  488. break
  489. }
  490. resp.Alarms = append(resp.Alarms, m)
  491. activated := oldCount == 0 && len(a.s.alarmStore.Get(m.Alarm)) == 1
  492. if !activated {
  493. break
  494. }
  495. switch m.Alarm {
  496. case pb.AlarmType_NOSPACE:
  497. plog.Warningf("alarm raised %+v", m)
  498. a.s.applyV3 = newApplierV3Capped(a)
  499. default:
  500. plog.Errorf("unimplemented alarm activation (%+v)", m)
  501. }
  502. case pb.AlarmRequest_DEACTIVATE:
  503. m := a.s.alarmStore.Deactivate(types.ID(ar.MemberID), ar.Alarm)
  504. if m == nil {
  505. break
  506. }
  507. resp.Alarms = append(resp.Alarms, m)
  508. deactivated := oldCount > 0 && len(a.s.alarmStore.Get(ar.Alarm)) == 0
  509. if !deactivated {
  510. break
  511. }
  512. switch m.Alarm {
  513. case pb.AlarmType_NOSPACE:
  514. plog.Infof("alarm disarmed %+v", ar)
  515. a.s.applyV3 = a.s.newApplierV3()
  516. default:
  517. plog.Errorf("unimplemented alarm deactivation (%+v)", m)
  518. }
  519. default:
  520. return nil, nil
  521. }
  522. return resp, nil
  523. }
  524. type applierV3Capped struct {
  525. applierV3
  526. q backendQuota
  527. }
  528. // newApplierV3Capped creates an applyV3 that will reject Puts and transactions
  529. // with Puts so that the number of keys in the store is capped.
  530. func newApplierV3Capped(base applierV3) applierV3 { return &applierV3Capped{applierV3: base} }
  531. func (a *applierV3Capped) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  532. return nil, ErrNoSpace
  533. }
  534. func (a *applierV3Capped) Txn(r *pb.TxnRequest) (*pb.TxnResponse, error) {
  535. if a.q.Cost(r) > 0 {
  536. return nil, ErrNoSpace
  537. }
  538. return a.applierV3.Txn(r)
  539. }
  540. func (a *applierV3Capped) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  541. return nil, ErrNoSpace
  542. }
  543. func (a *applierV3backend) AuthEnable() (*pb.AuthEnableResponse, error) {
  544. err := a.s.AuthStore().AuthEnable()
  545. if err != nil {
  546. return nil, err
  547. }
  548. return &pb.AuthEnableResponse{}, nil
  549. }
  550. func (a *applierV3backend) AuthDisable() (*pb.AuthDisableResponse, error) {
  551. a.s.AuthStore().AuthDisable()
  552. return &pb.AuthDisableResponse{}, nil
  553. }
  554. func (a *applierV3backend) Authenticate(r *pb.InternalAuthenticateRequest) (*pb.AuthenticateResponse, error) {
  555. ctx := context.WithValue(context.WithValue(context.TODO(), "index", a.s.consistIndex.ConsistentIndex()), "simpleToken", r.SimpleToken)
  556. return a.s.AuthStore().Authenticate(ctx, r.Name, r.Password)
  557. }
  558. func (a *applierV3backend) UserAdd(r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) {
  559. return a.s.AuthStore().UserAdd(r)
  560. }
  561. func (a *applierV3backend) UserDelete(r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) {
  562. return a.s.AuthStore().UserDelete(r)
  563. }
  564. func (a *applierV3backend) UserChangePassword(r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) {
  565. return a.s.AuthStore().UserChangePassword(r)
  566. }
  567. func (a *applierV3backend) UserGrantRole(r *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error) {
  568. return a.s.AuthStore().UserGrantRole(r)
  569. }
  570. func (a *applierV3backend) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) {
  571. return a.s.AuthStore().UserGet(r)
  572. }
  573. func (a *applierV3backend) UserRevokeRole(r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error) {
  574. return a.s.AuthStore().UserRevokeRole(r)
  575. }
  576. func (a *applierV3backend) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) {
  577. return a.s.AuthStore().RoleAdd(r)
  578. }
  579. func (a *applierV3backend) RoleGrantPermission(r *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error) {
  580. return a.s.AuthStore().RoleGrantPermission(r)
  581. }
  582. func (a *applierV3backend) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) {
  583. return a.s.AuthStore().RoleGet(r)
  584. }
  585. func (a *applierV3backend) RoleRevokePermission(r *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error) {
  586. return a.s.AuthStore().RoleRevokePermission(r)
  587. }
  588. func (a *applierV3backend) RoleDelete(r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error) {
  589. return a.s.AuthStore().RoleDelete(r)
  590. }
  591. func (a *applierV3backend) UserList(r *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error) {
  592. return a.s.AuthStore().UserList(r)
  593. }
  594. func (a *applierV3backend) RoleList(r *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) {
  595. return a.s.AuthStore().RoleList(r)
  596. }
  597. type quotaApplierV3 struct {
  598. applierV3
  599. q Quota
  600. }
  601. func newQuotaApplierV3(s *EtcdServer, app applierV3) applierV3 {
  602. return &quotaApplierV3{app, NewBackendQuota(s)}
  603. }
  604. func (a *quotaApplierV3) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  605. ok := a.q.Available(p)
  606. resp, err := a.applierV3.Put(txnID, p)
  607. if err == nil && !ok {
  608. err = ErrNoSpace
  609. }
  610. return resp, err
  611. }
  612. func (a *quotaApplierV3) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
  613. ok := a.q.Available(rt)
  614. resp, err := a.applierV3.Txn(rt)
  615. if err == nil && !ok {
  616. err = ErrNoSpace
  617. }
  618. return resp, err
  619. }
  620. func (a *quotaApplierV3) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  621. ok := a.q.Available(lc)
  622. resp, err := a.applierV3.LeaseGrant(lc)
  623. if err == nil && !ok {
  624. err = ErrNoSpace
  625. }
  626. return resp, err
  627. }
  628. type kvSort struct{ kvs []mvccpb.KeyValue }
  629. func (s *kvSort) Swap(i, j int) {
  630. t := s.kvs[i]
  631. s.kvs[i] = s.kvs[j]
  632. s.kvs[j] = t
  633. }
  634. func (s *kvSort) Len() int { return len(s.kvs) }
  635. type kvSortByKey struct{ *kvSort }
  636. func (s *kvSortByKey) Less(i, j int) bool {
  637. return bytes.Compare(s.kvs[i].Key, s.kvs[j].Key) < 0
  638. }
  639. type kvSortByVersion struct{ *kvSort }
  640. func (s *kvSortByVersion) Less(i, j int) bool {
  641. return (s.kvs[i].Version - s.kvs[j].Version) < 0
  642. }
  643. type kvSortByCreate struct{ *kvSort }
  644. func (s *kvSortByCreate) Less(i, j int) bool {
  645. return (s.kvs[i].CreateRevision - s.kvs[j].CreateRevision) < 0
  646. }
  647. type kvSortByMod struct{ *kvSort }
  648. func (s *kvSortByMod) Less(i, j int) bool {
  649. return (s.kvs[i].ModRevision - s.kvs[j].ModRevision) < 0
  650. }
  651. type kvSortByValue struct{ *kvSort }
  652. func (s *kvSortByValue) Less(i, j int) bool {
  653. return bytes.Compare(s.kvs[i].Value, s.kvs[j].Value) < 0
  654. }
  655. func (a *applierV3backend) checkRequestLeases(reqs []*pb.RequestOp) error {
  656. for _, requ := range reqs {
  657. tv, ok := requ.Request.(*pb.RequestOp_RequestPut)
  658. if !ok {
  659. continue
  660. }
  661. preq := tv.RequestPut
  662. if preq == nil || lease.LeaseID(preq.Lease) == lease.NoLease {
  663. continue
  664. }
  665. if l := a.s.lessor.Lookup(lease.LeaseID(preq.Lease)); l == nil {
  666. return lease.ErrLeaseNotFound
  667. }
  668. }
  669. return nil
  670. }
  671. func (a *applierV3backend) checkRequestRange(reqs []*pb.RequestOp) error {
  672. for _, requ := range reqs {
  673. tv, ok := requ.Request.(*pb.RequestOp_RequestRange)
  674. if !ok {
  675. continue
  676. }
  677. greq := tv.RequestRange
  678. if greq == nil || greq.Revision == 0 {
  679. continue
  680. }
  681. if greq.Revision > a.s.KV().Rev() {
  682. return mvcc.ErrFutureRev
  683. }
  684. if greq.Revision < a.s.KV().FirstRev() {
  685. return mvcc.ErrCompacted
  686. }
  687. }
  688. return nil
  689. }
  690. func compareInt64(a, b int64) int {
  691. switch {
  692. case a < b:
  693. return -1
  694. case a > b:
  695. return 1
  696. default:
  697. return 0
  698. }
  699. }
  700. // isGteRange determines if the range end is a >= range. This works around grpc
  701. // sending empty byte strings as nil; >= is encoded in the range end as '\0'.
  702. func isGteRange(rangeEnd []byte) bool {
  703. return len(rangeEnd) == 1 && rangeEnd[0] == 0
  704. }
  705. func noSideEffect(r *pb.InternalRaftRequest) bool {
  706. return r.Range != nil || r.AuthUserGet != nil || r.AuthRoleGet != nil
  707. }
  708. func removeNeedlessRangeReqs(txn *pb.TxnRequest) {
  709. f := func(ops []*pb.RequestOp) []*pb.RequestOp {
  710. j := 0
  711. for i := 0; i < len(ops); i++ {
  712. if _, ok := ops[i].Request.(*pb.RequestOp_RequestRange); ok {
  713. continue
  714. }
  715. ops[j] = ops[i]
  716. j++
  717. }
  718. return ops[:j]
  719. }
  720. txn.Success = f(txn.Success)
  721. txn.Failure = f(txn.Failure)
  722. }
  723. func pruneKVs(rr *mvcc.RangeResult, isPrunable func(*mvccpb.KeyValue) bool) {
  724. j := 0
  725. for i := range rr.KVs {
  726. rr.KVs[j] = rr.KVs[i]
  727. if !isPrunable(&rr.KVs[i]) {
  728. j++
  729. }
  730. }
  731. rr.KVs = rr.KVs[:j]
  732. }