2021-03-01 08:22:27 +08:00
package command
import (
"fmt"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/replication/source"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
"google.golang.org/grpc"
"time"
)
type FilerBackupOptions struct {
isActivePassive * bool
filer * string
path * string
debug * bool
proxyByFiler * bool
timeAgo * time . Duration
}
var (
filerBackupOptions FilerBackupOptions
)
func init ( ) {
cmdFilerBackup . Run = runFilerBackup // break init cycle
filerBackupOptions . filer = cmdFilerBackup . Flag . String ( "filer" , "localhost:8888" , "filer of one SeaweedFS cluster" )
filerBackupOptions . path = cmdFilerBackup . Flag . String ( "filerPath" , "/" , "directory to sync on filer" )
filerBackupOptions . proxyByFiler = cmdFilerBackup . Flag . Bool ( "filerProxy" , false , "read and write file chunks by filer instead of volume servers" )
filerBackupOptions . debug = cmdFilerBackup . Flag . Bool ( "debug" , false , "debug mode to print out received files" )
filerBackupOptions . timeAgo = cmdFilerBackup . Flag . Duration ( "timeAgo" , 0 , "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"" )
}
var cmdFilerBackup = & Command {
UsageLine : "filer.backup -filer=<filerHost>:<filerPort> " ,
Short : "resume-able continuously replicate files from a SeaweedFS cluster to another location defined in replication.toml" ,
Long : ` resume - able continuously replicate files from a SeaweedFS cluster to another location defined in replication . toml
filer . backup listens on filer notifications . If any file is updated , it will fetch the updated content ,
and write to the destination . This is to replace filer . replicate command since additional message queue is not needed .
If restarted and "-timeAgo" is not set , the synchronization will resume from the previous checkpoints , persisted every minute .
A fresh sync will start from the earliest metadata logs . To reset the checkpoints , just set "-timeAgo" to a high value .
` ,
}
func runFilerBackup ( cmd * Command , args [ ] string ) bool {
util . LoadConfiguration ( "security" , false )
util . LoadConfiguration ( "replication" , true )
2021-08-03 16:30:35 +08:00
grpcDialOption := security . LoadClientTLS ( util . GetViper ( ) , "grpc.client" )
2021-12-30 16:23:57 +08:00
clientId := util . RandomInt32 ( )
2022-07-24 01:50:28 +08:00
var clientEpoch int32
2021-12-30 16:23:57 +08:00
2021-03-01 08:22:27 +08:00
for {
2022-07-24 01:50:28 +08:00
clientEpoch ++
err := doFilerBackup ( grpcDialOption , & filerBackupOptions , clientId , clientEpoch )
2021-03-01 08:22:27 +08:00
if err != nil {
glog . Errorf ( "backup from %s: %v" , * filerBackupOptions . filer , err )
time . Sleep ( 1747 * time . Millisecond )
}
}
return true
}
const (
BackupKeyPrefix = "backup."
)
2022-07-24 01:50:28 +08:00
func doFilerBackup ( grpcDialOption grpc . DialOption , backupOption * FilerBackupOptions , clientId int32 , clientEpoch int32 ) error {
2021-03-01 08:22:27 +08:00
// find data sink
config := util . GetViper ( )
dataSink := findSink ( config )
if dataSink == nil {
return fmt . Errorf ( "no data sink configured in replication.toml" )
}
2021-09-13 13:47:52 +08:00
sourceFiler := pb . ServerAddress ( * backupOption . filer )
2021-03-01 08:22:27 +08:00
sourcePath := * backupOption . path
timeAgo := * backupOption . timeAgo
targetPath := dataSink . GetSinkToDirectory ( )
debug := * backupOption . debug
// get start time for the data sink
startFrom := time . Unix ( 0 , 0 )
sinkId := util . HashStringToLong ( dataSink . GetName ( ) + dataSink . GetSinkToDirectory ( ) )
if timeAgo . Milliseconds ( ) == 0 {
lastOffsetTsNs , err := getOffset ( grpcDialOption , sourceFiler , BackupKeyPrefix , int32 ( sinkId ) )
if err != nil {
glog . V ( 0 ) . Infof ( "starting from %v" , startFrom )
} else {
startFrom = time . Unix ( 0 , lastOffsetTsNs )
glog . V ( 0 ) . Infof ( "resuming from %v" , startFrom )
}
} else {
startFrom = time . Now ( ) . Add ( - timeAgo )
glog . V ( 0 ) . Infof ( "start time is set to %v" , startFrom )
}
// create filer sink
filerSource := & source . FilerSource { }
2021-09-13 13:47:52 +08:00
filerSource . DoInitialize ( sourceFiler . ToHttpAddress ( ) , sourceFiler . ToGrpcAddress ( ) , sourcePath , * backupOption . proxyByFiler )
2021-03-01 08:22:27 +08:00
dataSink . SetSourceFiler ( filerSource )
processEventFn := genProcessFunction ( sourcePath , targetPath , dataSink , debug )
2021-08-09 13:30:36 +08:00
processEventFnWithOffset := pb . AddOffsetFunc ( processEventFn , 3 * time . Second , func ( counter int64 , lastTsNs int64 ) error {
2021-08-05 07:25:46 +08:00
glog . V ( 0 ) . Infof ( "backup %s progressed to %v %0.2f/sec" , sourceFiler , time . Unix ( 0 , lastTsNs ) , float64 ( counter ) / float64 ( 3 ) )
return setOffset ( grpcDialOption , sourceFiler , BackupKeyPrefix , int32 ( sinkId ) , lastTsNs )
2021-03-01 08:22:27 +08:00
} )
2022-07-24 01:50:28 +08:00
return pb . FollowMetadata ( sourceFiler , grpcDialOption , "backup_" + dataSink . GetName ( ) , clientId , clientEpoch , sourcePath , nil , startFrom . UnixNano ( ) , 0 , 0 , processEventFnWithOffset , pb . TrivialOnError )
2021-08-05 07:25:46 +08:00
2021-03-01 08:22:27 +08:00
}