mongodb_store.go 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229
  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. return store.UpdateEntry(ctx, entry)
  79. }
  80. func (store *MongodbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  81. dir, name := entry.FullPath.DirAndName()
  82. meta, err := entry.EncodeAttributesAndChunks()
  83. if err != nil {
  84. return fmt.Errorf("encode %s: %s", entry.FullPath, err)
  85. }
  86. if len(entry.Chunks) > 50 {
  87. meta = util.MaybeGzipData(meta)
  88. }
  89. c := store.connect.Database(store.database).Collection(store.collectionName)
  90. opts := options.Update().SetUpsert(true)
  91. filter := bson.D{{"directory", dir}, {"name", name}}
  92. update := bson.D{{"$set", bson.D{{"meta", meta}}}}
  93. _, err = c.UpdateOne(ctx, filter, update, opts)
  94. if err != nil {
  95. return fmt.Errorf("UpdateEntry %s: %v", entry.FullPath, err)
  96. }
  97. return nil
  98. }
  99. func (store *MongodbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
  100. dir, name := fullpath.DirAndName()
  101. var data Model
  102. var where = bson.M{"directory": dir, "name": name}
  103. err = store.connect.Database(store.database).Collection(store.collectionName).FindOne(ctx, where).Decode(&data)
  104. if err != mongo.ErrNoDocuments && err != nil {
  105. glog.Errorf("find %s: %v", fullpath, err)
  106. return nil, filer_pb.ErrNotFound
  107. }
  108. if len(data.Meta) == 0 {
  109. return nil, filer_pb.ErrNotFound
  110. }
  111. entry = &filer.Entry{
  112. FullPath: fullpath,
  113. }
  114. err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data.Meta))
  115. if err != nil {
  116. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  117. }
  118. return entry, nil
  119. }
  120. func (store *MongodbStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) error {
  121. dir, name := fullpath.DirAndName()
  122. where := bson.M{"directory": dir, "name": name}
  123. _, err := store.connect.Database(store.database).Collection(store.collectionName).DeleteOne(ctx, where)
  124. if err != nil {
  125. return fmt.Errorf("delete %s : %v", fullpath, err)
  126. }
  127. return nil
  128. }
  129. func (store *MongodbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
  130. where := bson.M{"directory": fullpath}
  131. _, err := store.connect.Database(store.database).Collection(store.collectionName).DeleteMany(ctx, where)
  132. if err != nil {
  133. return fmt.Errorf("delete %s : %v", fullpath, err)
  134. }
  135. return nil
  136. }
  137. 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) {
  138. return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
  139. }
  140. func (store *MongodbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  141. var where = bson.M{"directory": string(dirPath), "name": bson.M{"$gt": startFileName}}
  142. if includeStartFile {
  143. where["name"] = bson.M{
  144. "$gte": startFileName,
  145. }
  146. }
  147. optLimit := int64(limit)
  148. opts := &options.FindOptions{Limit: &optLimit, Sort: bson.M{"name": 1}}
  149. cur, err := store.connect.Database(store.database).Collection(store.collectionName).Find(ctx, where, opts)
  150. for cur.Next(ctx) {
  151. var data Model
  152. err := cur.Decode(&data)
  153. if err != nil && err != mongo.ErrNoDocuments {
  154. return lastFileName, err
  155. }
  156. entry := &filer.Entry{
  157. FullPath: util.NewFullPath(string(dirPath), data.Name),
  158. }
  159. lastFileName = data.Name
  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. if !eachEntryFunc(entry) {
  166. break
  167. }
  168. }
  169. if err := cur.Close(ctx); err != nil {
  170. glog.V(0).Infof("list iterator close: %v", err)
  171. }
  172. return lastFileName, err
  173. }
  174. func (store *MongodbStore) Shutdown() {
  175. ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
  176. store.connect.Disconnect(ctx)
  177. }