apply.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685
  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. "github.com/coreos/etcd/auth"
  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. )
  27. const (
  28. // noTxn is an invalid txn ID.
  29. // To apply with independent Range, Put, Delete, you can pass noTxn
  30. // to apply functions instead of a valid txn ID.
  31. noTxn = -1
  32. )
  33. type applyResult struct {
  34. resp proto.Message
  35. err error
  36. // physc signals the physical effect of the request has completed in addition
  37. // to being logically reflected by the node. Currently only used for
  38. // Compaction requests.
  39. physc <-chan struct{}
  40. }
  41. // applierV3 is the interface for processing V3 raft messages
  42. type applierV3 interface {
  43. Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error)
  44. Range(txnID int64, r *pb.RangeRequest) (*pb.RangeResponse, error)
  45. DeleteRange(txnID int64, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error)
  46. Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error)
  47. Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, error)
  48. LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error)
  49. LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error)
  50. Alarm(*pb.AlarmRequest) (*pb.AlarmResponse, error)
  51. AuthEnable() (*pb.AuthEnableResponse, error)
  52. AuthDisable() (*pb.AuthDisableResponse, error)
  53. Authenticate(r *pb.AuthenticateRequest) (*pb.AuthenticateResponse, error)
  54. UserAdd(ua *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error)
  55. UserDelete(ua *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error)
  56. UserChangePassword(ua *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error)
  57. UserGrant(ua *pb.AuthUserGrantRequest) (*pb.AuthUserGrantResponse, error)
  58. UserGet(ua *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error)
  59. RoleAdd(ua *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error)
  60. RoleGrant(ua *pb.AuthRoleGrantRequest) (*pb.AuthRoleGrantResponse, error)
  61. RoleGet(ua *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error)
  62. }
  63. type applierV3backend struct {
  64. s *EtcdServer
  65. }
  66. func (s *EtcdServer) applyV3Request(r *pb.InternalRaftRequest) *applyResult {
  67. ar := &applyResult{}
  68. switch {
  69. case r.Range != nil:
  70. if s.AuthStore().IsRangePermitted(r.Header, string(r.Range.Key)) {
  71. ar.resp, ar.err = s.applyV3.Range(noTxn, r.Range)
  72. } else {
  73. ar.err = auth.ErrPermissionDenied
  74. }
  75. case r.Put != nil:
  76. if s.AuthStore().IsPutPermitted(r.Header, string(r.Put.Key)) {
  77. ar.resp, ar.err = s.applyV3.Put(noTxn, r.Put)
  78. } else {
  79. ar.err = auth.ErrPermissionDenied
  80. }
  81. case r.DeleteRange != nil:
  82. ar.resp, ar.err = s.applyV3.DeleteRange(noTxn, r.DeleteRange)
  83. case r.Txn != nil:
  84. ar.resp, ar.err = s.applyV3.Txn(r.Txn)
  85. case r.Compaction != nil:
  86. ar.resp, ar.physc, ar.err = s.applyV3.Compaction(r.Compaction)
  87. case r.LeaseGrant != nil:
  88. ar.resp, ar.err = s.applyV3.LeaseGrant(r.LeaseGrant)
  89. case r.LeaseRevoke != nil:
  90. ar.resp, ar.err = s.applyV3.LeaseRevoke(r.LeaseRevoke)
  91. case r.Alarm != nil:
  92. ar.resp, ar.err = s.applyV3.Alarm(r.Alarm)
  93. case r.AuthEnable != nil:
  94. ar.resp, ar.err = s.applyV3.AuthEnable()
  95. case r.AuthDisable != nil:
  96. ar.resp, ar.err = s.applyV3.AuthDisable()
  97. case r.Authenticate != nil:
  98. ar.resp, ar.err = s.applyV3.Authenticate(r.Authenticate)
  99. case r.AuthUserAdd != nil:
  100. ar.resp, ar.err = s.applyV3.UserAdd(r.AuthUserAdd)
  101. case r.AuthUserDelete != nil:
  102. ar.resp, ar.err = s.applyV3.UserDelete(r.AuthUserDelete)
  103. case r.AuthUserChangePassword != nil:
  104. ar.resp, ar.err = s.applyV3.UserChangePassword(r.AuthUserChangePassword)
  105. case r.AuthUserGrant != nil:
  106. ar.resp, ar.err = s.applyV3.UserGrant(r.AuthUserGrant)
  107. case r.AuthUserGet != nil:
  108. ar.resp, ar.err = s.applyV3.UserGet(r.AuthUserGet)
  109. case r.AuthRoleAdd != nil:
  110. ar.resp, ar.err = s.applyV3.RoleAdd(r.AuthRoleAdd)
  111. case r.AuthRoleGrant != nil:
  112. ar.resp, ar.err = s.applyV3.RoleGrant(r.AuthRoleGrant)
  113. case r.AuthRoleGet != nil:
  114. ar.resp, ar.err = s.applyV3.RoleGet(r.AuthRoleGet)
  115. default:
  116. panic("not implemented")
  117. }
  118. return ar
  119. }
  120. func (a *applierV3backend) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  121. resp := &pb.PutResponse{}
  122. resp.Header = &pb.ResponseHeader{}
  123. var (
  124. rev int64
  125. err error
  126. )
  127. if txnID != noTxn {
  128. rev, err = a.s.KV().TxnPut(txnID, p.Key, p.Value, lease.LeaseID(p.Lease))
  129. if err != nil {
  130. return nil, err
  131. }
  132. } else {
  133. leaseID := lease.LeaseID(p.Lease)
  134. if leaseID != lease.NoLease {
  135. if l := a.s.lessor.Lookup(leaseID); l == nil {
  136. return nil, lease.ErrLeaseNotFound
  137. }
  138. }
  139. rev = a.s.KV().Put(p.Key, p.Value, leaseID)
  140. }
  141. resp.Header.Revision = rev
  142. return resp, nil
  143. }
  144. func (a *applierV3backend) DeleteRange(txnID int64, dr *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
  145. resp := &pb.DeleteRangeResponse{}
  146. resp.Header = &pb.ResponseHeader{}
  147. var (
  148. n int64
  149. rev int64
  150. err error
  151. )
  152. if isGteRange(dr.RangeEnd) {
  153. dr.RangeEnd = []byte{}
  154. }
  155. if txnID != noTxn {
  156. n, rev, err = a.s.KV().TxnDeleteRange(txnID, dr.Key, dr.RangeEnd)
  157. if err != nil {
  158. return nil, err
  159. }
  160. } else {
  161. n, rev = a.s.KV().DeleteRange(dr.Key, dr.RangeEnd)
  162. }
  163. resp.Deleted = n
  164. resp.Header.Revision = rev
  165. return resp, nil
  166. }
  167. func (a *applierV3backend) Range(txnID int64, r *pb.RangeRequest) (*pb.RangeResponse, error) {
  168. resp := &pb.RangeResponse{}
  169. resp.Header = &pb.ResponseHeader{}
  170. var (
  171. kvs []mvccpb.KeyValue
  172. rev int64
  173. err error
  174. )
  175. if isGteRange(r.RangeEnd) {
  176. r.RangeEnd = []byte{}
  177. }
  178. limit := r.Limit
  179. if r.SortOrder != pb.RangeRequest_NONE {
  180. // fetch everything; sort and truncate afterwards
  181. limit = 0
  182. }
  183. if limit > 0 {
  184. // fetch one extra for 'more' flag
  185. limit = limit + 1
  186. }
  187. if txnID != noTxn {
  188. kvs, rev, err = a.s.KV().TxnRange(txnID, r.Key, r.RangeEnd, limit, r.Revision)
  189. if err != nil {
  190. return nil, err
  191. }
  192. } else {
  193. kvs, rev, err = a.s.KV().Range(r.Key, r.RangeEnd, limit, r.Revision)
  194. if err != nil {
  195. return nil, err
  196. }
  197. }
  198. if r.SortOrder != pb.RangeRequest_NONE {
  199. var sorter sort.Interface
  200. switch {
  201. case r.SortTarget == pb.RangeRequest_KEY:
  202. sorter = &kvSortByKey{&kvSort{kvs}}
  203. case r.SortTarget == pb.RangeRequest_VERSION:
  204. sorter = &kvSortByVersion{&kvSort{kvs}}
  205. case r.SortTarget == pb.RangeRequest_CREATE:
  206. sorter = &kvSortByCreate{&kvSort{kvs}}
  207. case r.SortTarget == pb.RangeRequest_MOD:
  208. sorter = &kvSortByMod{&kvSort{kvs}}
  209. case r.SortTarget == pb.RangeRequest_VALUE:
  210. sorter = &kvSortByValue{&kvSort{kvs}}
  211. }
  212. switch {
  213. case r.SortOrder == pb.RangeRequest_ASCEND:
  214. sort.Sort(sorter)
  215. case r.SortOrder == pb.RangeRequest_DESCEND:
  216. sort.Sort(sort.Reverse(sorter))
  217. }
  218. }
  219. if r.Limit > 0 && len(kvs) > int(r.Limit) {
  220. kvs = kvs[:r.Limit]
  221. resp.More = true
  222. }
  223. resp.Header.Revision = rev
  224. for i := range kvs {
  225. resp.Kvs = append(resp.Kvs, &kvs[i])
  226. }
  227. return resp, nil
  228. }
  229. func (a *applierV3backend) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
  230. ok := true
  231. for _, c := range rt.Compare {
  232. if _, ok = a.applyCompare(c); !ok {
  233. break
  234. }
  235. }
  236. var reqs []*pb.RequestUnion
  237. if ok {
  238. reqs = rt.Success
  239. } else {
  240. reqs = rt.Failure
  241. }
  242. if err := a.checkRequestLeases(reqs); err != nil {
  243. return nil, err
  244. }
  245. if err := a.checkRequestRange(reqs); err != nil {
  246. return nil, err
  247. }
  248. revision := a.s.KV().Rev()
  249. // When executing the operations of txn, we need to hold the txn lock.
  250. // So the reader will not see any intermediate results.
  251. txnID := a.s.KV().TxnBegin()
  252. defer func() {
  253. err := a.s.KV().TxnEnd(txnID)
  254. if err != nil {
  255. panic(fmt.Sprint("unexpected error when closing txn", txnID))
  256. }
  257. }()
  258. resps := make([]*pb.ResponseUnion, len(reqs))
  259. changedKV := false
  260. for i := range reqs {
  261. if reqs[i].GetRequestRange() == nil {
  262. changedKV = true
  263. }
  264. resps[i] = a.applyUnion(txnID, reqs[i])
  265. }
  266. if changedKV {
  267. revision += 1
  268. }
  269. txnResp := &pb.TxnResponse{}
  270. txnResp.Header = &pb.ResponseHeader{}
  271. txnResp.Header.Revision = revision
  272. txnResp.Responses = resps
  273. txnResp.Succeeded = ok
  274. return txnResp, nil
  275. }
  276. // applyCompare applies the compare request.
  277. // It returns the revision at which the comparison happens. If the comparison
  278. // succeeds, the it returns true. Otherwise it returns false.
  279. func (a *applierV3backend) applyCompare(c *pb.Compare) (int64, bool) {
  280. ckvs, rev, err := a.s.KV().Range(c.Key, nil, 1, 0)
  281. if err != nil {
  282. if err == mvcc.ErrTxnIDMismatch {
  283. panic("unexpected txn ID mismatch error")
  284. }
  285. return rev, false
  286. }
  287. var ckv mvccpb.KeyValue
  288. if len(ckvs) != 0 {
  289. ckv = ckvs[0]
  290. } else {
  291. // Use the zero value of ckv normally. However...
  292. if c.Target == pb.Compare_VALUE {
  293. // Always fail if we're comparing a value on a key that doesn't exist.
  294. // We can treat non-existence as the empty set explicitly, such that
  295. // even a key with a value of length 0 bytes is still a real key
  296. // that was written that way
  297. return rev, false
  298. }
  299. }
  300. // -1 is less, 0 is equal, 1 is greater
  301. var result int
  302. switch c.Target {
  303. case pb.Compare_VALUE:
  304. tv, _ := c.TargetUnion.(*pb.Compare_Value)
  305. if tv != nil {
  306. result = bytes.Compare(ckv.Value, tv.Value)
  307. }
  308. case pb.Compare_CREATE:
  309. tv, _ := c.TargetUnion.(*pb.Compare_CreateRevision)
  310. if tv != nil {
  311. result = compareInt64(ckv.CreateRevision, tv.CreateRevision)
  312. }
  313. case pb.Compare_MOD:
  314. tv, _ := c.TargetUnion.(*pb.Compare_ModRevision)
  315. if tv != nil {
  316. result = compareInt64(ckv.ModRevision, tv.ModRevision)
  317. }
  318. case pb.Compare_VERSION:
  319. tv, _ := c.TargetUnion.(*pb.Compare_Version)
  320. if tv != nil {
  321. result = compareInt64(ckv.Version, tv.Version)
  322. }
  323. }
  324. switch c.Result {
  325. case pb.Compare_EQUAL:
  326. if result != 0 {
  327. return rev, false
  328. }
  329. case pb.Compare_GREATER:
  330. if result != 1 {
  331. return rev, false
  332. }
  333. case pb.Compare_LESS:
  334. if result != -1 {
  335. return rev, false
  336. }
  337. }
  338. return rev, true
  339. }
  340. func (a *applierV3backend) applyUnion(txnID int64, union *pb.RequestUnion) *pb.ResponseUnion {
  341. switch tv := union.Request.(type) {
  342. case *pb.RequestUnion_RequestRange:
  343. if tv.RequestRange != nil {
  344. resp, err := a.Range(txnID, tv.RequestRange)
  345. if err != nil {
  346. panic("unexpected error during txn")
  347. }
  348. return &pb.ResponseUnion{Response: &pb.ResponseUnion_ResponseRange{ResponseRange: resp}}
  349. }
  350. case *pb.RequestUnion_RequestPut:
  351. if tv.RequestPut != nil {
  352. resp, err := a.Put(txnID, tv.RequestPut)
  353. if err != nil {
  354. panic("unexpected error during txn")
  355. }
  356. return &pb.ResponseUnion{Response: &pb.ResponseUnion_ResponsePut{ResponsePut: resp}}
  357. }
  358. case *pb.RequestUnion_RequestDeleteRange:
  359. if tv.RequestDeleteRange != nil {
  360. resp, err := a.DeleteRange(txnID, tv.RequestDeleteRange)
  361. if err != nil {
  362. panic("unexpected error during txn")
  363. }
  364. return &pb.ResponseUnion{Response: &pb.ResponseUnion_ResponseDeleteRange{ResponseDeleteRange: resp}}
  365. }
  366. default:
  367. // empty union
  368. return nil
  369. }
  370. return nil
  371. }
  372. func (a *applierV3backend) Compaction(compaction *pb.CompactionRequest) (*pb.CompactionResponse, <-chan struct{}, error) {
  373. resp := &pb.CompactionResponse{}
  374. resp.Header = &pb.ResponseHeader{}
  375. ch, err := a.s.KV().Compact(compaction.Revision)
  376. if err != nil {
  377. return nil, ch, err
  378. }
  379. // get the current revision. which key to get is not important.
  380. _, resp.Header.Revision, _ = a.s.KV().Range([]byte("compaction"), nil, 1, 0)
  381. return resp, ch, err
  382. }
  383. func (a *applierV3backend) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  384. l, err := a.s.lessor.Grant(lease.LeaseID(lc.ID), lc.TTL)
  385. resp := &pb.LeaseGrantResponse{}
  386. if err == nil {
  387. resp.ID = int64(l.ID)
  388. resp.TTL = l.TTL
  389. resp.Header = &pb.ResponseHeader{Revision: a.s.KV().Rev()}
  390. }
  391. return resp, err
  392. }
  393. func (a *applierV3backend) LeaseRevoke(lc *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) {
  394. err := a.s.lessor.Revoke(lease.LeaseID(lc.ID))
  395. return &pb.LeaseRevokeResponse{Header: &pb.ResponseHeader{Revision: a.s.KV().Rev()}}, err
  396. }
  397. func (a *applierV3backend) Alarm(ar *pb.AlarmRequest) (*pb.AlarmResponse, error) {
  398. resp := &pb.AlarmResponse{}
  399. oldCount := len(a.s.alarmStore.Get(ar.Alarm))
  400. switch ar.Action {
  401. case pb.AlarmRequest_GET:
  402. resp.Alarms = a.s.alarmStore.Get(ar.Alarm)
  403. case pb.AlarmRequest_ACTIVATE:
  404. m := a.s.alarmStore.Activate(types.ID(ar.MemberID), ar.Alarm)
  405. if m == nil {
  406. break
  407. }
  408. resp.Alarms = append(resp.Alarms, m)
  409. activated := oldCount == 0 && len(a.s.alarmStore.Get(m.Alarm)) == 1
  410. if !activated {
  411. break
  412. }
  413. switch m.Alarm {
  414. case pb.AlarmType_NOSPACE:
  415. plog.Warningf("alarm raised %+v", m)
  416. a.s.applyV3 = newApplierV3Capped(a)
  417. default:
  418. plog.Errorf("unimplemented alarm activation (%+v)", m)
  419. }
  420. case pb.AlarmRequest_DEACTIVATE:
  421. m := a.s.alarmStore.Deactivate(types.ID(ar.MemberID), ar.Alarm)
  422. if m == nil {
  423. break
  424. }
  425. resp.Alarms = append(resp.Alarms, m)
  426. deactivated := oldCount > 0 && len(a.s.alarmStore.Get(ar.Alarm)) == 0
  427. if !deactivated {
  428. break
  429. }
  430. switch m.Alarm {
  431. case pb.AlarmType_NOSPACE:
  432. plog.Infof("alarm disarmed %+v", ar)
  433. a.s.applyV3 = newQuotaApplierV3(a.s, &applierV3backend{a.s})
  434. default:
  435. plog.Errorf("unimplemented alarm deactivation (%+v)", m)
  436. }
  437. default:
  438. return nil, nil
  439. }
  440. return resp, nil
  441. }
  442. type applierV3Capped struct {
  443. applierV3
  444. q backendQuota
  445. }
  446. // newApplierV3Capped creates an applyV3 that will reject Puts and transactions
  447. // with Puts so that the number of keys in the store is capped.
  448. func newApplierV3Capped(base applierV3) applierV3 { return &applierV3Capped{applierV3: base} }
  449. func (a *applierV3Capped) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  450. return nil, ErrNoSpace
  451. }
  452. func (a *applierV3Capped) Txn(r *pb.TxnRequest) (*pb.TxnResponse, error) {
  453. if a.q.Cost(r) > 0 {
  454. return nil, ErrNoSpace
  455. }
  456. return a.applierV3.Txn(r)
  457. }
  458. func (a *applierV3Capped) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  459. return nil, ErrNoSpace
  460. }
  461. func (a *applierV3backend) AuthEnable() (*pb.AuthEnableResponse, error) {
  462. a.s.AuthStore().AuthEnable()
  463. return &pb.AuthEnableResponse{}, nil
  464. }
  465. func (a *applierV3backend) AuthDisable() (*pb.AuthDisableResponse, error) {
  466. a.s.AuthStore().AuthDisable()
  467. return &pb.AuthDisableResponse{}, nil
  468. }
  469. func (a *applierV3backend) Authenticate(r *pb.AuthenticateRequest) (*pb.AuthenticateResponse, error) {
  470. return a.s.AuthStore().Authenticate(r.Name, r.Password)
  471. }
  472. func (a *applierV3backend) UserAdd(r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) {
  473. return a.s.AuthStore().UserAdd(r)
  474. }
  475. func (a *applierV3backend) UserDelete(r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) {
  476. return a.s.AuthStore().UserDelete(r)
  477. }
  478. func (a *applierV3backend) UserChangePassword(r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) {
  479. return a.s.AuthStore().UserChangePassword(r)
  480. }
  481. func (a *applierV3backend) UserGrant(r *pb.AuthUserGrantRequest) (*pb.AuthUserGrantResponse, error) {
  482. return a.s.AuthStore().UserGrant(r)
  483. }
  484. func (a *applierV3backend) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) {
  485. return a.s.AuthStore().UserGet(r)
  486. }
  487. func (a *applierV3backend) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) {
  488. return a.s.AuthStore().RoleAdd(r)
  489. }
  490. func (a *applierV3backend) RoleGrant(r *pb.AuthRoleGrantRequest) (*pb.AuthRoleGrantResponse, error) {
  491. return a.s.AuthStore().RoleGrant(r)
  492. }
  493. func (a *applierV3backend) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) {
  494. return a.s.AuthStore().RoleGet(r)
  495. }
  496. type quotaApplierV3 struct {
  497. applierV3
  498. q Quota
  499. }
  500. func newQuotaApplierV3(s *EtcdServer, app applierV3) applierV3 {
  501. return &quotaApplierV3{app, NewBackendQuota(s)}
  502. }
  503. func (a *quotaApplierV3) Put(txnID int64, p *pb.PutRequest) (*pb.PutResponse, error) {
  504. ok := a.q.Available(p)
  505. resp, err := a.applierV3.Put(txnID, p)
  506. if err == nil && !ok {
  507. err = ErrNoSpace
  508. }
  509. return resp, err
  510. }
  511. func (a *quotaApplierV3) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
  512. ok := a.q.Available(rt)
  513. resp, err := a.applierV3.Txn(rt)
  514. if err == nil && !ok {
  515. err = ErrNoSpace
  516. }
  517. return resp, err
  518. }
  519. func (a *quotaApplierV3) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
  520. ok := a.q.Available(lc)
  521. resp, err := a.applierV3.LeaseGrant(lc)
  522. if err == nil && !ok {
  523. err = ErrNoSpace
  524. }
  525. return resp, err
  526. }
  527. type kvSort struct{ kvs []mvccpb.KeyValue }
  528. func (s *kvSort) Swap(i, j int) {
  529. t := s.kvs[i]
  530. s.kvs[i] = s.kvs[j]
  531. s.kvs[j] = t
  532. }
  533. func (s *kvSort) Len() int { return len(s.kvs) }
  534. type kvSortByKey struct{ *kvSort }
  535. func (s *kvSortByKey) Less(i, j int) bool {
  536. return bytes.Compare(s.kvs[i].Key, s.kvs[j].Key) < 0
  537. }
  538. type kvSortByVersion struct{ *kvSort }
  539. func (s *kvSortByVersion) Less(i, j int) bool {
  540. return (s.kvs[i].Version - s.kvs[j].Version) < 0
  541. }
  542. type kvSortByCreate struct{ *kvSort }
  543. func (s *kvSortByCreate) Less(i, j int) bool {
  544. return (s.kvs[i].CreateRevision - s.kvs[j].CreateRevision) < 0
  545. }
  546. type kvSortByMod struct{ *kvSort }
  547. func (s *kvSortByMod) Less(i, j int) bool {
  548. return (s.kvs[i].ModRevision - s.kvs[j].ModRevision) < 0
  549. }
  550. type kvSortByValue struct{ *kvSort }
  551. func (s *kvSortByValue) Less(i, j int) bool {
  552. return bytes.Compare(s.kvs[i].Value, s.kvs[j].Value) < 0
  553. }
  554. func (a *applierV3backend) checkRequestLeases(reqs []*pb.RequestUnion) error {
  555. for _, requ := range reqs {
  556. tv, ok := requ.Request.(*pb.RequestUnion_RequestPut)
  557. if !ok {
  558. continue
  559. }
  560. preq := tv.RequestPut
  561. if preq == nil || lease.LeaseID(preq.Lease) == lease.NoLease {
  562. continue
  563. }
  564. if l := a.s.lessor.Lookup(lease.LeaseID(preq.Lease)); l == nil {
  565. return lease.ErrLeaseNotFound
  566. }
  567. }
  568. return nil
  569. }
  570. func (a *applierV3backend) checkRequestRange(reqs []*pb.RequestUnion) error {
  571. for _, requ := range reqs {
  572. tv, ok := requ.Request.(*pb.RequestUnion_RequestRange)
  573. if !ok {
  574. continue
  575. }
  576. greq := tv.RequestRange
  577. if greq == nil || greq.Revision == 0 {
  578. continue
  579. }
  580. if greq.Revision > a.s.KV().Rev() {
  581. return mvcc.ErrFutureRev
  582. }
  583. if greq.Revision < a.s.KV().FirstRev() {
  584. return mvcc.ErrCompacted
  585. }
  586. }
  587. return nil
  588. }
  589. func compareInt64(a, b int64) int {
  590. switch {
  591. case a < b:
  592. return -1
  593. case a > b:
  594. return 1
  595. default:
  596. return 0
  597. }
  598. }
  599. // isGteRange determines if the range end is a >= range. This works around grpc
  600. // sending empty byte strings as nil; >= is encoded in the range end as '\0'.
  601. func isGteRange(rangeEnd []byte) bool {
  602. return len(rangeEnd) == 1 && rangeEnd[0] == 0
  603. }