file_system.go 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287
  1. package fileSystem
  2. import (
  3. "fmt"
  4. "path"
  5. "strings"
  6. "time"
  7. etcdErr "github.com/coreos/etcd/error"
  8. )
  9. type FileSystem struct {
  10. Root *Node
  11. EventHistory *EventHistory
  12. WatcherHub *watcherHub
  13. Index uint64
  14. Term uint64
  15. }
  16. func New() *FileSystem {
  17. return &FileSystem{
  18. Root: newDir("/", 0, 0, nil, "", Permanent),
  19. WatcherHub: newWatchHub(1000),
  20. }
  21. }
  22. func (fs *FileSystem) Get(keyPath string, recusive bool, index uint64, term uint64) (*Event, error) {
  23. // TODO: add recursive get
  24. n, err := fs.InternalGet(keyPath, index, term)
  25. if err != nil {
  26. return nil, err
  27. }
  28. e := newEvent(Get, keyPath, index, term)
  29. if n.IsDir() { // node is dir
  30. e.Dir = true
  31. children, _ := n.List()
  32. e.KVPairs = make([]KeyValuePair, len(children))
  33. // we do not use the index in the children slice directly
  34. // we need to skip the hidden one
  35. i := 0
  36. for _, child := range children {
  37. if child.IsHidden() { // get will not list hidden node
  38. continue
  39. }
  40. e.KVPairs[i] = child.Pair(recusive)
  41. i++
  42. }
  43. // eliminate hidden nodes
  44. e.KVPairs = e.KVPairs[:i]
  45. } else { // node is file
  46. e.Value = n.Value
  47. }
  48. return e, nil
  49. }
  50. func (fs *FileSystem) Create(keyPath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
  51. keyPath = path.Clean("/" + keyPath)
  52. // make sure we can create the node
  53. _, err := fs.InternalGet(keyPath, index, term)
  54. if err == nil { // key already exists
  55. return nil, etcdErr.NewError(105, keyPath)
  56. }
  57. etcdError, _ := err.(etcdErr.Error)
  58. if etcdError.ErrorCode == 104 { // we cannot create the key due to meet a file while walking
  59. return nil, err
  60. }
  61. dir, _ := path.Split(keyPath)
  62. // walk through the keyPath, create dirs and get the last directory node
  63. d, err := fs.walk(dir, fs.checkDir)
  64. if err != nil {
  65. return nil, err
  66. }
  67. e := newEvent(Set, keyPath, fs.Index, fs.Term)
  68. var n *Node
  69. if len(value) != 0 { // create file
  70. e.Value = value
  71. n = newFile(keyPath, value, fs.Index, fs.Term, d, "", expireTime)
  72. } else { // create directory
  73. e.Dir = true
  74. n = newDir(keyPath, fs.Index, fs.Term, d, "", expireTime)
  75. }
  76. err = d.Add(n)
  77. if err != nil {
  78. return nil, err
  79. }
  80. // Node with TTL
  81. if expireTime != Permanent {
  82. go n.Expire()
  83. e.Expiration = &n.ExpireTime
  84. e.TTL = int64(expireTime.Sub(time.Now()) / time.Second)
  85. }
  86. return e, nil
  87. }
  88. func (fs *FileSystem) Update(keyPath string, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
  89. n, err := fs.InternalGet(keyPath, index, term)
  90. if err != nil { // if node does not exist, return error
  91. return nil, err
  92. }
  93. e := newEvent(Set, keyPath, fs.Index, fs.Term)
  94. if n.IsDir() { // if the node is a directory, we can only update ttl
  95. if len(value) != 0 {
  96. return nil, etcdErr.NewError(102, keyPath)
  97. }
  98. if n.ExpireTime != Permanent && expireTime != Permanent {
  99. n.stopExpire <- true
  100. }
  101. } else { // if the node is a file, we can update value and ttl
  102. e.PrevValue = n.Value
  103. if len(value) != 0 {
  104. e.Value = value
  105. }
  106. n.Write(value, index, term)
  107. if n.ExpireTime != Permanent && expireTime != Permanent {
  108. n.stopExpire <- true
  109. }
  110. }
  111. // update ttl
  112. if expireTime != Permanent {
  113. go n.Expire()
  114. e.Expiration = &n.ExpireTime
  115. e.TTL = int64(expireTime.Sub(time.Now()) / time.Second)
  116. }
  117. return e, nil
  118. }
  119. func (fs *FileSystem) TestAndSet(keyPath string, prevValue string, prevIndex uint64, value string, expireTime time.Time, index uint64, term uint64) (*Event, error) {
  120. f, err := fs.InternalGet(keyPath, index, term)
  121. if err != nil {
  122. return nil, err
  123. }
  124. if f.IsDir() { // can only test and set file
  125. return nil, etcdErr.NewError(102, keyPath)
  126. }
  127. if f.Value == prevValue || f.ModifiedIndex == prevIndex {
  128. // if test succeed, write the value
  129. e := newEvent(TestAndSet, keyPath, index, term)
  130. e.PrevValue = f.Value
  131. e.Value = value
  132. f.Write(value, index, term)
  133. return e, nil
  134. }
  135. cause := fmt.Sprintf("[%v/%v] [%v/%v]", prevValue, f.Value, prevIndex, f.ModifiedIndex)
  136. return nil, etcdErr.NewError(101, cause)
  137. }
  138. func (fs *FileSystem) Delete(keyPath string, recurisive bool, index uint64, term uint64) (*Event, error) {
  139. n, err := fs.InternalGet(keyPath, index, term)
  140. if err != nil {
  141. return nil, err
  142. }
  143. err = n.Remove(recurisive)
  144. if err != nil {
  145. return nil, err
  146. }
  147. e := newEvent(Delete, keyPath, index, term)
  148. if n.IsDir() {
  149. e.Dir = true
  150. } else {
  151. e.PrevValue = n.Value
  152. }
  153. return e, nil
  154. }
  155. // walk function walks all the keyPath and apply the walkFunc on each directory
  156. func (fs *FileSystem) walk(keyPath string, walkFunc func(prev *Node, component string) (*Node, error)) (*Node, error) {
  157. components := strings.Split(keyPath, "/")
  158. curr := fs.Root
  159. var err error
  160. for i := 1; i < len(components); i++ {
  161. if len(components[i]) == 0 { // ignore empty string
  162. return curr, nil
  163. }
  164. curr, err = walkFunc(curr, components[i])
  165. if err != nil {
  166. return nil, err
  167. }
  168. }
  169. return curr, nil
  170. }
  171. // InternalGet function get the node of the given keyPath.
  172. func (fs *FileSystem) InternalGet(keyPath string, index uint64, term uint64) (*Node, error) {
  173. keyPath = path.Clean("/" + keyPath)
  174. // update file system known index and term
  175. fs.Index, fs.Term = index, term
  176. walkFunc := func(parent *Node, name string) (*Node, error) {
  177. if !parent.IsDir() {
  178. return nil, etcdErr.NewError(104, parent.Path)
  179. }
  180. child, ok := parent.Children[name]
  181. if ok {
  182. return child, nil
  183. }
  184. return nil, etcdErr.NewError(100, path.Join(parent.Path, name))
  185. }
  186. f, err := fs.walk(keyPath, walkFunc)
  187. if err != nil {
  188. return nil, err
  189. }
  190. return f, nil
  191. }
  192. // checkDir function will check whether the component is a directory under parent node.
  193. // If it is a directory, this function will return the pointer to that node.
  194. // If it does not exist, this function will create a new directory and return the pointer to that node.
  195. // If it is a file, this function will return error.
  196. func (fs *FileSystem) checkDir(parent *Node, dirName string) (*Node, error) {
  197. subDir, ok := parent.Children[dirName]
  198. if ok {
  199. return subDir, nil
  200. }
  201. n := newDir(path.Join(parent.Path, dirName), fs.Index, fs.Term, parent, parent.ACL, Permanent)
  202. parent.Children[dirName] = n
  203. return n, nil
  204. }