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.

230 lines
7.2 KiB

3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. package filer
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
  7. "github.com/chrislusf/seaweedfs/weed/remote_storage"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. "github.com/golang/protobuf/proto"
  10. "google.golang.org/grpc"
  11. "math"
  12. "strings"
  13. "github.com/chrislusf/seaweedfs/weed/glog"
  14. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  15. "github.com/viant/ptrie"
  16. )
  17. const REMOTE_STORAGE_CONF_SUFFIX = ".conf"
  18. const REMOTE_STORAGE_MOUNT_FILE = "mount.mapping"
  19. type FilerRemoteStorage struct {
  20. rules ptrie.Trie
  21. storageNameToConf map[string]*remote_pb.RemoteConf
  22. }
  23. func NewFilerRemoteStorage() (rs *FilerRemoteStorage) {
  24. rs = &FilerRemoteStorage{
  25. rules: ptrie.New(),
  26. storageNameToConf: make(map[string]*remote_pb.RemoteConf),
  27. }
  28. return rs
  29. }
  30. func (rs *FilerRemoteStorage) LoadRemoteStorageConfigurationsAndMapping(filer *Filer) (err error) {
  31. // execute this on filer
  32. limit := int64(math.MaxInt32)
  33. entries, _, err := filer.ListDirectoryEntries(context.Background(), DirectoryEtcRemote, "", false, limit, "", "", "")
  34. if err != nil {
  35. if err == filer_pb.ErrNotFound {
  36. return nil
  37. }
  38. glog.Errorf("read remote storage %s: %v", DirectoryEtcRemote, err)
  39. return
  40. }
  41. for _, entry := range entries {
  42. if entry.Name() == REMOTE_STORAGE_MOUNT_FILE {
  43. if err := rs.loadRemoteStorageMountMapping(entry.Content); err != nil {
  44. return err
  45. }
  46. continue
  47. }
  48. if !strings.HasSuffix(entry.Name(), REMOTE_STORAGE_CONF_SUFFIX) {
  49. return nil
  50. }
  51. conf := &remote_pb.RemoteConf{}
  52. if err := proto.Unmarshal(entry.Content, conf); err != nil {
  53. return fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, entry.Name(), err)
  54. }
  55. rs.storageNameToConf[conf.Name] = conf
  56. }
  57. return nil
  58. }
  59. func (rs *FilerRemoteStorage) loadRemoteStorageMountMapping(data []byte) (err error) {
  60. mappings := &remote_pb.RemoteStorageMapping{}
  61. if err := proto.Unmarshal(data, mappings); err != nil {
  62. return fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, err)
  63. }
  64. for dir, storageLocation := range mappings.Mappings {
  65. rs.mapDirectoryToRemoteStorage(util.FullPath(dir), storageLocation)
  66. }
  67. return nil
  68. }
  69. func (rs *FilerRemoteStorage) mapDirectoryToRemoteStorage(dir util.FullPath, loc *remote_pb.RemoteStorageLocation) {
  70. rs.rules.Put([]byte(dir+"/"), loc)
  71. }
  72. func (rs *FilerRemoteStorage) FindMountDirectory(p util.FullPath) (mountDir util.FullPath, remoteLocation *remote_pb.RemoteStorageLocation) {
  73. rs.rules.MatchPrefix([]byte(p), func(key []byte, value interface{}) bool {
  74. mountDir = util.FullPath(string(key[:len(key)-1]))
  75. remoteLocation = value.(*remote_pb.RemoteStorageLocation)
  76. return true
  77. })
  78. return
  79. }
  80. func (rs *FilerRemoteStorage) FindRemoteStorageClient(p util.FullPath) (client remote_storage.RemoteStorageClient, remoteConf *remote_pb.RemoteConf, found bool) {
  81. var storageLocation *remote_pb.RemoteStorageLocation
  82. rs.rules.MatchPrefix([]byte(p), func(key []byte, value interface{}) bool {
  83. storageLocation = value.(*remote_pb.RemoteStorageLocation)
  84. return true
  85. })
  86. if storageLocation == nil {
  87. found = false
  88. return
  89. }
  90. return rs.GetRemoteStorageClient(storageLocation.Name)
  91. }
  92. func (rs *FilerRemoteStorage) GetRemoteStorageClient(storageName string) (client remote_storage.RemoteStorageClient, remoteConf *remote_pb.RemoteConf, found bool) {
  93. remoteConf, found = rs.storageNameToConf[storageName]
  94. if !found {
  95. return
  96. }
  97. var err error
  98. if client, err = remote_storage.GetRemoteStorage(remoteConf); err == nil {
  99. found = true
  100. return
  101. }
  102. return
  103. }
  104. func UnmarshalRemoteStorageMappings(oldContent []byte) (mappings *remote_pb.RemoteStorageMapping, err error) {
  105. mappings = &remote_pb.RemoteStorageMapping{
  106. Mappings: make(map[string]*remote_pb.RemoteStorageLocation),
  107. }
  108. if len(oldContent) > 0 {
  109. if err = proto.Unmarshal(oldContent, mappings); err != nil {
  110. glog.Warningf("unmarshal existing mappings: %v", err)
  111. }
  112. }
  113. return
  114. }
  115. func AddRemoteStorageMapping(oldContent []byte, dir string, storageLocation *remote_pb.RemoteStorageLocation) (newContent []byte, err error) {
  116. mappings, unmarshalErr := UnmarshalRemoteStorageMappings(oldContent)
  117. if unmarshalErr != nil {
  118. // skip
  119. }
  120. // set the new mapping
  121. mappings.Mappings[dir] = storageLocation
  122. if newContent, err = proto.Marshal(mappings); err != nil {
  123. return oldContent, fmt.Errorf("marshal mappings: %v", err)
  124. }
  125. return
  126. }
  127. func RemoveRemoteStorageMapping(oldContent []byte, dir string) (newContent []byte, err error) {
  128. mappings, unmarshalErr := UnmarshalRemoteStorageMappings(oldContent)
  129. if unmarshalErr != nil {
  130. return nil, unmarshalErr
  131. }
  132. // set the new mapping
  133. delete(mappings.Mappings, dir)
  134. if newContent, err = proto.Marshal(mappings); err != nil {
  135. return oldContent, fmt.Errorf("marshal mappings: %v", err)
  136. }
  137. return
  138. }
  139. func ReadMountMappings(grpcDialOption grpc.DialOption, filerAddress string) (mappings *remote_pb.RemoteStorageMapping, readErr error) {
  140. var oldContent []byte
  141. if readErr = pb.WithFilerClient(filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  142. oldContent, readErr = ReadInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE)
  143. return readErr
  144. }); readErr != nil {
  145. return nil, readErr
  146. }
  147. mappings, readErr = UnmarshalRemoteStorageMappings(oldContent)
  148. if readErr != nil {
  149. return nil, fmt.Errorf("unmarshal mappings: %v", readErr)
  150. }
  151. return
  152. }
  153. func ReadRemoteStorageConf(grpcDialOption grpc.DialOption, filerAddress string, storageName string) (conf *remote_pb.RemoteConf, readErr error) {
  154. var oldContent []byte
  155. if readErr = pb.WithFilerClient(filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  156. oldContent, readErr = ReadInsideFiler(client, DirectoryEtcRemote, storageName+REMOTE_STORAGE_CONF_SUFFIX)
  157. return readErr
  158. }); readErr != nil {
  159. return nil, readErr
  160. }
  161. // unmarshal storage configuration
  162. conf = &remote_pb.RemoteConf{}
  163. if unMarshalErr := proto.Unmarshal(oldContent, conf); unMarshalErr != nil {
  164. readErr = fmt.Errorf("unmarshal %s/%s: %v", DirectoryEtcRemote, storageName+REMOTE_STORAGE_CONF_SUFFIX, unMarshalErr)
  165. return
  166. }
  167. return
  168. }
  169. func DetectMountInfo(grpcDialOption grpc.DialOption, filerAddress string, dir string) (*remote_pb.RemoteStorageMapping, string, *remote_pb.RemoteStorageLocation, *remote_pb.RemoteConf, error) {
  170. mappings, listErr := ReadMountMappings(grpcDialOption, filerAddress)
  171. if listErr != nil {
  172. return nil, "", nil, nil, listErr
  173. }
  174. if dir == "" {
  175. return mappings, "", nil, nil, fmt.Errorf("need to specify '-dir' option")
  176. }
  177. var localMountedDir string
  178. var remoteStorageMountedLocation *remote_pb.RemoteStorageLocation
  179. for k, loc := range mappings.Mappings {
  180. if strings.HasPrefix(dir, k) {
  181. localMountedDir, remoteStorageMountedLocation = k, loc
  182. }
  183. }
  184. if localMountedDir == "" {
  185. return mappings, localMountedDir, remoteStorageMountedLocation, nil, fmt.Errorf("%s is not mounted", dir)
  186. }
  187. // find remote storage configuration
  188. remoteStorageConf, err := ReadRemoteStorageConf(grpcDialOption, filerAddress, remoteStorageMountedLocation.Name)
  189. if err != nil {
  190. return mappings, localMountedDir, remoteStorageMountedLocation, remoteStorageConf, err
  191. }
  192. return mappings, localMountedDir, remoteStorageMountedLocation, remoteStorageConf, nil
  193. }