apply.go 23 KB

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