store.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898
  1. // Copyright 2015 CoreOS, Inc.
  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 store
  15. import (
  16. "encoding/json"
  17. "fmt"
  18. "path"
  19. "strconv"
  20. "strings"
  21. "sync"
  22. "time"
  23. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/jonboulle/clockwork"
  24. etcdErr "github.com/coreos/etcd/error"
  25. "github.com/coreos/etcd/pkg/testutil"
  26. "github.com/coreos/etcd/pkg/types"
  27. )
  28. // The default version to set when the store is first initialized.
  29. const defaultVersion = 2
  30. var minExpireTime time.Time
  31. func init() {
  32. minExpireTime, _ = time.Parse(time.RFC3339, "2000-01-01T00:00:00Z")
  33. }
  34. type Store interface {
  35. Version() int
  36. Index() uint64
  37. Get(nodePath string, recursive, sorted bool) (*Event, error)
  38. Set(nodePath string, dir bool, value string, expireOpts TTLOptionSet) (*Event, error)
  39. Update(nodePath string, newValue string, expireOpts TTLOptionSet) (*Event, error)
  40. Create(nodePath string, dir bool, value string, unique bool,
  41. expireOpts TTLOptionSet) (*Event, error)
  42. CompareAndSwap(nodePath string, prevValue string, prevIndex uint64,
  43. value string, expireOpts TTLOptionSet) (*Event, error)
  44. Delete(nodePath string, dir, recursive bool) (*Event, error)
  45. CompareAndDelete(nodePath string, prevValue string, prevIndex uint64) (*Event, error)
  46. Watch(prefix string, recursive, stream bool, sinceIndex uint64) (Watcher, error)
  47. Save() ([]byte, error)
  48. Recovery(state []byte) error
  49. Clone() Store
  50. SaveNoCopy() ([]byte, error)
  51. JsonStats() []byte
  52. DeleteExpiredKeys(cutoff time.Time)
  53. }
  54. type TTLOptionSet struct {
  55. ExpireTime time.Time
  56. Refresh bool
  57. }
  58. type store struct {
  59. Root *node
  60. WatcherHub *watcherHub
  61. CurrentIndex uint64
  62. Stats *Stats
  63. CurrentVersion int
  64. ttlKeyHeap *ttlKeyHeap // need to recovery manually
  65. worldLock sync.RWMutex // stop the world lock
  66. clock clockwork.Clock
  67. readonlySet types.Set
  68. }
  69. // The given namespaces will be created as initial directories in the returned store.
  70. func New(namespaces ...string) Store {
  71. s := newStore(namespaces...)
  72. s.clock = clockwork.NewRealClock()
  73. return s
  74. }
  75. func newStore(namespaces ...string) *store {
  76. s := new(store)
  77. s.CurrentVersion = defaultVersion
  78. s.Root = newDir(s, "/", s.CurrentIndex, nil, Permanent)
  79. for _, namespace := range namespaces {
  80. s.Root.Add(newDir(s, namespace, s.CurrentIndex, s.Root, Permanent))
  81. }
  82. s.Stats = newStats()
  83. s.WatcherHub = newWatchHub(1000)
  84. s.ttlKeyHeap = newTtlKeyHeap()
  85. s.readonlySet = types.NewUnsafeSet(append(namespaces, "/")...)
  86. return s
  87. }
  88. // Version retrieves current version of the store.
  89. func (s *store) Version() int {
  90. return s.CurrentVersion
  91. }
  92. // Retrieves current of the store
  93. func (s *store) Index() uint64 {
  94. s.worldLock.RLock()
  95. defer s.worldLock.RUnlock()
  96. return s.CurrentIndex
  97. }
  98. // Get returns a get event.
  99. // If recursive is true, it will return all the content under the node path.
  100. // If sorted is true, it will sort the content by keys.
  101. func (s *store) Get(nodePath string, recursive, sorted bool) (*Event, error) {
  102. var err *etcdErr.Error
  103. s.worldLock.Lock()
  104. defer s.worldLock.Unlock()
  105. defer func() {
  106. if err == nil {
  107. s.Stats.Inc(GetSuccess)
  108. if recursive {
  109. reportReadSuccess(GetRecursive)
  110. } else {
  111. reportReadSuccess(Get)
  112. }
  113. return
  114. }
  115. s.Stats.Inc(GetFail)
  116. if recursive {
  117. reportReadFailure(GetRecursive)
  118. } else {
  119. reportReadFailure(Get)
  120. }
  121. }()
  122. nodePath = path.Clean(path.Join("/", nodePath))
  123. n, err := s.internalGet(nodePath)
  124. if err != nil {
  125. return nil, err
  126. }
  127. e := newEvent(Get, nodePath, n.ModifiedIndex, n.CreatedIndex)
  128. e.EtcdIndex = s.CurrentIndex
  129. e.Node.loadInternalNode(n, recursive, sorted, s.clock)
  130. return e, nil
  131. }
  132. // Create creates the node at nodePath. Create will help to create intermediate directories with no ttl.
  133. // If the node has already existed, create will fail.
  134. // If any node on the path is a file, create will fail.
  135. func (s *store) Create(nodePath string, dir bool, value string, unique bool, expireOpts TTLOptionSet) (*Event, error) {
  136. var err *etcdErr.Error
  137. s.worldLock.Lock()
  138. defer s.worldLock.Unlock()
  139. defer func() {
  140. if err == nil {
  141. s.Stats.Inc(CreateSuccess)
  142. reportWriteSuccess(Create)
  143. return
  144. }
  145. s.Stats.Inc(CreateFail)
  146. reportWriteFailure(Create)
  147. }()
  148. e, err := s.internalCreate(nodePath, dir, value, unique, false, expireOpts.ExpireTime, Create)
  149. if err != nil {
  150. return nil, err
  151. }
  152. e.EtcdIndex = s.CurrentIndex
  153. s.WatcherHub.notify(e)
  154. return e, nil
  155. }
  156. // Set creates or replace the node at nodePath.
  157. func (s *store) Set(nodePath string, dir bool, value string, expireOpts TTLOptionSet) (*Event, error) {
  158. var err *etcdErr.Error
  159. s.worldLock.Lock()
  160. defer s.worldLock.Unlock()
  161. defer func() {
  162. if err == nil {
  163. s.Stats.Inc(SetSuccess)
  164. reportWriteSuccess(Set)
  165. return
  166. }
  167. s.Stats.Inc(SetFail)
  168. reportWriteFailure(Set)
  169. }()
  170. // Get prevNode value
  171. n, getErr := s.internalGet(nodePath)
  172. if getErr != nil && getErr.ErrorCode != etcdErr.EcodeKeyNotFound {
  173. err = getErr
  174. return nil, err
  175. }
  176. if expireOpts.Refresh {
  177. if getErr != nil {
  178. err = getErr
  179. return nil, err
  180. } else {
  181. value = n.Value
  182. }
  183. }
  184. // Set new value
  185. e, err := s.internalCreate(nodePath, dir, value, false, true, expireOpts.ExpireTime, Set)
  186. if err != nil {
  187. return nil, err
  188. }
  189. e.EtcdIndex = s.CurrentIndex
  190. // Put prevNode into event
  191. if getErr == nil {
  192. prev := newEvent(Get, nodePath, n.ModifiedIndex, n.CreatedIndex)
  193. prev.Node.loadInternalNode(n, false, false, s.clock)
  194. e.PrevNode = prev.Node
  195. }
  196. if !expireOpts.Refresh {
  197. s.WatcherHub.notify(e)
  198. }
  199. return e, nil
  200. }
  201. // returns user-readable cause of failed comparison
  202. func getCompareFailCause(n *node, which int, prevValue string, prevIndex uint64) string {
  203. switch which {
  204. case CompareIndexNotMatch:
  205. return fmt.Sprintf("[%v != %v]", prevIndex, n.ModifiedIndex)
  206. case CompareValueNotMatch:
  207. return fmt.Sprintf("[%v != %v]", prevValue, n.Value)
  208. default:
  209. return fmt.Sprintf("[%v != %v] [%v != %v]", prevValue, n.Value, prevIndex, n.ModifiedIndex)
  210. }
  211. }
  212. func (s *store) CompareAndSwap(nodePath string, prevValue string, prevIndex uint64,
  213. value string, expireOpts TTLOptionSet) (*Event, error) {
  214. var err *etcdErr.Error
  215. s.worldLock.Lock()
  216. defer s.worldLock.Unlock()
  217. defer func() {
  218. if err == nil {
  219. s.Stats.Inc(CompareAndSwapSuccess)
  220. reportWriteSuccess(CompareAndSwap)
  221. return
  222. }
  223. s.Stats.Inc(CompareAndSwapFail)
  224. reportWriteFailure(CompareAndSwap)
  225. }()
  226. nodePath = path.Clean(path.Join("/", nodePath))
  227. // we do not allow the user to change "/"
  228. if s.readonlySet.Contains(nodePath) {
  229. return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", s.CurrentIndex)
  230. }
  231. n, err := s.internalGet(nodePath)
  232. if err != nil {
  233. return nil, err
  234. }
  235. if n.IsDir() { // can only compare and swap file
  236. err = etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, s.CurrentIndex)
  237. return nil, err
  238. }
  239. // If both of the prevValue and prevIndex are given, we will test both of them.
  240. // Command will be executed, only if both of the tests are successful.
  241. if ok, which := n.Compare(prevValue, prevIndex); !ok {
  242. cause := getCompareFailCause(n, which, prevValue, prevIndex)
  243. err = etcdErr.NewError(etcdErr.EcodeTestFailed, cause, s.CurrentIndex)
  244. return nil, err
  245. }
  246. // update etcd index
  247. s.CurrentIndex++
  248. e := newEvent(CompareAndSwap, nodePath, s.CurrentIndex, n.CreatedIndex)
  249. e.EtcdIndex = s.CurrentIndex
  250. e.PrevNode = n.Repr(false, false, s.clock)
  251. eNode := e.Node
  252. // if test succeed, write the value
  253. n.Write(value, s.CurrentIndex)
  254. n.UpdateTTL(expireOpts.ExpireTime)
  255. // copy the value for safety
  256. valueCopy := value
  257. eNode.Value = &valueCopy
  258. eNode.Expiration, eNode.TTL = n.expirationAndTTL(s.clock)
  259. if !expireOpts.Refresh {
  260. s.WatcherHub.notify(e)
  261. }
  262. return e, nil
  263. }
  264. // Delete deletes the node at the given path.
  265. // If the node is a directory, recursive must be true to delete it.
  266. func (s *store) Delete(nodePath string, dir, recursive bool) (*Event, error) {
  267. var err *etcdErr.Error
  268. s.worldLock.Lock()
  269. defer s.worldLock.Unlock()
  270. defer func() {
  271. if err == nil {
  272. s.Stats.Inc(DeleteSuccess)
  273. reportWriteSuccess(Delete)
  274. return
  275. }
  276. s.Stats.Inc(DeleteFail)
  277. reportWriteFailure(Delete)
  278. }()
  279. nodePath = path.Clean(path.Join("/", nodePath))
  280. // we do not allow the user to change "/"
  281. if s.readonlySet.Contains(nodePath) {
  282. return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", s.CurrentIndex)
  283. }
  284. // recursive implies dir
  285. if recursive == true {
  286. dir = true
  287. }
  288. n, err := s.internalGet(nodePath)
  289. if err != nil { // if the node does not exist, return error
  290. return nil, err
  291. }
  292. nextIndex := s.CurrentIndex + 1
  293. e := newEvent(Delete, nodePath, nextIndex, n.CreatedIndex)
  294. e.EtcdIndex = nextIndex
  295. e.PrevNode = n.Repr(false, false, s.clock)
  296. eNode := e.Node
  297. if n.IsDir() {
  298. eNode.Dir = true
  299. }
  300. callback := func(path string) { // notify function
  301. // notify the watchers with deleted set true
  302. s.WatcherHub.notifyWatchers(e, path, true)
  303. }
  304. err = n.Remove(dir, recursive, callback)
  305. if err != nil {
  306. return nil, err
  307. }
  308. // update etcd index
  309. s.CurrentIndex++
  310. s.WatcherHub.notify(e)
  311. return e, nil
  312. }
  313. func (s *store) CompareAndDelete(nodePath string, prevValue string, prevIndex uint64) (*Event, error) {
  314. var err *etcdErr.Error
  315. s.worldLock.Lock()
  316. defer s.worldLock.Unlock()
  317. defer func() {
  318. if err == nil {
  319. s.Stats.Inc(CompareAndDeleteSuccess)
  320. reportWriteSuccess(CompareAndDelete)
  321. return
  322. }
  323. s.Stats.Inc(CompareAndDeleteFail)
  324. reportWriteFailure(CompareAndDelete)
  325. }()
  326. nodePath = path.Clean(path.Join("/", nodePath))
  327. n, err := s.internalGet(nodePath)
  328. if err != nil { // if the node does not exist, return error
  329. return nil, err
  330. }
  331. if n.IsDir() { // can only compare and delete file
  332. return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, s.CurrentIndex)
  333. }
  334. // If both of the prevValue and prevIndex are given, we will test both of them.
  335. // Command will be executed, only if both of the tests are successful.
  336. if ok, which := n.Compare(prevValue, prevIndex); !ok {
  337. cause := getCompareFailCause(n, which, prevValue, prevIndex)
  338. return nil, etcdErr.NewError(etcdErr.EcodeTestFailed, cause, s.CurrentIndex)
  339. }
  340. // update etcd index
  341. s.CurrentIndex++
  342. e := newEvent(CompareAndDelete, nodePath, s.CurrentIndex, n.CreatedIndex)
  343. e.EtcdIndex = s.CurrentIndex
  344. e.PrevNode = n.Repr(false, false, s.clock)
  345. callback := func(path string) { // notify function
  346. // notify the watchers with deleted set true
  347. s.WatcherHub.notifyWatchers(e, path, true)
  348. }
  349. err = n.Remove(false, false, callback)
  350. if err != nil {
  351. return nil, err
  352. }
  353. s.WatcherHub.notify(e)
  354. return e, nil
  355. }
  356. func (s *store) Watch(key string, recursive, stream bool, sinceIndex uint64) (Watcher, error) {
  357. s.worldLock.RLock()
  358. defer s.worldLock.RUnlock()
  359. key = path.Clean(path.Join("/", key))
  360. if sinceIndex == 0 {
  361. sinceIndex = s.CurrentIndex + 1
  362. }
  363. // WatcherHub does not know about the current index, so we need to pass it in
  364. w, err := s.WatcherHub.watch(key, recursive, stream, sinceIndex, s.CurrentIndex)
  365. if err != nil {
  366. return nil, err
  367. }
  368. return w, nil
  369. }
  370. // walk walks all the nodePath and apply the walkFunc on each directory
  371. func (s *store) walk(nodePath string, walkFunc func(prev *node, component string) (*node, *etcdErr.Error)) (*node, *etcdErr.Error) {
  372. components := strings.Split(nodePath, "/")
  373. curr := s.Root
  374. var err *etcdErr.Error
  375. for i := 1; i < len(components); i++ {
  376. if len(components[i]) == 0 { // ignore empty string
  377. return curr, nil
  378. }
  379. curr, err = walkFunc(curr, components[i])
  380. if err != nil {
  381. return nil, err
  382. }
  383. }
  384. return curr, nil
  385. }
  386. // Update updates the value/ttl of the node.
  387. // If the node is a file, the value and the ttl can be updated.
  388. // If the node is a directory, only the ttl can be updated.
  389. func (s *store) Update(nodePath string, newValue string, expireOpts TTLOptionSet) (*Event, error) {
  390. var err *etcdErr.Error
  391. s.worldLock.Lock()
  392. defer s.worldLock.Unlock()
  393. defer func() {
  394. if err == nil {
  395. s.Stats.Inc(UpdateSuccess)
  396. reportWriteSuccess(Update)
  397. return
  398. }
  399. s.Stats.Inc(UpdateFail)
  400. reportWriteFailure(Update)
  401. }()
  402. nodePath = path.Clean(path.Join("/", nodePath))
  403. // we do not allow the user to change "/"
  404. if s.readonlySet.Contains(nodePath) {
  405. return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", s.CurrentIndex)
  406. }
  407. currIndex, nextIndex := s.CurrentIndex, s.CurrentIndex+1
  408. n, err := s.internalGet(nodePath)
  409. if err != nil { // if the node does not exist, return error
  410. return nil, err
  411. }
  412. if n.IsDir() && len(newValue) != 0 {
  413. // if the node is a directory, we cannot update value to non-empty
  414. return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, currIndex)
  415. }
  416. if expireOpts.Refresh {
  417. newValue = n.Value
  418. }
  419. e := newEvent(Update, nodePath, nextIndex, n.CreatedIndex)
  420. e.EtcdIndex = nextIndex
  421. e.PrevNode = n.Repr(false, false, s.clock)
  422. eNode := e.Node
  423. n.Write(newValue, nextIndex)
  424. if n.IsDir() {
  425. eNode.Dir = true
  426. } else {
  427. // copy the value for safety
  428. newValueCopy := newValue
  429. eNode.Value = &newValueCopy
  430. }
  431. // update ttl
  432. n.UpdateTTL(expireOpts.ExpireTime)
  433. eNode.Expiration, eNode.TTL = n.expirationAndTTL(s.clock)
  434. if !expireOpts.Refresh {
  435. s.WatcherHub.notify(e)
  436. }
  437. s.CurrentIndex = nextIndex
  438. return e, nil
  439. }
  440. func (s *store) internalCreate(nodePath string, dir bool, value string, unique, replace bool,
  441. expireTime time.Time, action string) (*Event, *etcdErr.Error) {
  442. currIndex, nextIndex := s.CurrentIndex, s.CurrentIndex+1
  443. if unique { // append unique item under the node path
  444. nodePath += "/" + fmt.Sprintf("%020s", strconv.FormatUint(nextIndex, 10))
  445. }
  446. nodePath = path.Clean(path.Join("/", nodePath))
  447. // we do not allow the user to change "/"
  448. if s.readonlySet.Contains(nodePath) {
  449. return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, "/", currIndex)
  450. }
  451. // Assume expire times that are way in the past are
  452. // This can occur when the time is serialized to JS
  453. if expireTime.Before(minExpireTime) {
  454. expireTime = Permanent
  455. }
  456. dirName, nodeName := path.Split(nodePath)
  457. // walk through the nodePath, create dirs and get the last directory node
  458. d, err := s.walk(dirName, s.checkDir)
  459. if err != nil {
  460. s.Stats.Inc(SetFail)
  461. reportWriteFailure(action)
  462. err.Index = currIndex
  463. return nil, err
  464. }
  465. e := newEvent(action, nodePath, nextIndex, nextIndex)
  466. eNode := e.Node
  467. n, _ := d.GetChild(nodeName)
  468. // force will try to replace an existing file
  469. if n != nil {
  470. if replace {
  471. if n.IsDir() {
  472. return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, currIndex)
  473. }
  474. e.PrevNode = n.Repr(false, false, s.clock)
  475. n.Remove(false, false, nil)
  476. } else {
  477. return nil, etcdErr.NewError(etcdErr.EcodeNodeExist, nodePath, currIndex)
  478. }
  479. }
  480. if !dir { // create file
  481. // copy the value for safety
  482. valueCopy := value
  483. eNode.Value = &valueCopy
  484. n = newKV(s, nodePath, value, nextIndex, d, expireTime)
  485. } else { // create directory
  486. eNode.Dir = true
  487. n = newDir(s, nodePath, nextIndex, d, expireTime)
  488. }
  489. // we are sure d is a directory and does not have the children with name n.Name
  490. d.Add(n)
  491. // node with TTL
  492. if !n.IsPermanent() {
  493. s.ttlKeyHeap.push(n)
  494. eNode.Expiration, eNode.TTL = n.expirationAndTTL(s.clock)
  495. }
  496. s.CurrentIndex = nextIndex
  497. return e, nil
  498. }
  499. // InternalGet gets the node of the given nodePath.
  500. func (s *store) internalGet(nodePath string) (*node, *etcdErr.Error) {
  501. nodePath = path.Clean(path.Join("/", nodePath))
  502. walkFunc := func(parent *node, name string) (*node, *etcdErr.Error) {
  503. if !parent.IsDir() {
  504. err := etcdErr.NewError(etcdErr.EcodeNotDir, parent.Path, s.CurrentIndex)
  505. return nil, err
  506. }
  507. child, ok := parent.Children[name]
  508. if ok {
  509. return child, nil
  510. }
  511. return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, path.Join(parent.Path, name), s.CurrentIndex)
  512. }
  513. f, err := s.walk(nodePath, walkFunc)
  514. if err != nil {
  515. return nil, err
  516. }
  517. return f, nil
  518. }
  519. // DeleteExpiredKeys will delete all expired keys
  520. func (s *store) DeleteExpiredKeys(cutoff time.Time) {
  521. s.worldLock.Lock()
  522. defer s.worldLock.Unlock()
  523. for {
  524. node := s.ttlKeyHeap.top()
  525. if node == nil || node.ExpireTime.After(cutoff) {
  526. break
  527. }
  528. s.CurrentIndex++
  529. e := newEvent(Expire, node.Path, s.CurrentIndex, node.CreatedIndex)
  530. e.EtcdIndex = s.CurrentIndex
  531. e.PrevNode = node.Repr(false, false, s.clock)
  532. callback := func(path string) { // notify function
  533. // notify the watchers with deleted set true
  534. s.WatcherHub.notifyWatchers(e, path, true)
  535. }
  536. s.ttlKeyHeap.pop()
  537. node.Remove(true, true, callback)
  538. reportExpiredKey()
  539. s.Stats.Inc(ExpireCount)
  540. s.WatcherHub.notify(e)
  541. }
  542. }
  543. // checkDir will check whether the component is a directory under parent node.
  544. // If it is a directory, this function will return the pointer to that node.
  545. // If it does not exist, this function will create a new directory and return the pointer to that node.
  546. // If it is a file, this function will return error.
  547. func (s *store) checkDir(parent *node, dirName string) (*node, *etcdErr.Error) {
  548. node, ok := parent.Children[dirName]
  549. if ok {
  550. if node.IsDir() {
  551. return node, nil
  552. }
  553. return nil, etcdErr.NewError(etcdErr.EcodeNotDir, node.Path, s.CurrentIndex)
  554. }
  555. n := newDir(s, path.Join(parent.Path, dirName), s.CurrentIndex+1, parent, Permanent)
  556. parent.Children[dirName] = n
  557. return n, nil
  558. }
  559. // Save saves the static state of the store system.
  560. // It will not be able to save the state of watchers.
  561. // It will not save the parent field of the node. Or there will
  562. // be cyclic dependencies issue for the json package.
  563. func (s *store) Save() ([]byte, error) {
  564. b, err := json.Marshal(s.Clone())
  565. if err != nil {
  566. return nil, err
  567. }
  568. return b, nil
  569. }
  570. func (s *store) SaveNoCopy() ([]byte, error) {
  571. b, err := json.Marshal(s)
  572. if err != nil {
  573. return nil, err
  574. }
  575. return b, nil
  576. }
  577. func (s *store) Clone() Store {
  578. s.worldLock.Lock()
  579. clonedStore := newStore()
  580. clonedStore.CurrentIndex = s.CurrentIndex
  581. clonedStore.Root = s.Root.Clone()
  582. clonedStore.WatcherHub = s.WatcherHub.clone()
  583. clonedStore.Stats = s.Stats.clone()
  584. clonedStore.CurrentVersion = s.CurrentVersion
  585. s.worldLock.Unlock()
  586. return clonedStore
  587. }
  588. // Recovery recovers the store system from a static state
  589. // It needs to recover the parent field of the nodes.
  590. // It needs to delete the expired nodes since the saved time and also
  591. // needs to create monitoring go routines.
  592. func (s *store) Recovery(state []byte) error {
  593. s.worldLock.Lock()
  594. defer s.worldLock.Unlock()
  595. err := json.Unmarshal(state, s)
  596. if err != nil {
  597. return err
  598. }
  599. s.ttlKeyHeap = newTtlKeyHeap()
  600. s.Root.recoverAndclean()
  601. return nil
  602. }
  603. func (s *store) JsonStats() []byte {
  604. s.Stats.Watchers = uint64(s.WatcherHub.count)
  605. return s.Stats.toJson()
  606. }
  607. // StoreRecorder provides a Store interface with a testutil.Recorder
  608. type StoreRecorder struct {
  609. Store
  610. testutil.Recorder
  611. }
  612. // storeRecorder records all the methods it receives.
  613. // storeRecorder DOES NOT work as a actual store.
  614. // It always returns invalid empty response and no error.
  615. type storeRecorder struct {
  616. Store
  617. testutil.Recorder
  618. }
  619. func NewNop() Store { return &storeRecorder{Recorder: &testutil.RecorderBuffered{}} }
  620. func NewRecorder() *StoreRecorder {
  621. sr := &storeRecorder{Recorder: &testutil.RecorderBuffered{}}
  622. return &StoreRecorder{Store: sr, Recorder: sr.Recorder}
  623. }
  624. func NewRecorderStream() *StoreRecorder {
  625. sr := &storeRecorder{Recorder: testutil.NewRecorderStream()}
  626. return &StoreRecorder{Store: sr, Recorder: sr.Recorder}
  627. }
  628. func (s *storeRecorder) Version() int { return 0 }
  629. func (s *storeRecorder) Index() uint64 { return 0 }
  630. func (s *storeRecorder) Get(path string, recursive, sorted bool) (*Event, error) {
  631. s.Record(testutil.Action{
  632. Name: "Get",
  633. Params: []interface{}{path, recursive, sorted},
  634. })
  635. return &Event{}, nil
  636. }
  637. func (s *storeRecorder) Set(path string, dir bool, val string, expireOpts TTLOptionSet) (*Event, error) {
  638. s.Record(testutil.Action{
  639. Name: "Set",
  640. Params: []interface{}{path, dir, val, expireOpts},
  641. })
  642. return &Event{}, nil
  643. }
  644. func (s *storeRecorder) Update(path, val string, expireOpts TTLOptionSet) (*Event, error) {
  645. s.Record(testutil.Action{
  646. Name: "Update",
  647. Params: []interface{}{path, val, expireOpts},
  648. })
  649. return &Event{}, nil
  650. }
  651. func (s *storeRecorder) Create(path string, dir bool, val string, uniq bool, expireOpts TTLOptionSet) (*Event, error) {
  652. s.Record(testutil.Action{
  653. Name: "Create",
  654. Params: []interface{}{path, dir, val, uniq, expireOpts},
  655. })
  656. return &Event{}, nil
  657. }
  658. func (s *storeRecorder) CompareAndSwap(path, prevVal string, prevIdx uint64, val string, expireOpts TTLOptionSet) (*Event, error) {
  659. s.Record(testutil.Action{
  660. Name: "CompareAndSwap",
  661. Params: []interface{}{path, prevVal, prevIdx, val, expireOpts},
  662. })
  663. return &Event{}, nil
  664. }
  665. func (s *storeRecorder) Delete(path string, dir, recursive bool) (*Event, error) {
  666. s.Record(testutil.Action{
  667. Name: "Delete",
  668. Params: []interface{}{path, dir, recursive},
  669. })
  670. return &Event{}, nil
  671. }
  672. func (s *storeRecorder) CompareAndDelete(path, prevVal string, prevIdx uint64) (*Event, error) {
  673. s.Record(testutil.Action{
  674. Name: "CompareAndDelete",
  675. Params: []interface{}{path, prevVal, prevIdx},
  676. })
  677. return &Event{}, nil
  678. }
  679. func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (Watcher, error) {
  680. s.Record(testutil.Action{Name: "Watch"})
  681. return NewNopWatcher(), nil
  682. }
  683. func (s *storeRecorder) Save() ([]byte, error) {
  684. s.Record(testutil.Action{Name: "Save"})
  685. return nil, nil
  686. }
  687. func (s *storeRecorder) Recovery(b []byte) error {
  688. s.Record(testutil.Action{Name: "Recovery"})
  689. return nil
  690. }
  691. func (s *storeRecorder) SaveNoCopy() ([]byte, error) {
  692. s.Record(testutil.Action{Name: "SaveNoCopy"})
  693. return nil, nil
  694. }
  695. func (s *storeRecorder) Clone() Store {
  696. s.Record(testutil.Action{Name: "Clone"})
  697. return s
  698. }
  699. func (s *storeRecorder) JsonStats() []byte { return nil }
  700. func (s *storeRecorder) DeleteExpiredKeys(cutoff time.Time) {
  701. s.Record(testutil.Action{
  702. Name: "DeleteExpiredKeys",
  703. Params: []interface{}{cutoff},
  704. })
  705. }
  706. // errStoreRecorder is a storeRecorder, but returns the given error on
  707. // Get, Watch methods.
  708. type errStoreRecorder struct {
  709. storeRecorder
  710. err error
  711. }
  712. func NewErrRecorder(err error) *StoreRecorder {
  713. sr := &errStoreRecorder{err: err}
  714. sr.Recorder = &testutil.RecorderBuffered{}
  715. return &StoreRecorder{Store: sr, Recorder: sr.Recorder}
  716. }
  717. func (s *errStoreRecorder) Get(path string, recursive, sorted bool) (*Event, error) {
  718. s.storeRecorder.Get(path, recursive, sorted)
  719. return nil, s.err
  720. }
  721. func (s *errStoreRecorder) Watch(path string, recursive, sorted bool, index uint64) (Watcher, error) {
  722. s.storeRecorder.Watch(path, recursive, sorted, index)
  723. return nil, s.err
  724. }