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.

397 lines
14 KiB

  1. package command
  2. import (
  3. "fmt"
  4. "github.com/chrislusf/seaweedfs/weed/filer"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/pb"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
  9. "github.com/chrislusf/seaweedfs/weed/remote_storage"
  10. "github.com/chrislusf/seaweedfs/weed/replication/source"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. "github.com/golang/protobuf/proto"
  13. "math"
  14. "math/rand"
  15. "path/filepath"
  16. "strings"
  17. "time"
  18. )
  19. func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSource *source.FilerSource) error {
  20. // read filer remote storage mount mappings
  21. if detectErr := option.collectRemoteStorageConf(); detectErr != nil {
  22. return fmt.Errorf("read mount info: %v", detectErr)
  23. }
  24. eachEntryFunc, err := option.makeBucketedEventProcessor(filerSource)
  25. if err != nil {
  26. return err
  27. }
  28. processEventFnWithOffset := pb.AddOffsetFunc(eachEntryFunc, 3*time.Second, func(counter int64, lastTsNs int64) error {
  29. lastTime := time.Unix(0, lastTsNs)
  30. glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, lastTime, float64(counter)/float64(3))
  31. return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, lastTsNs)
  32. })
  33. lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo)
  34. return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync",
  35. option.bucketsDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
  36. }
  37. func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {
  38. handleCreateBucket := func(entry *filer_pb.Entry) error {
  39. if !entry.IsDirectory {
  40. return nil
  41. }
  42. if entry.RemoteEntry != nil {
  43. // this directory is imported from "remote.mount.buckets" or "remote.mount"
  44. return nil
  45. }
  46. if option.mappings.PrimaryBucketStorageName != "" && *option.createBucketAt == "" {
  47. *option.createBucketAt = option.mappings.PrimaryBucketStorageName
  48. glog.V(0).Infof("%s is set as the primary remote storage", *option.createBucketAt)
  49. }
  50. if len(option.mappings.Mappings) == 1 && *option.createBucketAt == "" {
  51. for k := range option.mappings.Mappings {
  52. *option.createBucketAt = k
  53. glog.V(0).Infof("%s is set as the only remote storage", *option.createBucketAt)
  54. }
  55. }
  56. if *option.createBucketAt == "" {
  57. return nil
  58. }
  59. remoteConf, found := option.remoteConfs[*option.createBucketAt]
  60. if !found {
  61. return fmt.Errorf("un-configured remote storage %s", *option.createBucketAt)
  62. }
  63. client, err := remote_storage.GetRemoteStorage(remoteConf)
  64. if err != nil {
  65. return err
  66. }
  67. bucketName := strings.ToLower(entry.Name)
  68. if *option.include != "" {
  69. if ok, _ := filepath.Match(*option.include, entry.Name); !ok {
  70. return nil
  71. }
  72. }
  73. if *option.exclude != "" {
  74. if ok, _ := filepath.Match(*option.exclude, entry.Name); ok {
  75. return nil
  76. }
  77. }
  78. bucketPath := util.FullPath(option.bucketsDir).Child(entry.Name)
  79. remoteLocation, found := option.mappings.Mappings[string(bucketPath)]
  80. if !found {
  81. if *option.createBucketRandomSuffix {
  82. // https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
  83. if len(bucketName)+5 > 63 {
  84. bucketName = bucketName[:58]
  85. }
  86. bucketName = fmt.Sprintf("%s-%04d", bucketName, rand.Uint32()%10000)
  87. }
  88. remoteLocation = &remote_pb.RemoteStorageLocation{
  89. Name: *option.createBucketAt,
  90. Bucket: bucketName,
  91. Path: "/",
  92. }
  93. // need to add new mapping here before getting updates from metadata tailing
  94. option.mappings.Mappings[string(bucketPath)] = remoteLocation
  95. } else {
  96. bucketName = remoteLocation.Bucket
  97. }
  98. glog.V(0).Infof("create bucket %s", bucketName)
  99. if err := client.CreateBucket(bucketName); err != nil {
  100. return fmt.Errorf("create bucket %s in %s: %v", bucketName, remoteConf.Name, err)
  101. }
  102. return filer.InsertMountMapping(option, string(bucketPath), remoteLocation)
  103. }
  104. handleDeleteBucket := func(entry *filer_pb.Entry) error {
  105. if !entry.IsDirectory {
  106. return nil
  107. }
  108. client, remoteStorageMountLocation, err := option.findRemoteStorageClient(entry.Name)
  109. if err != nil {
  110. return fmt.Errorf("findRemoteStorageClient %s: %v", entry.Name, err)
  111. }
  112. glog.V(0).Infof("delete remote bucket %s", remoteStorageMountLocation.Bucket)
  113. if err := client.DeleteBucket(remoteStorageMountLocation.Bucket); err != nil {
  114. return fmt.Errorf("delete remote bucket %s: %v", remoteStorageMountLocation.Bucket, err)
  115. }
  116. bucketPath := util.FullPath(option.bucketsDir).Child(entry.Name)
  117. return filer.DeleteMountMapping(option, string(bucketPath))
  118. }
  119. handleEtcRemoteChanges := func(resp *filer_pb.SubscribeMetadataResponse) error {
  120. message := resp.EventNotification
  121. if message.NewEntry != nil {
  122. // update
  123. if message.NewEntry.Name == filer.REMOTE_STORAGE_MOUNT_FILE {
  124. newMappings, readErr := filer.UnmarshalRemoteStorageMappings(message.NewEntry.Content)
  125. if readErr != nil {
  126. return fmt.Errorf("unmarshal mappings: %v", readErr)
  127. }
  128. option.mappings = newMappings
  129. }
  130. if strings.HasSuffix(message.NewEntry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) {
  131. conf := &remote_pb.RemoteConf{}
  132. if err := proto.Unmarshal(message.NewEntry.Content, conf); err != nil {
  133. return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, message.NewEntry.Name, err)
  134. }
  135. option.remoteConfs[conf.Name] = conf
  136. }
  137. } else if message.OldEntry != nil {
  138. // deletion
  139. if strings.HasSuffix(message.OldEntry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) {
  140. conf := &remote_pb.RemoteConf{}
  141. if err := proto.Unmarshal(message.OldEntry.Content, conf); err != nil {
  142. return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, message.OldEntry.Name, err)
  143. }
  144. delete(option.remoteConfs, conf.Name)
  145. }
  146. }
  147. return nil
  148. }
  149. eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
  150. message := resp.EventNotification
  151. if strings.HasPrefix(resp.Directory, filer.DirectoryEtcRemote) {
  152. return handleEtcRemoteChanges(resp)
  153. }
  154. if message.OldEntry == nil && message.NewEntry == nil {
  155. return nil
  156. }
  157. if message.OldEntry == nil && message.NewEntry != nil {
  158. if message.NewParentPath == option.bucketsDir {
  159. return handleCreateBucket(message.NewEntry)
  160. }
  161. if !filer.HasData(message.NewEntry) {
  162. return nil
  163. }
  164. bucket, remoteStorageMountLocation, remoteStorage, ok := option.detectBucketInfo(message.NewParentPath)
  165. if !ok {
  166. return nil
  167. }
  168. client, err := remote_storage.GetRemoteStorage(remoteStorage)
  169. if err != nil {
  170. return err
  171. }
  172. glog.V(2).Infof("create: %+v", resp)
  173. if !shouldSendToRemote(message.NewEntry) {
  174. glog.V(2).Infof("skipping creating: %+v", resp)
  175. return nil
  176. }
  177. dest := toRemoteStorageLocation(bucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
  178. if message.NewEntry.IsDirectory {
  179. glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest))
  180. return client.WriteDirectory(dest, message.NewEntry)
  181. }
  182. glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
  183. remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, dest)
  184. if writeErr != nil {
  185. return writeErr
  186. }
  187. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  188. }
  189. if message.OldEntry != nil && message.NewEntry == nil {
  190. if resp.Directory == option.bucketsDir {
  191. return handleDeleteBucket(message.OldEntry)
  192. }
  193. bucket, remoteStorageMountLocation, remoteStorage, ok := option.detectBucketInfo(resp.Directory)
  194. if !ok {
  195. return nil
  196. }
  197. client, err := remote_storage.GetRemoteStorage(remoteStorage)
  198. if err != nil {
  199. return err
  200. }
  201. glog.V(2).Infof("delete: %+v", resp)
  202. dest := toRemoteStorageLocation(bucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
  203. if message.OldEntry.IsDirectory {
  204. glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest))
  205. return client.RemoveDirectory(dest)
  206. }
  207. glog.V(0).Infof("delete %s", remote_storage.FormatLocation(dest))
  208. return client.DeleteFile(dest)
  209. }
  210. if message.OldEntry != nil && message.NewEntry != nil {
  211. if resp.Directory == option.bucketsDir {
  212. if message.NewParentPath == option.bucketsDir {
  213. if message.OldEntry.Name == message.NewEntry.Name {
  214. return nil
  215. }
  216. if err := handleCreateBucket(message.NewEntry); err != nil {
  217. return err
  218. }
  219. if err := handleDeleteBucket(message.OldEntry); err != nil {
  220. return err
  221. }
  222. }
  223. }
  224. oldBucket, oldRemoteStorageMountLocation, oldRemoteStorage, oldOk := option.detectBucketInfo(resp.Directory)
  225. newBucket, newRemoteStorageMountLocation, newRemoteStorage, newOk := option.detectBucketInfo(message.NewParentPath)
  226. if oldOk && newOk {
  227. if !shouldSendToRemote(message.NewEntry) {
  228. glog.V(2).Infof("skipping updating: %+v", resp)
  229. return nil
  230. }
  231. client, err := remote_storage.GetRemoteStorage(oldRemoteStorage)
  232. if err != nil {
  233. return err
  234. }
  235. if resp.Directory == message.NewParentPath && message.OldEntry.Name == message.NewEntry.Name {
  236. // update the same entry
  237. if message.NewEntry.IsDirectory {
  238. // update directory property
  239. return nil
  240. }
  241. if filer.IsSameData(message.OldEntry, message.NewEntry) {
  242. glog.V(2).Infof("update meta: %+v", resp)
  243. oldDest := toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation)
  244. return client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry)
  245. } else {
  246. newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation)
  247. remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest)
  248. if writeErr != nil {
  249. return writeErr
  250. }
  251. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  252. }
  253. }
  254. }
  255. // the following is entry rename
  256. if oldOk {
  257. client, err := remote_storage.GetRemoteStorage(oldRemoteStorage)
  258. if err != nil {
  259. return err
  260. }
  261. oldDest := toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation)
  262. if message.OldEntry.IsDirectory {
  263. return client.RemoveDirectory(oldDest)
  264. }
  265. glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
  266. if err := client.DeleteFile(oldDest); err != nil {
  267. return err
  268. }
  269. }
  270. if newOk {
  271. if !shouldSendToRemote(message.NewEntry) {
  272. glog.V(2).Infof("skipping updating: %+v", resp)
  273. return nil
  274. }
  275. client, err := remote_storage.GetRemoteStorage(newRemoteStorage)
  276. if err != nil {
  277. return err
  278. }
  279. newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation)
  280. if message.NewEntry.IsDirectory {
  281. return client.WriteDirectory(newDest, message.NewEntry)
  282. }
  283. remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewEntry, newDest)
  284. if writeErr != nil {
  285. return writeErr
  286. }
  287. return updateLocalEntry(&remoteSyncOptions, message.NewParentPath, message.NewEntry, remoteEntry)
  288. }
  289. }
  290. return nil
  291. }
  292. return eachEntryFunc, nil
  293. }
  294. func (option *RemoteGatewayOptions) findRemoteStorageClient(bucketName string) (client remote_storage.RemoteStorageClient, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, err error) {
  295. bucket := util.FullPath(option.bucketsDir).Child(bucketName)
  296. var isMounted bool
  297. remoteStorageMountLocation, isMounted = option.mappings.Mappings[string(bucket)]
  298. if !isMounted {
  299. return nil, remoteStorageMountLocation, fmt.Errorf("%s is not mounted", bucket)
  300. }
  301. remoteConf, hasClient := option.remoteConfs[remoteStorageMountLocation.Name]
  302. if !hasClient {
  303. return nil, remoteStorageMountLocation, fmt.Errorf("%s mounted to un-configured %+v", bucket, remoteStorageMountLocation)
  304. }
  305. client, err = remote_storage.GetRemoteStorage(remoteConf)
  306. if err != nil {
  307. return nil, remoteStorageMountLocation, err
  308. }
  309. return client, remoteStorageMountLocation, nil
  310. }
  311. func (option *RemoteGatewayOptions) detectBucketInfo(actualDir string) (bucket util.FullPath, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, remoteConf *remote_pb.RemoteConf, ok bool) {
  312. bucket, ok = extractBucketPath(option.bucketsDir, actualDir)
  313. if !ok {
  314. return "", nil, nil, false
  315. }
  316. var isMounted bool
  317. remoteStorageMountLocation, isMounted = option.mappings.Mappings[string(bucket)]
  318. if !isMounted {
  319. glog.Warningf("%s is not mounted", bucket)
  320. return "", nil, nil, false
  321. }
  322. var hasClient bool
  323. remoteConf, hasClient = option.remoteConfs[remoteStorageMountLocation.Name]
  324. if !hasClient {
  325. glog.Warningf("%s mounted to un-configured %+v", bucket, remoteStorageMountLocation)
  326. return "", nil, nil, false
  327. }
  328. return bucket, remoteStorageMountLocation, remoteConf, true
  329. }
  330. func extractBucketPath(bucketsDir, dir string) (util.FullPath, bool) {
  331. if !strings.HasPrefix(dir, bucketsDir+"/") {
  332. return "", false
  333. }
  334. parts := strings.SplitN(dir[len(bucketsDir)+1:], "/", 2)
  335. return util.FullPath(bucketsDir).Child(parts[0]), true
  336. }
  337. func (option *RemoteGatewayOptions) collectRemoteStorageConf() (err error) {
  338. if mappings, err := filer.ReadMountMappings(option.grpcDialOption, pb.ServerAddress(*option.filerAddress)); err != nil {
  339. return err
  340. } else {
  341. option.mappings = mappings
  342. }
  343. option.remoteConfs = make(map[string]*remote_pb.RemoteConf)
  344. var lastConfName string
  345. err = filer_pb.List(option, filer.DirectoryEtcRemote, "", func(entry *filer_pb.Entry, isLast bool) error {
  346. if !strings.HasSuffix(entry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) {
  347. return nil
  348. }
  349. conf := &remote_pb.RemoteConf{}
  350. if err := proto.Unmarshal(entry.Content, conf); err != nil {
  351. return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, entry.Name, err)
  352. }
  353. option.remoteConfs[conf.Name] = conf
  354. lastConfName = conf.Name
  355. return nil
  356. }, "", false, math.MaxUint32)
  357. if option.mappings.PrimaryBucketStorageName == "" && len(option.remoteConfs) == 1 {
  358. glog.V(0).Infof("%s is set to the default remote storage", lastConfName)
  359. option.mappings.PrimaryBucketStorageName = lastConfName
  360. }
  361. return
  362. }