You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

223 lines
6.1 KiB

5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
  1. package mongodb
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/filer"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. "go.mongodb.org/mongo-driver/bson"
  10. "go.mongodb.org/mongo-driver/mongo"
  11. "go.mongodb.org/mongo-driver/mongo/options"
  12. "go.mongodb.org/mongo-driver/x/bsonx"
  13. "time"
  14. )
  15. func init() {
  16. filer.Stores = append(filer.Stores, &MongodbStore{})
  17. }
  18. type MongodbStore struct {
  19. connect *mongo.Client
  20. database string
  21. collectionName string
  22. }
  23. type Model struct {
  24. Directory string `bson:"directory"`
  25. Name string `bson:"name"`
  26. Meta []byte `bson:"meta"`
  27. }
  28. func (store *MongodbStore) GetName() string {
  29. return "mongodb"
  30. }
  31. func (store *MongodbStore) Initialize(configuration util.Configuration, prefix string) (err error) {
  32. store.database = configuration.GetString(prefix + "database")
  33. store.collectionName = "filemeta"
  34. poolSize := configuration.GetInt(prefix + "option_pool_size")
  35. return store.connection(configuration.GetString(prefix+"uri"), uint64(poolSize))
  36. }
  37. func (store *MongodbStore) connection(uri string, poolSize uint64) (err error) {
  38. ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
  39. opts := options.Client().ApplyURI(uri)
  40. if poolSize > 0 {
  41. opts.SetMaxPoolSize(poolSize)
  42. }
  43. client, err := mongo.Connect(ctx, opts)
  44. if err != nil {
  45. return err
  46. }
  47. c := client.Database(store.database).Collection(store.collectionName)
  48. err = store.indexUnique(c)
  49. store.connect = client
  50. return err
  51. }
  52. func (store *MongodbStore) createIndex(c *mongo.Collection, index mongo.IndexModel, opts *options.CreateIndexesOptions) error {
  53. _, err := c.Indexes().CreateOne(context.Background(), index, opts)
  54. return err
  55. }
  56. func (store *MongodbStore) indexUnique(c *mongo.Collection) error {
  57. opts := options.CreateIndexes().SetMaxTime(10 * time.Second)
  58. unique := new(bool)
  59. *unique = true
  60. index := mongo.IndexModel{
  61. Keys: bsonx.Doc{{Key: "directory", Value: bsonx.Int32(1)}, {Key: "name", Value: bsonx.Int32(1)}},
  62. Options: &options.IndexOptions{
  63. Unique: unique,
  64. },
  65. }
  66. return store.createIndex(c, index, opts)
  67. }
  68. func (store *MongodbStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  69. return ctx, nil
  70. }
  71. func (store *MongodbStore) CommitTransaction(ctx context.Context) error {
  72. return nil
  73. }
  74. func (store *MongodbStore) RollbackTransaction(ctx context.Context) error {
  75. return nil
  76. }
  77. func (store *MongodbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  78. dir, name := entry.FullPath.DirAndName()
  79. meta, err := entry.EncodeAttributesAndChunks()
  80. if err != nil {
  81. return fmt.Errorf("encode %s: %s", entry.FullPath, err)
  82. }
  83. if len(entry.Chunks) > 50 {
  84. meta = util.MaybeGzipData(meta)
  85. }
  86. c := store.connect.Database(store.database).Collection(store.collectionName)
  87. _, err = c.InsertOne(ctx, Model{
  88. Directory: dir,
  89. Name: name,
  90. Meta: meta,
  91. })
  92. if err != nil {
  93. return fmt.Errorf("InsertEntry %st: %v", entry.FullPath, err)
  94. }
  95. return nil
  96. }
  97. func (store *MongodbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  98. return store.InsertEntry(ctx, entry)
  99. }
  100. func (store *MongodbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
  101. dir, name := fullpath.DirAndName()
  102. var data Model
  103. var where = bson.M{"directory": dir, "name": name}
  104. err = store.connect.Database(store.database).Collection(store.collectionName).FindOne(ctx, where).Decode(&data)
  105. if err != mongo.ErrNoDocuments && err != nil {
  106. glog.Errorf("find %s: %v", fullpath, err)
  107. return nil, filer_pb.ErrNotFound
  108. }
  109. if len(data.Meta) == 0 {
  110. return nil, filer_pb.ErrNotFound
  111. }
  112. entry = &filer.Entry{
  113. FullPath: fullpath,
  114. }
  115. err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data.Meta))
  116. if err != nil {
  117. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  118. }
  119. return entry, nil
  120. }
  121. func (store *MongodbStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) error {
  122. dir, name := fullpath.DirAndName()
  123. where := bson.M{"directory": dir, "name": name}
  124. _, err := store.connect.Database(store.database).Collection(store.collectionName).DeleteOne(ctx, where)
  125. if err != nil {
  126. return fmt.Errorf("delete %s : %v", fullpath, err)
  127. }
  128. return nil
  129. }
  130. func (store *MongodbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
  131. where := bson.M{"directory": fullpath}
  132. _, err := store.connect.Database(store.database).Collection(store.collectionName).DeleteMany(ctx, where)
  133. if err != nil {
  134. return fmt.Errorf("delete %s : %v", fullpath, err)
  135. }
  136. return nil
  137. }
  138. func (store *MongodbStore) ListDirectoryPrefixedEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool, limit int, prefix string) (entries []*filer.Entry, err error) {
  139. return nil, filer.ErrUnsupportedListDirectoryPrefixed
  140. }
  141. func (store *MongodbStore) ListDirectoryEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool, limit int) (entries []*filer.Entry, err error) {
  142. var where = bson.M{"directory": string(fullpath), "name": bson.M{"$gt": startFileName}}
  143. if inclusive {
  144. where["name"] = bson.M{
  145. "$gte": startFileName,
  146. }
  147. }
  148. optLimit := int64(limit)
  149. opts := &options.FindOptions{Limit: &optLimit, Sort: bson.M{"name": 1}}
  150. cur, err := store.connect.Database(store.database).Collection(store.collectionName).Find(ctx, where, opts)
  151. for cur.Next(ctx) {
  152. var data Model
  153. err := cur.Decode(&data)
  154. if err != nil && err != mongo.ErrNoDocuments {
  155. return nil, err
  156. }
  157. entry := &filer.Entry{
  158. FullPath: util.NewFullPath(string(fullpath), data.Name),
  159. }
  160. if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data.Meta)); decodeErr != nil {
  161. err = decodeErr
  162. glog.V(0).Infof("list %s : %v", entry.FullPath, err)
  163. break
  164. }
  165. entries = append(entries, entry)
  166. }
  167. if err := cur.Close(ctx); err != nil {
  168. glog.V(0).Infof("list iterator close: %v", err)
  169. }
  170. return entries, err
  171. }
  172. func (store *MongodbStore) Shutdown() {
  173. ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
  174. store.connect.Disconnect(ctx)
  175. }