2.70

Previous implementation append filer logs into one file. So one file is not always sorted, which can lead to miss reading some entries, especially when different filers have different write throughput.

.github/workflows/binaries_dev.yml

@@ -6,57 +6,72 @@ on:
 
 jobs:
 
-
-  build-latest-docker-image:
-    runs-on: [ubuntu-latest]
+  cleanup:
+    runs-on: ubuntu-latest
 
     steps:
-      -
-        name: Checkout
+
+      - name: Delete old release assets
+        uses: mknejp/delete-release-assets@v1
+        with:
+          token: ${{ github.token }}
+          tag: dev
+          fail-if-no-assets: false
+          assets: |
+            weed-*
+
+  build_dev:
+    needs: cleanup
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        goos: [linux, windows, darwin, freebsd]
+        goarch: [amd64, arm, arm64]
+        exclude:
+          - goarch: arm
+            goos: darwin
+          - goarch: 386
+            goos: darwin
+          - goarch: arm
+            goos: windows
+          - goarch: arm64
+            goos: windows
+
+    steps:
+
+      - name: Check out code into the Go module directory
         uses: actions/checkout@v2
-      -
-        name: Docker meta
-        id: docker_meta
-        uses: docker/metadata-action@v3
+
+      - name: Set BUILD_TIME env
+        run: echo BUILD_TIME=$(date -u +%Y%m%d-%H%M) >> ${GITHUB_ENV}
+
+      - name: Go Release Binaries Large Disk
+        uses: wangyoucao577/go-release-action@v1.20
         with:
-          images: |
-            chrislusf/seaweedfs
-            ghcr.io/chrislusf/seaweedfs
-          tags: |
-            type=raw,value=latest
-          labels: |
-            org.opencontainers.image.title=seaweedfs
-            org.opencontainers.image.vendor=Chris Lu
-      -
-        name: Set up QEMU
-        uses: docker/setup-qemu-action@v1
-      -
-        name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          goos: ${{ matrix.goos }}
+          goarch: ${{ matrix.goarch }}
+          release_tag: dev
+          overwrite: true
+          pre_command: export CGO_ENABLED=0
+          build_flags: -tags 5BytesOffset # optional, default is
+          ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
+          # Where to run `go build .`
+          project_path: weed
+          binary_name: weed-large-disk
+          asset_name: "weed-large-disk-${{ env.BUILD_TIME }}-${{ matrix.goos }}-${{ matrix.goarch }}"
+
+      - name: Go Release Binaries Normal Volume Size
+        uses: wangyoucao577/go-release-action@v1.20
         with:
-          buildkitd-flags: "--debug"
-      -
-        name: Login to Docker Hub
-        if: github.event_name != 'pull_request'
-        uses: docker/login-action@v1
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
-      -
-        name: Login to GHCR
-        if: github.event_name != 'pull_request'
-        uses: docker/login-action@v1
-        with:
-          registry: ghcr.io
-          username: ${{ secrets.GHCR_USERNAME }}
-          password: ${{ secrets.GHCR_TOKEN }}
-      -
-        name: Build
-        uses: docker/build-push-action@v2
-        with:
-          context: ./docker
-          push: ${{ github.event_name != 'pull_request' }}
-          file: ./docker/Dockerfile
-          platforms: linux/amd64, linux/arm, linux/arm64
-          tags: ${{ steps.docker_meta.outputs.tags }}
-          labels: ${{ steps.docker_meta.outputs.labels }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          goos: ${{ matrix.goos }}
+          goarch: ${{ matrix.goarch }}
+          release_tag: dev
+          overwrite: true
+          pre_command: export CGO_ENABLED=0
+          ldflags: -extldflags -static -X github.com/chrislusf/seaweedfs/weed/util.COMMIT=${{github.sha}}
+          # Where to run `go build .`
+          project_path: weed
+          binary_name: weed-normal-disk
+          asset_name: "weed-${{ env.BUILD_TIME }}-${{ matrix.goos }}-${{ matrix.goarch }}"

README.md

@@ -147,6 +147,7 @@ Faster and Cheaper than direct cloud storage!
 * [AES256-GCM Encrypted Storage][FilerDataEncryption] safely stores the encrypted data.
 * [Super Large Files][SuperLargeFiles] stores large or super large files in tens of TB.
 * [Cloud Drive][CloudDrive] mounts cloud storage to local cluster, cached for fast read and write with asynchronous write back.
+* [Gateway to Remote Object Store][GatewayToRemoteObjectStore] mirrors bucket operations to remote object storage, in addition to [Cloud Drive][CloudDrive]
 
 ## Kubernetes ##
 * [Kubernetes CSI Driver][SeaweedFsCsiDriver] A Container Storage Interface (CSI) Driver. [![Docker Pulls](https://img.shields.io/docker/pulls/chrislusf/seaweedfs-csi-driver.svg?maxAge=4800)](https://hub.docker.com/r/chrislusf/seaweedfs-csi-driver/)
@@ -170,6 +171,7 @@ Faster and Cheaper than direct cloud storage!
 [FilerStoreReplication]: https://github.com/chrislusf/seaweedfs/wiki/Filer-Store-Replication
 [KeyLargeValueStore]: https://github.com/chrislusf/seaweedfs/wiki/Filer-as-a-Key-Large-Value-Store
 [CloudDrive]: https://github.com/chrislusf/seaweedfs/wiki/Cloud-Drive-Architecture
+[GatewayToRemoteObjectStore]: https://github.com/chrislusf/seaweedfs/wiki/Gateway-to-Remote-Object-Storage
 
 
 [Back to TOC](#table-of-contents)

k8s/helm_charts2/Chart.yaml

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

test/s3/multipart/aws_upload.go

@@ -19,10 +19,10 @@ import (
 const (
 	maxPartSize        = int64(5 * 1024 * 1024)
 	maxRetries         = 3
-	awsAccessKeyID     = "Your access key"
-	awsSecretAccessKey = "Your secret key"
-	awsBucketRegion    = "S3 bucket region"
-	awsBucketName      = "newBucket"
+	awsAccessKeyID     = "any"
+	awsSecretAccessKey = "any"
+	awsBucketRegion    = "us‑west‑1"
+	awsBucketName      = "bucket1"
 )
 
 var (
@@ -37,7 +37,7 @@ func main() {
 	if err != nil {
 		fmt.Printf("bad credentials: %s", err)
 	}
-	cfg := aws.NewConfig().WithRegion(awsBucketRegion).WithCredentials(creds).WithDisableSSL(true).WithEndpoint("localhost:8333")
+	cfg := aws.NewConfig().WithRegion(awsBucketRegion).WithCredentials(creds).WithDisableSSL(true).WithEndpoint("localhost:8333").WithS3ForcePathStyle(true)
 	svc := s3.New(session.New(), cfg)
 
 	file, err := os.Open(*filename)

weed/command/command.go

@@ -21,6 +21,7 @@ var Commands = []*Command{
 	cmdFilerCopy,
 	cmdFilerMetaBackup,
 	cmdFilerMetaTail,
+	cmdFilerRemoteGateway,
 	cmdFilerRemoteSynchronize,
 	cmdFilerReplicate,
 	cmdFilerSynchronize,

weed/command/filer.go

@@ -62,7 +62,7 @@ func init() {
 	f.ip = cmdFiler.Flag.String("ip", util.DetectedHostAddress(), "filer server http listen ip address")
 	f.bindIp = cmdFiler.Flag.String("ip.bind", "", "ip address to bind to")
 	f.port = cmdFiler.Flag.Int("port", 8888, "filer server http listen port")
-	f.portGrpc = cmdFiler.Flag.Int("port.grpc", 18888, "filer server grpc listen port")
+	f.portGrpc = cmdFiler.Flag.Int("port.grpc", 0, "filer server grpc listen port")
 	f.publicPort = cmdFiler.Flag.Int("port.readonly", 0, "readonly port opened to public")
 	f.defaultReplicaPlacement = cmdFiler.Flag.String("defaultReplicaPlacement", "", "default replication type. If not specified, use master setting.")
 	f.disableDirListing = cmdFiler.Flag.Bool("disableDirListing", false, "turn off directory listing")
@@ -180,6 +180,9 @@ func (fo *FilerOptions) startFiler() {
 	if *fo.publicPort != 0 {
 		publicVolumeMux = http.NewServeMux()
 	}
+	if *fo.portGrpc == 0 {
+		*fo.portGrpc = 10000 + *fo.port
+	}
 
 	defaultLevelDbDirectory := util.ResolvePath(*fo.defaultLevelDbDirectory + "/filerldb2")
 

weed/command/filer_meta_tail.go

@@ -71,12 +71,12 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
 	}
 
 	shouldPrint := func(resp *filer_pb.SubscribeMetadataResponse) bool {
-		if filterFunc == nil {
-			return true
-		}
 		if resp.EventNotification.OldEntry == nil && resp.EventNotification.NewEntry == nil {
 			return false
 		}
+		if filterFunc == nil {
+			return true
+		}
 		if resp.EventNotification.OldEntry != nil && filterFunc(resp.Directory, resp.EventNotification.OldEntry.Name) {
 			return true
 		}

weed/command/filer_remote_gateway.go

@@ -0,0 +1,117 @@
+package command
+
+import (
+	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+	"os"
+	"time"
+)
+
+type RemoteGatewayOptions struct {
+	filerAddress             *string
+	grpcDialOption           grpc.DialOption
+	readChunkFromFiler       *bool
+	timeAgo                  *time.Duration
+	createBucketAt           *string
+	createBucketRandomSuffix *bool
+	include                  *string
+	exclude                  *string
+
+	mappings    *remote_pb.RemoteStorageMapping
+	remoteConfs map[string]*remote_pb.RemoteConf
+	bucketsDir  string
+}
+
+var _ = filer_pb.FilerClient(&RemoteGatewayOptions{})
+
+func (option *RemoteGatewayOptions) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+	return pb.WithFilerClient(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		return fn(client)
+	})
+}
+func (option *RemoteGatewayOptions) AdjustedUrl(location *filer_pb.Location) string {
+	return location.Url
+}
+
+var (
+	remoteGatewayOptions RemoteGatewayOptions
+)
+
+func init() {
+	cmdFilerRemoteGateway.Run = runFilerRemoteGateway // break init cycle
+	remoteGatewayOptions.filerAddress = cmdFilerRemoteGateway.Flag.String("filer", "localhost:8888", "filer of the SeaweedFS cluster")
+	remoteGatewayOptions.createBucketAt = cmdFilerRemoteGateway.Flag.String("createBucketAt", "", "one remote storage name to create new buckets in")
+	remoteGatewayOptions.createBucketRandomSuffix = cmdFilerRemoteGateway.Flag.Bool("createBucketWithRandomSuffix", true, "add randomized suffix to bucket name to avoid conflicts")
+	remoteGatewayOptions.readChunkFromFiler = cmdFilerRemoteGateway.Flag.Bool("filerProxy", false, "read file chunks from filer instead of volume servers")
+	remoteGatewayOptions.timeAgo = cmdFilerRemoteGateway.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
+	remoteGatewayOptions.include = cmdFilerRemoteGateway.Flag.String("include", "", "pattens of new bucket names, e.g., s3*")
+	remoteGatewayOptions.exclude = cmdFilerRemoteGateway.Flag.String("exclude", "", "pattens of new bucket names, e.g., local*")
+}
+
+var cmdFilerRemoteGateway = &Command{
+	UsageLine: "filer.remote.gateway",
+	Short:     "resumable continuously write back bucket creation, deletion, and other local updates to remote object store",
+	Long: `resumable continuously write back bucket creation, deletion, and other local updates to remote object store
+
+	filer.remote.gateway listens on filer local buckets update events. 
+	If any bucket is created, deleted, or updated, it will mirror the changes to remote object store.
+
+		weed filer.remote.sync -createBucketAt=cloud1
+
+`,
+}
+
+func runFilerRemoteGateway(cmd *Command, args []string) bool {
+
+	util.LoadConfiguration("security", false)
+	grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
+	remoteGatewayOptions.grpcDialOption = grpcDialOption
+
+	filerAddress := pb.ServerAddress(*remoteGatewayOptions.filerAddress)
+
+	filerSource := &source.FilerSource{}
+	filerSource.DoInitialize(
+		filerAddress.ToHttpAddress(),
+		filerAddress.ToGrpcAddress(),
+		"/", // does not matter
+		*remoteGatewayOptions.readChunkFromFiler,
+	)
+
+	remoteGatewayOptions.bucketsDir = "/buckets"
+	// check buckets again
+	remoteGatewayOptions.WithFilerClient(func(filerClient filer_pb.SeaweedFilerClient) error {
+		resp, err := filerClient.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		if err != nil {
+			return err
+		}
+		remoteGatewayOptions.bucketsDir = resp.DirBuckets
+		return nil
+	})
+
+	// read filer remote storage mount mappings
+	if detectErr := remoteGatewayOptions.collectRemoteStorageConf(); detectErr != nil {
+		fmt.Fprintf(os.Stderr, "read mount info: %v\n", detectErr)
+		return true
+	}
+
+	// synchronize /buckets folder
+	fmt.Printf("synchronize buckets in %s ...\n", remoteGatewayOptions.bucketsDir)
+	util.RetryForever("filer.remote.sync buckets", func() error {
+		return remoteGatewayOptions.followBucketUpdatesAndUploadToRemote(filerSource)
+	}, func(err error) bool {
+		if err != nil {
+			glog.Errorf("synchronize %s: %v", remoteGatewayOptions.bucketsDir, err)
+		}
+		return true
+	})
+	return true
+
+}

weed/command/filer_remote_gateway_buckets.go

@@ -13,11 +13,12 @@ import (
 	"github.com/golang/protobuf/proto"
 	"math"
 	"math/rand"
+	"path/filepath"
 	"strings"
 	"time"
 )
 
-func (option *RemoteSyncOptions) followBucketUpdatesAndUploadToRemote(filerSource *source.FilerSource) error {
+func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSource *source.FilerSource) error {
 
 	// read filer remote storage mount mappings
 	if detectErr := option.collectRemoteStorageConf(); detectErr != nil {
@@ -35,13 +36,13 @@ func (option *RemoteSyncOptions) followBucketUpdatesAndUploadToRemote(filerSourc
 		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, lastTsNs)
 	})
 
-	lastOffsetTs := collectLastSyncOffset(option, option.bucketsDir)
+	lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo)
 
 	return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync",
 		option.bucketsDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
 }
 
-func (option *RemoteSyncOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {
+func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {
 
 	handleCreateBucket := func(entry *filer_pb.Entry) error {
 		if !entry.IsDirectory {
@@ -75,6 +76,16 @@ func (option *RemoteSyncOptions) makeBucketedEventProcessor(filerSource *source.
 		}
 
 		bucketName := strings.ToLower(entry.Name)
+		if *option.include != "" {
+			if ok, _ := filepath.Match(*option.include, entry.Name); !ok {
+				return nil
+			}
+		}
+		if *option.exclude != "" {
+			if ok, _ := filepath.Match(*option.exclude, entry.Name); ok {
+				return nil
+			}
+		}
 		if *option.createBucketRandomSuffix {
 			// https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
 			if len(bucketName)+5 > 63 {
@@ -307,7 +318,7 @@ func (option *RemoteSyncOptions) makeBucketedEventProcessor(filerSource *source.
 	return eachEntryFunc, nil
 }
 
-func (option *RemoteSyncOptions) findRemoteStorageClient(bucketName string) (client remote_storage.RemoteStorageClient, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, err error) {
+func (option *RemoteGatewayOptions) findRemoteStorageClient(bucketName string) (client remote_storage.RemoteStorageClient, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, err error) {
 	bucket := util.FullPath(option.bucketsDir).Child(bucketName)
 
 	var isMounted bool
@@ -327,7 +338,7 @@ func (option *RemoteSyncOptions) findRemoteStorageClient(bucketName string) (cli
 	return client, remoteStorageMountLocation, nil
 }
 
-func (option *RemoteSyncOptions) detectBucketInfo(actualDir string) (bucket util.FullPath, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, remoteConf *remote_pb.RemoteConf, ok bool) {
+func (option *RemoteGatewayOptions) detectBucketInfo(actualDir string) (bucket util.FullPath, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, remoteConf *remote_pb.RemoteConf, ok bool) {
 	bucket, ok = extractBucketPath(option.bucketsDir, actualDir)
 	if !ok {
 		return "", nil, nil, false
@@ -355,7 +366,7 @@ func extractBucketPath(bucketsDir, dir string) (util.FullPath, bool) {
 	return util.FullPath(bucketsDir).Child(parts[0]), true
 }
 
-func (option *RemoteSyncOptions) collectRemoteStorageConf() (err error) {
+func (option *RemoteGatewayOptions) collectRemoteStorageConf() (err error) {
 
 	if mappings, err := filer.ReadMountMappings(option.grpcDialOption, pb.ServerAddress(*option.filerAddress)); err != nil {
 		return err

weed/command/filer_remote_sync.go

@@ -1,33 +1,23 @@
 package command
 
 import (
-	"context"
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/remote_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc"
-	"os"
 	"time"
 )
 
 type RemoteSyncOptions struct {
-	filerAddress             *string
-	grpcDialOption           grpc.DialOption
-	readChunkFromFiler       *bool
-	debug                    *bool
-	timeAgo                  *time.Duration
-	dir                      *string
-	createBucketAt           *string
-	createBucketRandomSuffix *bool
-
-	mappings    *remote_pb.RemoteStorageMapping
-	remoteConfs map[string]*remote_pb.RemoteConf
-	bucketsDir  string
+	filerAddress       *string
+	grpcDialOption     grpc.DialOption
+	readChunkFromFiler *bool
+	timeAgo            *time.Duration
+	dir                *string
 }
 
 var _ = filer_pb.FilerClient(&RemoteSyncOptions{})
@@ -49,10 +39,7 @@ func init() {
 	cmdFilerRemoteSynchronize.Run = runFilerRemoteSynchronize // break init cycle
 	remoteSyncOptions.filerAddress = cmdFilerRemoteSynchronize.Flag.String("filer", "localhost:8888", "filer of the SeaweedFS cluster")
 	remoteSyncOptions.dir = cmdFilerRemoteSynchronize.Flag.String("dir", "", "a mounted directory on filer")
-	remoteSyncOptions.createBucketAt = cmdFilerRemoteSynchronize.Flag.String("createBucketAt", "", "one remote storage name to create new buckets in")
-	remoteSyncOptions.createBucketRandomSuffix = cmdFilerRemoteSynchronize.Flag.Bool("createBucketWithRandomSuffix", true, "add randomized suffix to bucket name to avoid conflicts")
 	remoteSyncOptions.readChunkFromFiler = cmdFilerRemoteSynchronize.Flag.Bool("filerProxy", false, "read file chunks from filer instead of volume servers")
-	remoteSyncOptions.debug = cmdFilerRemoteSynchronize.Flag.Bool("debug", false, "debug mode to print out filer updated remote files")
 	remoteSyncOptions.timeAgo = cmdFilerRemoteSynchronize.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
 }
 
@@ -65,21 +52,8 @@ var cmdFilerRemoteSynchronize = &Command{
 	If any mounted remote file is updated, it will fetch the updated content,
 	and write to the remote storage.
 
-	There are two modes:
-
-	1)By default, watch /buckets folder and write back all changes.
-
-		# if there is only one remote storage configured
-		weed filer.remote.sync
-		# if there are multiple remote storages configured
-		# specify a remote storage to create new buckets.
-		weed filer.remote.sync -createBucketAt=cloud1
-
-	2)Write back one mounted folder to remote storage
-
 		weed filer.remote.sync -dir=/mount/s3_on_cloud
 
-
 `,
 }
 
@@ -100,18 +74,7 @@ func runFilerRemoteSynchronize(cmd *Command, args []string) bool {
 		*remoteSyncOptions.readChunkFromFiler,
 	)
 
-	remoteSyncOptions.bucketsDir = "/buckets"
-	// check buckets again
-	remoteSyncOptions.WithFilerClient(func(filerClient filer_pb.SeaweedFilerClient) error {
-		resp, err := filerClient.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
-		if err != nil {
-			return err
-		}
-		remoteSyncOptions.bucketsDir = resp.DirBuckets
-		return nil
-	})
-
-	if dir != "" && dir != remoteSyncOptions.bucketsDir {
+	if dir != "" {
 		fmt.Printf("synchronize %s to remote storage...\n", dir)
 		util.RetryForever("filer.remote.sync "+dir, func() error {
 			return followUpdatesAndUploadToRemote(&remoteSyncOptions, filerSource, dir)
@@ -124,22 +87,6 @@ func runFilerRemoteSynchronize(cmd *Command, args []string) bool {
 		return true
 	}
 
-	// read filer remote storage mount mappings
-	if detectErr := remoteSyncOptions.collectRemoteStorageConf(); detectErr != nil {
-		fmt.Fprintf(os.Stderr, "read mount info: %v\n", detectErr)
-		return true
-	}
-
-	// synchronize /buckets folder
-	fmt.Printf("synchronize buckets in %s ...\n", remoteSyncOptions.bucketsDir)
-	util.RetryForever("filer.remote.sync buckets", func() error {
-		return remoteSyncOptions.followBucketUpdatesAndUploadToRemote(filerSource)
-	}, func(err error) bool {
-		if err != nil {
-			glog.Errorf("synchronize %s: %v", remoteSyncOptions.bucketsDir, err)
-		}
-		return true
-	})
 	return true
 
 }

weed/command/filer_remote_sync_dir.go

@@ -12,6 +12,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/golang/protobuf/proto"
+	"google.golang.org/grpc"
 	"os"
 	"strings"
 	"time"
@@ -36,7 +37,7 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, lastTsNs)
 	})
 
-	lastOffsetTs := collectLastSyncOffset(option, mountedDir)
+	lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo)
 
 	return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync",
 		mountedDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
@@ -159,19 +160,19 @@ func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string,
 	return eachEntryFunc, nil
 }
 
-func collectLastSyncOffset(option *RemoteSyncOptions, mountedDir string) time.Time {
+func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time {
 	// 1. specified by timeAgo
 	// 2. last offset timestamp for this directory
 	// 3. directory creation time
 	var lastOffsetTs time.Time
-	if *option.timeAgo == 0 {
-		mountedDirEntry, err := filer_pb.GetEntry(option, util.FullPath(mountedDir))
+	if timeAgo == 0 {
+		mountedDirEntry, err := filer_pb.GetEntry(filerClient, util.FullPath(mountedDir))
 		if err != nil {
 			glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err)
 			return time.Now()
 		}
 
-		lastOffsetTsNs, err := remote_storage.GetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir)
+		lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir)
 		if mountedDirEntry != nil {
 			if err == nil && mountedDirEntry.Attributes.Crtime < lastOffsetTsNs/1000000 {
 				lastOffsetTs = time.Unix(0, lastOffsetTsNs)
@@ -183,7 +184,7 @@ func collectLastSyncOffset(option *RemoteSyncOptions, mountedDir string) time.Ti
 			lastOffsetTs = time.Now()
 		}
 	} else {
-		lastOffsetTs = time.Now().Add(-*option.timeAgo)
+		lastOffsetTs = time.Now().Add(-timeAgo)
 	}
 	return lastOffsetTs
 }

weed/command/master.go

@@ -47,7 +47,7 @@ type MasterOptions struct {
 func init() {
 	cmdMaster.Run = runMaster // break init cycle
 	m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
-	m.portGrpc = cmdMaster.Flag.Int("port.grpc", 19333, "grpc listen port")
+	m.portGrpc = cmdMaster.Flag.Int("port.grpc", 0, "grpc listen port")
 	m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
 	m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to")
 	m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
@@ -113,6 +113,10 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 
 	backend.LoadConfiguration(util.GetViper())
 
+	if *masterOption.portGrpc == 0 {
+		*masterOption.portGrpc = 10000 + *masterOption.port
+	}
+
 	myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.portGrpc, *masterOption.peers)
 
 	r := mux.NewRouter()

weed/command/master_follower.go

@@ -23,7 +23,7 @@ var (
 func init() {
 	cmdMasterFollower.Run = runMasterFollower // break init cycle
 	mf.port = cmdMasterFollower.Flag.Int("port", 9334, "http listen port")
-	mf.portGrpc = cmdMasterFollower.Flag.Int("port.grpc", 19334, "grpc listen port")
+	mf.portGrpc = cmdMasterFollower.Flag.Int("port.grpc", 0, "grpc listen port")
 	mf.ipBind = cmdMasterFollower.Flag.String("ip.bind", "", "ip address to bind to")
 	mf.peers = cmdMasterFollower.Flag.String("masters", "localhost:9333", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095")
 
@@ -70,6 +70,10 @@ func runMasterFollower(cmd *Command, args []string) bool {
 	util.LoadConfiguration("security", false)
 	util.LoadConfiguration("master", false)
 
+	if *mf.portGrpc == 0 {
+		*mf.portGrpc = 10000 + *mf.port
+	}
+
 	startMasterFollower(mf)
 
 	return true

weed/command/server.go

@@ -86,7 +86,7 @@ func init() {
 	serverOptions.debugPort = cmdServer.Flag.Int("debug.port", 6060, "http port for debugging")
 
 	masterOptions.port = cmdServer.Flag.Int("master.port", 9333, "master server http listen port")
-	masterOptions.portGrpc = cmdServer.Flag.Int("master.port.grpc", 19333, "master server grpc listen port")
+	masterOptions.portGrpc = cmdServer.Flag.Int("master.port.grpc", 0, "master server grpc listen port")
 	masterOptions.metaFolder = cmdServer.Flag.String("master.dir", "", "data directory to store meta data, default to same as -dir specified")
 	masterOptions.peers = cmdServer.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list")
 	masterOptions.volumeSizeLimitMB = cmdServer.Flag.Uint("master.volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
@@ -99,7 +99,7 @@ func init() {
 
 	filerOptions.collection = cmdServer.Flag.String("filer.collection", "", "all data will be stored in this collection")
 	filerOptions.port = cmdServer.Flag.Int("filer.port", 8888, "filer server http listen port")
-	filerOptions.portGrpc = cmdServer.Flag.Int("filer.port.grpc", 18888, "filer server grpc listen port")
+	filerOptions.portGrpc = cmdServer.Flag.Int("filer.port.grpc", 0, "filer server grpc listen port")
 	filerOptions.publicPort = cmdServer.Flag.Int("filer.port.public", 0, "filer server public http listen port")
 	filerOptions.defaultReplicaPlacement = cmdServer.Flag.String("filer.defaultReplicaPlacement", "", "default replication type. If not specified, use master setting.")
 	filerOptions.disableDirListing = cmdServer.Flag.Bool("filer.disableDirListing", false, "turn off directory listing")
@@ -111,7 +111,7 @@ func init() {
 	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
 
 	serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
-	serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 18080, "volume server grpc listen port")
+	serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 0, "volume server grpc listen port")
 	serverOptions.v.publicPort = cmdServer.Flag.Int("volume.port.public", 0, "volume server public port")
 	serverOptions.v.indexType = cmdServer.Flag.String("volume.index", "memory", "Choose [memory|leveldb|leveldbMedium|leveldbLarge] mode for memory~performance balance.")
 	serverOptions.v.diskType = cmdServer.Flag.String("volume.disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")

weed/command/volume.go

@@ -70,7 +70,7 @@ type VolumeServerOptions struct {
 func init() {
 	cmdVolume.Run = runVolume // break init cycle
 	v.port = cmdVolume.Flag.Int("port", 8080, "http listen port")
-	v.portGrpc = cmdVolume.Flag.Int("port.grpc", 18080, "grpc listen port")
+	v.portGrpc = cmdVolume.Flag.Int("port.grpc", 0, "grpc listen port")
 	v.publicPort = cmdVolume.Flag.Int("port.public", 0, "port opened to public")
 	v.ip = cmdVolume.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
 	v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
@@ -197,6 +197,9 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 	if *v.publicPort == 0 {
 		*v.publicPort = *v.port
 	}
+	if *v.portGrpc == 0 {
+		*v.portGrpc = 10000 + *v.port
+	}
 	if *v.publicUrl == "" {
 		*v.publicUrl = util.JoinHostPort(*v.ip, *v.publicPort)
 	}

weed/command/webdav.go

@@ -103,16 +103,16 @@ func (wo *WebDavOption) startWebDav() bool {
 	}
 
 	ws, webdavServer_err := weed_server.NewWebDavServer(&weed_server.WebDavOption{
-		Filer:            filerAddress,
-		GrpcDialOption:   grpcDialOption,
-		Collection:       *wo.collection,
-		Replication:      *wo.replication,
-		DiskType:         *wo.disk,
-		Uid:              uid,
-		Gid:              gid,
-		Cipher:           cipher,
-		CacheDir:         util.ResolvePath(*wo.cacheDir),
-		CacheSizeMB:      *wo.cacheSizeMB,
+		Filer:          filerAddress,
+		GrpcDialOption: grpcDialOption,
+		Collection:     *wo.collection,
+		Replication:    *wo.replication,
+		DiskType:       *wo.disk,
+		Uid:            uid,
+		Gid:            gid,
+		Cipher:         cipher,
+		CacheDir:       util.ResolvePath(*wo.cacheDir),
+		CacheSizeMB:    *wo.cacheSizeMB,
 	})
 	if webdavServer_err != nil {
 		glog.Fatalf("WebDav Server startup error: %v", webdavServer_err)

weed/filer/filer.go

@@ -44,6 +44,7 @@ type Filer struct {
 	Signature           int32
 	FilerConf           *FilerConf
 	RemoteStorage       *FilerRemoteStorage
+	UniqueFileId        uint32
 }
 
 func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
@@ -54,6 +55,7 @@ func NewFiler(masters []pb.ServerAddress, grpcDialOption grpc.DialOption,
 		GrpcDialOption:      grpcDialOption,
 		FilerConf:           NewFilerConf(),
 		RemoteStorage:       NewFilerRemoteStorage(),
+		UniqueFileId:        uint32(util.RandomInt32()),
 	}
 	f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, notifyFn)
 	f.metaLogCollection = collection

weed/filer/filer_notify.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"strings"
 	"time"
 
@@ -92,8 +93,8 @@ func (f *Filer) logFlushFunc(startTime, stopTime time.Time, buf []byte) {
 
 	startTime, stopTime = startTime.UTC(), stopTime.UTC()
 
-	targetFile := fmt.Sprintf("%s/%04d-%02d-%02d/%02d-%02d.segment", SystemLogDir,
-		startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(),
+	targetFile := fmt.Sprintf("%s/%04d-%02d-%02d/%02d-%02d.%08x", SystemLogDir,
+		startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(), f.UniqueFileId,
 		// startTime.Second(), startTime.Nanosecond(),
 	)
 
@@ -111,7 +112,7 @@ func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(
 
 	startTime = startTime.UTC()
 	startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
-	startHourMinute := fmt.Sprintf("%02d-%02d.segment", startTime.Hour(), startTime.Minute())
+	startHourMinute := fmt.Sprintf("%02d-%02d", startTime.Hour(), startTime.Minute())
 
 	sizeBuf := make([]byte, 4)
 	startTsNs := startTime.UnixNano()
@@ -122,14 +123,15 @@ func (f *Filer) ReadPersistedLogBuffer(startTime time.Time, eachLogEntryFn func(
 	}
 	for _, dayEntry := range dayEntries {
 		// println("checking day", dayEntry.FullPath)
-		hourMinuteEntries, _, listHourMinuteErr := f.ListDirectoryEntries(context.Background(), util.NewFullPath(SystemLogDir, dayEntry.Name()), "", false, 24*60, "", "", "")
+		hourMinuteEntries, _, listHourMinuteErr := f.ListDirectoryEntries(context.Background(), util.NewFullPath(SystemLogDir, dayEntry.Name()), "", false, math.MaxInt32, "", "", "")
 		if listHourMinuteErr != nil {
 			return lastTsNs, fmt.Errorf("fail to list log %s by day: %v", dayEntry.Name(), listHourMinuteErr)
 		}
 		for _, hourMinuteEntry := range hourMinuteEntries {
 			// println("checking hh-mm", hourMinuteEntry.FullPath)
 			if dayEntry.Name() == startDate {
-				if strings.Compare(hourMinuteEntry.Name(), startHourMinute) < 0 {
+				hourMinute := util.FileNameBase(hourMinuteEntry.Name())
+				if strings.Compare(hourMinute, startHourMinute) < 0 {
 					continue
 				}
 			}

weed/filer/meta_aggregator.go

@@ -110,7 +110,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self pb.ServerAddress, p
 		}
 		dir := event.Directory
 		// println("received meta change", dir, "size", len(data))
-		ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, 0)
+		ma.MetaLogBuffer.AddToBuffer([]byte(dir), data, event.TsNs)
 		if maybeReplicateMetadataChange != nil {
 			maybeReplicateMetadataChange(event)
 		}

weed/filer/redis/universal_redis_store.go

@@ -120,6 +120,8 @@ func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, full
 		if err != nil {
 			return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
 		}
+		// not efficient, but need to remove if it is a directory
+		store.Client.Del(ctx, genDirectoryListKey(string(path)))
 	}
 
 	return nil

weed/filer/redis2/universal_redis_store.go

@@ -144,6 +144,8 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
 		if err != nil {
 			return fmt.Errorf("DeleteFolderChildren %s in parent dir: %v", fullpath, err)
 		}
+		// not efficient, but need to remove if it is a directory
+		store.Client.Del(ctx, genDirectoryListKey(string(path)))
 	}
 
 	return nil

weed/filesys/xattr.go

@@ -2,6 +2,7 @@ package filesys
 
 import (
 	"context"
+	"strings"
 
 	"github.com/seaweedfs/fuse"
 
@@ -10,6 +11,10 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
+const (
+	XATTR_PREFIX = "xattr-" // same as filer
+)
+
 func getxattr(entry *filer_pb.Entry, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
 
 	if entry == nil {
@@ -18,7 +23,7 @@ func getxattr(entry *filer_pb.Entry, req *fuse.GetxattrRequest, resp *fuse.Getxa
 	if entry.Extended == nil {
 		return fuse.ErrNoXattr
 	}
-	data, found := entry.Extended[req.Name]
+	data, found := entry.Extended[XATTR_PREFIX+req.Name]
 	if !found {
 		return fuse.ErrNoXattr
 	}
@@ -47,7 +52,7 @@ func setxattr(entry *filer_pb.Entry, req *fuse.SetxattrRequest) error {
 	if entry.Extended == nil {
 		entry.Extended = make(map[string][]byte)
 	}
-	data, _ := entry.Extended[req.Name]
+	data, _ := entry.Extended[XATTR_PREFIX+req.Name]
 
 	newData := make([]byte, int(req.Position)+len(req.Xattr))
 
@@ -55,7 +60,7 @@ func setxattr(entry *filer_pb.Entry, req *fuse.SetxattrRequest) error {
 
 	copy(newData[int(req.Position):], req.Xattr)
 
-	entry.Extended[req.Name] = newData
+	entry.Extended[XATTR_PREFIX+req.Name] = newData
 
 	return nil
 
@@ -71,13 +76,13 @@ func removexattr(entry *filer_pb.Entry, req *fuse.RemovexattrRequest) error {
 		return fuse.ErrNoXattr
 	}
 
-	_, found := entry.Extended[req.Name]
+	_, found := entry.Extended[XATTR_PREFIX+req.Name]
 
 	if !found {
 		return fuse.ErrNoXattr
 	}
 
-	delete(entry.Extended, req.Name)
+	delete(entry.Extended, XATTR_PREFIX+req.Name)
 
 	return nil
 
@@ -90,7 +95,9 @@ func listxattr(entry *filer_pb.Entry, req *fuse.ListxattrRequest, resp *fuse.Lis
 	}
 
 	for k := range entry.Extended {
-		resp.Append(k)
+		if strings.HasPrefix(k, XATTR_PREFIX) {
+			resp.Append(k[len(XATTR_PREFIX):])
+		}
 	}
 
 	size := req.Size

weed/ftpd/ftp_server.go

@@ -3,8 +3,8 @@ package ftpd
 import (
 	"crypto/tls"
 	"errors"
-	"net"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"net"
 
 	ftpserver "github.com/fclairamb/ftpserverlib"
 	"google.golang.org/grpc"

weed/iamapi/iamapi_server.go

@@ -33,10 +33,10 @@ type IamS3ApiConfigure struct {
 }
 
 type IamServerOption struct {
-	Masters          []pb.ServerAddress
-	Filer            pb.ServerAddress
-	Port             int
-	GrpcDialOption   grpc.DialOption
+	Masters        []pb.ServerAddress
+	Filer          pb.ServerAddress
+	Port           int
+	GrpcDialOption grpc.DialOption
 }
 
 type IamApiServer struct {
@@ -49,7 +49,7 @@ var s3ApiConfigure IamS3ApiConfig
 func NewIamApiServer(router *mux.Router, option *IamServerOption) (iamApiServer *IamApiServer, err error) {
 	s3ApiConfigure = IamS3ApiConfigure{
 		option:       option,
-		masterClient: wdclient.NewMasterClient(option.GrpcDialOption, pb.AdminShellClient, "", "",option.Masters),
+		masterClient: wdclient.NewMasterClient(option.GrpcDialOption, pb.AdminShellClient, "", "", option.Masters),
 	}
 	s3Option := s3api.S3ApiServerOption{Filer: option.Filer}
 	iamApiServer = &IamApiServer{

weed/messaging/broker/broker_grpc_server_subscribe.go

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/log_buffer"
 	"io"
 	"strings"
@@ -141,7 +142,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTime time.Time, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (err error) {
 	startTime = startTime.UTC()
 	startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
-	startHourMinute := fmt.Sprintf("%02d-%02d.segment", startTime.Hour(), startTime.Minute())
+	startHourMinute := fmt.Sprintf("%02d-%02d", startTime.Hour(), startTime.Minute())
 
 	sizeBuf := make([]byte, 4)
 	startTsNs := startTime.UnixNano()
@@ -153,7 +154,8 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
 		dayDir := fmt.Sprintf("%s/%s", topicDir, dayEntry.Name)
 		return filer_pb.List(broker, dayDir, "", func(hourMinuteEntry *filer_pb.Entry, isLast bool) error {
 			if dayEntry.Name == startDate {
-				if strings.Compare(hourMinuteEntry.Name, startHourMinute) < 0 {
+				hourMinute := util.FileNameBase(hourMinuteEntry.Name)
+				if strings.Compare(hourMinute, startHourMinute) < 0 {
 					return nil
 				}
 			}

weed/operation/lookup.go

@@ -96,7 +96,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 				locations = append(locations, Location{
 					Url:       loc.Url,
 					PublicUrl: loc.PublicUrl,
-					GrpcPort: int(loc.GrpcPort),
+					GrpcPort:  int(loc.GrpcPort),
 				})
 			}
 			if vidLocations.Error != "" {

weed/pb/grpc_client_server.go

@@ -244,4 +244,4 @@ func WithOneOfGrpcFilerClients(filerAddresses []ServerAddress, grpcDialOption gr
 	}
 
 	return err
-}
+}

weed/pb/server_address.go

@@ -13,7 +13,7 @@ type ServerAddress string
 type ServerAddresses string
 
 func NewServerAddress(host string, port int, grpcPort int) ServerAddress {
-	if grpcPort == port+10000 {
+	if grpcPort == 0 || grpcPort == port+10000 {
 		return ServerAddress(util.JoinHostPort(host, port))
 	}
 	return ServerAddress(util.JoinHostPort(host, port) + "." + strconv.Itoa(grpcPort))

weed/remote_storage/s3/s3_storage_client.go

@@ -269,7 +269,7 @@ func (s *s3RemoteStorageClient) CreateBucket(name string) (err error) {
 
 func (s *s3RemoteStorageClient) DeleteBucket(name string) (err error) {
 	_, err = s.conn.DeleteBucket(&s3.DeleteBucketInput{
-		Bucket:                     aws.String(name),
+		Bucket: aws.String(name),
 	})
 	if err != nil {
 		return fmt.Errorf("delete bucket %s: %v", name, err)

weed/s3api/s3api_bucket_handlers.go

@@ -27,6 +27,8 @@ type ListAllMyBucketsResult struct {
 
 func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Request) {
 
+	glog.V(3).Infof("ListBucketsHandler")
+
 	var identity *Identity
 	var s3Err s3err.ErrorCode
 	if s3a.iam.isEnabled() {
@@ -75,6 +77,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("PutBucketHandler %s", bucket)
 
 	// avoid duplicated buckets
 	errCode := s3err.ErrNone
@@ -128,6 +131,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("DeleteBucketHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, err, r)
@@ -162,6 +166,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("HeadBucketHandler %s", bucket)
 
 	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, err, r)

weed/s3api/s3api_object_copy_handlers.go

@@ -27,6 +27,8 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 
 	srcBucket, srcObject := pathToBucketAndObject(cpSrcPath)
 
+	glog.V(3).Infof("CopyObjectHandler %s %s => %s %s", srcBucket, srcObject, dstBucket, dstObject)
+
 	if (srcBucket == dstBucket && srcObject == dstObject || cpSrcPath == "") && isReplace(r) {
 		fullPath := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, dstBucket, dstObject))
 		dir, name := fullPath.DirAndName()
@@ -139,6 +141,8 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 
+	glog.V(3).Infof("CopyObjectPartHandler %s %s => %s part %d", srcBucket, srcObject, dstBucket, partID)
+
 	// check partID with maximum part ID for multipart objects
 	if partID > globalMaxPartID {
 		s3err.WriteErrorResponse(w, s3err.ErrInvalidMaxParts, r)

weed/s3api/s3api_object_handlers.go

@@ -41,6 +41,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 	// http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("PutObjectHandler %s %s", bucket, object)
 
 	_, err := validateContentMd5(r.Header)
 	if err != nil {
@@ -118,6 +119,7 @@ func urlPathEscape(object string) string {
 func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("GetObjectHandler %s %s", bucket, object)
 
 	if strings.HasSuffix(r.URL.Path, "/") {
 		s3err.WriteErrorResponse(w, s3err.ErrNotImplemented, r)
@@ -134,6 +136,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object)
 
 	destUrl := fmt.Sprintf("http://%s%s/%s%s",
 		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
@@ -145,6 +148,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("DeleteObjectHandler %s %s", bucket, object)
 
 	destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
 		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
@@ -192,6 +196,7 @@ type DeleteObjectsResponse struct {
 func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("DeleteMultipleObjectsHandler %s", bucket)
 
 	deleteXMLBytes, err := ioutil.ReadAll(r.Body)
 	if err != nil {
@@ -291,7 +296,7 @@ var passThroughHeaders = []string{
 
 func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResponse *http.Response, w http.ResponseWriter)) {
 
-	glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
+	glog.V(3).Infof("s3 proxying %s to %s", r.Method, destUrl)
 
 	proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
 

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/s3api/policy"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
 	"github.com/dustin/go-humanize"
@@ -24,6 +25,8 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 
 	bucket := mux.Vars(r)["bucket"]
 
+	glog.V(3).Infof("PostPolicyBucketHandler %s", bucket)
+
 	reader, err := r.MultipartReader()
 	if err != nil {
 		s3err.WriteErrorResponse(w, s3err.ErrMalformedPOSTRequest, r)

weed/s3api/s3api_object_skip_handlers.go

@@ -0,0 +1,37 @@
+package s3api
+
+import (
+	"net/http"
+)
+
+// PutObjectAclHandler Put object ACL
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectAcl.html
+func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Request) {
+
+	w.WriteHeader(http.StatusNoContent)
+
+}
+
+// PutObjectRetentionHandler Put object Retention
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectRetention.html
+func (s3a *S3ApiServer) PutObjectRetentionHandler(w http.ResponseWriter, r *http.Request) {
+
+	w.WriteHeader(http.StatusNoContent)
+
+}
+
+// PutObjectLegalHoldHandler Put object Legal Hold
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectLegalHold.html
+func (s3a *S3ApiServer) PutObjectLegalHoldHandler(w http.ResponseWriter, r *http.Request) {
+
+	w.WriteHeader(http.StatusNoContent)
+
+}
+
+// PutObjectLockConfigurationHandler Put object Lock configuration
+// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectLockConfiguration.html
+func (s3a *S3ApiServer) PutObjectLockConfigurationHandler(w http.ResponseWriter, r *http.Request) {
+
+	w.WriteHeader(http.StatusNoContent)
+
+}

weed/s3api/s3api_object_tagging_handlers.go

@@ -17,6 +17,7 @@ import (
 func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("GetObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
 	dir, name := target.DirAndName()
@@ -42,6 +43,7 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
 func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("PutObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
 	dir, name := target.DirAndName()
@@ -97,6 +99,7 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
 func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := getBucketAndObject(r)
+	glog.V(3).Infof("DeleteObjectTaggingHandler %s %s", bucket, object)
 
 	target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
 	dir, name := target.DirAndName()

weed/s3api/s3api_objects_list_handlers.go

@@ -40,6 +40,7 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 
 	// collect parameters
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("ListObjectsV2Handler %s", bucket)
 
 	originalPrefix, continuationToken, startAfter, delimiter, _, maxKeys := getListObjectsV2Args(r.URL.Query())
 
@@ -95,6 +96,7 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
 
 	// collect parameters
 	bucket, _ := getBucketAndObject(r)
+	glog.V(3).Infof("ListObjectsV1Handler %s", bucket)
 
 	originalPrefix, marker, delimiter, maxKeys := getListObjectsV1Args(r.URL.Query())
 

weed/s3api/s3api_server.go

@@ -90,6 +90,15 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 		// DeleteObjectTagging
 		bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteObjectTaggingHandler, ACTION_TAGGING), "DELETE")).Queries("tagging", "")
 
+		// PutObjectACL
+		bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectAclHandler, ACTION_WRITE), "PUT")).Queries("acl", "")
+		// PutObjectRetention
+		bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectRetentionHandler, ACTION_WRITE), "PUT")).Queries("retention", "")
+		// PutObjectLegalHold
+		bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectLegalHoldHandler, ACTION_WRITE), "PUT")).Queries("legal-hold", "")
+		// PutObjectLockConfiguration
+		bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectLockConfigurationHandler, ACTION_WRITE), "PUT")).Queries("object-lock", "")
+
 		// CopyObject
 		bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectHandler, ACTION_WRITE), "COPY"))
 		// PutObject

weed/server/filer_grpc_server_remote.go

@@ -116,7 +116,7 @@ func (fs *FilerServer) DownloadToLocal(ctx context.Context, req *filer_pb.Downlo
 				replicas = append(replicas, &volume_server_pb.FetchAndWriteNeedleRequest_Replica{
 					Url:       r.Url,
 					PublicUrl: r.PublicUrl,
-					GrpcPort: int32(r.GrpcPort),
+					GrpcPort:  int32(r.GrpcPort),
 				})
 			}
 

weed/server/filer_grpc_server_sub_meta.go

@@ -49,11 +49,6 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 		if processedTsNs != 0 {
 			lastReadTime = time.Unix(0, processedTsNs)
-		} else {
-			if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
-				time.Sleep(1127 * time.Millisecond)
-				continue
-			}
 		}
 
 		glog.V(4).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
@@ -66,6 +61,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		}, eachLogEntryFn)
 		if readInMemoryLogErr != nil {
 			if readInMemoryLogErr == log_buffer.ResumeFromDiskError {
+				time.Sleep(1127 * time.Millisecond)
 				continue
 			}
 			glog.Errorf("processed to %v: %v", lastReadTime, readInMemoryLogErr)

weed/server/filer_server_handlers_read.go

@@ -64,7 +64,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 
 	// set etag
 	etag := filer.ETagEntry(entry)
-	if ifm := r.Header.Get("If-Match"); ifm != "" && (ifm != "\""+etag+"\"" && ifm != etag){
+	if ifm := r.Header.Get("If-Match"); ifm != "" && (ifm != "\""+etag+"\"" && ifm != etag) {
 		w.WriteHeader(http.StatusPreconditionFailed)
 		return
 	}
@@ -97,7 +97,10 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 
 	// print out the header from extended properties
 	for k, v := range entry.Extended {
-		w.Header().Set(k, string(v))
+		if !strings.HasPrefix(k, "xattr-") {
+			// "xattr-" prefix is set in filesys.XATTR_PREFIX
+			w.Header().Set(k, string(v))
+		}
 	}
 
 	//Seaweed custom header are not visible to Vue or javascript
@@ -174,6 +177,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 				Directory: dir,
 				Name:      name,
 			}); err != nil {
+				glog.Errorf("DownloadToLocal %s: %v", entry.FullPath, err)
 				return fmt.Errorf("cache %s: %v", entry.FullPath, err)
 			} else {
 				chunks = resp.Entry.Chunks

weed/server/master_grpc_server.go

@@ -24,7 +24,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 	defer func() {
 		if dn != nil {
 			dn.Counter--
-			if dn.Counter >  0 {
+			if dn.Counter > 0 {
 				glog.V(0).Infof("disconnect phantom volume server %s:%d remaining %d", dn.Ip, dn.Port, dn.Counter)
 				return
 			}

weed/server/master_grpc_server_volume.go

@@ -155,19 +155,19 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 				replicas = append(replicas, &master_pb.Location{
 					Url:       r.Url(),
 					PublicUrl: r.PublicUrl,
-					GrpcPort: uint32(r.GrpcPort),
+					GrpcPort:  uint32(r.GrpcPort),
 				})
 			}
 			return &master_pb.AssignResponse{
-				Fid:       fid,
+				Fid: fid,
 				Location: &master_pb.Location{
 					Url:       dn.Url(),
 					PublicUrl: dn.PublicUrl,
 					GrpcPort:  uint32(dn.GrpcPort),
 				},
-				Count:     count,
-				Auth:      string(security.GenJwt(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
-				Replicas:  replicas,
+				Count:    count,
+				Auth:     string(security.GenJwt(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
+				Replicas: replicas,
 			}, nil
 		}
 		//glog.V(4).Infoln("waiting for volume growing...")

weed/server/volume_grpc_remote.go

@@ -53,7 +53,7 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 			}
 		}
 	}()
-	if len(req.Replicas)>0{
+	if len(req.Replicas) > 0 {
 		fileId := needle.NewFileId(v.Id, req.NeedleId, req.Cookie)
 		for _, replica := range req.Replicas {
 			wg.Add(1)

weed/server/webdav_server.go

@@ -27,18 +27,18 @@ import (
 )
 
 type WebDavOption struct {
-	Filer            pb.ServerAddress
-	DomainName       string
-	BucketsPath      string
-	GrpcDialOption   grpc.DialOption
-	Collection       string
-	Replication      string
-	DiskType         string
-	Uid              uint32
-	Gid              uint32
-	Cipher           bool
-	CacheDir         string
-	CacheSizeMB      int64
+	Filer          pb.ServerAddress
+	DomainName     string
+	BucketsPath    string
+	GrpcDialOption grpc.DialOption
+	Collection     string
+	Replication    string
+	DiskType       string
+	Uid            uint32
+	Gid            uint32
+	Cipher         bool
+	CacheDir       string
+	CacheSizeMB    int64
 }
 
 type WebDavServer struct {

weed/shell/command_collection_delete.go

@@ -29,10 +29,6 @@ func (c *commandCollectionDelete) Help() string {
 
 func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	colDeleteCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	collectionName := colDeleteCommand.String("collection", "", "collection to delete. Use '_default_' for the empty-named collection.")
 	applyBalancing := colDeleteCommand.Bool("force", false, "apply the collection")
@@ -40,6 +36,10 @@ func (c *commandCollectionDelete) Do(args []string, commandEnv *CommandEnv, writ
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	if *collectionName == "" {
 		return fmt.Errorf("empty collection name is not allowed")
 	}

weed/shell/command_ec_balance.go

@@ -100,10 +100,6 @@ func (c *commandEcBalance) Help() string {
 
 func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection")
 	dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
@@ -112,6 +108,10 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	// collect all ec nodes
 	allEcNodes, totalFreeEcSlots, err := collectEcNodes(commandEnv, *dc)
 	if err != nil {

weed/shell/command_ec_decode.go

@@ -37,11 +37,6 @@ func (c *commandEcDecode) Help() string {
 }
 
 func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
-
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
 	collection := encodeCommand.String("collection", "", "the collection name")
@@ -49,6 +44,10 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	vid := needle.VolumeId(*volumeId)
 
 	// collect topology information

weed/shell/command_ec_encode.go

@@ -55,10 +55,6 @@ func (c *commandEcEncode) Help() string {
 
 func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := encodeCommand.Int("volumeId", 0, "the volume id")
 	collection := encodeCommand.String("collection", "", "the collection name")
@@ -69,6 +65,10 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	vid := needle.VolumeId(*volumeId)
 
 	// volumeId is provided

weed/shell/command_ec_rebuild.go

@@ -57,10 +57,6 @@ func (c *commandEcRebuild) Help() string {
 
 func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	fixCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	collection := fixCommand.String("collection", "EACH_COLLECTION", "collection name, or \"EACH_COLLECTION\" for each collection")
 	applyChanges := fixCommand.Bool("force", false, "apply the changes")
@@ -68,6 +64,10 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	// collect all ec nodes
 	allEcNodes, _, err := collectEcNodes(commandEnv, "")
 	if err != nil {

weed/shell/command_remote_cache.go

@@ -79,7 +79,7 @@ func (c *commandRemoteCache) Do(args []string, commandEnv *CommandEnv, writer io
 	return nil
 }
 
-func (c *commandRemoteCache) doCacheOneDirectory(commandEnv *CommandEnv, writer io.Writer, dir string, fileFiler *FileFilter, concurrency int) (error) {
+func (c *commandRemoteCache) doCacheOneDirectory(commandEnv *CommandEnv, writer io.Writer, dir string, fileFiler *FileFilter, concurrency int) error {
 	mappings, localMountedDir, remoteStorageMountedLocation, remoteStorageConf, detectErr := detectMountInfo(commandEnv, writer, dir)
 	if detectErr != nil {
 		jsonPrintln(writer, mappings)

weed/shell/command_remote_unmount.go

@@ -118,4 +118,3 @@ func (c *commandRemoteUnmount) purgeMountedData(commandEnv *CommandEnv, dir stri
 
 	return nil
 }
-

weed/shell/command_volume_balance.go

@@ -63,10 +63,6 @@ func (c *commandVolumeBalance) Help() string {
 
 func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	balanceCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	collection := balanceCommand.String("collection", "EACH_COLLECTION", "collection name, or use \"ALL_COLLECTIONS\" across collections, \"EACH_COLLECTION\" for each collection")
 	dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter")
@@ -75,6 +71,10 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	// collect topology information
 	topologyInfo, volumeSizeLimitMb, err := collectTopologyInfo(commandEnv)
 	if err != nil {

weed/shell/command_volume_check_disk.go

@@ -42,10 +42,6 @@ func (c *commandVolumeCheckDisk) Help() string {
 
 func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	slowMode := fsckCommand.Bool("slow", false, "slow mode checks all replicas even file counts are the same")
 	verbose := fsckCommand.Bool("v", false, "verbose mode")
@@ -55,6 +51,10 @@ func (c *commandVolumeCheckDisk) Do(args []string, commandEnv *CommandEnv, write
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	c.env = commandEnv
 
 	// collect topology information

weed/shell/command_volume_configure_replication.go

@@ -36,10 +36,6 @@ func (c *commandVolumeConfigureReplication) Help() string {
 
 func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	configureReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
 	replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
@@ -47,6 +43,10 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	if *replicationString == "" {
 		return fmt.Errorf("empty replication value")
 	}

weed/shell/command_volume_copy.go

@@ -33,10 +33,6 @@ func (c *commandVolumeCopy) Help() string {
 
 func (c *commandVolumeCopy) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volCopyCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volCopyCommand.Int("volumeId", 0, "the volume id")
 	sourceNodeStr := volCopyCommand.String("source", "", "the source volume server <host>:<port>")
@@ -45,6 +41,10 @@ func (c *commandVolumeCopy) Do(args []string, commandEnv *CommandEnv, writer io.
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer, targetVolumeServer := pb.ServerAddress(*sourceNodeStr), pb.ServerAddress(*targetNodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_delete.go

@@ -31,10 +31,6 @@ func (c *commandVolumeDelete) Help() string {
 
 func (c *commandVolumeDelete) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volDeleteCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volDeleteCommand.Int("volumeId", 0, "the volume id")
 	nodeStr := volDeleteCommand.String("node", "", "the volume server <host>:<port>")
@@ -42,6 +38,10 @@ func (c *commandVolumeDelete) Do(args []string, commandEnv *CommandEnv, writer i
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer := pb.ServerAddress(*nodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_delete_empty.go

@@ -33,10 +33,6 @@ func (c *commandVolumeDeleteEmpty) Help() string {
 
 func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volDeleteCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	quietPeriod := volDeleteCommand.Duration("quietFor", 24*time.Hour, "select empty volumes with no recent writes, avoid newly created ones")
 	applyBalancing := volDeleteCommand.Bool("force", false, "apply to delete empty volumes")
@@ -44,6 +40,10 @@ func (c *commandVolumeDeleteEmpty) Do(args []string, commandEnv *CommandEnv, wri
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	// collect topology information
 	topologyInfo, _, err := collectTopologyInfo(commandEnv)
 	if err != nil {

weed/shell/command_volume_fix_replication.go

@@ -52,10 +52,6 @@ func (c *commandVolumeFixReplication) Help() string {
 
 func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	skipChange := volFixReplicationCommand.Bool("n", false, "skip the changes")
@@ -64,6 +60,10 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	takeAction := !*skipChange
 
 	// collect topology information

weed/shell/command_volume_fsck.go

@@ -58,10 +58,6 @@ func (c *commandVolumeFsck) Help() string {
 
 func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	verbose := fsckCommand.Bool("v", false, "verbose mode")
 	findMissingChunksInFiler := fsckCommand.Bool("findMissingChunksInFiler", false, "see \"help volume.fsck\"")
@@ -71,6 +67,10 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	c.env = commandEnv
 
 	// create a temp folder

weed/shell/command_volume_mark.go

@@ -29,10 +29,6 @@ func (c *commandVolumeMark) Help() string {
 
 func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volMarkCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volMarkCommand.Int("volumeId", 0, "the volume id")
 	nodeStr := volMarkCommand.String("node", "", "the volume server <host>:<port>")
@@ -48,6 +44,10 @@ func (c *commandVolumeMark) Do(args []string, commandEnv *CommandEnv, writer io.
 		markWritable = true
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer := pb.ServerAddress(*nodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_mount.go

@@ -35,10 +35,6 @@ func (c *commandVolumeMount) Help() string {
 
 func (c *commandVolumeMount) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volMountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volMountCommand.Int("volumeId", 0, "the volume id")
 	nodeStr := volMountCommand.String("node", "", "the volume server <host>:<port>")
@@ -46,6 +42,10 @@ func (c *commandVolumeMount) Do(args []string, commandEnv *CommandEnv, writer io
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer := pb.ServerAddress(*nodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_move.go

@@ -50,10 +50,6 @@ func (c *commandVolumeMove) Help() string {
 
 func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volMoveCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volMoveCommand.Int("volumeId", 0, "the volume id")
 	sourceNodeStr := volMoveCommand.String("source", "", "the source volume server <host>:<port>")
@@ -63,6 +59,10 @@ func (c *commandVolumeMove) Do(args []string, commandEnv *CommandEnv, writer io.
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer, targetVolumeServer := pb.ServerAddress(*sourceNodeStr), pb.ServerAddress(*targetNodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_server_evacuate.go

@@ -44,10 +44,6 @@ func (c *commandVolumeServerEvacuate) Help() string {
 
 func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	vsEvacuateCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeServer := vsEvacuateCommand.String("node", "", "<host>:<port> of the volume server")
 	skipNonMoveable := vsEvacuateCommand.Bool("skipNonMoveable", false, "skip volumes that can not be moved")
@@ -57,6 +53,10 @@ func (c *commandVolumeServerEvacuate) Do(args []string, commandEnv *CommandEnv,
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	if *volumeServer == "" {
 		return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
 	}

weed/shell/command_volume_server_leave.go

@@ -37,16 +37,16 @@ func (c *commandVolumeServerLeave) Help() string {
 
 func (c *commandVolumeServerLeave) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	vsLeaveCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeServer := vsLeaveCommand.String("node", "", "<host>:<port> of the volume server")
 	if err = vsLeaveCommand.Parse(args); err != nil {
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	if *volumeServer == "" {
 		return fmt.Errorf("need to specify volume server by -node=<host>:<port>")
 	}

weed/shell/command_volume_tier_download.go

@@ -43,10 +43,6 @@ func (c *commandVolumeTierDownload) Help() string {
 
 func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := tierCommand.Int("volumeId", 0, "the volume id")
 	collection := tierCommand.String("collection", "", "the collection name")
@@ -54,6 +50,10 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	vid := needle.VolumeId(*volumeId)
 
 	// collect topology information

weed/shell/command_volume_tier_move.go

@@ -46,10 +46,6 @@ func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer
 	c.activeServers = make(map[pb.ServerAddress]struct{})
 	c.activeServersCond = sync.NewCond(new(sync.Mutex))
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	collectionPattern := tierCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	fullPercentage := tierCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size")
@@ -61,6 +57,10 @@ func (c *commandVolumeTierMove) Do(args []string, commandEnv *CommandEnv, writer
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	fromDiskType := types.ToDiskType(*source)
 	toDiskType := types.ToDiskType(*target)
 

weed/shell/command_volume_tier_upload.go

@@ -57,10 +57,6 @@ func (c *commandVolumeTierUpload) Help() string {
 
 func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := tierCommand.Int("volumeId", 0, "the volume id")
 	collection := tierCommand.String("collection", "", "the collection name")
@@ -72,6 +68,10 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	vid := needle.VolumeId(*volumeId)
 
 	// volumeId is provided

weed/shell/command_volume_unmount.go

@@ -35,10 +35,6 @@ func (c *commandVolumeUnmount) Help() string {
 
 func (c *commandVolumeUnmount) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volUnmountCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeIdInt := volUnmountCommand.Int("volumeId", 0, "the volume id")
 	nodeStr := volUnmountCommand.String("node", "", "the volume server <host>:<port>")
@@ -46,6 +42,10 @@ func (c *commandVolumeUnmount) Do(args []string, commandEnv *CommandEnv, writer
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	sourceVolumeServer := pb.ServerAddress(*nodeStr)
 
 	volumeId := needle.VolumeId(*volumeIdInt)

weed/shell/command_volume_vacuum.go

@@ -29,16 +29,16 @@ func (c *commandVacuum) Help() string {
 
 func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
-	if err = commandEnv.confirmIsLocked(); err != nil {
-		return
-	}
-
 	volumeVacuumCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	garbageThreshold := volumeVacuumCommand.Float64("garbageThreshold", 0.3, "vacuum when garbage is more than this limit")
 	if err = volumeVacuumCommand.Parse(args); err != nil {
 		return nil
 	}
 
+	if err = commandEnv.confirmIsLocked(); err != nil {
+		return
+	}
+
 	err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
 		_, err = client.VacuumVolume(context.Background(), &master_pb.VacuumVolumeRequest{
 			GarbageThreshold: float32(*garbageThreshold),

weed/shell/commands.go

@@ -49,7 +49,7 @@ func NewCommandEnv(options ShellOptions) *CommandEnv {
 		MasterClient: wdclient.NewMasterClient(options.GrpcDialOption, pb.AdminShellClient, "", "", pb.ServerAddresses(*options.Masters).ToAddresses()),
 		option:       options,
 	}
-	ce.locker = exclusive_locks.NewExclusiveLocker(ce.MasterClient)
+	ce.locker = exclusive_locks.NewExclusiveLocker(ce.MasterClient, "admin")
 	return ce
 }
 

weed/topology/topology_test.go

@@ -31,7 +31,7 @@ func TestHandlingVolumeServerHeartbeat(t *testing.T) {
 	maxVolumeCounts := make(map[string]uint32)
 	maxVolumeCounts[""] = 25
 	maxVolumeCounts["ssd"] = 12
-	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0,"127.0.0.1", maxVolumeCounts)
+	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", maxVolumeCounts)
 
 	{
 		volumeCount := 7
@@ -177,7 +177,7 @@ func TestAddRemoveVolume(t *testing.T) {
 	maxVolumeCounts := make(map[string]uint32)
 	maxVolumeCounts[""] = 25
 	maxVolumeCounts["ssd"] = 12
-	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0,"127.0.0.1", maxVolumeCounts)
+	dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", maxVolumeCounts)
 
 	v := storage.VolumeInfo{
 		Id:               needle.VolumeId(1),

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.68)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.70)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )

weed/util/file_util.go

@@ -87,3 +87,11 @@ func ResolvePath(path string) string {
 
 	return path
 }
+
+func FileNameBase(filename string) string {
+	lastDotIndex := strings.LastIndex(filename, ".")
+	if lastDotIndex < 0 {
+		return filename
+	}
+	return filename[:lastDotIndex]
+}

weed/util/log_buffer/log_buffer.go

@@ -56,7 +56,7 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn func(startTi
 	return lb
 }
 
-func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64) {
+func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, processingTsNs int64) {
 
 	m.Lock()
 	defer func() {
@@ -68,20 +68,20 @@ func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64) {
 
 	// need to put the timestamp inside the lock
 	var ts time.Time
-	if eventTsNs == 0 {
+	if processingTsNs == 0 {
 		ts = time.Now()
-		eventTsNs = ts.UnixNano()
+		processingTsNs = ts.UnixNano()
 	} else {
-		ts = time.Unix(0, eventTsNs)
+		ts = time.Unix(0, processingTsNs)
 	}
-	if m.lastTsNs >= eventTsNs {
+	if m.lastTsNs >= processingTsNs {
 		// this is unlikely to happen, but just in case
-		eventTsNs = m.lastTsNs + 1
-		ts = time.Unix(0, eventTsNs)
+		processingTsNs = m.lastTsNs + 1
+		ts = time.Unix(0, processingTsNs)
 	}
-	m.lastTsNs = eventTsNs
+	m.lastTsNs = processingTsNs
 	logEntry := &filer_pb.LogEntry{
-		TsNs:             eventTsNs,
+		TsNs:             processingTsNs,
 		PartitionKeyHash: util.HashToInt32(partitionKey),
 		Data:             data,
 	}
@@ -189,7 +189,11 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
 	defer m.RUnlock()
 
 	if !m.lastFlushTime.IsZero() && m.lastFlushTime.After(lastReadTime) {
-		return nil, ResumeFromDiskError
+		if time.Now().Sub(m.lastFlushTime) < m.flushInterval * 2 {
+			diff := m.lastFlushTime.Sub(lastReadTime)
+			glog.V(4).Infof("lastFlush:%v lastRead:%v diff:%v", m.lastFlushTime, lastReadTime, diff)
+			return nil, ResumeFromDiskError
+		}
 	}
 
 	/*

weed/util/network.go

@@ -15,11 +15,11 @@ func DetectedHostAddress() string {
 		return ""
 	}
 
-	if v4Address := selectIpV4(netInterfaces, true); v4Address != ""{
+	if v4Address := selectIpV4(netInterfaces, true); v4Address != "" {
 		return v4Address
 	}
 
-	if v6Address := selectIpV4(netInterfaces, false); v6Address != ""{
+	if v6Address := selectIpV4(netInterfaces, false); v6Address != "" {
 		return v6Address
 	}
 
@@ -59,4 +59,4 @@ func JoinHostPort(host string, port int) string {
 		return host + ":" + portStr
 	}
 	return net.JoinHostPort(host, portStr)
-}
+}

weed/wdclient/exclusive_locks/exclusive_locker.go

@@ -14,7 +14,6 @@ const (
 	RenewInteval     = 4 * time.Second
 	SafeRenewInteval = 3 * time.Second
 	InitLockInteval  = 1 * time.Second
-	AdminLockName    = "admin"
 )
 
 type ExclusiveLocker struct {
@@ -22,13 +21,16 @@ type ExclusiveLocker struct {
 	lockTsNs     int64
 	isLocking    bool
 	masterClient *wdclient.MasterClient
+	lockName     string
 }
 
-func NewExclusiveLocker(masterClient *wdclient.MasterClient) *ExclusiveLocker {
+func NewExclusiveLocker(masterClient *wdclient.MasterClient, lockName string) *ExclusiveLocker {
 	return &ExclusiveLocker{
 		masterClient: masterClient,
+		lockName:     lockName,
 	}
 }
+
 func (l *ExclusiveLocker) IsLocking() bool {
 	return l.isLocking
 }
@@ -55,7 +57,7 @@ func (l *ExclusiveLocker) RequestLock(clientName string) {
 			resp, err := client.LeaseAdminToken(ctx, &master_pb.LeaseAdminTokenRequest{
 				PreviousToken:    atomic.LoadInt64(&l.token),
 				PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
-				LockName:         AdminLockName,
+				LockName:         l.lockName,
 				ClientName:       clientName,
 			})
 			if err == nil {
@@ -83,7 +85,7 @@ func (l *ExclusiveLocker) RequestLock(clientName string) {
 				resp, err := client.LeaseAdminToken(ctx2, &master_pb.LeaseAdminTokenRequest{
 					PreviousToken:    atomic.LoadInt64(&l.token),
 					PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
-					LockName:         AdminLockName,
+					LockName:         l.lockName,
 					ClientName:       clientName,
 				})
 				if err == nil {
@@ -114,7 +116,7 @@ func (l *ExclusiveLocker) ReleaseLock() {
 		client.ReleaseAdminToken(ctx, &master_pb.ReleaseAdminTokenRequest{
 			PreviousToken:    atomic.LoadInt64(&l.token),
 			PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
-			LockName:         AdminLockName,
+			LockName:         l.lockName,
 		})
 		return nil
 	})