seaweedfs/weed/replication/sink/filersink/filer_sink.go

255 lines
7.7 KiB
Go
Raw Normal View History

2018-09-23 07:40:36 +00:00
package filersink
2018-09-17 07:27:56 +00:00
import (
2018-09-21 08:54:29 +00:00
"context"
2018-09-21 08:56:43 +00:00
"fmt"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
2021-07-21 21:38:12 +00:00
"math"
2019-02-18 20:11:52 +00:00
"google.golang.org/grpc"
2018-09-21 08:54:29 +00:00
"github.com/seaweedfs/seaweedfs/weed/security"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/replication/sink"
"github.com/seaweedfs/seaweedfs/weed/replication/source"
"github.com/seaweedfs/seaweedfs/weed/util"
2018-09-17 07:27:56 +00:00
)
type FilerSink struct {
filerSource *source.FilerSource
grpcAddress string
dir string
replication string
collection string
ttlSec int32
2021-02-11 07:41:05 +00:00
diskType string
dataCenter string
grpcDialOption grpc.DialOption
address string
writeChunkByFiler bool
2021-03-01 04:34:14 +00:00
isIncremental bool
executor *util.LimitedConcurrentExecutor
2018-09-17 07:27:56 +00:00
}
2018-10-11 07:08:13 +00:00
func init() {
2018-10-04 06:36:52 +00:00
sink.Sinks = append(sink.Sinks, &FilerSink{})
}
func (fs *FilerSink) GetName() string {
return "filer"
}
2018-09-22 07:53:52 +00:00
func (fs *FilerSink) GetSinkToDirectory() string {
2018-09-17 08:37:24 +00:00
return fs.dir
}
2021-03-01 00:19:03 +00:00
func (fs *FilerSink) IsIncremental() bool {
return fs.isIncremental
}
func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
2021-03-01 04:34:14 +00:00
fs.isIncremental = configuration.GetBool(prefix + "is_incremental")
fs.dataCenter = configuration.GetString(prefix + "dataCenter")
2020-09-09 18:21:23 +00:00
return fs.DoInitialize(
"",
configuration.GetString(prefix+"grpcAddress"),
configuration.GetString(prefix+"directory"),
configuration.GetString(prefix+"replication"),
configuration.GetString(prefix+"collection"),
configuration.GetInt(prefix+"ttlSec"),
2020-12-13 20:05:31 +00:00
configuration.GetString(prefix+"disk"),
security.LoadClientTLS(util.GetViper(), "grpc.client"),
false)
2018-09-17 07:27:56 +00:00
}
2018-09-21 08:54:29 +00:00
func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
fs.filerSource = s
}
func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
fs.address = address
if fs.address == "" {
fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
}
2018-09-17 07:27:56 +00:00
fs.grpcAddress = grpcAddress
fs.dir = dir
2018-09-22 07:53:52 +00:00
fs.replication = replication
fs.collection = collection
fs.ttlSec = int32(ttlSec)
2020-12-13 19:59:32 +00:00
fs.diskType = diskType
2020-09-09 18:21:23 +00:00
fs.grpcDialOption = grpcDialOption
fs.writeChunkByFiler = writeChunkByFiler
fs.executor = util.NewLimitedConcurrentExecutor(32)
2018-09-17 07:27:56 +00:00
return nil
}
2020-09-09 18:21:23 +00:00
func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
2018-09-17 07:27:56 +00:00
2020-03-23 08:30:22 +00:00
dir, name := util.FullPath(key).DirAndName()
2018-09-17 07:27:56 +00:00
2020-09-09 18:21:23 +00:00
glog.V(4).Infof("delete entry: %v", key)
err := filer_pb.Remove(fs, dir, name, deleteIncludeChunks, true, true, true, signatures)
2020-03-23 08:30:22 +00:00
if err != nil {
glog.V(0).Infof("delete entry %s: %v", key, err)
return fmt.Errorf("delete entry %s: %v", key, err)
}
return nil
2018-09-17 07:27:56 +00:00
}
2020-09-09 18:21:23 +00:00
func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
2018-09-17 07:27:56 +00:00
return fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
2018-09-17 07:27:56 +00:00
2020-03-23 07:01:34 +00:00
dir, name := util.FullPath(key).DirAndName()
2018-09-17 07:27:56 +00:00
// look up existing entry
lookupRequest := &filer_pb.LookupDirectoryEntryRequest{
Directory: dir,
Name: name,
}
glog.V(1).Infof("lookup: %v", lookupRequest)
if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
2020-09-01 07:21:19 +00:00
if filer.ETag(resp.Entry) == filer.ETag(entry) {
2020-09-09 18:21:23 +00:00
glog.V(3).Infof("already replicated %s", key)
return nil
}
}
2018-09-21 08:54:29 +00:00
replicatedChunks, err := fs.replicateChunks(entry.Chunks, key)
2018-09-17 07:27:56 +00:00
if err != nil {
2020-09-09 18:21:23 +00:00
// only warning here since the source chunk may have been deleted already
glog.Warningf("replicate entry chunks %s: %v", key, err)
}
2020-09-09 18:21:23 +00:00
glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
2018-09-17 07:27:56 +00:00
request := &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: &filer_pb.Entry{
Name: name,
IsDirectory: entry.IsDirectory,
Attributes: entry.Attributes,
Extended: entry.Extended,
2018-09-17 07:27:56 +00:00
Chunks: replicatedChunks,
2020-11-30 12:34:04 +00:00
Content: entry.Content,
2021-07-27 05:53:44 +00:00
RemoteEntry: entry.RemoteEntry,
2018-09-17 07:27:56 +00:00
},
IsFromOtherCluster: true,
2020-09-09 18:21:23 +00:00
Signatures: signatures,
2018-09-17 07:27:56 +00:00
}
2020-09-09 18:21:23 +00:00
glog.V(3).Infof("create: %v", request)
if err := filer_pb.CreateEntry(client, request); err != nil {
2018-09-17 08:37:24 +00:00
glog.V(0).Infof("create entry %s: %v", key, err)
return fmt.Errorf("create entry %s: %v", key, err)
2018-09-17 07:27:56 +00:00
}
return nil
})
}
2020-09-09 18:21:23 +00:00
func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
2018-09-17 07:27:56 +00:00
2020-03-23 07:01:34 +00:00
dir, name := util.FullPath(key).DirAndName()
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
// read existing entry
2018-10-04 06:36:52 +00:00
var existingEntry *filer_pb.Entry
err = fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
request := &filer_pb.LookupDirectoryEntryRequest{
Directory: dir,
Name: name,
}
2018-09-17 07:27:56 +00:00
glog.V(4).Infof("lookup entry: %v", request)
resp, err := filer_pb.LookupEntry(client, request)
2018-09-21 08:54:29 +00:00
if err != nil {
glog.V(0).Infof("lookup %s: %v", key, err)
return err
}
2018-09-17 07:27:56 +00:00
2018-10-04 06:36:52 +00:00
existingEntry = resp.Entry
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
return nil
})
2018-09-17 07:27:56 +00:00
if err != nil {
2018-10-04 06:36:52 +00:00
return false, fmt.Errorf("lookup %s: %v", key, err)
2018-09-17 07:27:56 +00:00
}
2020-09-09 18:21:23 +00:00
glog.V(4).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
2018-11-01 04:48:05 +00:00
if existingEntry.Attributes.Mtime > newEntry.Attributes.Mtime {
// skip if already changed
// this usually happens when the messages are not ordered
2020-09-09 18:21:23 +00:00
glog.V(2).Infof("late updates %s", key)
} else {
// find out what changed
2020-09-01 07:21:19 +00:00
deletedChunks, newChunks, err := compareChunks(filer.LookupFn(fs), oldEntry, newEntry)
if err != nil {
return true, fmt.Errorf("replicate %s compare chunks error: %v", key, err)
}
// delete the chunks that are deleted from the source
if deleteIncludeChunks {
// remove the deleted chunks. Actual data deletion happens in filer UpdateEntry FindUnusedFileChunks
existingEntry.Chunks = filer.DoMinusChunksBySourceFileId(existingEntry.Chunks, deletedChunks)
}
2018-09-21 08:54:29 +00:00
// replicate the chunks that are new in the source
replicatedChunks, err := fs.replicateChunks(newChunks, key)
if err != nil {
return true, fmt.Errorf("replicate %s chunks error: %v", key, err)
}
existingEntry.Chunks = append(existingEntry.Chunks, replicatedChunks...)
existingEntry.Attributes = newEntry.Attributes
existingEntry.Extended = newEntry.Extended
existingEntry.HardLinkId = newEntry.HardLinkId
existingEntry.HardLinkCounter = newEntry.HardLinkCounter
existingEntry.Content = newEntry.Content
existingEntry.RemoteEntry = newEntry.RemoteEntry
}
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
// save updated meta data
return true, fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
2018-09-21 08:54:29 +00:00
request := &filer_pb.UpdateEntryRequest{
Directory: newParentPath,
Entry: existingEntry,
IsFromOtherCluster: true,
2020-09-09 18:21:23 +00:00
Signatures: signatures,
2018-09-21 08:54:29 +00:00
}
if _, err := client.UpdateEntry(context.Background(), request); err != nil {
return fmt.Errorf("update existingEntry %s: %v", key, err)
2018-09-21 08:54:29 +00:00
}
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
return nil
})
}
func compareChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) {
aData, aMeta, aErr := filer.ResolveChunkManifest(lookupFileIdFn, oldEntry.Chunks, 0, math.MaxInt64)
if aErr != nil {
return nil, nil, aErr
}
bData, bMeta, bErr := filer.ResolveChunkManifest(lookupFileIdFn, newEntry.Chunks, 0, math.MaxInt64)
if bErr != nil {
return nil, nil, bErr
}
2020-09-01 07:21:19 +00:00
deletedChunks = append(deletedChunks, filer.DoMinusChunks(aData, bData)...)
deletedChunks = append(deletedChunks, filer.DoMinusChunks(aMeta, bMeta)...)
2020-09-01 07:21:19 +00:00
newChunks = append(newChunks, filer.DoMinusChunks(bData, aData)...)
newChunks = append(newChunks, filer.DoMinusChunks(bMeta, aMeta)...)
2018-09-17 07:27:56 +00:00
return
}