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

229 lines
6 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"
2018-09-21 08:54:29 +00:00
2018-09-17 07:27:56 +00:00
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
2018-09-21 08:56:43 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2018-10-11 07:08:13 +00:00
"github.com/chrislusf/seaweedfs/weed/replication/sink"
2018-09-21 08:54:29 +00:00
"github.com/chrislusf/seaweedfs/weed/replication/source"
2018-09-21 08:56:43 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
2018-09-17 07:27:56 +00:00
)
type FilerSink struct {
2018-09-22 07:53:52 +00:00
filerSource *source.FilerSource
2018-09-17 07:27:56 +00:00
grpcAddress string
dir string
2018-09-21 08:54:29 +00:00
replication string
collection string
ttlSec int32
dataCenter string
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
}
2018-09-17 07:27:56 +00:00
func (fs *FilerSink) Initialize(configuration util.Configuration) error {
return fs.initialize(
configuration.GetString("grpcAddress"),
configuration.GetString("directory"),
2018-09-22 07:53:52 +00:00
configuration.GetString("replication"),
configuration.GetString("collection"),
configuration.GetInt("ttlSec"),
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
}
2018-09-22 07:53:52 +00:00
func (fs *FilerSink) initialize(grpcAddress string, dir string,
replication string, collection string, ttlSec int) (err error) {
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)
2018-09-17 07:27:56 +00:00
return nil
}
2018-10-04 06:36:52 +00:00
func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool) error {
2018-09-17 07:27:56 +00:00
return fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
2018-09-17 08:37:24 +00:00
dir, name := filer2.FullPath(key).DirAndName()
2018-09-17 07:27:56 +00:00
request := &filer_pb.DeleteEntryRequest{
Directory: dir,
Name: name,
2018-10-04 06:36:52 +00:00
IsDirectory: isDirectory,
2018-09-17 07:27:56 +00:00
IsDeleteData: deleteIncludeChunks,
}
glog.V(1).Infof("delete entry: %v", request)
_, err := client.DeleteEntry(context.Background(), request)
if err != nil {
2018-09-17 08:37:24 +00:00
glog.V(0).Infof("delete entry %s: %v", key, err)
return fmt.Errorf("delete entry %s: %v", key, err)
2018-09-17 07:27:56 +00:00
}
return nil
})
}
2018-09-17 08:37:24 +00:00
func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error {
2018-09-17 07:27:56 +00:00
return fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
2018-09-17 07:27:56 +00:00
dir, name := filer2.FullPath(key).DirAndName()
ctx := context.Background()
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 := client.LookupDirectoryEntry(ctx, lookupRequest); err == nil {
if filer2.ETag(resp.Entry.Chunks) == filer2.ETag(entry.Chunks) {
glog.V(0).Infof("already replicated %s", key)
return nil
}
}
2018-09-21 08:54:29 +00:00
replicatedChunks, err := fs.replicateChunks(entry.Chunks)
2018-09-17 07:27:56 +00:00
if err != nil {
glog.V(0).Infof("replicate entry chunks %s: %v", key, err)
return fmt.Errorf("replicate entry chunks %s: %v", key, err)
}
glog.V(0).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,
Chunks: replicatedChunks,
},
}
glog.V(1).Infof("create: %v", request)
if _, err := client.CreateEntry(ctx, 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
})
}
2018-10-04 06:36:52 +00:00
func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
ctx := context.Background()
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
dir, name := filer2.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
2018-09-21 08:54:29 +00:00
err = fs.withFilerClient(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)
2018-09-21 08:54:29 +00:00
resp, err := client.LookupDirectoryEntry(ctx, request)
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
}
glog.V(0).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
glog.V(0).Infof("late updates %s", key)
} else if filer2.ETag(newEntry.Chunks) == filer2.ETag(existingEntry.Chunks) {
// skip if no change
// this usually happens when retrying the replication
glog.V(0).Infof("already replicated %s", key)
} else {
// find out what changed
deletedChunks, newChunks := compareChunks(oldEntry, newEntry)
// 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 = minusChunks(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)
if err != nil {
2018-10-04 06:36:52 +00:00
return true, fmt.Errorf("replicte %s chunks error: %v", key, err)
}
existingEntry.Chunks = append(existingEntry.Chunks, replicatedChunks...)
}
2018-09-17 07:27:56 +00:00
2018-09-21 08:54:29 +00:00
// save updated meta data
2018-10-04 06:36:52 +00:00
return true, fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
2018-09-21 08:54:29 +00:00
request := &filer_pb.UpdateEntryRequest{
Directory: dir,
Entry: existingEntry,
2018-09-21 08:54:29 +00:00
}
if _, err := client.UpdateEntry(ctx, 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(oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk) {
deletedChunks = minusChunks(oldEntry.Chunks, newEntry.Chunks)
newChunks = minusChunks(newEntry.Chunks, oldEntry.Chunks)
return
}
func minusChunks(as, bs []*filer_pb.FileChunk) (delta []*filer_pb.FileChunk) {
for _, a := range as {
found := false
for _, b := range bs {
if a.FileId == b.FileId {
found = true
break
}
}
if !found {
delta = append(delta, a)
}
}
2018-09-17 07:27:56 +00:00
return
}