aboutsummaryrefslogtreecommitdiff
path: root/weed/replication
diff options
context:
space:
mode:
Diffstat (limited to 'weed/replication')
-rw-r--r--weed/replication/replicator.go26
-rw-r--r--weed/replication/sink/azuresink/azure_sink.go6
-rw-r--r--weed/replication/sink/b2sink/b2_sink.go6
-rw-r--r--weed/replication/sink/filersink/fetch_write.go9
-rw-r--r--weed/replication/sink/filersink/filer_sink.go42
-rw-r--r--weed/replication/sink/gcssink/gcs_sink.go6
-rw-r--r--weed/replication/sink/replication_sink.go6
-rw-r--r--weed/replication/sink/s3sink/s3_sink.go6
-rw-r--r--weed/replication/source/filer_source.go10
9 files changed, 66 insertions, 51 deletions
diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go
index 051199adb..c4228434f 100644
--- a/weed/replication/replicator.go
+++ b/weed/replication/replicator.go
@@ -3,6 +3,8 @@ package replication
import (
"context"
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/pb"
+ "google.golang.org/grpc"
"strings"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -43,28 +45,42 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
key = newKey
if message.OldEntry != nil && message.NewEntry == nil {
glog.V(4).Infof("deleting %v", key)
- return r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks)
+ return r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
}
if message.OldEntry == nil && message.NewEntry != nil {
glog.V(4).Infof("creating %v", key)
- return r.sink.CreateEntry(key, message.NewEntry)
+ return r.sink.CreateEntry(key, message.NewEntry, message.Signatures)
}
if message.OldEntry == nil && message.NewEntry == nil {
glog.V(0).Infof("weird message %+v", message)
return nil
}
- foundExisting, err := r.sink.UpdateEntry(key, message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks)
+ foundExisting, err := r.sink.UpdateEntry(key, message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks, message.Signatures)
if foundExisting {
glog.V(4).Infof("updated %v", key)
return err
}
- err = r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, false)
+ err = r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, false, message.Signatures)
if err != nil {
return fmt.Errorf("delete old entry %v: %v", key, err)
}
glog.V(4).Infof("creating missing %v", key)
- return r.sink.CreateEntry(key, message.NewEntry)
+ return r.sink.CreateEntry(key, message.NewEntry, message.Signatures)
+}
+
+func ReadFilerSignature(grpcDialOption grpc.DialOption, filer string) (filerSignature int32, readErr error) {
+ if readErr = pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+ if resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}); err != nil {
+ return fmt.Errorf("GetFilerConfiguration %s: %v", filer, err)
+ } else {
+ filerSignature = resp.Signature
+ }
+ return nil
+ }); readErr != nil {
+ return 0, readErr
+ }
+ return filerSignature, nil
}
diff --git a/weed/replication/sink/azuresink/azure_sink.go b/weed/replication/sink/azuresink/azure_sink.go
index 64c3d46ea..dab5cf4f4 100644
--- a/weed/replication/sink/azuresink/azure_sink.go
+++ b/weed/replication/sink/azuresink/azure_sink.go
@@ -70,7 +70,7 @@ func (g *AzureSink) initialize(accountName, accountKey, container, dir string) e
return nil
}
-func (g *AzureSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *AzureSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
key = cleanKey(key)
@@ -87,7 +87,7 @@ func (g *AzureSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks boo
}
-func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry) error {
+func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
key = cleanKey(key)
@@ -132,7 +132,7 @@ func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry) error {
}
-func (g *AzureSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *AzureSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
key = cleanKey(key)
// TODO improve efficiency
return false, nil
diff --git a/weed/replication/sink/b2sink/b2_sink.go b/weed/replication/sink/b2sink/b2_sink.go
index d0b3e7a34..cf212f129 100644
--- a/weed/replication/sink/b2sink/b2_sink.go
+++ b/weed/replication/sink/b2sink/b2_sink.go
@@ -57,7 +57,7 @@ func (g *B2Sink) initialize(accountId, accountKey, bucket, dir string) error {
return nil
}
-func (g *B2Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *B2Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
key = cleanKey(key)
@@ -76,7 +76,7 @@ func (g *B2Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool)
}
-func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
+func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
key = cleanKey(key)
@@ -123,7 +123,7 @@ func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
}
-func (g *B2Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *B2Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
key = cleanKey(key)
diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go
index bde29176c..d33669447 100644
--- a/weed/replication/sink/filersink/fetch_write.go
+++ b/weed/replication/sink/filersink/fetch_write.go
@@ -3,6 +3,7 @@ package filersink
import (
"context"
"fmt"
+ "github.com/chrislusf/seaweedfs/weed/util"
"sync"
"google.golang.org/grpc"
@@ -59,11 +60,11 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, dir stri
func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, dir string) (fileId string, err error) {
- filename, header, readCloser, err := fs.filerSource.ReadPart(sourceChunk.GetFileIdString())
+ filename, header, resp, err := fs.filerSource.ReadPart(sourceChunk.GetFileIdString())
if err != nil {
return "", fmt.Errorf("read part %s: %v", sourceChunk.GetFileIdString(), err)
}
- defer readCloser.Close()
+ defer util.CloseResponse(resp)
var host string
var auth security.EncodedJwt
@@ -100,9 +101,9 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, dir string)
glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
// fetch data as is, regardless whether it is encrypted or not
- uploadResult, err, _ := operation.Upload(fileUrl, filename, false, readCloser, "gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
+ uploadResult, err, _ := operation.Upload(fileUrl, filename, false, resp.Body, "gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
if err != nil {
- glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
+ glog.V(0).Infof("upload source data %v to %s: %v", sourceChunk.GetFileIdString(), fileUrl, err)
return "", fmt.Errorf("upload data: %v", err)
}
if uploadResult.Error != "" {
diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go
index 7ba1670e0..f1d8ff840 100644
--- a/weed/replication/sink/filersink/filer_sink.go
+++ b/weed/replication/sink/filersink/filer_sink.go
@@ -25,7 +25,6 @@ type FilerSink struct {
ttlSec int32
dataCenter string
grpcDialOption grpc.DialOption
- signature int32
}
func init() {
@@ -41,37 +40,36 @@ func (fs *FilerSink) GetSinkToDirectory() string {
}
func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
- return fs.initialize(
+ return fs.DoInitialize(
configuration.GetString(prefix+"grpcAddress"),
configuration.GetString(prefix+"directory"),
configuration.GetString(prefix+"replication"),
configuration.GetString(prefix+"collection"),
configuration.GetInt(prefix+"ttlSec"),
- )
+ security.LoadClientTLS(util.GetViper(), "grpc.client"))
}
func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
fs.filerSource = s
}
-func (fs *FilerSink) initialize(grpcAddress string, dir string,
- replication string, collection string, ttlSec int) (err error) {
+func (fs *FilerSink) DoInitialize(grpcAddress string, dir string,
+ replication string, collection string, ttlSec int, grpcDialOption grpc.DialOption) (err error) {
fs.grpcAddress = grpcAddress
fs.dir = dir
fs.replication = replication
fs.collection = collection
fs.ttlSec = int32(ttlSec)
- fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
- fs.signature = util.RandomInt32()
+ fs.grpcDialOption = grpcDialOption
return nil
}
-func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
+func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
dir, name := util.FullPath(key).DirAndName()
- glog.V(1).Infof("delete entry: %v", key)
- err := filer_pb.Remove(fs, dir, name, deleteIncludeChunks, false, false, true, fs.signature)
+ glog.V(4).Infof("delete entry: %v", key)
+ err := filer_pb.Remove(fs, dir, name, deleteIncludeChunks, true, true, true, signatures)
if err != nil {
glog.V(0).Infof("delete entry %s: %v", key, err)
return fmt.Errorf("delete entry %s: %v", key, err)
@@ -79,7 +77,7 @@ func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bo
return nil
}
-func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
+func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
@@ -93,7 +91,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
glog.V(1).Infof("lookup: %v", lookupRequest)
if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
if filer.ETag(resp.Entry) == filer.ETag(entry) {
- glog.V(0).Infof("already replicated %s", key)
+ glog.V(3).Infof("already replicated %s", key)
return nil
}
}
@@ -101,11 +99,11 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
replicatedChunks, err := fs.replicateChunks(entry.Chunks, dir)
if err != nil {
- glog.V(0).Infof("replicate entry chunks %s: %v", key, err)
- return fmt.Errorf("replicate entry chunks %s: %v", key, err)
+ // only warning here since the source chunk may have been deleted already
+ glog.Warningf("replicate entry chunks %s: %v", key, err)
}
- glog.V(0).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
+ glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
request := &filer_pb.CreateEntryRequest{
Directory: dir,
@@ -116,10 +114,10 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
Chunks: replicatedChunks,
},
IsFromOtherCluster: true,
- Signatures: []int32{fs.signature},
+ Signatures: signatures,
}
- glog.V(1).Infof("create: %v", request)
+ glog.V(3).Infof("create: %v", request)
if err := filer_pb.CreateEntry(client, request); err != nil {
glog.V(0).Infof("create entry %s: %v", key, err)
return fmt.Errorf("create entry %s: %v", key, err)
@@ -129,7 +127,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
})
}
-func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
dir, name := util.FullPath(key).DirAndName()
@@ -158,16 +156,16 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
return false, fmt.Errorf("lookup %s: %v", key, err)
}
- glog.V(0).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
+ glog.V(4).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
if existingEntry.Attributes.Mtime > newEntry.Attributes.Mtime {
// skip if already changed
// this usually happens when the messages are not ordered
- glog.V(0).Infof("late updates %s", key)
+ glog.V(2).Infof("late updates %s", key)
} else if filer.ETag(newEntry) == filer.ETag(existingEntry) {
// skip if no change
// this usually happens when retrying the replication
- glog.V(0).Infof("already replicated %s", key)
+ glog.V(3).Infof("already replicated %s", key)
} else {
// find out what changed
deletedChunks, newChunks, err := compareChunks(filer.LookupFn(fs), oldEntry, newEntry)
@@ -196,7 +194,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
Directory: newParentPath,
Entry: existingEntry,
IsFromOtherCluster: true,
- Signatures: []int32{fs.signature},
+ Signatures: signatures,
}
if _, err := client.UpdateEntry(context.Background(), request); err != nil {
diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go
index 2e09a87f9..c6bfa212a 100644
--- a/weed/replication/sink/gcssink/gcs_sink.go
+++ b/weed/replication/sink/gcssink/gcs_sink.go
@@ -69,7 +69,7 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
return nil
}
-func (g *GcsSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *GcsSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
if isDirectory {
key = key + "/"
@@ -83,7 +83,7 @@ func (g *GcsSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool)
}
-func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry) error {
+func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
if entry.IsDirectory {
return nil
@@ -119,7 +119,7 @@ func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry) error {
}
-func (g *GcsSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *GcsSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
// TODO improve efficiency
return false, nil
}
diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go
index 6d85f660a..cfc6e0a4d 100644
--- a/weed/replication/sink/replication_sink.go
+++ b/weed/replication/sink/replication_sink.go
@@ -9,9 +9,9 @@ import (
type ReplicationSink interface {
GetName() string
Initialize(configuration util.Configuration, prefix string) error
- DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error
- CreateEntry(key string, entry *filer_pb.Entry) error
- UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error)
+ DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error
+ CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error
+ UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error)
GetSinkToDirectory() string
SetSourceFiler(s *source.FilerSource)
}
diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go
index 4e7df8ff2..58432ee6b 100644
--- a/weed/replication/sink/s3sink/s3_sink.go
+++ b/weed/replication/sink/s3sink/s3_sink.go
@@ -83,7 +83,7 @@ func (s3sink *S3Sink) initialize(awsAccessKeyId, awsSecretAccessKey, region, buc
return nil
}
-func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
+func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
key = cleanKey(key)
@@ -95,7 +95,7 @@ func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks b
}
-func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
+func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
key = cleanKey(key)
if entry.IsDirectory {
@@ -136,7 +136,7 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
}
-func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
key = cleanKey(key)
// TODO improve efficiency
return false, nil
diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go
index 69c23fe82..9106ee98b 100644
--- a/weed/replication/source/filer_source.go
+++ b/weed/replication/source/filer_source.go
@@ -28,13 +28,13 @@ type FilerSource struct {
}
func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
- return fs.initialize(
+ return fs.DoInitialize(
configuration.GetString(prefix+"grpcAddress"),
configuration.GetString(prefix+"directory"),
)
}
-func (fs *FilerSource) initialize(grpcAddress string, dir string) (err error) {
+func (fs *FilerSource) DoInitialize(grpcAddress string, dir string) (err error) {
fs.grpcAddress = grpcAddress
fs.Dir = dir
fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
@@ -79,16 +79,16 @@ func (fs *FilerSource) LookupFileId(part string) (fileUrl string, err error) {
return
}
-func (fs *FilerSource) ReadPart(part string) (filename string, header http.Header, readCloser io.ReadCloser, err error) {
+func (fs *FilerSource) ReadPart(part string) (filename string, header http.Header, resp *http.Response, err error) {
fileUrl, err := fs.LookupFileId(part)
if err != nil {
return "", nil, nil, err
}
- filename, header, readCloser, err = util.DownloadFile(fileUrl)
+ filename, header, resp, err = util.DownloadFile(fileUrl)
- return filename, header, readCloser, err
+ return filename, header, resp, err
}
var _ = filer_pb.FilerClient(&FilerSource{})
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 17:22:41 +0000