mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2025-09-20 06:59:23 +08:00
added context to filer_client method calls (#6808)
Co-authored-by: akosov <a.kosov@kryptonite.ru>
This commit is contained in:
@@ -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,
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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()
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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,
|
||||
})
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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) {
|
||||
|
@@ -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
|
||||
|
@@ -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)
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
||||
|
@@ -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)
|
||||
}
|
||||
|
||||
|
@@ -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,
|
||||
})
|
||||
|
@@ -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,
|
||||
})
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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)
|
||||
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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)
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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,
|
||||
|
@@ -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)
|
||||
|
@@ -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)
|
||||
|
||||
}
|
||||
|
@@ -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,
|
||||
|
@@ -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)
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
|
||||
|
@@ -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 {
|
||||
|
@@ -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
|
||||
|
@@ -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 {
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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 {
|
||||
|
@@ -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 {
|
||||
|
@@ -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
|
||||
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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,
|
||||
})
|
||||
|
@@ -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)
|
||||
|
@@ -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()
|
||||
|
@@ -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 {
|
||||
|
@@ -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)
|
||||
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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 {
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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)
|
||||
|
@@ -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)
|
||||
|
Reference in New Issue
Block a user