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"
|
2022-07-29 07:17:28 +00:00
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
2021-07-21 21:38:12 +00:00
|
|
|
"math"
|
2020-01-29 17:09:55 +00:00
|
|
|
|
2019-02-18 20:11:52 +00:00
|
|
|
"google.golang.org/grpc"
|
2018-09-21 08:54:29 +00:00
|
|
|
|
2022-07-29 07:17:28 +00:00
|
|
|
"github.com/seaweedfs/seaweedfs/weed/security"
|
2020-01-29 17:09:55 +00:00
|
|
|
|
2022-07-29 07:17:28 +00:00
|
|
|
"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 {
|
2021-01-24 08:01:44 +00:00
|
|
|
filerSource *source.FilerSource
|
|
|
|
grpcAddress string
|
|
|
|
dir string
|
|
|
|
replication string
|
|
|
|
collection string
|
|
|
|
ttlSec int32
|
2021-02-11 07:41:05 +00:00
|
|
|
diskType string
|
2021-01-24 08:01:44 +00:00
|
|
|
dataCenter string
|
|
|
|
grpcDialOption grpc.DialOption
|
|
|
|
address string
|
|
|
|
writeChunkByFiler bool
|
2021-03-01 04:34:14 +00:00
|
|
|
isIncremental bool
|
2022-10-04 18:35:07 +00:00
|
|
|
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
|
|
|
|
}
|
|
|
|
|
2020-01-29 17:09:55 +00:00
|
|
|
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")
|
2022-08-05 00:35:00 +00:00
|
|
|
fs.dataCenter = configuration.GetString(prefix + "dataCenter")
|
2022-10-04 18:35:07 +00:00
|
|
|
fs.executor = util.NewLimitedConcurrentExecutor(32)
|
2020-09-09 18:21:23 +00:00
|
|
|
return fs.DoInitialize(
|
2021-01-24 08:01:44 +00:00
|
|
|
"",
|
2020-01-29 17:09:55 +00:00
|
|
|
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"),
|
2021-01-24 08:01:44 +00:00
|
|
|
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
|
|
|
|
}
|
|
|
|
|
2021-01-24 08:01:44 +00:00
|
|
|
func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
|
2021-02-09 19:37:07 +00:00
|
|
|
replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
|
2021-01-24 08:01:44 +00:00
|
|
|
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
|
2021-01-24 08:01:44 +00:00
|
|
|
fs.writeChunkByFiler = writeChunkByFiler
|
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
|
|
|
|
2021-12-26 08:15:03 +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
|
|
|
|
2018-09-22 18:14:04 +00:00
|
|
|
// look up existing entry
|
|
|
|
lookupRequest := &filer_pb.LookupDirectoryEntryRequest{
|
|
|
|
Directory: dir,
|
|
|
|
Name: name,
|
|
|
|
}
|
|
|
|
glog.V(1).Infof("lookup: %v", lookupRequest)
|
2020-03-08 00:51:46 +00:00
|
|
|
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)
|
2018-09-22 18:14:04 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
2018-09-21 08:54:29 +00:00
|
|
|
|
2020-10-25 22:46:29 +00:00
|
|
|
replicatedChunks, err := fs.replicateChunks(entry.Chunks, key)
|
2018-09-17 07:27:56 +00:00
|
|
|
|
2018-09-22 18:14:04 +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)
|
2018-09-22 18:14:04 +00:00
|
|
|
}
|
|
|
|
|
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,
|
2022-09-05 01:26:36 +00:00
|
|
|
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
|
|
|
},
|
2020-07-01 05:53:53 +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)
|
2020-02-26 05:50:12 +00:00
|
|
|
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
|
2021-12-26 08:15:03 +00:00
|
|
|
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
|
|
|
|
2018-09-25 16:27:03 +00:00
|
|
|
glog.V(4).Infof("lookup entry: %v", request)
|
2020-03-08 00:51:46 +00:00
|
|
|
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-09-25 16:27:03 +00:00
|
|
|
|
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)
|
2018-09-22 18:14:04 +00:00
|
|
|
} else {
|
|
|
|
// find out what changed
|
2020-09-01 07:21:19 +00:00
|
|
|
deletedChunks, newChunks, err := compareChunks(filer.LookupFn(fs), oldEntry, newEntry)
|
2020-07-20 00:59:43 +00:00
|
|
|
if err != nil {
|
2022-09-14 17:01:18 +00:00
|
|
|
return true, fmt.Errorf("replicate %s compare chunks error: %v", key, err)
|
2020-07-20 00:59:43 +00:00
|
|
|
}
|
2018-09-22 18:14:04 +00:00
|
|
|
|
|
|
|
// delete the chunks that are deleted from the source
|
|
|
|
if deleteIncludeChunks {
|
|
|
|
// remove the deleted chunks. Actual data deletion happens in filer UpdateEntry FindUnusedFileChunks
|
2022-02-07 11:46:28 +00:00
|
|
|
existingEntry.Chunks = filer.DoMinusChunksBySourceFileId(existingEntry.Chunks, deletedChunks)
|
2018-09-22 18:14:04 +00:00
|
|
|
}
|
2018-09-21 08:54:29 +00:00
|
|
|
|
2018-09-22 18:14:04 +00:00
|
|
|
// replicate the chunks that are new in the source
|
2020-10-25 22:46:29 +00:00
|
|
|
replicatedChunks, err := fs.replicateChunks(newChunks, key)
|
2018-09-22 18:14:04 +00:00
|
|
|
if err != nil {
|
2022-09-14 17:01:18 +00:00
|
|
|
return true, fmt.Errorf("replicate %s chunks error: %v", key, err)
|
2018-09-22 18:14:04 +00:00
|
|
|
}
|
2018-09-25 16:27:03 +00:00
|
|
|
existingEntry.Chunks = append(existingEntry.Chunks, replicatedChunks...)
|
2022-05-06 10:54:12 +00:00
|
|
|
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-22 18:14:04 +00:00
|
|
|
}
|
2018-09-17 07:27:56 +00:00
|
|
|
|
2018-09-21 08:54:29 +00:00
|
|
|
// save updated meta data
|
2021-12-26 08:15:03 +00:00
|
|
|
return true, fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
2018-09-21 08:54:29 +00:00
|
|
|
|
|
|
|
request := &filer_pb.UpdateEntryRequest{
|
2020-07-01 05:53:53 +00:00
|
|
|
Directory: newParentPath,
|
|
|
|
Entry: existingEntry,
|
|
|
|
IsFromOtherCluster: true,
|
2020-09-09 18:21:23 +00:00
|
|
|
Signatures: signatures,
|
2018-09-21 08:54:29 +00:00
|
|
|
}
|
|
|
|
|
2020-02-26 05:50:12 +00:00
|
|
|
if _, err := client.UpdateEntry(context.Background(), request); err != nil {
|
2018-09-25 16:27:03 +00:00
|
|
|
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
|
|
|
|
})
|
|
|
|
|
|
|
|
}
|
2021-01-06 12:21:34 +00:00
|
|
|
func compareChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) {
|
2021-07-20 06:07:22 +00:00
|
|
|
aData, aMeta, aErr := filer.ResolveChunkManifest(lookupFileIdFn, oldEntry.Chunks, 0, math.MaxInt64)
|
2020-07-20 00:59:43 +00:00
|
|
|
if aErr != nil {
|
|
|
|
return nil, nil, aErr
|
|
|
|
}
|
2021-07-20 06:07:22 +00:00
|
|
|
bData, bMeta, bErr := filer.ResolveChunkManifest(lookupFileIdFn, newEntry.Chunks, 0, math.MaxInt64)
|
2020-07-20 00:59:43 +00:00
|
|
|
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-07-20 00:59:43 +00:00
|
|
|
|
2020-09-01 07:21:19 +00:00
|
|
|
newChunks = append(newChunks, filer.DoMinusChunks(bData, aData)...)
|
|
|
|
newChunks = append(newChunks, filer.DoMinusChunks(bMeta, aMeta)...)
|
2020-07-20 00:59:43 +00:00
|
|
|
|
2018-09-17 07:27:56 +00:00
|
|
|
return
|
|
|
|
}
|