store.go 22 KB

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