2.22

This reverts commit 096e088d7b.

docker/Makefile

@@ -12,6 +12,9 @@ build:
 dev: build
 	docker-compose -f local-dev-compose.yml -p seaweedfs up
 
+dev_mount: build
+	docker-compose -f local-mount-compose.yml -p seaweedfs up
+
 k8s: build
 	docker-compose -f local-k8s-compose.yml -p seaweedfs up
 
@@ -21,5 +24,8 @@ dev_registry: build
 cluster: build
 	docker-compose -f local-cluster-compose.yml -p seaweedfs up
 
+2clusters: build
+	docker-compose -f local-clusters-compose.yml -p seaweedfs up
+
 clean:
 	rm ./weed

docker/local-clusters-compose.yml

@@ -0,0 +1,23 @@
+version: '2'
+
+services:
+  server1:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+      - 8084:8080
+      - 18084:18080
+      - 8888:8888
+      - 18888:18888
+    command: "server -ip=server1 -filer -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1"
+  server2:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9334:9333
+      - 19334:19333
+      - 8085:8080
+      - 18085:18080
+      - 8889:8888
+      - 18889:18888
+    command: "server -ip=server2 -filer -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1"

docker/local-mount-compose.yml

@@ -0,0 +1,47 @@
+version: '2'
+
+services:
+  master:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+    command: "master -ip=master"
+  volume:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8080:8080
+      - 18080:18080
+      - 9325:9325
+    command: 'volume -mserver="master:9333" -port=8080  -metricsPort=9325 -preStopSeconds=1'
+    depends_on:
+      - master
+  filer:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8888:8888
+      - 18888:18888
+      - 9326:9326
+    command: 'filer -master="master:9333"  -metricsPort=9326'
+    tty: true
+    stdin_open: true
+    depends_on:
+      - master
+      - volume
+  mount_1:
+    image: chrislusf/seaweedfs:local
+    privileged: true
+    entrypoint: '/bin/sh -c "mkdir -p t1 && mkdir -p cache/t1 && weed -v=4 mount -filer=filer:8888 -cacheDir=./cache/t1 -dir=./t1 -filer.path=/"'
+    depends_on:
+      - master
+      - volume
+      - filer
+  mount_2:
+    image: chrislusf/seaweedfs:local
+    privileged: true
+    entrypoint: '/bin/sh -c "mkdir -p t2 && mkdir -p cache/t2 && weed -v=4  mount -filer=filer:8888 -cacheDir=./cache/t2 -dir=./t2 -filer.path=/"'
+    depends_on:
+      - master
+      - volume
+      - filer
+      - mount_1

k8s/seaweedfs/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "2.21"
-version: 2.21
+appVersion: "2.22"
+version: 2.22

k8s/seaweedfs/values.yaml

@@ -4,7 +4,7 @@ global:
   registry: ""
   repository: ""
   imageName: chrislusf/seaweedfs
-  # imageTag: "2.21" - started using {.Chart.appVersion}
+  # imageTag: "2.22" - started using {.Chart.appVersion}
   imagePullPolicy: IfNotPresent
   imagePullSecrets: imagepullsecret
   restartPolicy: Always

weed/command/filer_meta_tail.go

@@ -3,9 +3,11 @@ package command
 import (
 	"context"
 	"fmt"
+	"github.com/golang/protobuf/jsonpb"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/olivere/elastic/v7"
 	"io"
+	"os"
 	"path/filepath"
 	"strings"
 	"time"
@@ -25,6 +27,10 @@ var cmdFilerMetaTail = &Command{
 	Short:     "see recent changes on a filer",
 	Long: `See recent changes on a filer.
 
+	weed filer.meta.tail -timeAgo=30h | grep truncate
+	weed filer.meta.tail -timeAgo=30h | jq .
+	weed filer.meta.tail -timeAgo=30h | jq .eventNotification.newEntry.name
+
   `,
 }
 
@@ -80,8 +86,12 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
 		return false
 	}
 
+	jsonpbMarshaler := jsonpb.Marshaler{
+		EmitDefaults: false,
+	}
 	eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
-		fmt.Printf("dir:%s %+v\n", resp.Directory, resp.EventNotification)
+		jsonpbMarshaler.Marshal(os.Stdout, resp)
+		fmt.Fprintln(os.Stdout)
 		return nil
 	}
 	if *esServers != "" {

weed/command/filer_sync.go

@@ -20,19 +20,21 @@ import (
 )
 
 type SyncOptions struct {
-	isActivePassive *bool
-	filerA          *string
-	filerB          *string
-	aPath           *string
-	bPath           *string
-	aReplication    *string
-	bReplication    *string
-	aCollection     *string
-	bCollection     *string
-	aTtlSec         *int
-	bTtlSec         *int
-	aDebug          *bool
-	bDebug          *bool
+	isActivePassive     *bool
+	filerA              *string
+	filerB              *string
+	aPath               *string
+	bPath         *string
+	aReplication  *string
+	bReplication  *string
+	aCollection   *string
+	bCollection   *string
+	aTtlSec       *int
+	bTtlSec       *int
+	aDebug        *bool
+	bDebug        *bool
+	aProxyByFiler *bool
+	bProxyByFiler *bool
 }
 
 var (
@@ -43,7 +45,7 @@ var (
 
 func init() {
 	cmdFilerSynchronize.Run = runFilerSynchronize // break init cycle
-	syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow if true")
+	syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow from A to B if true")
 	syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
 	syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
 	syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
@@ -54,6 +56,8 @@ func init() {
 	syncOptions.bCollection = cmdFilerSynchronize.Flag.String("b.collection", "", "collection on filer B")
 	syncOptions.aTtlSec = cmdFilerSynchronize.Flag.Int("a.ttlSec", 0, "ttl in seconds on filer A")
 	syncOptions.bTtlSec = cmdFilerSynchronize.Flag.Int("b.ttlSec", 0, "ttl in seconds on filer B")
+	syncOptions.aProxyByFiler = cmdFilerSynchronize.Flag.Bool("a.filerProxy", true, "read and write file chunks by filer A instead of volume servers")
+	syncOptions.bProxyByFiler = cmdFilerSynchronize.Flag.Bool("b.filerProxy", true, "read and write file chunks by filer B instead of volume servers")
 	syncOptions.aDebug = cmdFilerSynchronize.Flag.Bool("a.debug", false, "debug mode to print out filer A received files")
 	syncOptions.bDebug = cmdFilerSynchronize.Flag.Bool("b.debug", false, "debug mode to print out filer B received files")
 	syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file")
@@ -62,8 +66,8 @@ func init() {
 
 var cmdFilerSynchronize = &Command{
 	UsageLine: "filer.sync -a=<oneFilerHost>:<oneFilerPort> -b=<otherFilerHost>:<otherFilerPort>",
-	Short:     "continuously synchronize between two active-active or active-passive SeaweedFS clusters",
-	Long: `continuously synchronize file changes between two active-active or active-passive filers
+	Short:     "resumeable continuous synchronization between two active-active or active-passive SeaweedFS clusters",
+	Long: `resumeable continuous synchronization for file changes between two active-active or active-passive filers
 
 	filer.sync listens on filer notifications. If any file is updated, it will fetch the updated content,
 	and write to the other destination. Different from filer.replicate:
@@ -86,8 +90,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 
 	go func() {
 		for {
-			err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.filerB,
-				*syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bDebug)
+			err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.aProxyByFiler,
+				*syncOptions.filerB, *syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bProxyByFiler,
+				*syncOptions.bDebug)
 			if err != nil {
 				glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
 				time.Sleep(1747 * time.Millisecond)
@@ -98,8 +103,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	if !*syncOptions.isActivePassive {
 		go func() {
 			for {
-				err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.filerA,
-					*syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aDebug)
+				err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.bProxyByFiler,
+					*syncOptions.filerA, *syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aProxyByFiler,
+					*syncOptions.aDebug)
 				if err != nil {
 					glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
 					time.Sleep(2147 * time.Millisecond)
@@ -113,8 +119,8 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	return true
 }
 
-func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath, targetFiler, targetPath string,
-	replicationStr, collection string, ttlSec int, debug bool) error {
+func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath string, sourceReadChunkFromFiler bool, targetFiler, targetPath string,
+	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler, debug bool) error {
 
 	// read source filer signature
 	sourceFilerSignature, sourceErr := replication.ReadFilerSignature(grpcDialOption, sourceFiler)
@@ -138,9 +144,9 @@ func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, so
 
 	// create filer sink
 	filerSource := &source.FilerSource{}
-	filerSource.DoInitialize(pb.ServerToGrpcAddress(sourceFiler), sourcePath)
+	filerSource.DoInitialize(sourceFiler, pb.ServerToGrpcAddress(sourceFiler), sourcePath, sourceReadChunkFromFiler)
 	filerSink := &filersink.FilerSink{}
-	filerSink.DoInitialize(pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption)
+	filerSink.DoInitialize(targetFiler, pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption, sinkWriteChunkByFiler)
 	filerSink.SetSourceFiler(filerSource)
 
 	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {

weed/filesys/file.go

@@ -91,6 +91,9 @@ func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp
 func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
 
 	glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
+	if USE_DIRECT_IO {
+		resp.Flags |= fuse.OpenDirectIO
+	}
 
 	handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
 

weed/filesys/file_darwin.go

@@ -0,0 +1,8 @@
+//+build darwin
+
+package filesys
+
+const (
+	USE_DIRECT_IO = false
+)
+

weed/filesys/file_other.go

@@ -0,0 +1,7 @@
+//+build !darwin
+
+package filesys
+
+const (
+	USE_DIRECT_IO = true
+)

weed/filesys/filehandle.go

@@ -90,8 +90,9 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus
 		glog.Warningf("%s FileHandle Read %d: [%d,%d) size %d totalRead %d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, totalRead)
 		totalRead = min(int64(len(buff)), totalRead)
 	}
-	// resp.Data = buff[:totalRead]
-	resp.Data = buff
+	if err == nil {
+		resp.Data = buff[:totalRead]
+	}
 
 	return err
 }

weed/ftpd/ftp_fs.go

@@ -1,61 +0,0 @@
-package ftpd
-
-import (
-	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/spf13/afero"
-	"net"
-	"os"
-	"time"
-)
-
-type FtpFileSystem struct {
-	option      *FtpServerOption
-	ftpListener net.Listener
-}
-
-var (
-	_ = afero.Fs(&FtpFileSystem{})
-)
-
-// NewServer returns a new FTP server driver
-func NewFtpFileSystem(option *FtpServerOption) (*FtpFileSystem, error) {
-	return &FtpFileSystem{
-		option: option,
-	}, nil
-}
-
-func (fs *FtpFileSystem) Create(name string) (afero.File, error) {
-
-}
-func (fs *FtpFileSystem) Mkdir(name string, perm os.FileMode) error {
-
-}
-func (fs *FtpFileSystem) MkdirAll(path string, perm os.FileMode) error {
-
-}
-func (fs *FtpFileSystem) Open(name string) (afero.File, error) {
-
-}
-func (fs *FtpFileSystem) OpenFile(name string, flag int, perm os.FileMode) (afero.File, error) {
-
-}
-func (fs *FtpFileSystem) Remove(name string) error {
-
-}
-func (fs *FtpFileSystem) RemoveAll(path string) error {
-
-}
-func (fs *FtpFileSystem) Rename(oldname, newname string) error {
-
-}
-func (fs *FtpFileSystem) Stat(name string) (os.FileInfo, error) {
-
-}
-func (fs *FtpFileSystem) Name() string {
-	return "SeaweedFS FTP Server " + util.Version()
-}
-func (fs *FtpFileSystem) Chmod(name string, mode os.FileMode) error {
-}
-func (fs *FtpFileSystem) Chtimes(name string, atime time.Time, mtime time.Time) error {
-
-}

weed/ftpd/ftp_server.go

@@ -4,7 +4,6 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
-	"github.com/spf13/afero"
 	"net"
 
 	ftpserver "github.com/fclairamb/ftpserverlib"
@@ -26,18 +25,16 @@ type FtpServerOption struct {
 type SftpServer struct {
 	option      *FtpServerOption
 	ftpListener net.Listener
-	fs          *FtpFileSystem
 }
 
 var _ = ftpserver.MainDriver(&SftpServer{})
 
 // NewServer returns a new FTP server driver
 func NewFtpServer(ftpListener net.Listener, option *FtpServerOption) (*SftpServer, error) {
-	fs, err := NewFtpFileSystem(option)
+	var err error
 	server := &SftpServer{
 		option:      option,
 		ftpListener: ftpListener,
-		fs:          fs,
 	}
 	return server, err
 }
@@ -74,9 +71,7 @@ func (s *SftpServer) ClientDisconnected(cc ftpserver.ClientContext) {
 
 // AuthUser authenticates the user and selects an handling driver
 func (s *SftpServer) AuthUser(cc ftpserver.ClientContext, username, password string) (ftpserver.ClientDriver, error) {
-	return &ClientDriver{
-		Fs: s.fs,
-	}, nil
+	return nil, nil
 }
 
 // GetTLSConfig returns a TLS Certificate to use
@@ -84,7 +79,3 @@ func (s *SftpServer) AuthUser(cc ftpserver.ClientContext, username, password str
 func (s *SftpServer) GetTLSConfig() (*tls.Config, error) {
 	return nil, errors.New("no TLS certificate configured")
 }
-
-type ClientDriver struct {
-	afero.Fs
-}

weed/pb/grpc_client_server.go

@@ -138,6 +138,22 @@ func ServerToGrpcAddress(server string) (serverGrpcAddress string) {
 	return fmt.Sprintf("%s:%d", hostnameAndPort[0], grpcPort)
 }
 
+func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) {
+	hostnameAndPort := strings.Split(grpcAddress, ":")
+	if len(hostnameAndPort) != 2 {
+		return fmt.Sprintf("unexpected grpcAddress: %s", grpcAddress)
+	}
+
+	grpcPort, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
+	if parseErr != nil {
+		return fmt.Sprintf("failed to parse port for %s:%s", hostnameAndPort[0], hostnameAndPort[1])
+	}
+
+	port := int(grpcPort) - 10000
+
+	return fmt.Sprintf("%s:%d", hostnameAndPort[0], port)
+}
+
 func WithMasterClient(master string, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) error {
 
 	masterGrpcAddress, parseErr := ParseServerToGrpcAddress(master)

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)
@@ -97,6 +98,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,7 @@ package filersink
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 
 	"google.golang.org/grpc"
@@ -18,14 +19,16 @@ import (
 )
 
 type FilerSink struct {
-	filerSource    *source.FilerSource
-	grpcAddress    string
-	dir            string
-	replication    string
-	collection     string
-	ttlSec         int32
-	dataCenter     string
-	grpcDialOption grpc.DialOption
+	filerSource       *source.FilerSource
+	grpcAddress       string
+	dir               string
+	replication       string
+	collection        string
+	ttlSec            int32
+	dataCenter        string
+	grpcDialOption    grpc.DialOption
+	address           string
+	writeChunkByFiler bool
 }
 
 func init() {
@@ -42,26 +45,33 @@ 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"),
-		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, grpcDialOption grpc.DialOption) (err error) {
+func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
+	replication string, collection string, ttlSec int, 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
 	fs.collection = collection
 	fs.ttlSec = int32(ttlSec)
 	fs.grpcDialOption = grpcDialOption
+	fs.writeChunkByFiler = writeChunkByFiler
 	return nil
 }
 

weed/replication/source/filer_source.go

@@ -25,19 +25,28 @@ type FilerSource struct {
 	grpcAddress    string
 	grpcDialOption grpc.DialOption
 	Dir            string
+	address        string
+	proxyByFiler   bool
 }
 
 func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
 	return fs.DoInitialize(
+		"",
 		configuration.GetString(prefix+"grpcAddress"),
 		configuration.GetString(prefix+"directory"),
+		false,
 	)
 }
 
-func (fs *FilerSource) DoInitialize(grpcAddress string, dir string) (err error) {
+func (fs *FilerSource) DoInitialize(address, grpcAddress string, dir string, readChunkFromFiler bool) (err error) {
+	fs.address = address
+	if fs.address == "" {
+		fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.Dir = dir
 	fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
+	fs.proxyByFiler = readChunkFromFiler
 	return nil
 }
 
@@ -81,9 +90,13 @@ func (fs *FilerSource) LookupFileId(part string) (fileUrls []string, err error)
 	return
 }
 
-func (fs *FilerSource) ReadPart(part string) (filename string, header http.Header, resp *http.Response, err error) {
+func (fs *FilerSource) ReadPart(fileId string) (filename string, header http.Header, resp *http.Response, err error) {
 
-	fileUrls, err := fs.LookupFileId(part)
+	if fs.proxyByFiler {
+		return util.DownloadFile("http://" + fs.address + "/?proxyChunkId=" + fileId)
+	}
+
+	fileUrls, err := fs.LookupFileId(fileId)
 	if err != nil {
 		return "", nil, nil, err
 	}

weed/s3api/s3api_objects_list_handlers.go

@@ -238,7 +238,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 				return
 			}
 		}
-		if counter >= maxKeys + 1 {
+		if counter >= maxKeys {
 			isTruncated = true
 			return
 		}

weed/server/filer_server_handlers.go

@@ -3,18 +3,33 @@ package weed_server
 import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/stats"
 )
 
 func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
+
+	start := time.Now()
+
+	// proxy to volume servers
+	var fileId string
+	if strings.HasPrefix(r.RequestURI, "/?proxyChunkId=") {
+		fileId = r.RequestURI[len("/?proxyChunkId="):]
+	}
+	if fileId != "" {
+		stats.FilerRequestCounter.WithLabelValues("proxy").Inc()
+		fs.proxyToVolumeServer(w,r,fileId)
+		stats.FilerRequestHistogram.WithLabelValues("proxy").Observe(time.Since(start).Seconds())
+		return
+	}
+
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 	if r.Header.Get("Origin") != "" {
 		w.Header().Set("Access-Control-Allow-Origin", "*")
 		w.Header().Set("Access-Control-Allow-Credentials", "true")
 	}
-	start := time.Now()
 	switch r.Method {
 	case "GET":
 		stats.FilerRequestCounter.WithLabelValues("get").Inc()

weed/server/filer_server_handlers_proxy.go

@@ -0,0 +1,66 @@
+package weed_server
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io"
+	"math/rand"
+	"net/http"
+)
+
+var (
+	client *http.Client
+)
+
+func init() {
+	client = &http.Client{Transport: &http.Transport{
+		MaxIdleConnsPerHost: 1024,
+	}}
+}
+
+func (fs *FilerServer) proxyToVolumeServer(w http.ResponseWriter, r *http.Request, fileId string) {
+
+	urlStrings, err := fs.filer.MasterClient.GetLookupFileIdFunction()(fileId)
+	if err != nil {
+		glog.Errorf("locate %s: %v", fileId, err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	if len(urlStrings) == 0 {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	proxyReq, err := http.NewRequest(r.Method, urlStrings[rand.Intn(len(urlStrings))], r.Body)
+	if err != nil {
+		glog.Errorf("NewRequest %s: %v", urlStrings[0], err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	proxyReq.Header.Set("Host", r.Host)
+	proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
+
+	for header, values := range r.Header {
+		for _, value := range values {
+			proxyReq.Header.Add(header, value)
+		}
+	}
+
+	proxyResponse, postErr := client.Do(proxyReq)
+
+	if postErr != nil {
+		glog.Errorf("post to filer: %v", postErr)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	defer util.CloseResponse(proxyResponse)
+
+	for k, v := range proxyResponse.Header {
+		w.Header()[k] = v
+	}
+	w.WriteHeader(proxyResponse.StatusCode)
+	io.Copy(w, proxyResponse.Body)
+
+}

weed/server/filer_server_handlers_read.go

@@ -159,10 +159,10 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
 	}
 
 	processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64, httpStatusCode int) error {
+		if httpStatusCode != 0 {
+			w.WriteHeader(httpStatusCode)
+		}
 		if offset+size <= int64(len(entry.Content)) {
-			if httpStatusCode != 0 {
-				w.WriteHeader(httpStatusCode)
-			}
 			_, err := writer.Write(entry.Content[offset : offset+size])
 			return err
 		}

weed/server/filer_server_handlers_write_autochunk.go

@@ -204,7 +204,7 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 
 	filerResult = &FilerPostResult{
 		Name: fileName,
-		Size: chunkOffset,
+		Size: int64(entry.FileSize),
 	}
 
 	if entry.Extended == nil {

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 21)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 22)
 	COMMIT  = ""
 )