Merge branch 'master' into support_ssd_volume

weed/replication/sink/filersink/fetch_write.go

@@ -30,6 +30,7 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
 			replicatedChunk, e := fs.replicateOneChunk(chunk, path)
 			if e != nil {
 				err = e
+				return
 			}
 			replicatedChunks[index] = replicatedChunk
 		}(sourceChunk, chunkIndex)
@@ -98,6 +99,9 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string)
 	}
 
 	fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
+	if fs.writeChunkByFiler {
+		fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)
+	}
 
 	glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
 

weed/replication/sink/filersink/filer_sink.go

@@ -3,6 +3,8 @@ package filersink
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/wdclient"
 
 	"google.golang.org/grpc"
 
@@ -17,15 +19,17 @@ import (
 )
 
 type FilerSink struct {
-	filerSource    *source.FilerSource
-	grpcAddress    string
-	dir            string
-	replication    string
-	collection     string
-	ttlSec         int32
+	filerSource       *source.FilerSource
+	grpcAddress       string
+	dir               string
+	replication       string
+	collection        string
+	ttlSec            int32
 	diskType       string
-	dataCenter     string
-	grpcDialOption grpc.DialOption
+	dataCenter        string
+	grpcDialOption    grpc.DialOption
+	address           string
+	writeChunkByFiler bool
 }
 
 func init() {
@@ -42,21 +46,27 @@ func (fs *FilerSink) GetSinkToDirectory() string {
 
 func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
 	return fs.DoInitialize(
+		"",
 		configuration.GetString(prefix+"grpcAddress"),
 		configuration.GetString(prefix+"directory"),
 		configuration.GetString(prefix+"replication"),
 		configuration.GetString(prefix+"collection"),
 		configuration.GetInt(prefix+"ttlSec"),
 		configuration.GetString(prefix+"disk"),
-		security.LoadClientTLS(util.GetViper(), "grpc.client"))
+		security.LoadClientTLS(util.GetViper(), "grpc.client"),
+		false)
 }
 
 func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
 	fs.filerSource = s
 }
 
-func (fs *FilerSink) DoInitialize(grpcAddress string, dir string,
-	replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption) (err error) {
+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)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.dir = dir
 	fs.replication = replication
@@ -64,6 +74,7 @@ func (fs *FilerSink) DoInitialize(grpcAddress string, dir string,
 	fs.ttlSec = int32(ttlSec)
 	fs.diskType = diskType
 	fs.grpcDialOption = grpcDialOption
+	fs.writeChunkByFiler = writeChunkByFiler
 	return nil
 }
 
@@ -209,7 +220,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 	})
 
 }
-func compareChunks(lookupFileIdFn filer.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) {
+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)
 	if aErr != nil {
 		return nil, nil, aErr

weed/replication/sink/localsink/local_incremental_sink.go

@@ -0,0 +1,17 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+)
+
+type LocalIncSink struct {
+	LocalSink
+}
+
+func (localincsink *LocalIncSink) GetName() string {
+	return "local_incremental"
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalIncSink{})
+}

weed/replication/sink/localsink/local_sink.go

@@ -0,0 +1,101 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+type LocalSink struct {
+	Dir         string
+	filerSource *source.FilerSource
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalSink{})
+}
+
+func (localsink *LocalSink) SetSourceFiler(s *source.FilerSource) {
+	localsink.filerSource = s
+}
+
+func (localsink *LocalSink) GetName() string {
+	return "local"
+}
+
+func (localsink *LocalSink) isMultiPartEntry(key string) bool {
+	return strings.HasSuffix(key, ".part") && strings.Contains(key, "/.uploads/")
+}
+
+func (localsink *LocalSink) initialize(dir string) error {
+	localsink.Dir = dir
+	return nil
+}
+
+func (localsink *LocalSink) Initialize(configuration util.Configuration, prefix string) error {
+	dir := configuration.GetString(prefix + "directory")
+	glog.V(4).Infof("sink.local.directory: %v", dir)
+	return localsink.initialize(dir)
+}
+
+func (localsink *LocalSink) GetSinkToDirectory() string {
+	return localsink.Dir
+}
+
+func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
+	if localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Delete Entry key: %s", key)
+	if err := os.Remove(key); err != nil {
+		glog.V(0).Infof("remove entry key %s: %s", key, err)
+	}
+	return nil
+}
+
+func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
+	if entry.IsDirectory || localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Create Entry key: %s", key)
+
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(localsink.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
+
+	dir := filepath.Dir(key)
+
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		glog.V(4).Infof("Create Direcotry key: %s", dir)
+		if err = os.MkdirAll(dir, 0); err != nil {
+			return err
+		}
+	}
+
+	writeFunc := func(data []byte) error {
+		writeErr := ioutil.WriteFile(key, data, 0)
+		return writeErr
+	}
+
+	if err := repl_util.CopyFromChunkViews(chunkViews, localsink.filerSource, writeFunc); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (localsink *LocalSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
+	if localsink.isMultiPartEntry(key) {
+		return true, nil
+	}
+	glog.V(4).Infof("Update Entry key: %s", key)
+	// do delete and create
+	return false, nil
+}