seaweedfs/weed/shell/command_remote_mount.go

250 lines
7.3 KiB
Go
Raw Normal View History

2021-07-27 05:53:44 +00:00
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"
2021-08-01 05:39:38 +00:00
"github.com/golang/protobuf/jsonpb"
2021-08-09 05:26:37 +00:00
"github.com/golang/protobuf/proto"
2021-07-27 05:53:44 +00:00
"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
2021-08-10 09:48:41 +00:00
# assume a remote storage is configured to name "cloud1"
remote.configure -name=cloud1 -type=s3 -access_key=xxx -secret_key=yyy
2021-07-27 05:53:44 +00:00
# mount and pull one bucket
2021-08-10 09:48:41 +00:00
remote.mount -dir=xxx -remote=cloud1/bucket
2021-07-27 05:53:44 +00:00
# mount and pull one directory in the bucket
2021-08-10 09:48:41 +00:00
remote.mount -dir=xxx -remote=cloud1/bucket/dir1
2021-07-27 05:53:44 +00:00
2021-08-09 05:30:12 +00:00
# after mount, start a separate process to write updates to remote storage
weed filer.remote.sync -filer=<filerHost>:<filerPort> -dir=xxx
2021-07-27 05:53:44 +00:00
`
}
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
}
2021-08-01 05:39:38 +00:00
if *dir == "" {
2021-08-09 05:26:37 +00:00
_, err = listExistingRemoteStorageMounts(commandEnv, writer)
return err
2021-08-01 05:39:38 +00:00
}
remoteStorageLocation := remote_storage.ParseLocation(*remote)
2021-07-27 10:26:35 +00:00
2021-07-27 05:53:44 +00:00
// find configuration for remote storage
2021-07-27 10:26:35 +00:00
// remotePath is /<bucket>/path/to/dir
remoteConf, err := c.findRemoteStorageConfiguration(commandEnv, writer, remoteStorageLocation)
2021-07-27 05:53:44 +00:00
if err != nil {
return fmt.Errorf("find configuration for %s: %v", *remote, err)
}
// pull metadata from remote
2021-07-27 10:26:35 +00:00
if err = c.pullMetadata(commandEnv, writer, *dir, *nonEmpty, remoteConf, remoteStorageLocation); err != nil {
2021-07-27 05:53:44 +00:00
return fmt.Errorf("pull metadata: %v", err)
}
// store a mount configuration in filer
2021-07-27 10:26:35 +00:00
if err = c.saveMountMapping(commandEnv, writer, *dir, remoteStorageLocation); err != nil {
return fmt.Errorf("save mount mapping: %v", err)
}
2021-07-27 05:53:44 +00:00
return nil
}
2021-08-09 05:26:37 +00:00
func listExistingRemoteStorageMounts(commandEnv *CommandEnv, writer io.Writer) (mappings *filer_pb.RemoteStorageMapping, err error) {
2021-08-01 05:39:38 +00:00
// read current mapping
2021-08-09 05:26:37 +00:00
mappings, err = filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress)
if err != nil {
return mappings, err
2021-08-01 05:39:38 +00:00
}
2021-08-09 05:26:37 +00:00
jsonPrintln(writer, mappings)
return
}
func jsonPrintln(writer io.Writer, message proto.Message) error {
2021-08-01 05:39:38 +00:00
m := jsonpb.Marshaler{
EmitDefaults: false,
Indent: " ",
}
2021-08-09 05:26:37 +00:00
err := m.Marshal(writer, message)
2021-08-08 08:21:42 +00:00
fmt.Fprintln(writer)
2021-08-09 05:26:37 +00:00
return err
2021-08-01 05:39:38 +00:00
}
func (c *commandRemoteMount) findRemoteStorageConfiguration(commandEnv *CommandEnv, writer io.Writer, remote *filer_pb.RemoteStorageLocation) (conf *filer_pb.RemoteConf, err error) {
2021-07-27 05:53:44 +00:00
2021-08-04 23:25:46 +00:00
return filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, remote.Name)
2021-07-27 05:53:44 +00:00
}
func (c *commandRemoteMount) pullMetadata(commandEnv *CommandEnv, writer io.Writer, dir string, nonEmpty bool, remoteConf *filer_pb.RemoteConf, remote *filer_pb.RemoteStorageLocation) error {
2021-07-27 05:53:44 +00:00
// find existing directory, and ensure the directory is empty
err := commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
parent, name := util.FullPath(dir).DirAndName()
2021-08-04 07:31:06 +00:00
_, lookupErr := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
2021-07-27 05:53:44 +00:00
Directory: parent,
Name: name,
})
if lookupErr != nil {
return fmt.Errorf("lookup %s: %v", dir, lookupErr)
}
2021-07-27 10:26:35 +00:00
mountToDirIsEmpty := true
2021-07-27 05:53:44 +00:00
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()
2021-07-27 10:26:35 +00:00
err = remoteStorage.Traverse(remote, func(remoteDir, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error {
2021-07-27 05:53:44 +00:00
localDir := dir + remoteDir
println(util.NewFullPath(localDir, name))
lookupResponse, lookupErr := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
Directory: localDir,
Name: name,
})
2021-07-27 10:26:35 +00:00
var existingEntry *filer_pb.Entry
2021-07-27 05:53:44 +00:00
if lookupErr != nil {
if lookupErr != filer_pb.ErrNotFound {
return lookupErr
}
2021-07-27 10:26:35 +00:00
} else {
existingEntry = lookupResponse.Entry
2021-07-27 05:53:44 +00:00
}
if existingEntry == nil {
_, createErr := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
Directory: localDir,
Entry: &filer_pb.Entry{
Name: name,
IsDirectory: isDirectory,
Attributes: &filer_pb.FuseAttributes{
2021-08-09 21:35:18 +00:00
FileSize: uint64(remoteEntry.RemoteSize),
Mtime: remoteEntry.RemoteMtime,
2021-07-27 05:53:44 +00:00
FileMode: uint32(0644),
},
RemoteEntry: remoteEntry,
},
})
return createErr
} else {
2021-08-09 21:35:18 +00:00
if existingEntry.RemoteEntry == nil || existingEntry.RemoteEntry.RemoteETag != remoteEntry.RemoteETag {
return doSaveRemoteEntry(client, localDir, existingEntry, remoteEntry)
2021-07-27 05:53:44 +00:00
}
}
return nil
})
return err
})
if err != nil {
return err
}
return nil
}
2021-07-27 10:26:35 +00:00
func (c *commandRemoteMount) saveMountMapping(commandEnv *CommandEnv, writer io.Writer, dir string, remoteStorageLocation *filer_pb.RemoteStorageLocation) (err error) {
2021-07-27 10:26:35 +00:00
// 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
2021-08-01 05:39:38 +00:00
newContent, err = filer.AddRemoteStorageMapping(oldContent, dir, remoteStorageLocation)
2021-07-27 10:26:35 +00:00
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
}
2021-08-09 21:35:18 +00:00
func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, existingEntry *filer_pb.Entry, remoteEntry *filer_pb.RemoteEntry) error {
existingEntry.RemoteEntry = remoteEntry
existingEntry.Attributes.FileSize = uint64(remoteEntry.RemoteSize)
existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime
_, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
Directory: localDir,
Entry: existingEntry,
})
if updateErr != nil {
return updateErr
}
return nil
2021-08-09 21:37:34 +00:00
}