node.go 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411
  1. package store
  2. import (
  3. "path"
  4. "sort"
  5. "sync"
  6. "time"
  7. etcdErr "github.com/coreos/etcd/error"
  8. )
  9. var (
  10. Permanent time.Time
  11. )
  12. const (
  13. normal = iota
  14. removed
  15. )
  16. // Node is the basic element in the store system.
  17. // A key-value pair will have a string value
  18. // A directory will have a children map
  19. type Node struct {
  20. Path string
  21. CreateIndex uint64
  22. CreateTerm uint64
  23. ModifiedIndex uint64
  24. ModifiedTerm uint64
  25. Parent *Node `json:"-"` // should not encode this field! avoid cyclical dependency.
  26. ExpireTime time.Time
  27. ACL string
  28. Value string // for key-value pair
  29. Children map[string]*Node // for directory
  30. // a ttl node will have an expire routine associated with it.
  31. // we need a channel to stop that routine when the expiration changes.
  32. stopExpire chan bool
  33. // ensure we only delete the node once
  34. // expire and remove may try to delete a node twice
  35. once sync.Once
  36. }
  37. // newKV creates a Key-Value pair
  38. func newKV(nodePath string, value string, createIndex uint64,
  39. createTerm uint64, parent *Node, ACL string, expireTime time.Time) *Node {
  40. return &Node{
  41. Path: nodePath,
  42. CreateIndex: createIndex,
  43. CreateTerm: createTerm,
  44. ModifiedIndex: createIndex,
  45. ModifiedTerm: createTerm,
  46. Parent: parent,
  47. ACL: ACL,
  48. stopExpire: make(chan bool, 1),
  49. ExpireTime: expireTime,
  50. Value: value,
  51. }
  52. }
  53. // newDir creates a directory
  54. func newDir(nodePath string, createIndex uint64, createTerm uint64,
  55. parent *Node, ACL string, expireTime time.Time) *Node {
  56. return &Node{
  57. Path: nodePath,
  58. CreateIndex: createIndex,
  59. CreateTerm: createTerm,
  60. Parent: parent,
  61. ACL: ACL,
  62. stopExpire: make(chan bool, 1),
  63. ExpireTime: expireTime,
  64. Children: make(map[string]*Node),
  65. }
  66. }
  67. // IsHidden function checks if the node is a hidden node. A hidden node
  68. // will begin with '_'
  69. // A hidden node will not be shown via get command under a directory
  70. // For example if we have /foo/_hidden and /foo/notHidden, get "/foo"
  71. // will only return /foo/notHidden
  72. func (n *Node) IsHidden() bool {
  73. _, name := path.Split(n.Path)
  74. return name[0] == '_'
  75. }
  76. // IsPermanent function checks if the node is a permanent one.
  77. func (n *Node) IsPermanent() bool {
  78. return n.ExpireTime.Sub(Permanent) == 0
  79. }
  80. // IsExpired function checks if the node has been expired.
  81. func (n *Node) IsExpired() (bool, time.Duration) {
  82. if n.IsPermanent() {
  83. return false, 0
  84. }
  85. duration := n.ExpireTime.Sub(time.Now())
  86. if duration <= 0 {
  87. return true, 0
  88. }
  89. return false, duration
  90. }
  91. // IsDir function checks whether the node is a directory.
  92. // If the node is a directory, the function will return true.
  93. // Otherwise the function will return false.
  94. func (n *Node) IsDir() bool {
  95. return !(n.Children == nil)
  96. }
  97. // Read function gets the value of the node.
  98. // If the receiver node is not a key-value pair, a "Not A File" error will be returned.
  99. func (n *Node) Read() (string, *etcdErr.Error) {
  100. if n.IsDir() {
  101. return "", etcdErr.NewError(etcdErr.EcodeNotFile, "", UndefIndex, UndefTerm)
  102. }
  103. return n.Value, nil
  104. }
  105. // Write function set the value of the node to the given value.
  106. // If the receiver node is a directory, a "Not A File" error will be returned.
  107. func (n *Node) Write(value string, index uint64, term uint64) *etcdErr.Error {
  108. if n.IsDir() {
  109. return etcdErr.NewError(etcdErr.EcodeNotFile, "", UndefIndex, UndefTerm)
  110. }
  111. n.Value = value
  112. n.ModifiedIndex = index
  113. n.ModifiedTerm = term
  114. return nil
  115. }
  116. func (n *Node) ExpirationAndTTL() (*time.Time, int64) {
  117. if n.ExpireTime.Sub(Permanent) != 0 {
  118. return &n.ExpireTime, int64(n.ExpireTime.Sub(time.Now())/time.Second) + 1
  119. }
  120. return nil, 0
  121. }
  122. // List function return a slice of nodes under the receiver node.
  123. // If the receiver node is not a directory, a "Not A Directory" error will be returned.
  124. func (n *Node) List() ([]*Node, *etcdErr.Error) {
  125. if !n.IsDir() {
  126. return nil, etcdErr.NewError(etcdErr.EcodeNotDir, "", UndefIndex, UndefTerm)
  127. }
  128. nodes := make([]*Node, len(n.Children))
  129. i := 0
  130. for _, node := range n.Children {
  131. nodes[i] = node
  132. i++
  133. }
  134. return nodes, nil
  135. }
  136. // GetChild function returns the child node under the directory node.
  137. // On success, it returns the file node
  138. func (n *Node) GetChild(name string) (*Node, *etcdErr.Error) {
  139. if !n.IsDir() {
  140. return nil, etcdErr.NewError(etcdErr.EcodeNotDir, n.Path, UndefIndex, UndefTerm)
  141. }
  142. child, ok := n.Children[name]
  143. if ok {
  144. return child, nil
  145. }
  146. return nil, nil
  147. }
  148. // Add function adds a node to the receiver node.
  149. // If the receiver is not a directory, a "Not A Directory" error will be returned.
  150. // If there is a existing node with the same name under the directory, a "Already Exist"
  151. // error will be returned
  152. func (n *Node) Add(child *Node) *etcdErr.Error {
  153. if !n.IsDir() {
  154. return etcdErr.NewError(etcdErr.EcodeNotDir, "", UndefIndex, UndefTerm)
  155. }
  156. _, name := path.Split(child.Path)
  157. _, ok := n.Children[name]
  158. if ok {
  159. return etcdErr.NewError(etcdErr.EcodeNodeExist, "", UndefIndex, UndefTerm)
  160. }
  161. n.Children[name] = child
  162. return nil
  163. }
  164. // Remove function remove the node.
  165. func (n *Node) Remove(recursive bool, callback func(path string)) *etcdErr.Error {
  166. if n.IsDir() && !recursive {
  167. // cannot delete a directory without set recursive to true
  168. return etcdErr.NewError(etcdErr.EcodeNotFile, "", UndefIndex, UndefTerm)
  169. }
  170. onceBody := func() {
  171. n.internalRemove(recursive, callback)
  172. }
  173. // this function might be entered multiple times by expire and delete
  174. // every node will only be deleted once.
  175. n.once.Do(onceBody)
  176. return nil
  177. }
  178. // internalRemove function will be called by remove()
  179. func (n *Node) internalRemove(recursive bool, callback func(path string)) {
  180. if !n.IsDir() { // key-value pair
  181. _, name := path.Split(n.Path)
  182. // find its parent and remove the node from the map
  183. if n.Parent != nil && n.Parent.Children[name] == n {
  184. delete(n.Parent.Children, name)
  185. }
  186. if callback != nil {
  187. callback(n.Path)
  188. }
  189. // the stop channel has a buffer. just send to it!
  190. n.stopExpire <- true
  191. return
  192. }
  193. for _, child := range n.Children { // delete all children
  194. child.Remove(true, callback)
  195. }
  196. // delete self
  197. _, name := path.Split(n.Path)
  198. if n.Parent != nil && n.Parent.Children[name] == n {
  199. delete(n.Parent.Children, name)
  200. if callback != nil {
  201. callback(n.Path)
  202. }
  203. n.stopExpire <- true
  204. }
  205. }
  206. // Expire function will test if the node is expired.
  207. // if the node is already expired, delete the node and return.
  208. // if the node is permanent (this shouldn't happen), return at once.
  209. // else wait for a period time, then remove the node. and notify the watchhub.
  210. func (n *Node) Expire(s *Store) {
  211. expired, duration := n.IsExpired()
  212. if expired { // has been expired
  213. // since the parent function of Expire() runs serially,
  214. // there is no need for lock here
  215. e := newEvent(Expire, n.Path, UndefIndex, UndefTerm)
  216. s.WatcherHub.notify(e)
  217. n.Remove(true, nil)
  218. s.Stats.Inc(ExpireCount)
  219. return
  220. }
  221. if duration == 0 { // Permanent Node
  222. return
  223. }
  224. go func() { // do monitoring
  225. select {
  226. // if timeout, delete the node
  227. case <-time.After(duration):
  228. // before expire get the lock, the expiration time
  229. // of the node may be updated.
  230. // we have to check again when get the lock
  231. s.worldLock.Lock()
  232. defer s.worldLock.Unlock()
  233. expired, _ := n.IsExpired()
  234. if expired {
  235. e := newEvent(Expire, n.Path, UndefIndex, UndefTerm)
  236. s.WatcherHub.notify(e)
  237. n.Remove(true, nil)
  238. s.Stats.Inc(ExpireCount)
  239. }
  240. return
  241. // if stopped, return
  242. case <-n.stopExpire:
  243. return
  244. }
  245. }()
  246. }
  247. func (n *Node) Pair(recurisive, sorted bool) KeyValuePair {
  248. if n.IsDir() {
  249. pair := KeyValuePair{
  250. Key: n.Path,
  251. Dir: true,
  252. }
  253. if !recurisive {
  254. return pair
  255. }
  256. children, _ := n.List()
  257. pair.KVPairs = make([]KeyValuePair, len(children))
  258. // we do not use the index in the children slice directly
  259. // we need to skip the hidden one
  260. i := 0
  261. for _, child := range children {
  262. if child.IsHidden() { // get will not list hidden node
  263. continue
  264. }
  265. pair.KVPairs[i] = child.Pair(recurisive, sorted)
  266. i++
  267. }
  268. // eliminate hidden nodes
  269. pair.KVPairs = pair.KVPairs[:i]
  270. if sorted {
  271. sort.Sort(pair.KVPairs)
  272. }
  273. return pair
  274. }
  275. return KeyValuePair{
  276. Key: n.Path,
  277. Value: n.Value,
  278. }
  279. }
  280. func (n *Node) UpdateTTL(expireTime time.Time, s *Store) {
  281. if !n.IsPermanent() {
  282. // check if the node has been expired
  283. // if the node is not expired, we need to stop the go routine associated with
  284. // that node.
  285. expired, _ := n.IsExpired()
  286. if !expired {
  287. n.stopExpire <- true // suspend it to modify the expiration
  288. }
  289. }
  290. if expireTime.Sub(Permanent) != 0 {
  291. n.ExpireTime = expireTime
  292. n.Expire(s)
  293. }
  294. }
  295. // Clone function clone the node recursively and return the new node.
  296. // If the node is a directory, it will clone all the content under this directory.
  297. // If the node is a key-value pair, it will clone the pair.
  298. func (n *Node) Clone() *Node {
  299. if !n.IsDir() {
  300. return newKV(n.Path, n.Value, n.CreateIndex, n.CreateTerm, n.Parent, n.ACL, n.ExpireTime)
  301. }
  302. clone := newDir(n.Path, n.CreateIndex, n.CreateTerm, n.Parent, n.ACL, n.ExpireTime)
  303. for key, child := range n.Children {
  304. clone.Children[key] = child.Clone()
  305. }
  306. return clone
  307. }
  308. // recoverAndclean function help to do recovery.
  309. // Two things need to be done: 1. recovery structure; 2. delete expired nodes
  310. // If the node is a directory, it will help recover children's parent pointer and recursively
  311. // call this function on its children.
  312. // We check the expire last since we need to recover the whole structure first and add all the
  313. // notifications into the event history.
  314. func (n *Node) recoverAndclean(s *Store) {
  315. if n.IsDir() {
  316. for _, child := range n.Children {
  317. child.Parent = n
  318. child.recoverAndclean(s)
  319. }
  320. }
  321. n.stopExpire = make(chan bool, 1)
  322. n.Expire(s)
  323. }