aboutsummaryrefslogtreecommitdiff
path: root/weed/command
diff options
context:
space:
mode:
Diffstat (limited to 'weed/command')
-rw-r--r--weed/command/filer_backup.go20
-rw-r--r--weed/command/filer_meta_backup.go8
-rw-r--r--weed/command/filer_meta_tail.go2
-rw-r--r--weed/command/filer_remote_gateway.go1
-rw-r--r--weed/command/filer_remote_gateway_buckets.go3
-rw-r--r--weed/command/filer_remote_sync.go1
-rw-r--r--weed/command/filer_remote_sync_dir.go3
-rw-r--r--weed/command/filer_sync.go63
-rw-r--r--weed/command/master.go2
-rw-r--r--weed/command/scaffold/replication.toml2
10 files changed, 82 insertions, 23 deletions
diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go
index d191c693b..62477227b 100644
--- a/weed/command/filer_backup.go
+++ b/weed/command/filer_backup.go
@@ -8,6 +8,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
"google.golang.org/grpc"
+ "strings"
"time"
)
@@ -15,6 +16,7 @@ type FilerBackupOptions struct {
isActivePassive *bool
filer *string
path *string
+ excludePaths *string
debug *bool
proxyByFiler *bool
timeAgo *time.Duration
@@ -28,6 +30,7 @@ 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.excludePaths = cmdFilerBackup.Flag.String("filerExcludePaths", "", "exclude directories 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\"")
@@ -55,9 +58,11 @@ func runFilerBackup(cmd *Command, args []string) bool {
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
clientId := util.RandomInt32()
+ var clientEpoch int32
for {
- err := doFilerBackup(grpcDialOption, &filerBackupOptions, clientId)
+ clientEpoch++
+ err := doFilerBackup(grpcDialOption, &filerBackupOptions, clientId, clientEpoch)
if err != nil {
glog.Errorf("backup from %s: %v", *filerBackupOptions.filer, err)
time.Sleep(1747 * time.Millisecond)
@@ -71,7 +76,7 @@ const (
BackupKeyPrefix = "backup."
)
-func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOptions, clientId int32) error {
+func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOptions, clientId int32, clientEpoch int32) error {
// find data sink
config := util.GetViper()
@@ -82,6 +87,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
sourceFiler := pb.ServerAddress(*backupOption.filer)
sourcePath := *backupOption.path
+ excludePaths := strings.Split(*backupOption.excludePaths, ",")
timeAgo := *backupOption.timeAgo
targetPath := dataSink.GetSinkToDirectory()
debug := *backupOption.debug
@@ -104,16 +110,20 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
// create filer sink
filerSource := &source.FilerSource{}
- filerSource.DoInitialize(sourceFiler.ToHttpAddress(), sourceFiler.ToGrpcAddress(), sourcePath, *backupOption.proxyByFiler)
+ filerSource.DoInitialize(
+ sourceFiler.ToHttpAddress(),
+ sourceFiler.ToGrpcAddress(),
+ sourcePath,
+ *backupOption.proxyByFiler)
dataSink.SetSourceFiler(filerSource)
- processEventFn := genProcessFunction(sourcePath, targetPath, dataSink, debug)
+ processEventFn := genProcessFunction(sourcePath, targetPath, excludePaths, dataSink, debug)
processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error {
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)
})
- return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(), clientId, sourcePath, nil, startFrom.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
+ return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(), clientId, clientEpoch, sourcePath, nil, startFrom.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
}
diff --git a/weed/command/filer_meta_backup.go b/weed/command/filer_meta_backup.go
index cf679885d..54cfc31b7 100644
--- a/weed/command/filer_meta_backup.go
+++ b/weed/command/filer_meta_backup.go
@@ -27,8 +27,9 @@ type FilerMetaBackupOptions struct {
restart *bool
backupFilerConfig *string
- store filer.FilerStore
- clientId int32
+ store filer.FilerStore
+ clientId int32
+ clientEpoch int32
}
func init() {
@@ -194,7 +195,8 @@ func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup() error {
return metaBackup.setOffset(lastTime)
})
- return pb.FollowMetadata(pb.ServerAddress(*metaBackup.filerAddress), metaBackup.grpcDialOption, "meta_backup", metaBackup.clientId,
+ metaBackup.clientEpoch++
+ return pb.FollowMetadata(pb.ServerAddress(*metaBackup.filerAddress), metaBackup.grpcDialOption, "meta_backup", metaBackup.clientId, metaBackup.clientEpoch,
*metaBackup.filerDirectory, nil, startTime.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
}
diff --git a/weed/command/filer_meta_tail.go b/weed/command/filer_meta_tail.go
index 66a87c3d9..e319e93e6 100644
--- a/weed/command/filer_meta_tail.go
+++ b/weed/command/filer_meta_tail.go
@@ -110,7 +110,7 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
untilTsNs = time.Now().Add(-*tailStop).UnixNano()
}
- tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpcDialOption, "tail", clientId, *tailTarget, nil,
+ tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpcDialOption, "tail", clientId, 0, *tailTarget, nil,
time.Now().Add(-*tailStart).UnixNano(), untilTsNs, 0, func(resp *filer_pb.SubscribeMetadataResponse) error {
if !shouldPrint(resp) {
return nil
diff --git a/weed/command/filer_remote_gateway.go b/weed/command/filer_remote_gateway.go
index 33454f378..9389ff79b 100644
--- a/weed/command/filer_remote_gateway.go
+++ b/weed/command/filer_remote_gateway.go
@@ -29,6 +29,7 @@ type RemoteGatewayOptions struct {
remoteConfs map[string]*remote_pb.RemoteConf
bucketsDir string
clientId int32
+ clientEpoch int32
}
var _ = filer_pb.FilerClient(&RemoteGatewayOptions{})
diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go
index 9fe0e29df..121f46114 100644
--- a/weed/command/filer_remote_gateway_buckets.go
+++ b/weed/command/filer_remote_gateway_buckets.go
@@ -39,7 +39,8 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo)
- return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId,
+ option.clientEpoch++
+ return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId, option.clientEpoch,
option.bucketsDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
}
diff --git a/weed/command/filer_remote_sync.go b/weed/command/filer_remote_sync.go
index d6ccf7b79..49334f13d 100644
--- a/weed/command/filer_remote_sync.go
+++ b/weed/command/filer_remote_sync.go
@@ -19,6 +19,7 @@ type RemoteSyncOptions struct {
timeAgo *time.Duration
dir *string
clientId int32
+ clientEpoch int32
}
var _ = filer_pb.FilerClient(&RemoteSyncOptions{})
diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go
index 5fc20be9a..b54bfcf71 100644
--- a/weed/command/filer_remote_sync_dir.go
+++ b/weed/command/filer_remote_sync_dir.go
@@ -40,7 +40,8 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo)
- return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId,
+ option.clientEpoch++
+ return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId, option.clientEpoch,
mountedDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
}
diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go
index 1550d155a..abf13a81d 100644
--- a/weed/command/filer_sync.go
+++ b/weed/command/filer_sync.go
@@ -26,7 +26,9 @@ type SyncOptions struct {
filerA *string
filerB *string
aPath *string
+ aExcludePaths *string
bPath *string
+ bExcludePaths *string
aReplication *string
bReplication *string
aCollection *string
@@ -43,6 +45,7 @@ type SyncOptions struct {
bProxyByFiler *bool
metricsHttpPort *int
clientId int32
+ clientEpoch int32
}
var (
@@ -57,7 +60,9 @@ func init() {
syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
+ syncOptions.aExcludePaths = cmdFilerSynchronize.Flag.String("a.excludePaths", "", "exclude directories to sync on filer A")
syncOptions.bPath = cmdFilerSynchronize.Flag.String("b.path", "/", "directory to sync on filer B")
+ syncOptions.bExcludePaths = cmdFilerSynchronize.Flag.String("b.excludePaths", "", "exclude directories to sync on filer B")
syncOptions.aReplication = cmdFilerSynchronize.Flag.String("a.replication", "", "replication on filer A")
syncOptions.bReplication = cmdFilerSynchronize.Flag.String("b.replication", "", "replication on filer B")
syncOptions.aCollection = cmdFilerSynchronize.Flag.String("a.collection", "", "collection on filer A")
@@ -131,9 +136,25 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
os.Exit(2)
}
for {
- err := doSubscribeFilerMetaChanges(syncOptions.clientId, grpcDialOption, filerA, *syncOptions.aPath, *syncOptions.aProxyByFiler, filerB,
- *syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bProxyByFiler, *syncOptions.bDiskType,
- *syncOptions.bDebug, aFilerSignature, bFilerSignature)
+ syncOptions.clientEpoch++
+ err := doSubscribeFilerMetaChanges(
+ syncOptions.clientId,
+ syncOptions.clientEpoch,
+ grpcDialOption,
+ filerA,
+ *syncOptions.aPath,
+ strings.Split(*syncOptions.aExcludePaths, ","),
+ *syncOptions.aProxyByFiler,
+ filerB,
+ *syncOptions.bPath,
+ *syncOptions.bReplication,
+ *syncOptions.bCollection,
+ *syncOptions.bTtlSec,
+ *syncOptions.bProxyByFiler,
+ *syncOptions.bDiskType,
+ *syncOptions.bDebug,
+ aFilerSignature,
+ bFilerSignature)
if err != nil {
glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
time.Sleep(1747 * time.Millisecond)
@@ -151,9 +172,25 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
}
go func() {
for {
- err := doSubscribeFilerMetaChanges(syncOptions.clientId, grpcDialOption, filerB, *syncOptions.bPath, *syncOptions.bProxyByFiler, filerA,
- *syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aProxyByFiler, *syncOptions.aDiskType,
- *syncOptions.aDebug, bFilerSignature, aFilerSignature)
+ syncOptions.clientEpoch++
+ err := doSubscribeFilerMetaChanges(
+ syncOptions.clientId,
+ syncOptions.clientEpoch,
+ grpcDialOption,
+ filerB,
+ *syncOptions.bPath,
+ strings.Split(*syncOptions.bExcludePaths, ","),
+ *syncOptions.bProxyByFiler,
+ filerA,
+ *syncOptions.aPath,
+ *syncOptions.aReplication,
+ *syncOptions.aCollection,
+ *syncOptions.aTtlSec,
+ *syncOptions.aProxyByFiler,
+ *syncOptions.aDiskType,
+ *syncOptions.aDebug,
+ bFilerSignature,
+ aFilerSignature)
if err != nil {
glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
time.Sleep(2147 * time.Millisecond)
@@ -183,7 +220,7 @@ func initOffsetFromTsMs(grpcDialOption grpc.DialOption, targetFiler pb.ServerAdd
return nil
}
-func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sourcePath string, sourceReadChunkFromFiler bool, targetFiler pb.ServerAddress, targetPath string,
+func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, grpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sourcePath string, sourceExcludePaths []string, sourceReadChunkFromFiler bool, targetFiler pb.ServerAddress, targetPath string,
replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, sourceFilerSignature int32, targetFilerSignature int32) error {
// if first time, start from now
@@ -202,7 +239,7 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
filerSink.DoInitialize(targetFiler.ToHttpAddress(), targetFiler.ToGrpcAddress(), targetPath, replicationStr, collection, ttlSec, diskType, grpcDialOption, sinkWriteChunkByFiler)
filerSink.SetSourceFiler(filerSource)
- persistEventFn := genProcessFunction(sourcePath, targetPath, filerSink, debug)
+ persistEventFn := genProcessFunction(sourcePath, targetPath, sourceExcludePaths, filerSink, debug)
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
message := resp.EventNotification
@@ -226,7 +263,7 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
return setOffset(grpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature, lastTsNs)
})
- return pb.FollowMetadata(sourceFiler, grpcDialOption, clientName, clientId,
+ return pb.FollowMetadata(sourceFiler, grpcDialOption, clientName, clientId, clientEpoch,
sourcePath, nil, sourceFilerOffsetTsNs, 0, targetFilerSignature, processEventFnWithOffset, pb.RetryForeverOnError)
}
@@ -299,7 +336,7 @@ func setOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signature
}
-func genProcessFunction(sourcePath string, targetPath string, dataSink sink.ReplicationSink, debug bool) func(resp *filer_pb.SubscribeMetadataResponse) error {
+func genProcessFunction(sourcePath string, targetPath string, excludePaths []string, dataSink sink.ReplicationSink, debug bool) func(resp *filer_pb.SubscribeMetadataResponse) error {
// process function
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
message := resp.EventNotification
@@ -319,7 +356,11 @@ func genProcessFunction(sourcePath string, targetPath string, dataSink sink.Repl
if !strings.HasPrefix(resp.Directory, sourcePath) {
return nil
}
-
+ for _, excludePath := range excludePaths {
+ if strings.HasPrefix(resp.Directory, excludePath) {
+ return nil
+ }
+ }
// handle deletions
if filer_pb.IsDelete(resp) {
if !strings.HasPrefix(string(sourceOldKey), sourcePath) {
diff --git a/weed/command/master.go b/weed/command/master.go
index ab8466d47..448c3491f 100644
--- a/weed/command/master.go
+++ b/weed/command/master.go
@@ -9,9 +9,9 @@ import (
"strings"
"time"
- "github.com/chrislusf/raft/protobuf"
stats_collect "github.com/chrislusf/seaweedfs/weed/stats"
"github.com/gorilla/mux"
+ "github.com/seaweedfs/raft/protobuf"
"github.com/spf13/viper"
"google.golang.org/grpc/reflection"
diff --git a/weed/command/scaffold/replication.toml b/weed/command/scaffold/replication.toml
index c463c8077..cffe1b76f 100644
--- a/weed/command/scaffold/replication.toml
+++ b/weed/command/scaffold/replication.toml
@@ -13,6 +13,8 @@ grpcAddress = "localhost:18888"
# this is not a directory on your hard drive, but on your filer.
# i.e., all files with this "prefix" are sent to notification message queue.
directory = "/buckets"
+# files from the directory separated by space are excluded from sending notifications
+excludeDirectories = "/buckets/tmp"
[sink.local]
enabled = false
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 17:30:16 +0000