aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAleksey Kosov <rusyak777@list.ru>2025-05-22 19:46:49 +0300
committerGitHub <noreply@github.com>2025-05-22 09:46:49 -0700
commit165af32d6b738daa14b84fb6e3c8891f4b46135b (patch)
tree7e6128d2fda753abc031c6461ae9709b284afcdf
parent5182d46e22d2458b16f1f2fb0358f6b5f3e18b5d (diff)
downloadseaweedfs-165af32d6b738daa14b84fb6e3c8891f4b46135b.tar.xz
seaweedfs-165af32d6b738daa14b84fb6e3c8891f4b46135b.zip
added context to filer_client method calls (#6808)
Co-authored-by: akosov <a.kosov@kryptonite.ru>
-rw-r--r--unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go5
-rw-r--r--weed/command/filer_cat.go2
-rw-r--r--weed/command/filer_copy.go4
-rw-r--r--weed/command/filer_remote_gateway_buckets.go3
-rw-r--r--weed/command/filer_remote_sync_dir.go2
-rw-r--r--weed/command/mount_std.go2
-rw-r--r--weed/filer/read_write.go11
-rw-r--r--weed/mount/meta_cache/meta_cache_init.go2
-rw-r--r--weed/mount/weedfs_dir_lookup.go2
-rw-r--r--weed/mount/weedfs_dir_mkrm.go4
-rw-r--r--weed/mount/weedfs_file_mkrm.go4
-rw-r--r--weed/mount/weedfs_file_sync.go2
-rw-r--r--weed/mount/weedfs_link.go4
-rw-r--r--weed/mount/weedfs_symlink.go2
-rw-r--r--weed/mq/broker/broker_write.go5
-rw-r--r--weed/mq/logstore/log_to_parquet.go11
-rw-r--r--weed/mq/logstore/read_log_from_disk.go3
-rw-r--r--weed/mq/logstore/read_parquet_to_log.go3
-rw-r--r--weed/pb/filer_pb/filer_client.go56
-rw-r--r--weed/pb/filer_pb/filer_client_bfs.go2
-rw-r--r--weed/pb/filer_pb/filer_pb_helper.go12
-rw-r--r--weed/replication/sink/filersink/filer_sink.go8
-rw-r--r--weed/s3api/bucket_metadata.go5
-rw-r--r--weed/s3api/filer_util.go14
-rw-r--r--weed/s3api/filer_util_tags.go11
-rw-r--r--weed/s3api/s3api_object_handlers_list.go2
-rw-r--r--weed/s3api/s3api_object_handlers_multipart.go1
-rw-r--r--weed/server/webdav_server.go14
-rw-r--r--weed/sftpd/sftp_filer.go2
-rw-r--r--weed/sftpd/user/homemanager.go2
-rw-r--r--weed/shell/command_fs_cat.go3
-rw-r--r--weed/shell/command_fs_du.go3
-rw-r--r--weed/shell/command_fs_log.go5
-rw-r--r--weed/shell/command_fs_ls.go3
-rw-r--r--weed/shell/command_fs_merge_volumes.go2
-rw-r--r--weed/shell/command_fs_meta_cat.go3
-rw-r--r--weed/shell/command_fs_meta_change_volume_id.go3
-rw-r--r--weed/shell/command_fs_meta_load.go5
-rw-r--r--weed/shell/command_fs_mv.go2
-rw-r--r--weed/shell/command_fs_rm.go2
-rw-r--r--weed/shell/command_fs_tree.go3
-rw-r--r--weed/shell/command_remote_cache.go3
-rw-r--r--weed/shell/command_remote_configure.go2
-rw-r--r--weed/shell/command_remote_meta_sync.go2
-rw-r--r--weed/shell/command_remote_mount.go2
-rw-r--r--weed/shell/command_remote_unmount.go5
-rw-r--r--weed/shell/command_s3_bucket_create.go2
-rw-r--r--weed/shell/command_s3_bucket_delete.go2
-rw-r--r--weed/shell/command_s3_bucket_list.go2
-rw-r--r--weed/shell/command_s3_bucket_quota.go2
-rw-r--r--weed/shell/command_s3_bucket_quota_check.go3
-rw-r--r--weed/shell/command_s3_clean_uploads.go5
-rw-r--r--weed/shell/commands.go2
53 files changed, 143 insertions, 123 deletions
diff --git a/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go b/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go
index 2b63d5d59..0544e5f2b 100644
--- a/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go
+++ b/unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go
@@ -1,18 +1,19 @@
package main
import (
+ "context"
"flag"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
+ util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"strconv"
"strings"
"time"
- util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
)
var (
@@ -60,7 +61,7 @@ func startGenerateMetadata() {
for i := 0; i < *n; i++ {
name := fmt.Sprintf("file%d", i)
glog.V(0).Infof("write %s/%s", *dir, name)
- if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ if err := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: *dir,
Entry: &filer_pb.Entry{
Name: name,
diff --git a/weed/command/filer_cat.go b/weed/command/filer_cat.go
index ba3625b0d..2c0f84ddc 100644
--- a/weed/command/filer_cat.go
+++ b/weed/command/filer_cat.go
@@ -102,7 +102,7 @@ func runFilerCat(cmd *Command, args []string) bool {
Name: name,
Directory: dir,
}
- respLookupEntry, err := filer_pb.LookupEntry(client, request)
+ respLookupEntry, err := filer_pb.LookupEntry(context.Background(), client, request)
if err != nil {
return err
}
diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go
index 9b195e4e8..547e46c4f 100644
--- a/weed/command/filer_copy.go
+++ b/weed/command/filer_copy.go
@@ -394,7 +394,7 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
},
}
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
return fmt.Errorf("update fh: %v", err)
}
return nil
@@ -511,7 +511,7 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
},
}
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
return fmt.Errorf("update fh: %v", err)
}
return nil
diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go
index f6fe9a99c..aa3869171 100644
--- a/weed/command/filer_remote_gateway_buckets.go
+++ b/weed/command/filer_remote_gateway_buckets.go
@@ -1,6 +1,7 @@
package command
import (
+ "context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -408,7 +409,7 @@ func (option *RemoteGatewayOptions) collectRemoteStorageConf() (err error) {
option.remoteConfs = make(map[string]*remote_pb.RemoteConf)
var lastConfName string
- err = filer_pb.List(option, filer.DirectoryEtcRemote, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.List(context.Background(), option, filer.DirectoryEtcRemote, "", func(entry *filer_pb.Entry, isLast bool) error {
if !strings.HasSuffix(entry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) {
return nil
}
diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go
index 186523e45..c85ba03af 100644
--- a/weed/command/filer_remote_sync_dir.go
+++ b/weed/command/filer_remote_sync_dir.go
@@ -230,7 +230,7 @@ func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc
// 3. directory creation time
var lastOffsetTs time.Time
if timeAgo == 0 {
- mountedDirEntry, err := filer_pb.GetEntry(filerClient, util.FullPath(mountedDir))
+ mountedDirEntry, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(mountedDir))
if err != nil {
glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err)
return time.Now()
diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go
index f00daba8a..631091e80 100644
--- a/weed/command/mount_std.go
+++ b/weed/command/mount_std.go
@@ -255,7 +255,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
// create mount root
mountRootPath := util.FullPath(mountRoot)
mountRootParent, mountDir := mountRootPath.DirAndName()
- if err = filer_pb.Mkdir(seaweedFileSystem, mountRootParent, mountDir, nil); err != nil {
+ if err = filer_pb.Mkdir(context.Background(), seaweedFileSystem, mountRootParent, mountDir, nil); err != nil {
fmt.Printf("failed to create dir %s on filer %s: %v\n", mountRoot, filerAddresses, err)
return false
}
diff --git a/weed/filer/read_write.go b/weed/filer/read_write.go
index dc4f3d0d8..66f4c0bf8 100644
--- a/weed/filer/read_write.go
+++ b/weed/filer/read_write.go
@@ -2,6 +2,7 @@ package filer
import (
"bytes"
+ "context"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -14,7 +15,7 @@ func ReadEntry(masterClient *wdclient.MasterClient, filerClient filer_pb.Seaweed
Directory: dir,
Name: name,
}
- respLookupEntry, err := filer_pb.LookupEntry(filerClient, request)
+ respLookupEntry, err := filer_pb.LookupEntry(context.Background(), filerClient, request)
if err != nil {
return err
}
@@ -32,7 +33,7 @@ func ReadInsideFiler(filerClient filer_pb.SeaweedFilerClient, dir, name string)
Directory: dir,
Name: name,
}
- respLookupEntry, err := filer_pb.LookupEntry(filerClient, request)
+ respLookupEntry, err := filer_pb.LookupEntry(context.Background(), filerClient, request)
if err != nil {
return
}
@@ -42,13 +43,13 @@ func ReadInsideFiler(filerClient filer_pb.SeaweedFilerClient, dir, name string)
func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, content []byte) error {
- resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir,
Name: name,
})
if err == filer_pb.ErrNotFound {
- err = filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ err = filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: &filer_pb.Entry{
Name: name,
@@ -68,7 +69,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte
entry.Content = content
entry.Attributes.Mtime = time.Now().Unix()
entry.Attributes.FileSize = uint64(len(content))
- err = filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
+ err = filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
Directory: dir,
Entry: entry,
})
diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go
index 1cab499e0..068daa3f9 100644
--- a/weed/mount/meta_cache/meta_cache_init.go
+++ b/weed/mount/meta_cache/meta_cache_init.go
@@ -44,7 +44,7 @@ func doEnsureVisited(mc *MetaCache, client filer_pb.FilerClient, path util.FullP
glog.V(4).Infof("ReadDirAllEntries %s ...", path)
err := util.Retry("ReadDirAllEntries", func() error {
- return filer_pb.ReadDirAllEntries(client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
+ return filer_pb.ReadDirAllEntries(context.Background(), client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
entry := filer.FromPbEntry(string(path), pbEntry)
if IsHiddenSystemEntry(string(path), entry.Name()) {
return nil
diff --git a/weed/mount/weedfs_dir_lookup.go b/weed/mount/weedfs_dir_lookup.go
index 7fc10ef28..7af989b80 100644
--- a/weed/mount/weedfs_dir_lookup.go
+++ b/weed/mount/weedfs_dir_lookup.go
@@ -41,7 +41,7 @@ func (wfs *WFS) Lookup(cancel <-chan struct{}, header *fuse.InHeader, name strin
if localEntry == nil {
// glog.V(3).Infof("dir Lookup cache miss %s", fullFilePath)
- entry, err := filer_pb.GetEntry(wfs, fullFilePath)
+ entry, err := filer_pb.GetEntry(context.Background(), wfs, fullFilePath)
if err != nil {
glog.V(1).Infof("dir GetEntry %s: %v", fullFilePath, err)
return fuse.ENOENT
diff --git a/weed/mount/weedfs_dir_mkrm.go b/weed/mount/weedfs_dir_mkrm.go
index e69c9796e..503d6a076 100644
--- a/weed/mount/weedfs_dir_mkrm.go
+++ b/weed/mount/weedfs_dir_mkrm.go
@@ -63,7 +63,7 @@ func (wfs *WFS) Mkdir(cancel <-chan struct{}, in *fuse.MkdirIn, name string, out
}
glog.V(1).Infof("mkdir: %v", request)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
glog.V(0).Infof("mkdir %s: %v", entryFullPath, err)
return err
}
@@ -107,7 +107,7 @@ func (wfs *WFS) Rmdir(cancel <-chan struct{}, header *fuse.InHeader, name string
glog.V(3).Infof("remove directory: %v", entryFullPath)
ignoreRecursiveErr := true // ignore recursion error since the OS should manage it
- err := filer_pb.Remove(wfs, string(dirFullPath), name, true, false, ignoreRecursiveErr, false, []int32{wfs.signature})
+ err := filer_pb.Remove(context.Background(), wfs, string(dirFullPath), name, true, false, ignoreRecursiveErr, false, []int32{wfs.signature})
if err != nil {
glog.V(0).Infof("remove %s: %v", entryFullPath, err)
if strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go
index e6139e88e..e48d5af40 100644
--- a/weed/mount/weedfs_file_mkrm.go
+++ b/weed/mount/weedfs_file_mkrm.go
@@ -83,7 +83,7 @@ func (wfs *WFS) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string, out
}
glog.V(1).Infof("mknod: %v", request)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
glog.V(0).Infof("mknod %s: %v", entryFullPath, err)
return err
}
@@ -137,7 +137,7 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin
// first, ensure the filer store can correctly delete
glog.V(3).Infof("remove file: %v", entryFullPath)
isDeleteData := entry != nil && entry.HardLinkCounter <= 1
- err := filer_pb.Remove(wfs, string(dirFullPath), name, isDeleteData, false, false, false, []int32{wfs.signature})
+ err := filer_pb.Remove(context.Background(), wfs, string(dirFullPath), name, isDeleteData, false, false, false, []int32{wfs.signature})
if err != nil {
glog.V(0).Infof("remove %s: %v", entryFullPath, err)
return fuse.OK
diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go
index e13ab198d..541270ff5 100644
--- a/weed/mount/weedfs_file_sync.go
+++ b/weed/mount/weedfs_file_sync.go
@@ -159,7 +159,7 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32) fuse.Status {
wfs.mapPbIdFromLocalToFiler(request.Entry)
defer wfs.mapPbIdFromFilerToLocal(request.Entry)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
glog.Errorf("fh flush create %s: %v", fileFullPath, err)
return fmt.Errorf("fh flush create %s: %v", fileFullPath, err)
}
diff --git a/weed/mount/weedfs_link.go b/weed/mount/weedfs_link.go
index 8c5b67ce2..344b907b4 100644
--- a/weed/mount/weedfs_link.go
+++ b/weed/mount/weedfs_link.go
@@ -88,12 +88,12 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
wfs.mapPbIdFromLocalToFiler(request.Entry)
defer wfs.mapPbIdFromFilerToLocal(request.Entry)
- if err := filer_pb.UpdateEntry(client, updateOldEntryRequest); err != nil {
+ if err := filer_pb.UpdateEntry(context.Background(), client, updateOldEntryRequest); err != nil {
return err
}
wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(updateOldEntryRequest.Directory, updateOldEntryRequest.Entry))
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
return err
}
diff --git a/weed/mount/weedfs_symlink.go b/weed/mount/weedfs_symlink.go
index 8842ec3e6..80081c31c 100644
--- a/weed/mount/weedfs_symlink.go
+++ b/weed/mount/weedfs_symlink.go
@@ -53,7 +53,7 @@ func (wfs *WFS) Symlink(cancel <-chan struct{}, header *fuse.InHeader, target st
wfs.mapPbIdFromLocalToFiler(request.Entry)
defer wfs.mapPbIdFromFilerToLocal(request.Entry)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
return fmt.Errorf("symlink %s: %v", entryFullPath, err)
}
diff --git a/weed/mq/broker/broker_write.go b/weed/mq/broker/broker_write.go
index 9999529fb..9f3c7b50f 100644
--- a/weed/mq/broker/broker_write.go
+++ b/weed/mq/broker/broker_write.go
@@ -1,6 +1,7 @@
package broker
import (
+ "context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/operation"
@@ -20,7 +21,7 @@ func (b *MessageQueueBroker) appendToFile(targetFile string, data []byte) error
// find out existing entry
fullpath := util.FullPath(targetFile)
dir, name := fullpath.DirAndName()
- entry, err := filer_pb.GetEntry(b, fullpath)
+ entry, err := filer_pb.GetEntry(context.Background(), b, fullpath)
var offset int64 = 0
if err == filer_pb.ErrNotFound {
entry = &filer_pb.Entry{
@@ -45,7 +46,7 @@ func (b *MessageQueueBroker) appendToFile(targetFile string, data []byte) error
// update the entry
return b.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- return filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ return filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: entry,
})
diff --git a/weed/mq/logstore/log_to_parquet.go b/weed/mq/logstore/log_to_parquet.go
index 30cad8cc1..b2ee2ae00 100644
--- a/weed/mq/logstore/log_to_parquet.go
+++ b/weed/mq/logstore/log_to_parquet.go
@@ -1,6 +1,7 @@
package logstore
import (
+ "context"
"encoding/binary"
"fmt"
"github.com/parquet-go/parquet-go"
@@ -50,7 +51,7 @@ func CompactTopicPartitions(filerClient filer_pb.FilerClient, t topic.Topic, tim
}
func collectTopicVersions(filerClient filer_pb.FilerClient, t topic.Topic, timeAgo time.Duration) (partitionVersions []time.Time, err error) {
- err = filer_pb.ReadDirAllEntries(filerClient, util.FullPath(t.Dir()), "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, util.FullPath(t.Dir()), "", func(entry *filer_pb.Entry, isLast bool) error {
t, err := topic.ParseTopicVersion(entry.Name)
if err != nil {
// skip non-partition directories
@@ -66,7 +67,7 @@ func collectTopicVersions(filerClient filer_pb.FilerClient, t topic.Topic, timeA
func collectTopicVersionsPartitions(filerClient filer_pb.FilerClient, t topic.Topic, topicVersion time.Time) (partitions []topic.Partition, err error) {
version := topicVersion.Format(topic.PartitionGenerationFormat)
- err = filer_pb.ReadDirAllEntries(filerClient, util.FullPath(t.Dir()).Child(version), "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, util.FullPath(t.Dir()).Child(version), "", func(entry *filer_pb.Entry, isLast bool) error {
if !entry.IsDirectory {
return nil
}
@@ -151,7 +152,7 @@ func groupFilesBySize(logFiles []*filer_pb.Entry, maxGroupSize int64) (logFileGr
}
func readAllLogFiles(filerClient filer_pb.FilerClient, partitionDir string, timeAgo time.Duration, minTsNs, maxTsNs int64) (logFiles []*filer_pb.Entry, err error) {
- err = filer_pb.ReadDirAllEntries(filerClient, util.FullPath(partitionDir), "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, util.FullPath(partitionDir), "", func(entry *filer_pb.Entry, isLast bool) error {
if strings.HasSuffix(entry.Name, ".parquet") {
return nil
}
@@ -173,7 +174,7 @@ func readAllLogFiles(filerClient filer_pb.FilerClient, partitionDir string, time
}
func readAllParquetFiles(filerClient filer_pb.FilerClient, partitionDir string) (minTsNs, maxTsNs int64, err error) {
- err = filer_pb.ReadDirAllEntries(filerClient, util.FullPath(partitionDir), "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, util.FullPath(partitionDir), "", func(entry *filer_pb.Entry, isLast bool) error {
if !strings.HasSuffix(entry.Name, ".parquet") {
return nil
}
@@ -354,7 +355,7 @@ func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile
// write the entry to partitionDir
if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- return filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ return filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: partitionDir,
Entry: entry,
})
diff --git a/weed/mq/logstore/read_log_from_disk.go b/weed/mq/logstore/read_log_from_disk.go
index 71ba58c1f..2c8a7c1de 100644
--- a/weed/mq/logstore/read_log_from_disk.go
+++ b/weed/mq/logstore/read_log_from_disk.go
@@ -1,6 +1,7 @@
package logstore
import (
+ "context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -113,7 +114,7 @@ func GenLogOnDiskReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p top
stopTime := time.Unix(0, stopTsNs)
var processedTsNs int64
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- return filer_pb.SeaweedList(client, partitionDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ return filer_pb.SeaweedList(context.Background(), client, partitionDir, "", func(entry *filer_pb.Entry, isLast bool) error {
if entry.IsDirectory {
return nil
}
diff --git a/weed/mq/logstore/read_parquet_to_log.go b/weed/mq/logstore/read_parquet_to_log.go
index 4811d6c80..6c69b9f12 100644
--- a/weed/mq/logstore/read_parquet_to_log.go
+++ b/weed/mq/logstore/read_parquet_to_log.go
@@ -1,6 +1,7 @@
package logstore
import (
+ "context"
"encoding/binary"
"fmt"
"github.com/parquet-go/parquet-go"
@@ -115,7 +116,7 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- return filer_pb.SeaweedList(client, partitionDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ return filer_pb.SeaweedList(context.Background(), client, partitionDir, "", func(entry *filer_pb.Entry, isLast bool) error {
if entry.IsDirectory {
return nil
}
diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go
index 35cea71b3..60a7c637c 100644
--- a/weed/pb/filer_pb/filer_client.go
+++ b/weed/pb/filer_pb/filer_client.go
@@ -20,12 +20,12 @@ var (
)
type FilerClient interface {
- WithFilerClient(streamingMode bool, fn func(SeaweedFilerClient) error) error
+ WithFilerClient(streamingMode bool, fn func(SeaweedFilerClient) error) error // 15 implementation
AdjustedUrl(location *Location) string
GetDataCenter() string
}
-func GetEntry(filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry, err error) {
+func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry, err error) {
dir, name := fullFilePath.DirAndName()
@@ -37,7 +37,7 @@ func GetEntry(filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry
}
// glog.V(3).Infof("read %s request: %v", fullFilePath, request)
- resp, err := LookupEntry(client, request)
+ resp, err := LookupEntry(ctx, client, request)
if err != nil {
glog.V(3).Infof("read %s %v: %v", fullFilePath, resp, err)
return err
@@ -57,7 +57,7 @@ func GetEntry(filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry
type EachEntryFunction func(entry *Entry, isLast bool) error
-func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction) (err error) {
+func ReadDirAllEntries(ctx context.Context, filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction) (err error) {
var counter uint32
var startFrom string
@@ -69,13 +69,13 @@ func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefi
var paginationLimit uint32 = 10000
- if err = doList(filerClient, fullDirPath, prefix, counterFunc, "", false, paginationLimit); err != nil {
+ if err = doList(ctx, filerClient, fullDirPath, prefix, counterFunc, "", false, paginationLimit); err != nil {
return err
}
for counter == paginationLimit {
counter = 0
- if err = doList(filerClient, fullDirPath, prefix, counterFunc, startFrom, false, paginationLimit); err != nil {
+ if err = doList(ctx, filerClient, fullDirPath, prefix, counterFunc, startFrom, false, paginationLimit); err != nil {
return err
}
}
@@ -83,23 +83,23 @@ func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefi
return nil
}
-func List(filerClient FilerClient, parentDirectoryPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
+func List(ctx context.Context, filerClient FilerClient, parentDirectoryPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
- return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
+ return doSeaweedList(ctx, client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
})
}
-func doList(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
+func doList(ctx context.Context, filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
- return doSeaweedList(client, fullDirPath, prefix, fn, startFrom, inclusive, limit)
+ return doSeaweedList(ctx, client, fullDirPath, prefix, fn, startFrom, inclusive, limit)
})
}
-func SeaweedList(client SeaweedFilerClient, parentDirectoryPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
- return doSeaweedList(client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
+func SeaweedList(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
+ return doSeaweedList(ctx, client, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
}
-func doSeaweedList(client SeaweedFilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
+func doSeaweedList(ctx context.Context, client SeaweedFilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction, startFrom string, inclusive bool, limit uint32) (err error) {
// Redundancy limit to make it correctly judge whether it is the last file.
redLimit := limit
@@ -118,7 +118,7 @@ func doSeaweedList(client SeaweedFilerClient, fullDirPath util.FullPath, prefix
}
glog.V(4).Infof("read directory: %v", request)
- ctx, cancel := context.WithCancel(context.Background())
+ ctx, cancel := context.WithCancel(ctx)
defer cancel()
stream, err := client.ListEntries(ctx, request)
if err != nil {
@@ -156,7 +156,7 @@ func doSeaweedList(client SeaweedFilerClient, fullDirPath util.FullPath, prefix
return nil
}
-func Exists(filerClient FilerClient, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
+func Exists(ctx context.Context, filerClient FilerClient, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
err = filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
@@ -166,7 +166,7 @@ func Exists(filerClient FilerClient, parentDirectoryPath string, entryName strin
}
glog.V(4).Infof("exists entry %v/%v: %v", parentDirectoryPath, entryName, request)
- resp, err := LookupEntry(client, request)
+ resp, err := LookupEntry(ctx, client, request)
if err != nil {
if err == ErrNotFound {
exists = false
@@ -184,7 +184,7 @@ func Exists(filerClient FilerClient, parentDirectoryPath string, entryName strin
return
}
-func Touch(filerClient FilerClient, parentDirectoryPath string, entryName string, entry *Entry) (err error) {
+func Touch(ctx context.Context, filerClient FilerClient, parentDirectoryPath string, entryName string, entry *Entry) (err error) {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
@@ -194,7 +194,7 @@ func Touch(filerClient FilerClient, parentDirectoryPath string, entryName string
}
glog.V(4).Infof("touch entry %v/%v: %v", parentDirectoryPath, entryName, request)
- if err := UpdateEntry(client, request); err != nil {
+ if err := UpdateEntry(ctx, client, request); err != nil {
glog.V(0).Infof("touch exists entry %v: %v", request, err)
return fmt.Errorf("touch exists entry %s/%s: %v", parentDirectoryPath, entryName, err)
}
@@ -204,13 +204,13 @@ func Touch(filerClient FilerClient, parentDirectoryPath string, entryName string
}
-func Mkdir(filerClient FilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
+func Mkdir(ctx context.Context, filerClient FilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
- return DoMkdir(client, parentDirectoryPath, dirName, fn)
+ return DoMkdir(ctx, client, parentDirectoryPath, dirName, fn)
})
}
-func DoMkdir(client SeaweedFilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
+func DoMkdir(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
entry := &Entry{
Name: dirName,
IsDirectory: true,
@@ -233,7 +233,7 @@ func DoMkdir(client SeaweedFilerClient, parentDirectoryPath string, dirName stri
}
glog.V(1).Infof("mkdir: %v", request)
- if err := CreateEntry(client, request); err != nil {
+ if err := CreateEntry(ctx, client, request); err != nil {
glog.V(0).Infof("mkdir %v: %v", request, err)
return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
}
@@ -241,7 +241,7 @@ func DoMkdir(client SeaweedFilerClient, parentDirectoryPath string, dirName stri
return nil
}
-func MkFile(filerClient FilerClient, parentDirectoryPath string, fileName string, chunks []*FileChunk, fn func(entry *Entry)) error {
+func MkFile(ctx context.Context, filerClient FilerClient, parentDirectoryPath string, fileName string, chunks []*FileChunk, fn func(entry *Entry)) error {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
entry := &Entry{
@@ -267,7 +267,7 @@ func MkFile(filerClient FilerClient, parentDirectoryPath string, fileName string
}
glog.V(1).Infof("create file: %s/%s", parentDirectoryPath, fileName)
- if err := CreateEntry(client, request); err != nil {
+ if err := CreateEntry(ctx, client, request); err != nil {
glog.V(0).Infof("create file %v:%v", request, err)
return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
}
@@ -276,13 +276,13 @@ func MkFile(filerClient FilerClient, parentDirectoryPath string, fileName string
})
}
-func Remove(filerClient FilerClient, parentDirectoryPath, name string, isDeleteData, isRecursive, ignoreRecursiveErr, isFromOtherCluster bool, signatures []int32) error {
+func Remove(ctx context.Context, filerClient FilerClient, parentDirectoryPath, name string, isDeleteData, isRecursive, ignoreRecursiveErr, isFromOtherCluster bool, signatures []int32) error {
return filerClient.WithFilerClient(false, func(client SeaweedFilerClient) error {
- return DoRemove(client, parentDirectoryPath, name, isDeleteData, isRecursive, ignoreRecursiveErr, isFromOtherCluster, signatures)
+ return DoRemove(ctx, client, parentDirectoryPath, name, isDeleteData, isRecursive, ignoreRecursiveErr, isFromOtherCluster, signatures)
})
}
-func DoRemove(client SeaweedFilerClient, parentDirectoryPath string, name string, isDeleteData bool, isRecursive bool, ignoreRecursiveErr bool, isFromOtherCluster bool, signatures []int32) error {
+func DoRemove(ctx context.Context, client SeaweedFilerClient, parentDirectoryPath string, name string, isDeleteData bool, isRecursive bool, ignoreRecursiveErr bool, isFromOtherCluster bool, signatures []int32) error {
deleteEntryRequest := &DeleteEntryRequest{
Directory: parentDirectoryPath,
Name: name,
@@ -292,7 +292,7 @@ func DoRemove(client SeaweedFilerClient, parentDirectoryPath string, name string
IsFromOtherCluster: isFromOtherCluster,
Signatures: signatures,
}
- if resp, err := client.DeleteEntry(context.Background(), deleteEntryRequest); err != nil {
+ if resp, err := client.DeleteEntry(ctx, deleteEntryRequest); err != nil {
if strings.Contains(err.Error(), ErrNotFound.Error()) {
return nil
}
diff --git a/weed/pb/filer_pb/filer_client_bfs.go b/weed/pb/filer_pb/filer_client_bfs.go
index e43443706..55d492aca 100644
--- a/weed/pb/filer_pb/filer_client_bfs.go
+++ b/weed/pb/filer_pb/filer_client_bfs.go
@@ -52,7 +52,7 @@ func TraverseBfs(filerClient FilerClient, parentPath util.FullPath, fn func(pare
func processOneDirectory(filerClient FilerClient, parentPath util.FullPath, queue *util.Queue[util.FullPath], jobQueueWg *sync.WaitGroup, fn func(parentPath util.FullPath, entry *Entry)) (err error) {
- return ReadDirAllEntries(filerClient, parentPath, "", func(entry *Entry, isLast bool) error {
+ return ReadDirAllEntries(context.Background(), filerClient, parentPath, "", func(entry *Entry, isLast bool) error {
fn(parentPath, entry)
diff --git a/weed/pb/filer_pb/filer_pb_helper.go b/weed/pb/filer_pb/filer_pb_helper.go
index 0ec31420c..b45d999ad 100644
--- a/weed/pb/filer_pb/filer_pb_helper.go
+++ b/weed/pb/filer_pb/filer_pb_helper.go
@@ -108,8 +108,8 @@ func AfterEntryDeserialization(chunks []*FileChunk) {
}
}
-func CreateEntry(client SeaweedFilerClient, request *CreateEntryRequest) error {
- resp, err := client.CreateEntry(context.Background(), request)
+func CreateEntry(ctx context.Context, client SeaweedFilerClient, request *CreateEntryRequest) error {
+ resp, err := client.CreateEntry(ctx, request)
if err != nil {
glog.V(1).Infof("create entry %s/%s %v: %v", request.Directory, request.Entry.Name, request.OExcl, err)
return fmt.Errorf("CreateEntry: %v", err)
@@ -121,8 +121,8 @@ func CreateEntry(client SeaweedFilerClient, request *CreateEntryRequest) error {
return nil
}
-func UpdateEntry(client SeaweedFilerClient, request *UpdateEntryRequest) error {
- _, err := client.UpdateEntry(context.Background(), request)
+func UpdateEntry(ctx context.Context, client SeaweedFilerClient, request *UpdateEntryRequest) error {
+ _, err := client.UpdateEntry(ctx, request)
if err != nil {
glog.V(1).Infof("update entry %s/%s :%v", request.Directory, request.Entry.Name, err)
return fmt.Errorf("UpdateEntry: %v", err)
@@ -130,8 +130,8 @@ func UpdateEntry(client SeaweedFilerClient, request *UpdateEntryRequest) error {
return nil
}
-func LookupEntry(client SeaweedFilerClient, request *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) {
- resp, err := client.LookupDirectoryEntry(context.Background(), request)
+func LookupEntry(ctx context.Context, client SeaweedFilerClient, request *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) {
+ resp, err := client.LookupDirectoryEntry(ctx, request)
if err != nil {
if err == ErrNotFound || strings.Contains(err.Error(), ErrNotFound.Error()) {
return nil, ErrNotFound
diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go
index 49f6877a0..2254ca2f7 100644
--- a/weed/replication/sink/filersink/filer_sink.go
+++ b/weed/replication/sink/filersink/filer_sink.go
@@ -95,7 +95,7 @@ func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bo
dir, name := util.FullPath(key).DirAndName()
glog.V(4).Infof("delete entry: %v", key)
- err := filer_pb.Remove(fs, dir, name, deleteIncludeChunks, true, true, true, signatures)
+ err := filer_pb.Remove(context.Background(), fs, dir, name, deleteIncludeChunks, true, true, true, signatures)
if err != nil {
glog.V(0).Infof("delete entry %s: %v", key, err)
return fmt.Errorf("delete entry %s: %v", key, err)
@@ -115,7 +115,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
Name: name,
}
// glog.V(1).Infof("lookup: %v", lookupRequest)
- if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
+ if resp, err := filer_pb.LookupEntry(context.Background(), client, lookupRequest); err == nil {
if filer.ETag(resp.Entry) == filer.ETag(entry) {
glog.V(3).Infof("already replicated %s", key)
return nil
@@ -152,7 +152,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
}
glog.V(3).Infof("create: %v", request)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
glog.V(0).Infof("create entry %s: %v", key, err)
return fmt.Errorf("create entry %s: %v", key, err)
}
@@ -175,7 +175,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
}
glog.V(4).Infof("lookup entry: %v", request)
- resp, err := filer_pb.LookupEntry(client, request)
+ resp, err := filer_pb.LookupEntry(context.Background(), client, request)
if err != nil {
glog.V(0).Infof("lookup %s: %v", key, err)
return err
diff --git a/weed/s3api/bucket_metadata.go b/weed/s3api/bucket_metadata.go
index d1f487104..a65fe5404 100644
--- a/weed/s3api/bucket_metadata.go
+++ b/weed/s3api/bucket_metadata.go
@@ -1,6 +1,7 @@
package s3api
import (
+ "context"
"encoding/json"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -13,7 +14,7 @@ import (
)
var loadBucketMetadataFromFiler = func(r *BucketRegistry, bucketName string) (*BucketMetaData, error) {
- entry, err := filer_pb.GetEntry(r.s3a, util.NewFullPath(r.s3a.option.BucketsPath, bucketName))
+ entry, err := filer_pb.GetEntry(context.Background(), r.s3a, util.NewFullPath(r.s3a.option.BucketsPath, bucketName))
if err != nil {
return nil, err
}
@@ -64,7 +65,7 @@ func NewBucketRegistry(s3a *S3ApiServer) *BucketRegistry {
}
func (r *BucketRegistry) init() error {
- err := filer_pb.List(r.s3a, r.s3a.option.BucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err := filer_pb.List(context.Background(), r.s3a, r.s3a.option.BucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
r.LoadBucketMetadata(entry)
return nil
}, "", false, math.MaxUint32)
diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go
index 8ae8f780a..38400140f 100644
--- a/weed/s3api/filer_util.go
+++ b/weed/s3api/filer_util.go
@@ -11,19 +11,19 @@ import (
func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {
- return filer_pb.Mkdir(s3a, parentDirectoryPath, dirName, fn)
+ return filer_pb.Mkdir(context.Background(), s3a, parentDirectoryPath, dirName, fn)
}
func (s3a *S3ApiServer) mkFile(parentDirectoryPath string, fileName string, chunks []*filer_pb.FileChunk, fn func(entry *filer_pb.Entry)) error {
- return filer_pb.MkFile(s3a, parentDirectoryPath, fileName, chunks, fn)
+ return filer_pb.MkFile(context.Background(), s3a, parentDirectoryPath, fileName, chunks, fn)
}
func (s3a *S3ApiServer) list(parentDirectoryPath, prefix, startFrom string, inclusive bool, limit uint32) (entries []*filer_pb.Entry, isLast bool, err error) {
- err = filer_pb.List(s3a, parentDirectoryPath, prefix, func(entry *filer_pb.Entry, isLastEntry bool) error {
+ err = filer_pb.List(context.Background(), s3a, parentDirectoryPath, prefix, func(entry *filer_pb.Entry, isLastEntry bool) error {
entries = append(entries, entry)
if isLastEntry {
isLast = true
@@ -76,19 +76,19 @@ func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath strin
func (s3a *S3ApiServer) exists(parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
- return filer_pb.Exists(s3a, parentDirectoryPath, entryName, isDirectory)
+ return filer_pb.Exists(context.Background(), s3a, parentDirectoryPath, entryName, isDirectory)
}
func (s3a *S3ApiServer) touch(parentDirectoryPath string, entryName string, entry *filer_pb.Entry) (err error) {
- return filer_pb.Touch(s3a, parentDirectoryPath, entryName, entry)
+ return filer_pb.Touch(context.Background(), s3a, parentDirectoryPath, entryName, entry)
}
func (s3a *S3ApiServer) getEntry(parentDirectoryPath, entryName string) (entry *filer_pb.Entry, err error) {
fullPath := util.NewFullPath(parentDirectoryPath, entryName)
- return filer_pb.GetEntry(s3a, fullPath)
+ return filer_pb.GetEntry(context.Background(), s3a, fullPath)
}
func (s3a *S3ApiServer) updateEntry(parentDirectoryPath string, newEntry *filer_pb.Entry) error {
@@ -98,7 +98,7 @@ func (s3a *S3ApiServer) updateEntry(parentDirectoryPath string, newEntry *filer_
}
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- err := filer_pb.UpdateEntry(client, updateEntryRequest)
+ err := filer_pb.UpdateEntry(context.Background(), client, updateEntryRequest)
if err != nil {
return err
}
diff --git a/weed/s3api/filer_util_tags.go b/weed/s3api/filer_util_tags.go
index c2bdc7294..d33e46c2e 100644
--- a/weed/s3api/filer_util_tags.go
+++ b/weed/s3api/filer_util_tags.go
@@ -1,6 +1,7 @@
package s3api
import (
+ "context"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"strings"
@@ -15,7 +16,7 @@ func (s3a *S3ApiServer) getTags(parentDirectoryPath string, entryName string) (t
err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
Directory: parentDirectoryPath,
Name: entryName,
})
@@ -37,7 +38,7 @@ func (s3a *S3ApiServer) setTags(parentDirectoryPath string, entryName string, ta
return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
Directory: parentDirectoryPath,
Name: entryName,
})
@@ -58,7 +59,7 @@ func (s3a *S3ApiServer) setTags(parentDirectoryPath string, entryName string, ta
resp.Entry.Extended[S3TAG_PREFIX+k] = []byte(v)
}
- return filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
+ return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
Directory: parentDirectoryPath,
Entry: resp.Entry,
IsFromOtherCluster: false,
@@ -73,7 +74,7 @@ func (s3a *S3ApiServer) rmTags(parentDirectoryPath string, entryName string) (er
return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
Directory: parentDirectoryPath,
Name: entryName,
})
@@ -93,7 +94,7 @@ func (s3a *S3ApiServer) rmTags(parentDirectoryPath string, entryName string) (er
return nil
}
- return filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
+ return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
Directory: parentDirectoryPath,
Entry: resp.Entry,
IsFromOtherCluster: false,
diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go
index 6a4740fef..5233b7c30 100644
--- a/weed/s3api/s3api_object_handlers_list.go
+++ b/weed/s3api/s3api_object_handlers_list.go
@@ -471,7 +471,7 @@ func (s3a *S3ApiServer) ensureDirectoryAllEmpty(filerClient filer_pb.SeaweedFile
var isExhausted bool
var foundEntry bool
for fileCounter == 0 && !isExhausted && err == nil {
- err = filer_pb.SeaweedList(filerClient, currentDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.SeaweedList(context.Background(), filerClient, currentDir, "", func(entry *filer_pb.Entry, isLast bool) error {
foundEntry = true
if entry.IsOlderDir() {
subDirs = append(subDirs, entry.Name)
diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go
index dfd9f5844..ccad9d942 100644
--- a/weed/s3api/s3api_object_handlers_multipart.go
+++ b/weed/s3api/s3api_object_handlers_multipart.go
@@ -93,7 +93,6 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r
}
stats_collect.RecordBucketActiveTime(bucket)
stats_collect.S3UploadedObjectsCounter.WithLabelValues(bucket).Inc()
-
writeSuccessResponseXML(w, r, response)
}
diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go
index dbe6dfed5..2b7493ee6 100644
--- a/weed/server/webdav_server.go
+++ b/weed/server/webdav_server.go
@@ -209,7 +209,7 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm
}
glog.V(1).Infof("mkdir: %v", request)
- if err := filer_pb.CreateEntry(client, request); err != nil {
+ if err := filer_pb.CreateEntry(context.Background(), client, request); err != nil {
return fmt.Errorf("mkdir %s/%s: %v", dir, name, err)
}
@@ -240,7 +240,7 @@ func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, f
dir, name := util.FullPath(fullFilePath).DirAndName()
err = fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
- if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ if err := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: dir,
Entry: &filer_pb.Entry{
Name: name,
@@ -299,7 +299,7 @@ func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string)
dir, name := util.FullPath(fullFilePath).DirAndName()
- return filer_pb.Remove(fs, dir, name, true, false, false, false, []int32{fs.signature})
+ return filer_pb.Remove(context.Background(), fs, dir, name, true, false, false, false, []int32{fs.signature})
}
@@ -371,7 +371,7 @@ func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.F
fullpath := util.FullPath(fullFilePath)
var fi FileInfo
- entry, err := filer_pb.GetEntry(fs, fullpath)
+ entry, err := filer_pb.GetEntry(context.Background(), fs, fullpath)
if err != nil {
if err == filer_pb.ErrNotFound {
return nil, os.ErrNotExist
@@ -452,7 +452,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
var getErr error
ctx := context.Background()
if f.entry == nil {
- f.entry, getErr = filer_pb.GetEntry(f.fs, fullPath)
+ f.entry, getErr = filer_pb.GetEntry(context.Background(), f.fs, fullPath)
}
if f.entry == nil {
@@ -543,7 +543,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
glog.V(2).Infof("WebDavFileSystem.Read %v", f.name)
if f.entry == nil {
- f.entry, err = filer_pb.GetEntry(f.fs, util.FullPath(f.name))
+ f.entry, err = filer_pb.GetEntry(context.Background(), f.fs, util.FullPath(f.name))
}
if f.entry == nil {
return 0, err
@@ -583,7 +583,7 @@ func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) {
dir, _ := util.FullPath(f.name).DirAndName()
- err = filer_pb.ReadDirAllEntries(f.fs, util.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), f.fs, util.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) error {
fi := FileInfo{
size: int64(filer.FileSize(entry)),
name: entry.Name,
diff --git a/weed/sftpd/sftp_filer.go b/weed/sftpd/sftp_filer.go
index dbe6a438d..4cc709390 100644
--- a/weed/sftpd/sftp_filer.go
+++ b/weed/sftpd/sftp_filer.go
@@ -322,7 +322,7 @@ func (fs *SftpServer) makeDir(r *sftp.Request) error {
return err
}
// default mode and ownership
- err := filer_pb.Mkdir(fs, string(dir), name, func(entry *filer_pb.Entry) {
+ err := filer_pb.Mkdir(context.Background(), fs, string(dir), name, func(entry *filer_pb.Entry) {
mode := uint32(0755 | os.ModeDir)
if strings.HasPrefix(r.Filepath, fs.user.HomeDir) {
mode = uint32(0700 | os.ModeDir)
diff --git a/weed/sftpd/user/homemanager.go b/weed/sftpd/user/homemanager.go
index c9051939c..c0e29ca73 100644
--- a/weed/sftpd/user/homemanager.go
+++ b/weed/sftpd/user/homemanager.go
@@ -104,7 +104,7 @@ func (hm *HomeManager) createSingleDirectory(dirPath string, user *User) error {
// Directory doesn't exist, create it
glog.V(0).Infof("Creating directory %s for user %s", dirPath, user.Username)
- err = filer_pb.Mkdir(hm, string(dir), name, func(entry *filer_pb.Entry) {
+ err = filer_pb.Mkdir(context.Background(), hm, string(dir), name, func(entry *filer_pb.Entry) {
// Set appropriate permissions
entry.Attributes.FileMode = uint32(0700 | os.ModeDir) // rwx------ for user
entry.Attributes.Uid = user.Uid
diff --git a/weed/shell/command_fs_cat.go b/weed/shell/command_fs_cat.go
index 47a7f3be8..cf1395a2f 100644
--- a/weed/shell/command_fs_cat.go
+++ b/weed/shell/command_fs_cat.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -49,7 +50,7 @@ func (c *commandFsCat) Do(args []string, commandEnv *CommandEnv, writer io.Write
Name: name,
Directory: dir,
}
- respLookupEntry, err := filer_pb.LookupEntry(client, request)
+ respLookupEntry, err := filer_pb.LookupEntry(context.Background(), client, request)
if err != nil {
return err
}
diff --git a/weed/shell/command_fs_du.go b/weed/shell/command_fs_du.go
index c35179f88..456f6bab6 100644
--- a/weed/shell/command_fs_du.go
+++ b/weed/shell/command_fs_du.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"fmt"
"io"
@@ -58,7 +59,7 @@ func (c *commandFsDu) Do(args []string, commandEnv *CommandEnv, writer io.Writer
func duTraverseDirectory(writer io.Writer, filerClient filer_pb.FilerClient, dir, name string) (blockCount, byteCount uint64, err error) {
- err = filer_pb.ReadDirAllEntries(filerClient, util.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, util.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) error {
var fileBlockCount, fileByteCount uint64
diff --git a/weed/shell/command_fs_log.go b/weed/shell/command_fs_log.go
index e6b7a6f9a..66f143030 100644
--- a/weed/shell/command_fs_log.go
+++ b/weed/shell/command_fs_log.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"flag"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -41,11 +42,11 @@ func (c *commandFsLogPurge) Do(args []string, commandEnv *CommandEnv, writer io.
}
modificationTimeAgo := time.Now().Add(-time.Hour * 24 * time.Duration(*daysAgo)).Unix()
- err = filer_pb.ReadDirAllEntries(commandEnv, filer.SystemLogDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), commandEnv, filer.SystemLogDir, "", func(entry *filer_pb.Entry, isLast bool) error {
if entry.Attributes.Mtime > modificationTimeAgo {
return nil
}
- if errDel := filer_pb.Remove(commandEnv, filer.SystemLogDir, entry.Name, true, true, true, false, nil); errDel != nil {
+ if errDel := filer_pb.Remove(context.Background(), commandEnv, filer.SystemLogDir, entry.Name, true, true, true, false, nil); errDel != nil {
return errDel
}
if *verbose {
diff --git a/weed/shell/command_fs_ls.go b/weed/shell/command_fs_ls.go
index e007b4c9e..442702693 100644
--- a/weed/shell/command_fs_ls.go
+++ b/weed/shell/command_fs_ls.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"fmt"
"io"
"os"
@@ -66,7 +67,7 @@ func (c *commandFsLs) Do(args []string, commandEnv *CommandEnv, writer io.Writer
dir, name := util.FullPath(path).DirAndName()
entryCount := 0
- err = filer_pb.ReadDirAllEntries(commandEnv, util.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(dir), name, func(entry *filer_pb.Entry, isLast bool) error {
if !showHidden && strings.HasPrefix(entry.Name, ".") {
return nil
diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go
index 31d9d6294..f31ec10f4 100644
--- a/weed/shell/command_fs_merge_volumes.go
+++ b/weed/shell/command_fs_merge_volumes.go
@@ -134,7 +134,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
continue
}
- if err = filer_pb.UpdateEntry(filerClient, &filer_pb.UpdateEntryRequest{
+ if err = filer_pb.UpdateEntry(context.Background(), filerClient, &filer_pb.UpdateEntryRequest{
Directory: string(parentPath),
Entry: entry,
}); err != nil {
diff --git a/weed/shell/command_fs_meta_cat.go b/weed/shell/command_fs_meta_cat.go
index e47d8faa6..2abb4d2b9 100644
--- a/weed/shell/command_fs_meta_cat.go
+++ b/weed/shell/command_fs_meta_cat.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"google.golang.org/protobuf/proto"
@@ -49,7 +50,7 @@ func (c *commandFsMetaCat) Do(args []string, commandEnv *CommandEnv, writer io.W
Name: name,
Directory: dir,
}
- respLookupEntry, err := filer_pb.LookupEntry(client, request)
+ respLookupEntry, err := filer_pb.LookupEntry(context.Background(), client, request)
if err != nil {
return err
}
diff --git a/weed/shell/command_fs_meta_change_volume_id.go b/weed/shell/command_fs_meta_change_volume_id.go
index f1c148f5b..167d64288 100644
--- a/weed/shell/command_fs_meta_change_volume_id.go
+++ b/weed/shell/command_fs_meta_change_volume_id.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"flag"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -86,7 +87,7 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv,
if hasChanges {
println("Updating", parentPath, entry.Name)
if *isForce {
- if updateErr := filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
+ if updateErr := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
Directory: string(parentPath),
Entry: entry,
}); updateErr != nil {
diff --git a/weed/shell/command_fs_meta_load.go b/weed/shell/command_fs_meta_load.go
index d56274362..f43574f49 100644
--- a/weed/shell/command_fs_meta_load.go
+++ b/weed/shell/command_fs_meta_load.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"flag"
"fmt"
"io"
@@ -117,7 +118,7 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x")
if fullEntry.Entry.IsDirectory {
wg.Wait()
- if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ if errEntry := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: fullEntry.Dir,
Entry: fullEntry.Entry,
}); errEntry != nil {
@@ -128,7 +129,7 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
wg.Add(1)
waitChan <- struct{}{}
go func(entry *filer_pb.FullEntry) {
- if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ if errEntry := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: entry.Dir,
Entry: entry.Entry,
}); errEntry != nil {
diff --git a/weed/shell/command_fs_mv.go b/weed/shell/command_fs_mv.go
index cb966571c..2d44e4b58 100644
--- a/weed/shell/command_fs_mv.go
+++ b/weed/shell/command_fs_mv.go
@@ -65,7 +65,7 @@ func (c *commandFsMv) Do(args []string, commandEnv *CommandEnv, writer io.Writer
Name: destinationDir,
Directory: destinationName,
}
- respDestinationLookupEntry, err := filer_pb.LookupEntry(client, destinationRequest)
+ respDestinationLookupEntry, err := filer_pb.LookupEntry(context.Background(), client, destinationRequest)
var targetDir, targetName string
diff --git a/weed/shell/command_fs_rm.go b/weed/shell/command_fs_rm.go
index 0af75f048..2e3f19121 100644
--- a/weed/shell/command_fs_rm.go
+++ b/weed/shell/command_fs_rm.go
@@ -74,7 +74,7 @@ func (c *commandFsRm) Do(args []string, commandEnv *CommandEnv, writer io.Writer
Directory: targetDir,
Name: targetName,
}
- _, err = filer_pb.LookupEntry(client, lookupRequest)
+ _, err = filer_pb.LookupEntry(context.Background(), client, lookupRequest)
if err != nil {
fmt.Fprintf(writer, "rm: %s: %v\n", targetPath, err)
continue
diff --git a/weed/shell/command_fs_tree.go b/weed/shell/command_fs_tree.go
index 21e352af2..628c95b30 100644
--- a/weed/shell/command_fs_tree.go
+++ b/weed/shell/command_fs_tree.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"fmt"
"io"
"strings"
@@ -55,7 +56,7 @@ func treeTraverseDirectory(writer io.Writer, filerClient filer_pb.FilerClient, d
prefix.addMarker(level)
- err = filer_pb.ReadDirAllEntries(filerClient, dir, name, func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, dir, name, func(entry *filer_pb.Entry, isLast bool) error {
if level < 0 && name != "" {
if entry.Name != name {
return nil
diff --git a/weed/shell/command_remote_cache.go b/weed/shell/command_remote_cache.go
index da19af0cf..987502c14 100644
--- a/weed/shell/command_remote_cache.go
+++ b/weed/shell/command_remote_cache.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"flag"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -100,7 +101,7 @@ func (c *commandRemoteCache) doCacheOneDirectory(commandEnv *CommandEnv, writer
func recursivelyTraverseDirectory(filerClient filer_pb.FilerClient, dirPath util.FullPath, visitEntry func(dir util.FullPath, entry *filer_pb.Entry) bool) (err error) {
- err = filer_pb.ReadDirAllEntries(filerClient, dirPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.ReadDirAllEntries(context.Background(), filerClient, dirPath, "", func(entry *filer_pb.Entry, isLast bool) error {
if entry.IsDirectory {
if !visitEntry(dirPath, entry) {
return nil
diff --git a/weed/shell/command_remote_configure.go b/weed/shell/command_remote_configure.go
index dbc44c8bf..aa2f3d1d4 100644
--- a/weed/shell/command_remote_configure.go
+++ b/weed/shell/command_remote_configure.go
@@ -141,7 +141,7 @@ func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, write
func (c *commandRemoteConfigure) listExistingRemoteStorages(commandEnv *CommandEnv, writer io.Writer) error {
- return filer_pb.ReadDirAllEntries(commandEnv, util.FullPath(filer.DirectoryEtcRemote), "", func(entry *filer_pb.Entry, isLast bool) error {
+ return filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(filer.DirectoryEtcRemote), "", func(entry *filer_pb.Entry, isLast bool) error {
if len(entry.Content) == 0 {
fmt.Fprintf(writer, "skipping %s\n", entry.Name)
return nil
diff --git a/weed/shell/command_remote_meta_sync.go b/weed/shell/command_remote_meta_sync.go
index 1b2e33c14..4d430bc76 100644
--- a/weed/shell/command_remote_meta_sync.go
+++ b/weed/shell/command_remote_meta_sync.go
@@ -136,7 +136,7 @@ func pullMetadata(commandEnv *CommandEnv, writer io.Writer, localMountedDir util
localDir := filer.MapRemoteStorageLocationPathToFullPath(localMountedDir, remoteMountedLocation, remoteDir)
fmt.Fprint(writer, localDir.Child(name))
- lookupResponse, lookupErr := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
+ lookupResponse, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
Directory: string(localDir),
Name: name,
})
diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go
index cfe89ddce..2892852e9 100644
--- a/weed/shell/command_remote_mount.go
+++ b/weed/shell/command_remote_mount.go
@@ -138,7 +138,7 @@ func syncMetadata(commandEnv *CommandEnv, writer io.Writer, dir string, nonEmpty
}
mountToDirIsEmpty := true
- listErr := filer_pb.SeaweedList(client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ listErr := filer_pb.SeaweedList(context.Background(), client, dir, "", func(entry *filer_pb.Entry, isLast bool) error {
mountToDirIsEmpty = false
return nil
}, "", false, 1)
diff --git a/weed/shell/command_remote_unmount.go b/weed/shell/command_remote_unmount.go
index 6de92e612..87487481a 100644
--- a/weed/shell/command_remote_unmount.go
+++ b/weed/shell/command_remote_unmount.go
@@ -88,6 +88,7 @@ func (c *commandRemoteUnmount) purgeMountedData(commandEnv *CommandEnv, dir stri
// find existing directory, and ensure the directory is empty
err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+ ctx := context.Background()
parent, name := util.FullPath(dir).DirAndName()
lookupResp, lookupErr := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
Directory: parent,
@@ -99,12 +100,12 @@ func (c *commandRemoteUnmount) purgeMountedData(commandEnv *CommandEnv, dir stri
oldEntry := lookupResp.Entry
- deleteError := filer_pb.DoRemove(client, parent, name, true, true, true, false, nil)
+ deleteError := filer_pb.DoRemove(ctx, client, parent, name, true, true, true, false, nil)
if deleteError != nil {
return fmt.Errorf("delete %s: %v", dir, deleteError)
}
- mkdirErr := filer_pb.DoMkdir(client, parent, name, func(entry *filer_pb.Entry) {
+ mkdirErr := filer_pb.DoMkdir(ctx, client, parent, name, func(entry *filer_pb.Entry) {
entry.Attributes = oldEntry.Attributes
entry.Extended = oldEntry.Extended
entry.Attributes.Crtime = time.Now().Unix()
diff --git a/weed/shell/command_s3_bucket_create.go b/weed/shell/command_s3_bucket_create.go
index 86c788b0d..be17591e4 100644
--- a/weed/shell/command_s3_bucket_create.go
+++ b/weed/shell/command_s3_bucket_create.go
@@ -71,7 +71,7 @@ func (c *commandS3BucketCreate) Do(args []string, commandEnv *CommandEnv, writer
},
}
- if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+ if err := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
Directory: filerBucketsPath,
Entry: entry,
}); err != nil {
diff --git a/weed/shell/command_s3_bucket_delete.go b/weed/shell/command_s3_bucket_delete.go
index 6f8c9c4b5..ecd83d1c2 100644
--- a/weed/shell/command_s3_bucket_delete.go
+++ b/weed/shell/command_s3_bucket_delete.go
@@ -66,6 +66,6 @@ func (c *commandS3BucketDelete) Do(args []string, commandEnv *CommandEnv, writer
return
}
- return filer_pb.Remove(commandEnv, filerBucketsPath, *bucketName, false, true, true, false, nil)
+ return filer_pb.Remove(context.Background(), commandEnv, filerBucketsPath, *bucketName, false, true, true, false, nil)
}
diff --git a/weed/shell/command_s3_bucket_list.go b/weed/shell/command_s3_bucket_list.go
index 7939f2086..e6fa44303 100644
--- a/weed/shell/command_s3_bucket_list.go
+++ b/weed/shell/command_s3_bucket_list.go
@@ -57,7 +57,7 @@ func (c *commandS3BucketList) Do(args []string, commandEnv *CommandEnv, writer i
return fmt.Errorf("read buckets: %v", err)
}
- err = filer_pb.List(commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.List(context.Background(), commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
if !entry.IsDirectory {
return nil
}
diff --git a/weed/shell/command_s3_bucket_quota.go b/weed/shell/command_s3_bucket_quota.go
index 3301a8d0e..f6562453f 100644
--- a/weed/shell/command_s3_bucket_quota.go
+++ b/weed/shell/command_s3_bucket_quota.go
@@ -83,7 +83,7 @@ func (c *commandS3BucketQuota) Do(args []string, commandEnv *CommandEnv, writer
}
}
- if err := filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
+ if err := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{
Directory: filerBucketsPath,
Entry: bucketEntry,
}); err != nil {
diff --git a/weed/shell/command_s3_bucket_quota_check.go b/weed/shell/command_s3_bucket_quota_check.go
index 04708a0e7..27b903405 100644
--- a/weed/shell/command_s3_bucket_quota_check.go
+++ b/weed/shell/command_s3_bucket_quota_check.go
@@ -2,6 +2,7 @@ package shell
import (
"bytes"
+ "context"
"flag"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -65,7 +66,7 @@ func (c *commandS3BucketQuotaEnforce) Do(args []string, commandEnv *CommandEnv,
// process each bucket
hasConfChanges := false
- err = filer_pb.List(commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.List(context.Background(), commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
if !entry.IsDirectory {
return nil
}
diff --git a/weed/shell/command_s3_clean_uploads.go b/weed/shell/command_s3_clean_uploads.go
index bfce5d54d..c77f3cd74 100644
--- a/weed/shell/command_s3_clean_uploads.go
+++ b/weed/shell/command_s3_clean_uploads.go
@@ -1,6 +1,7 @@
package shell
import (
+ "context"
"flag"
"fmt"
"io"
@@ -54,7 +55,7 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer
}
var buckets []string
- err = filer_pb.List(commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err = filer_pb.List(context.Background(), commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
buckets = append(buckets, entry.Name)
return nil
}, "", false, math.MaxUint32)
@@ -75,7 +76,7 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io
uploadsDir := filerBucketsPath + "/" + bucket + "/" + s3_constants.MultipartUploadsFolder
var staleUploads []string
now := time.Now()
- err := filer_pb.List(commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+ err := filer_pb.List(context.Background(), commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
ctime := time.Unix(entry.Attributes.Crtime, 0)
if ctime.Add(timeAgo).Before(now) {
staleUploads = append(staleUploads, entry.Name)
diff --git a/weed/shell/commands.go b/weed/shell/commands.go
index 264dd4818..40be210a2 100644
--- a/weed/shell/commands.go
+++ b/weed/shell/commands.go
@@ -91,7 +91,7 @@ func (ce *CommandEnv) checkDirectory(path string) error {
dir, name := util.FullPath(path).DirAndName()
- exists, err := filer_pb.Exists(ce, dir, name, true)
+ exists, err := filer_pb.Exists(context.Background(), ce, dir, name, true)
if !exists {
return fmt.Errorf("%s is not a directory", path)