mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-12-19 21:47:50 +08:00
254 lines
8.2 KiB
Go
254 lines
8.2 KiB
Go
package command
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/spf13/viper"
|
|
"google.golang.org/grpc"
|
|
"reflect"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/security"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
var (
|
|
metaBackup FilerMetaBackupOptions
|
|
)
|
|
|
|
type FilerMetaBackupOptions struct {
|
|
grpcDialOption grpc.DialOption
|
|
filerAddress *string
|
|
filerDirectory *string
|
|
restart *bool
|
|
backupFilerConfig *string
|
|
|
|
store filer.FilerStore
|
|
clientId int32
|
|
clientEpoch int32
|
|
}
|
|
|
|
func init() {
|
|
cmdFilerMetaBackup.Run = runFilerMetaBackup // break init cycle
|
|
metaBackup.filerAddress = cmdFilerMetaBackup.Flag.String("filer", "localhost:8888", "filer hostname:port")
|
|
metaBackup.filerDirectory = cmdFilerMetaBackup.Flag.String("filerDir", "/", "a folder on the filer")
|
|
metaBackup.restart = cmdFilerMetaBackup.Flag.Bool("restart", false, "copy the full metadata before async incremental backup")
|
|
metaBackup.backupFilerConfig = cmdFilerMetaBackup.Flag.String("config", "", "path to filer.toml specifying backup filer store")
|
|
metaBackup.clientId = util.RandomInt32()
|
|
}
|
|
|
|
var cmdFilerMetaBackup = &Command{
|
|
UsageLine: "filer.meta.backup [-filer=localhost:8888] [-filerDir=/] [-restart] -config=/path/to/backup_filer.toml",
|
|
Short: "continuously backup filer meta data changes to anther filer store specified in a backup_filer.toml",
|
|
Long: `continuously backup filer meta data changes.
|
|
The backup writes to another filer store specified in a backup_filer.toml.
|
|
|
|
weed filer.meta.backup -config=/path/to/backup_filer.toml -filer="localhost:8888"
|
|
weed filer.meta.backup -config=/path/to/backup_filer.toml -filer="localhost:8888" -restart
|
|
|
|
`,
|
|
}
|
|
|
|
func runFilerMetaBackup(cmd *Command, args []string) bool {
|
|
|
|
util.LoadConfiguration("security", false)
|
|
metaBackup.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
|
|
|
|
// load backup_filer.toml
|
|
v := viper.New()
|
|
v.SetConfigFile(*metaBackup.backupFilerConfig)
|
|
|
|
if err := v.ReadInConfig(); err != nil { // Handle errors reading the config file
|
|
glog.Fatalf("Failed to load %s file: %v\nPlease use this command to generate the a %s.toml file\n"+
|
|
" weed scaffold -config=%s -output=.\n\n\n",
|
|
*metaBackup.backupFilerConfig, err, "backup_filer", "filer")
|
|
}
|
|
|
|
if err := metaBackup.initStore(v); err != nil {
|
|
glog.V(0).Infof("init backup filer store: %v", err)
|
|
return true
|
|
}
|
|
|
|
missingPreviousBackup := false
|
|
_, err := metaBackup.getOffset()
|
|
if err != nil {
|
|
missingPreviousBackup = true
|
|
}
|
|
|
|
if *metaBackup.restart || missingPreviousBackup {
|
|
glog.V(0).Infof("traversing metadata tree...")
|
|
startTime := time.Now()
|
|
if err := metaBackup.traverseMetadata(); err != nil {
|
|
glog.Errorf("traverse meta data: %v", err)
|
|
return true
|
|
}
|
|
glog.V(0).Infof("metadata copied up to %v", startTime)
|
|
if err := metaBackup.setOffset(startTime); err != nil {
|
|
startTime = time.Now()
|
|
}
|
|
}
|
|
|
|
for {
|
|
err := metaBackup.streamMetadataBackup()
|
|
if err != nil {
|
|
glog.Errorf("filer meta backup from %s: %v", *metaBackup.filerAddress, err)
|
|
time.Sleep(1747 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) initStore(v *viper.Viper) error {
|
|
// load configuration for default filer store
|
|
hasDefaultStoreConfigured := false
|
|
for _, store := range filer.Stores {
|
|
if v.GetBool(store.GetName() + ".enabled") {
|
|
store = reflect.New(reflect.ValueOf(store).Elem().Type()).Interface().(filer.FilerStore)
|
|
if err := store.Initialize(v, store.GetName()+"."); err != nil {
|
|
glog.Fatalf("failed to initialize store for %s: %+v", store.GetName(), err)
|
|
}
|
|
glog.V(0).Infof("configured filer store to %s", store.GetName())
|
|
hasDefaultStoreConfigured = true
|
|
metaBackup.store = filer.NewFilerStoreWrapper(store)
|
|
break
|
|
}
|
|
}
|
|
if !hasDefaultStoreConfigured {
|
|
return fmt.Errorf("no filer store enabled in %s", v.ConfigFileUsed())
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) traverseMetadata() (err error) {
|
|
var saveErr error
|
|
|
|
traverseErr := filer_pb.TraverseBfs(metaBackup, util.FullPath(*metaBackup.filerDirectory), func(parentPath util.FullPath, entry *filer_pb.Entry) {
|
|
|
|
println("+", parentPath.Child(entry.Name))
|
|
if err := metaBackup.store.InsertEntry(context.Background(), filer.FromPbEntry(string(parentPath), entry)); err != nil {
|
|
saveErr = fmt.Errorf("insert entry error: %v\n", err)
|
|
return
|
|
}
|
|
|
|
})
|
|
|
|
if traverseErr != nil {
|
|
return fmt.Errorf("traverse: %v", traverseErr)
|
|
}
|
|
return saveErr
|
|
}
|
|
|
|
var (
|
|
MetaBackupKey = []byte("metaBackup")
|
|
)
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup() error {
|
|
|
|
startTime, err := metaBackup.getOffset()
|
|
if err != nil {
|
|
startTime = time.Now()
|
|
}
|
|
glog.V(0).Infof("streaming from %v", startTime)
|
|
|
|
store := metaBackup.store
|
|
|
|
eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
|
|
|
|
ctx := context.Background()
|
|
message := resp.EventNotification
|
|
|
|
if filer_pb.IsEmpty(resp) {
|
|
return nil
|
|
} else if filer_pb.IsCreate(resp) {
|
|
println("+", util.FullPath(message.NewParentPath).Child(message.NewEntry.Name))
|
|
entry := filer.FromPbEntry(message.NewParentPath, message.NewEntry)
|
|
return store.InsertEntry(ctx, entry)
|
|
} else if filer_pb.IsDelete(resp) {
|
|
println("-", util.FullPath(resp.Directory).Child(message.OldEntry.Name))
|
|
return store.DeleteEntry(ctx, util.FullPath(resp.Directory).Child(message.OldEntry.Name))
|
|
} else if filer_pb.IsUpdate(resp) {
|
|
println("~", util.FullPath(message.NewParentPath).Child(message.NewEntry.Name))
|
|
entry := filer.FromPbEntry(message.NewParentPath, message.NewEntry)
|
|
return store.UpdateEntry(ctx, entry)
|
|
} else {
|
|
// renaming
|
|
println("-", util.FullPath(resp.Directory).Child(message.OldEntry.Name))
|
|
if err := store.DeleteEntry(ctx, util.FullPath(resp.Directory).Child(message.OldEntry.Name)); err != nil {
|
|
return err
|
|
}
|
|
println("+", util.FullPath(message.NewParentPath).Child(message.NewEntry.Name))
|
|
return store.InsertEntry(ctx, filer.FromPbEntry(message.NewParentPath, message.NewEntry))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
processEventFnWithOffset := pb.AddOffsetFunc(eachEntryFunc, 3*time.Second, func(counter int64, lastTsNs int64) error {
|
|
lastTime := time.Unix(0, lastTsNs)
|
|
glog.V(0).Infof("meta backup %s progressed to %v %0.2f/sec", *metaBackup.filerAddress, lastTime, float64(counter)/float64(3))
|
|
return metaBackup.setOffset(lastTime)
|
|
})
|
|
|
|
metaBackup.clientEpoch++
|
|
|
|
metadataFollowOption := &pb.MetadataFollowOption{
|
|
ClientName: "meta_backup",
|
|
ClientId: metaBackup.clientId,
|
|
ClientEpoch: metaBackup.clientEpoch,
|
|
SelfSignature: 0,
|
|
PathPrefix: *metaBackup.filerDirectory,
|
|
AdditionalPathPrefixes: nil,
|
|
DirectoriesToWatch: nil,
|
|
StartTsNs: startTime.UnixNano(),
|
|
StopTsNs: 0,
|
|
EventErrorType: pb.RetryForeverOnError,
|
|
}
|
|
|
|
return pb.FollowMetadata(pb.ServerAddress(*metaBackup.filerAddress), metaBackup.grpcDialOption, metadataFollowOption, processEventFnWithOffset)
|
|
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) getOffset() (lastWriteTime time.Time, err error) {
|
|
value, err := metaBackup.store.KvGet(context.Background(), MetaBackupKey)
|
|
if err != nil {
|
|
return
|
|
}
|
|
tsNs := util.BytesToUint64(value)
|
|
|
|
return time.Unix(0, int64(tsNs)), nil
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) setOffset(lastWriteTime time.Time) error {
|
|
valueBuf := make([]byte, 8)
|
|
util.Uint64toBytes(valueBuf, uint64(lastWriteTime.UnixNano()))
|
|
|
|
if err := metaBackup.store.KvPut(context.Background(), MetaBackupKey, valueBuf); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var _ = filer_pb.FilerClient(&FilerMetaBackupOptions{})
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
|
|
|
|
return pb.WithFilerClient(streamingMode, metaBackup.clientId, pb.ServerAddress(*metaBackup.filerAddress), metaBackup.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
|
|
return fn(client)
|
|
})
|
|
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) AdjustedUrl(location *filer_pb.Location) string {
|
|
return location.Url
|
|
}
|
|
|
|
func (metaBackup *FilerMetaBackupOptions) GetDataCenter() string {
|
|
return ""
|
|
}
|