etcd_store.go 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219
  1. package etcd
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "strings"
  7. "time"
  8. "go.etcd.io/etcd/clientv3"
  9. "github.com/chrislusf/seaweedfs/weed/filer"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. weed_util "github.com/chrislusf/seaweedfs/weed/util"
  13. )
  14. const (
  15. DIR_FILE_SEPARATOR = byte(0x00)
  16. )
  17. func init() {
  18. filer.Stores = append(filer.Stores, &EtcdStore{})
  19. }
  20. type EtcdStore struct {
  21. client *clientv3.Client
  22. }
  23. func (store *EtcdStore) GetName() string {
  24. return "etcd"
  25. }
  26. func (store *EtcdStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
  27. servers := configuration.GetString(prefix + "servers")
  28. if servers == "" {
  29. servers = "localhost:2379"
  30. }
  31. timeout := configuration.GetString(prefix + "timeout")
  32. if timeout == "" {
  33. timeout = "3s"
  34. }
  35. return store.initialize(servers, timeout)
  36. }
  37. func (store *EtcdStore) initialize(servers string, timeout string) (err error) {
  38. glog.Infof("filer store etcd: %s", servers)
  39. to, err := time.ParseDuration(timeout)
  40. if err != nil {
  41. return fmt.Errorf("parse timeout %s: %s", timeout, err)
  42. }
  43. store.client, err = clientv3.New(clientv3.Config{
  44. Endpoints: strings.Split(servers, ","),
  45. DialTimeout: to,
  46. })
  47. if err != nil {
  48. return fmt.Errorf("connect to etcd %s: %s", servers, err)
  49. }
  50. return
  51. }
  52. func (store *EtcdStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  53. return ctx, nil
  54. }
  55. func (store *EtcdStore) CommitTransaction(ctx context.Context) error {
  56. return nil
  57. }
  58. func (store *EtcdStore) RollbackTransaction(ctx context.Context) error {
  59. return nil
  60. }
  61. func (store *EtcdStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  62. key := genKey(entry.DirAndName())
  63. meta, err := entry.EncodeAttributesAndChunks()
  64. if err != nil {
  65. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  66. }
  67. if len(entry.Chunks) > 50 {
  68. meta = weed_util.MaybeGzipData(meta)
  69. }
  70. if _, err := store.client.Put(ctx, string(key), string(meta)); err != nil {
  71. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  72. }
  73. return nil
  74. }
  75. func (store *EtcdStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  76. return store.InsertEntry(ctx, entry)
  77. }
  78. func (store *EtcdStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
  79. key := genKey(fullpath.DirAndName())
  80. resp, err := store.client.Get(ctx, string(key))
  81. if err != nil {
  82. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  83. }
  84. if len(resp.Kvs) == 0 {
  85. return nil, filer_pb.ErrNotFound
  86. }
  87. entry = &filer.Entry{
  88. FullPath: fullpath,
  89. }
  90. err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(resp.Kvs[0].Value))
  91. if err != nil {
  92. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  93. }
  94. return entry, nil
  95. }
  96. func (store *EtcdStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  97. key := genKey(fullpath.DirAndName())
  98. if _, err := store.client.Delete(ctx, string(key)); err != nil {
  99. return fmt.Errorf("delete %s : %v", fullpath, err)
  100. }
  101. return nil
  102. }
  103. func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  104. directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
  105. if _, err := store.client.Delete(ctx, string(directoryPrefix), clientv3.WithPrefix()); err != nil {
  106. return fmt.Errorf("deleteFolderChildren %s : %v", fullpath, err)
  107. }
  108. return nil
  109. }
  110. func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  111. return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
  112. }
  113. func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  114. directoryPrefix := genDirectoryKeyPrefix(dirPath, "")
  115. lastFileStart := directoryPrefix
  116. if startFileName != "" {
  117. lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
  118. }
  119. resp, err := store.client.Get(ctx, string(lastFileStart),
  120. clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
  121. if err != nil {
  122. return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
  123. }
  124. for _, kv := range resp.Kvs {
  125. if !bytes.HasPrefix(kv.Key, directoryPrefix) {
  126. break
  127. }
  128. fileName := getNameFromKey(kv.Key)
  129. if fileName == "" {
  130. continue
  131. }
  132. if fileName == startFileName && !includeStartFile {
  133. continue
  134. }
  135. limit--
  136. if limit < 0 {
  137. break
  138. }
  139. entry := &filer.Entry{
  140. FullPath: weed_util.NewFullPath(string(dirPath), fileName),
  141. }
  142. if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(kv.Value)); decodeErr != nil {
  143. err = decodeErr
  144. glog.V(0).Infof("list %s : %v", entry.FullPath, err)
  145. break
  146. }
  147. if !eachEntryFunc(entry) {
  148. break
  149. }
  150. lastFileName = fileName
  151. }
  152. return lastFileName, err
  153. }
  154. func genKey(dirPath, fileName string) (key []byte) {
  155. key = []byte(dirPath)
  156. key = append(key, DIR_FILE_SEPARATOR)
  157. key = append(key, []byte(fileName)...)
  158. return key
  159. }
  160. func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) {
  161. keyPrefix = []byte(string(fullpath))
  162. keyPrefix = append(keyPrefix, DIR_FILE_SEPARATOR)
  163. if len(startFileName) > 0 {
  164. keyPrefix = append(keyPrefix, []byte(startFileName)...)
  165. }
  166. return keyPrefix
  167. }
  168. func getNameFromKey(key []byte) string {
  169. sepIndex := len(key) - 1
  170. for sepIndex >= 0 && key[sepIndex] != DIR_FILE_SEPARATOR {
  171. sepIndex--
  172. }
  173. return string(key[sepIndex+1:])
  174. }
  175. func (store *EtcdStore) Shutdown() {
  176. store.client.Close()
  177. }