aboutsummaryrefslogtreecommitdiff
path: root/weed/shell/command_remote_mount.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/shell/command_remote_mount.go')
-rw-r--r--weed/shell/command_remote_mount.go227
1 files changed, 227 insertions, 0 deletions
diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go
new file mode 100644
index 000000000..8f16d27c8
--- /dev/null
+++ b/weed/shell/command_remote_mount.go
@@ -0,0 +1,227 @@
+package shell
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "github.com/chrislusf/seaweedfs/weed/filer"
+ "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+ "github.com/chrislusf/seaweedfs/weed/remote_storage"
+ "github.com/chrislusf/seaweedfs/weed/util"
+ "github.com/golang/protobuf/proto"
+ "io"
+)
+
+func init() {
+ Commands = append(Commands, &commandRemoteMount{})
+}
+
+type commandRemoteMount struct {
+}
+
+func (c *commandRemoteMount) Name() string {
+ return "remote.mount"
+}
+
+func (c *commandRemoteMount) Help() string {
+ return `mount remote storage and pull its metadata
+
+ # assume a remote storage is configured to name "s3_1"
+ remote.configure -name=s3_1 -type=s3 -access_key=xxx -secret_key=yyy
+
+ # mount and pull one bucket
+ remote.mount -dir=xxx -remote=s3_1/bucket
+ # mount and pull one directory in the bucket
+ remote.mount -dir=xxx -remote=s3_1/bucket/dir1
+
+`
+}
+
+func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+ remoteMountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+
+ dir := remoteMountCommand.String("dir", "", "a directory in filer")
+ nonEmpty := remoteMountCommand.Bool("nonempty", false, "allows the mounting over a non-empty directory")
+ remote := remoteMountCommand.String("remote", "", "a directory in remote storage, ex. <storageName>/<bucket>/path/to/dir")
+
+ if err = remoteMountCommand.Parse(args); err != nil {
+ return nil
+ }
+
+ remoteStorageLocation := remote_storage.ParseLocation(*remote)
+
+ // find configuration for remote storage
+ // remotePath is /<bucket>/path/to/dir
+ remoteConf, err := c.findRemoteStorageConfiguration(commandEnv, writer, remoteStorageLocation)
+ if err != nil {
+ return fmt.Errorf("find configuration for %s: %v", *remote, err)
+ }
+
+ // pull metadata from remote
+ if err = c.pullMetadata(commandEnv, writer, *dir, *nonEmpty, remoteConf, remoteStorageLocation); err != nil {
+ return fmt.Errorf("pull metadata: %v", err)
+ }
+
+ // store a mount configuration in filer
+ if err = c.saveMountMapping(commandEnv, writer, *dir, remoteStorageLocation); err != nil {
+ return fmt.Errorf("save mount mapping: %v", err)
+ }
+
+ return nil
+}
+
+func (c *commandRemoteMount) findRemoteStorageConfiguration(commandEnv *CommandEnv, writer io.Writer, remote *filer_pb.RemoteStorageLocation) (conf *filer_pb.RemoteConf, err error) {
+
+ // read storage configuration data
+ var confBytes []byte
+ err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ confBytes, err = filer.ReadInsideFiler(client, filer.DirectoryEtcRemote, remote.Name+filer.REMOTE_STORAGE_CONF_SUFFIX)
+ return err
+ })
+ if err != nil {
+ err = fmt.Errorf("no remote storage configuration for %s : %v", remote.Name, err)
+ return
+ }
+
+ // unmarshal storage configuration
+ conf = &filer_pb.RemoteConf{}
+ if unMarshalErr := proto.Unmarshal(confBytes, conf); unMarshalErr != nil {
+ err = fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, remote.Name, unMarshalErr)
+ return
+ }
+
+ return
+}
+
+func (c *commandRemoteMount) pullMetadata(commandEnv *CommandEnv, writer io.Writer, dir string, nonEmpty bool, remoteConf *filer_pb.RemoteConf, remote *filer_pb.RemoteStorageLocation) error {
+
+ // find existing directory, and ensure the directory is empty
+ var mountToDir *filer_pb.Entry
+ err := commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ parent, name := util.FullPath(dir).DirAndName()
+ resp, lookupErr := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+ Directory: parent,
+ Name: name,
+ })
+ if lookupErr != nil {
+ return fmt.Errorf("lookup %s: %v", dir, lookupErr)
+ }
+ mountToDir = resp.Entry
+
+ mountToDirIsEmpty := true
+ listErr := filer_pb.SeaweedList(client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ mountToDirIsEmpty = false
+ return nil
+ }, "", false, 1)
+
+ if listErr != nil {
+ return fmt.Errorf("list %s: %v", dir, listErr)
+ }
+
+ if !mountToDirIsEmpty {
+ if !nonEmpty {
+ return fmt.Errorf("dir %s is not empty", dir)
+ }
+ }
+
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+
+ // visit remote storage
+ remoteStorage, err := remote_storage.GetRemoteStorage(remoteConf)
+ if err != nil {
+ return err
+ }
+
+ err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ ctx := context.Background()
+ err = remoteStorage.Traverse(remote, func(remoteDir, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error {
+ localDir := dir + remoteDir
+ println(util.NewFullPath(localDir, name))
+
+ lookupResponse, lookupErr := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ Directory: localDir,
+ Name: name,
+ })
+ var existingEntry *filer_pb.Entry
+ if lookupErr != nil {
+ if lookupErr != filer_pb.ErrNotFound {
+ return lookupErr
+ }
+ } else {
+ existingEntry = lookupResponse.Entry
+ }
+
+ if existingEntry == nil {
+ _, createErr := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
+ Directory: localDir,
+ Entry: &filer_pb.Entry{
+ Name: name,
+ IsDirectory: isDirectory,
+ Attributes: &filer_pb.FuseAttributes{
+ FileSize: uint64(remoteEntry.Size),
+ Mtime: remoteEntry.LastModifiedAt,
+ FileMode: uint32(0644),
+ },
+ RemoteEntry: remoteEntry,
+ },
+ })
+ return createErr
+ } else {
+ if existingEntry.RemoteEntry == nil || existingEntry.RemoteEntry.ETag != remoteEntry.ETag {
+ existingEntry.RemoteEntry = remoteEntry
+ existingEntry.Attributes.FileSize = uint64(remoteEntry.Size)
+ existingEntry.Attributes.Mtime = remoteEntry.LastModifiedAt
+ _, updateErr := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+ Directory: localDir,
+ Entry: existingEntry,
+ })
+ return updateErr
+ }
+ }
+ return nil
+ })
+ return err
+ })
+
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+func (c *commandRemoteMount) saveMountMapping(commandEnv *CommandEnv, writer io.Writer, dir string, remoteStorageLocation *filer_pb.RemoteStorageLocation) (err error) {
+
+ // read current mapping
+ var oldContent, newContent []byte
+ err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ oldContent, err = filer.ReadInsideFiler(client, filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE)
+ return err
+ })
+ if err != nil {
+ if err != filer_pb.ErrNotFound {
+ return fmt.Errorf("read existing mapping: %v", err)
+ }
+ }
+
+ // add new mapping
+ newContent, err = filer.AddMapping(oldContent, dir, remoteStorageLocation)
+ if err != nil {
+ return fmt.Errorf("add mapping %s~%s: %v", dir, remoteStorageLocation, err)
+ }
+
+ // save back
+ err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ return filer.SaveInsideFiler(client, filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE, newContent)
+ })
+ if err != nil {
+ return fmt.Errorf("save mapping: %v", err)
+ }
+
+ return nil
+}