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)