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.

232 lines
6.4 KiB

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