diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index 2574de14e9..ed663ed6e8 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -182,7 +182,7 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc self: owner, lc: lc, } - if err := lock.retryUntilLocked(5 * time.Second); err != nil { + if err := lock.retryUntilLocked(5*time.Second, 0); err != nil { glog.Warningf("create lock %s: %v", key, err) return nil } @@ -194,6 +194,17 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc // synchronous acquisition of NewShortLivedLock with the auto-renewal of // StartLongLivedLock. Release with Stop(). func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.Duration) *LiveLock { + lock, err := lc.NewBlockingLongLivedLockWithin(key, owner, lockTTL, 0) + if err != nil { + glog.Warningf("create lock %s: %v", key, err) + return nil + } + return lock +} + +// NewBlockingLongLivedLockWithin is NewBlockingLongLivedLock that gives up once +// another owner has held the lock for maxWait. Zero waits without bound. +func (lc *LockClient) NewBlockingLongLivedLockWithin(key, owner string, lockTTL, maxWait time.Duration) (*LiveLock, error) { if lockTTL == 0 { lockTTL = lock_manager.LiveLockTTL } @@ -207,10 +218,8 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D lc: lc, lockTTL: lockTTL, } - // Block until acquired - if err := lock.retryUntilLocked(lockTTL); err != nil { - glog.Warningf("create lock %s: %v", key, err) - return nil + if err := lock.retryUntilLocked(lockTTL, maxWait); err != nil { + return nil, err } // Start renewal goroutine using a ticker for interruptible sleep lock.renewalDone = make(chan struct{}) @@ -231,7 +240,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D } } }() - return lock + return lock, nil } // StartLongLivedLock starts a goroutine to lock the key and returns immediately. @@ -310,7 +319,8 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh // several seconds): when a holder on another mount releases the lock, the // waiter must pick it up promptly, otherwise cross-mount write handoff stalls // long enough to time out clients. -func (lock *LiveLock) retryUntilLocked(lockDuration time.Duration) error { +func (lock *LiveLock) retryUntilLocked(lockDuration, maxWait time.Duration) error { + started := time.Now() // The retry window bounds the attempts against one route: the clock // starts before the first attempt and each attempt gets what is left // of it. A spent window on a non-seed target falls back to the seed @@ -356,7 +366,10 @@ func (lock *LiveLock) retryUntilLocked(lockDuration time.Duration) error { glog.V(1).Infof("create lock %s: %v", lock.key, err) if strings.Contains(err.Error(), "lock already owned") { // Ordinary contention: a reachable server holds the lock, so - // waiting is the point and has no bound. + // waiting is the point and is bounded only by maxWait. + if maxWait > 0 && time.Since(started) >= maxWait { + return err + } unavailableSince = time.Now() continue } diff --git a/weed/cluster/lock_client_test.go b/weed/cluster/lock_client_test.go index 796cf28021..7aaae0cbd3 100644 --- a/weed/cluster/lock_client_test.go +++ b/weed/cluster/lock_client_test.go @@ -225,9 +225,8 @@ func (s *contendedLockFiler) DistributedLock(ctx context.Context, req *filer_pb. return &filer_pb.LockResponse{Error: "lock already owned by someone"}, nil } -// A lock held by another owner is ordinary contention: acquisition waits it -// out rather than failing on the unavailability bound. -func TestNewShortLivedLockWaitsOutContention(t *testing.T) { +func newContendedLockClient(t *testing.T) *LockClient { + t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("listen: %v", err) @@ -235,14 +234,19 @@ func TestNewShortLivedLockWaitsOutContention(t *testing.T) { grpcServer := grpc.NewServer() filer_pb.RegisterSeaweedFilerServer(grpcServer, &contendedLockFiler{}) go grpcServer.Serve(listener) - defer grpcServer.Stop() + t.Cleanup(grpcServer.Stop) - dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) host, port, err := net.SplitHostPort(listener.Addr().String()) if err != nil { t.Fatalf("split host port: %v", err) } - lc := NewLockClient(dialOption, pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port))) + return NewLockClient(grpc.WithTransportCredentials(insecure.NewCredentials()), pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port))) +} + +// A lock held by another owner is ordinary contention: acquisition waits it +// out rather than failing on the unavailability bound. +func TestNewShortLivedLockWaitsOutContention(t *testing.T) { + lc := newContendedLockClient(t) lc.noLockServerRetryPeriod = 200 * time.Millisecond done := make(chan *LiveLock, 1) @@ -256,6 +260,22 @@ func TestNewShortLivedLockWaitsOutContention(t *testing.T) { } } +func TestNewBlockingLongLivedLockWithinGivesUpOnContention(t *testing.T) { + lc := newContendedLockClient(t) + + maxWait := 1500 * time.Millisecond + start := time.Now() + lock, err := lc.NewBlockingLongLivedLockWithin("test-key", "test-owner", 15*time.Second, maxWait) + elapsed := time.Since(start) + + if lock != nil || err == nil { + t.Fatalf("got lock %v err %v, want no lock and the contention error", lock, err) + } + if elapsed < maxWait || elapsed > maxWait+5*time.Second { + t.Fatalf("gave up after %v, want just past %v", elapsed, maxWait) + } +} + // A ring member that refuses connections is unavailability, not contention: // acquisition fails on the same bound as "no lock server found". func TestNewShortLivedLockFailsFastOnUnreachableFiler(t *testing.T) { @@ -519,27 +539,12 @@ func TestDoLockKeepsRenewTokenOnError(t *testing.T) { // A definitive rejection answers with an empty token, which clears the stale // one — only transport failures keep it. func TestDoLockClearsRenewTokenOnRejection(t *testing.T) { - listener, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - grpcServer := grpc.NewServer() - filer_pb.RegisterSeaweedFilerServer(grpcServer, &contendedLockFiler{}) - go grpcServer.Serve(listener) - defer grpcServer.Stop() - - dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) - host, port, err := net.SplitHostPort(listener.Addr().String()) - if err != nil { - t.Fatalf("split host port: %v", err) - } - filerAddr := pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)) - lc := NewLockClient(dialOption, filerAddr) + lc := newContendedLockClient(t) lock := &LiveLock{ key: "test-key", renewToken: "tok", - hostFiler: filerAddr, - grpcDialOption: dialOption, + hostFiler: lc.seedFiler, + grpcDialOption: lc.grpcDialOption, lc: lc, } errorMessage, _, err := lock.doLock(time.Second, time.Second) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 8d8acc670e..76c1086aab 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -1246,11 +1246,15 @@ func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s if owner == "" { return rmUploadDir(true) } - objectPath := s3a.toFilerPath(bucket, object) + commitPath, err := s3a.uploadCommitPath(bucket, object) + if err != nil { + glog.Errorf("bucket %s abort upload %s: %v", bucket, uploadId, err) + return s3err.ErrInternalError + } resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ - LockKey: objectPath, + LockKey: commitPath, RouteKey: s3a.objectRouteKey(bucket, object), - ConditionKey: objectPath, + ConditionKey: commitPath, Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ Kind: filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL, ExtKey: s3_constants.SeaweedFSUploadId, @@ -1278,6 +1282,21 @@ func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s return s3err.ErrNone } +// uploadCommitPath is the path a routed completion of object locks: the +// object path in a versioned bucket, else the entry it writes, which for a +// trailing-slash key lies inside the directory the key names. +func (s3a *S3ApiServer) uploadCommitPath(bucket, object string) (string, error) { + versioningState, err := s3a.getVersioningState(bucket) + if err != nil { + return "", err + } + if versioningState == s3_constants.VersioningEnabled { + return s3a.toFilerPath(bucket, object), nil + } + entryName, dirName := s3a.getEntryNameAndDir(&s3.CompleteMultipartUploadInput{Bucket: &bucket, Key: &object}) + return dirName + "/" + entryName, nil +} + // uploadCompleted reports whether the upload assembled into an object: the // object entry, or any version file under .versions, still carries the // upload id completion stamps on it. diff --git a/weed/s3api/filer_multipart_abort_test.go b/weed/s3api/filer_multipart_abort_test.go index fc1af1d300..aad44355f9 100644 --- a/weed/s3api/filer_multipart_abort_test.go +++ b/weed/s3api/filer_multipart_abort_test.go @@ -3,9 +3,12 @@ package s3api import ( "context" "testing" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -281,3 +284,52 @@ func TestLifecycleAbortUndecidableCheckRetriesLater(t *testing.T) { t.Fatalf("delete issued despite undecidable check: %+v", f.deleteReq) } } + +type fakeRoutedAbortFiler struct { + fakeAbortFiler + txnReq *filer_pb.ObjectTransactionRequest +} + +func (f *fakeRoutedAbortFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.txnReq = req + return &filer_pb.ObjectTransactionResponse{}, nil +} + +// A routed abort must take the path lock the completion commits under. For a +// trailing-slash key a non-versioned completion writes the entry inside the +// directory the key names, while a versioned one locks the object path. +func TestRoutedAbortLocksTheCompletionCommitPath(t *testing.T) { + for _, tc := range []struct { + versioning string + lockKey string + }{ + {"", "/dir/dir"}, + {s3_constants.VersioningSuspended, "/dir/dir"}, + {s3_constants.VersioningEnabled, "/dir/"}, + } { + f := &fakeRoutedAbortFiler{} + ownerAddr := startFakeFiler(t, f) + s3a := newFailoverTestServer(t, ownerAddr) + bucketDir := s3a.bucketDir("b") + f.uploadsDir = bucketDir + "/" + s3_constants.MultipartUploadsFolder + f.uploadEntry = uploadRecordEntry("up1") + f.uploadEntry.Extended[s3_constants.ExtMultipartObjectKey] = []byte("dir/") + f.objectDir = bucketDir + f.versionsDir = bucketDir + "/dir/dir" + s3_constants.VersionsFolder + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: tc.versioning}) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1) + + if _, code := s3a.abortMultipartUpload(&s3.AbortMultipartUploadInput{Bucket: aws.String("b"), Key: aws.String("dir/"), UploadId: aws.String("up1")}); code != s3err.ErrNone { + t.Fatalf("versioning %q: code = %v", tc.versioning, code) + } + req := f.txnReq + if req == nil { + t.Fatalf("versioning %q: no transaction", tc.versioning) + } + if req.LockKey != bucketDir+tc.lockKey || req.ConditionKey != req.LockKey || req.RouteKey != s3a.objectRouteKey("b", "dir/") { + t.Fatalf("versioning %q: lock %q condition %q route %q", tc.versioning, req.LockKey, req.ConditionKey, req.RouteKey) + } + } +} diff --git a/weed/shell/command_s3_clean_uploads.go b/weed/shell/command_s3_clean_uploads.go index 965d03749a..ed8c09188d 100644 --- a/weed/shell/command_s3_clean_uploads.go +++ b/weed/shell/command_s3_clean_uploads.go @@ -11,12 +11,14 @@ import ( "strings" "time" + "github.com/google/uuid" + + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" - "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" - util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) func init() { @@ -49,34 +51,34 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer return nil } - signingKey := util.GetViper().GetString("jwt.filer_signing.key") - var filerBucketsPath string filerBucketsPath, err = readFilerBucketsPath(commandEnv) if err != nil { return fmt.Errorf("read buckets: %w", err) } - var buckets []string + var buckets []*filer_pb.Entry err = filer_pb.List(context.Background(), commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error { - buckets = append(buckets, entry.Name) + buckets = append(buckets, entry) return nil }, "", false, math.MaxUint32) if err != nil { return fmt.Errorf("list buckets under %v: %w", filerBucketsPath, err) } + lock := newObjectWriteLock(commandEnv) for _, bucket := range buckets { - if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket, *uploadedTimeAgo, signingKey); err != nil { - fmt.Fprintf(writer, "failed cleanup uploads for bucket %s: %v", bucket, err) + if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket.Name, *uploadedTimeAgo, lock); err != nil { + fmt.Fprintf(writer, "failed cleanup uploads for bucket %s: %v\n", bucket.Name, err) } } return err } -func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, signingKey string) error { - uploadsDir := filerBucketsPath + "/" + bucket + "/" + s3_constants.MultipartUploadsFolder +func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, lock objectWriteLock) error { + bucketDir := filerBucketsPath + "/" + bucket + uploadsDir := bucketDir + "/" + s3_constants.MultipartUploadsFolder var staleUploads []*filer_pb.Entry now := time.Now() err := filer_pb.List(context.Background(), commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error { @@ -90,37 +92,166 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io return fmt.Errorf("list uploads under %v: %w", uploadsDir, err) } - var encodedJwt security.EncodedJwt - if signingKey != "" { - encodedJwt = security.GenJwtForFilerServer(security.SigningKey(signingKey), 15*60) - } - for _, staleUpload := range staleUploads { - // A completed upload's part entries share chunks with the finished - // object, so purging their data corrupts it. Completion normally - // removes this directory itself; a survivor means that cleanup failed - // and only the metadata should go. An undecidable lookup is left for - // the next run rather than risk live chunks. - completed, checkErr := c.uploadCompleted(commandEnv, filerBucketsPath+"/"+bucket, staleUpload) - if checkErr != nil { - fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, checkErr) + held, release, lockErr := lock(uploadObjectPath(bucketDir, staleUpload)) + if lockErr != nil { + fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, lockErr) continue } - deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.FilerAddress().ToHttpAddress(), uploadsDir, staleUpload.Name) - if completed { - deleteUrl += "&skipChunkDeletion=true" - } - fmt.Fprintf(writer, "purge %s\n", deleteUrl) - - err = util_http.Delete(deleteUrl, string(encodedJwt)) - if err != nil && err.Error() != "" { - return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload.Name, err) + err := c.removeUploadDir(commandEnv, writer, bucketDir, uploadsDir, staleUpload, held) + release() + if err != nil { + return err } } return nil } +// objectWriteLock takes the distributed lock the S3 gateway commits a +// completion under when it is not routed to the object's owner filer, and +// renews it until release. It gives up after a minute of contention. +type objectWriteLock func(objectPath string) (held func() bool, release func(), err error) + +func newObjectWriteLock(commandEnv *CommandEnv) objectWriteLock { + lockClient := cluster.NewLockClient(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress()) + owner := fmt.Sprintf("s3.clean.uploads-%d", util.RandomInt32()) + return func(objectPath string) (func() bool, func(), error) { + if objectPath == "" { + return func() bool { return true }, func() {}, nil + } + key := objectRouteKey(objectPath) + lock, err := lockClient.NewBlockingLongLivedLockWithin(key, owner, 15*time.Second, time.Minute) + if err != nil { + return nil, nil, fmt.Errorf("lock %s: %w", key, err) + } + return lock.IsLocked, func() { + if err := lock.Stop(); err != nil { + glog.Warningf("release lock %s: %v", key, err) + } + }, nil + } +} + +// removeUploadDir mirrors the S3 gateway's abort, under the object write lock. +// A completed upload's part entries share chunks with the finished object, so +// only the metadata goes; an open upload's chunks are freed by +// routedUploadDirDelete. An undecidable lookup, or a lost lock, is left for +// the next run. +func (c *commandS3CleanUploads) removeUploadDir(filerClient filer_pb.FilerClient, writer io.Writer, bucketDir, uploadsDir string, upload *filer_pb.Entry, lockHeld func() bool) error { + completed, err := c.uploadCompleted(filerClient, bucketDir, upload) + if err != nil { + fmt.Fprintf(writer, "skip %s: %v\n", upload.Name, err) + return nil + } + var commitPath string + if !completed { + if commitPath, err = uploadCommitPath(filerClient, bucketDir, upload); err != nil { + fmt.Fprintf(writer, "skip %s: %v\n", upload.Name, err) + return nil + } + } + if !lockHeld() { + fmt.Fprintf(writer, "skip %s: lost the object write lock\n", upload.Name) + return nil + } + if !completed { + freed, err := routedUploadDirDelete(filerClient, uploadObjectPath(bucketDir, upload), commitPath, uploadsDir, upload.Name) + if err != nil { + return fmt.Errorf("purge %s/%s: %v", uploadsDir, upload.Name, err) + } + if freed { + fmt.Fprintf(writer, "purge %s/%s\n", uploadsDir, upload.Name) + return nil + } + } + fmt.Fprintf(writer, "purge %s/%s metadata only\n", uploadsDir, upload.Name) + if err := filer_pb.Remove(context.Background(), filerClient, uploadsDir, upload.Name, false, true, true, false, nil); err != nil { + return fmt.Errorf("purge %s/%s: %v", uploadsDir, upload.Name, err) + } + return nil +} + +// routedUploadDirDelete frees an open upload's chunks in an ObjectTransaction +// on the object's owner filer, under the path lock a routed completion +// commits under. It deletes nothing and reports false when the object +// already carries the upload. +func routedUploadDirDelete(filerClient filer_pb.FilerClient, objectPath, commitPath, uploadsDir, uploadId string) (bool, error) { + if objectPath == "" { + return true, filer_pb.Remove(context.Background(), filerClient, uploadsDir, uploadId, true, true, true, false, nil) + } + var resp *filer_pb.ObjectTransactionResponse + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + var txnErr error + resp, txnErr = client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{ + LockKey: commitPath, + RouteKey: objectRouteKey(objectPath), + ConditionKey: commitPath, + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ + Kind: filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL, + ExtKey: s3_constants.SeaweedFSUploadId, + ExtValue: uploadId, + }}}, + Mutations: []*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_DELETE, + Directory: uploadsDir, + Name: uploadId, + IsDeleteData: true, + IsRecursive: true, + }}, + RequestId: uuid.NewString(), + }) + return txnErr + }) + if err != nil { + return false, err + } + if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED { + return false, nil + } + if resp.Error != "" { + return false, errors.New(resp.Error) + } + return true, nil +} + +// uploadObjectPath is the filer path of the object the upload completes into, as +// the gateway's toFilerPath derives it, or "" when the upload names none. +func uploadObjectPath(bucketDir string, upload *filer_pb.Entry) string { + objectKey := string(upload.Extended[s3_constants.ExtMultipartObjectKey]) + if objectKey == "" { + return "" + } + return bucketDir + "/" + s3_constants.NormalizeObjectKey(objectKey) +} + +// uploadCommitPath is the path a routed completion of the upload locks: the +// object path in a versioned bucket, else the entry it writes, which for a +// trailing-slash key lies inside the directory the key names. Like the +// gateway's, it reads the bucket's versioning state just before the delete. +func uploadCommitPath(filerClient filer_pb.FilerClient, bucketDir string, upload *filer_pb.Entry) (string, error) { + objectKey := string(upload.Extended[s3_constants.ExtMultipartObjectKey]) + if objectKey == "" { + return "", nil + } + bucket, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(bucketDir)) + if err != nil { + return "", fmt.Errorf("read bucket %s: %w", bucketDir, err) + } + if versioning, lockEnabled := getBucketVersioningState(bucket); versioning == s3_constants.VersioningEnabled || lockEnabled { + return uploadObjectPath(bucketDir, upload), nil + } + dir := path.Dir(objectKey) + if dir == "." { + dir = "" + } + return path.Join(bucketDir, dir, path.Base(objectKey)), nil +} + +func objectRouteKey(objectPath string) string { + return s3_constants.ObjectWriteRouteKeyPrefix + strings.TrimSuffix(objectPath, "/") +} + // uploadCompleted reports whether the upload assembled into an object: the // object entry, or any version file under .versions, still carries the // upload id completion stamps on it. diff --git a/weed/shell/command_s3_clean_uploads_test.go b/weed/shell/command_s3_clean_uploads_test.go index 459be2deb0..141191ccf8 100644 --- a/weed/shell/command_s3_clean_uploads_test.go +++ b/weed/shell/command_s3_clean_uploads_test.go @@ -26,6 +26,7 @@ func (s stubFilerClient) GetDataCenter() string { return "" } type stubFilerSvc struct { filer_pb.SeaweedFilerClient + entries map[string]*filer_pb.Entry lookupResp *filer_pb.LookupDirectoryEntryResponse lookupErr error listResp []*filer_pb.Entry @@ -33,9 +34,25 @@ type stubFilerSvc struct { listErr error lookupReq *filer_pb.LookupDirectoryEntryRequest listReq *filer_pb.ListEntriesRequest + txnResp *filer_pb.ObjectTransactionResponse + txnReqs []*filer_pb.ObjectTransactionRequest + deleteReqs []*filer_pb.DeleteEntryRequest +} + +func (s *stubFilerSvc) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest, opts ...grpc.CallOption) (*filer_pb.ObjectTransactionResponse, error) { + s.txnReqs = append(s.txnReqs, req) + return s.txnResp, nil +} + +func (s *stubFilerSvc) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest, opts ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + s.deleteReqs = append(s.deleteReqs, req) + return &filer_pb.DeleteEntryResponse{}, nil } func (s *stubFilerSvc) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + if entry, ok := s.entries[req.Directory+"/"+req.Name]; ok { + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil + } s.lookupReq = req return s.lookupResp, s.lookupErr } @@ -201,3 +218,144 @@ func TestUploadCompleted(t *testing.T) { } }) } + +func bucketEntry(versioning string) *filer_pb.Entry { + return &filer_pb.Entry{Name: "b", IsDirectory: true, Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(versioning)}} +} + +// openUploadSvc answers an open upload's lookups: no object, no versions, and +// a bucket with the given versioning state. +func openUploadSvc(versioning string, txnResp *filer_pb.ObjectTransactionResponse) *stubFilerSvc { + return &stubFilerSvc{ + entries: map[string]*filer_pb.Entry{"/buckets/b": bucketEntry(versioning)}, + lookupErr: filer_pb.ErrNotFound, + listErr: filer_pb.ErrNotFound, + txnResp: txnResp, + } +} + +func TestRemoveUploadDir(t *testing.T) { + c := &commandS3CleanUploads{} + const uploadsDir = "/buckets/b/" + s3_constants.MultipartUploadsFolder + lockHeld := func() bool { return true } + + t.Run("open upload is freed in a transaction on the object's lock", func(t *testing.T) { + svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{}) + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "/dir//obj"), lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 1 || len(svc.deleteReqs) != 0 { + t.Fatalf("%d transactions, %d deletes; want 1, 0", len(svc.txnReqs), len(svc.deleteReqs)) + } + req := svc.txnReqs[0] + if req.LockKey != "/buckets/b/dir/obj" || req.ConditionKey != req.LockKey || req.RouteKey != s3_constants.ObjectWriteRouteKeyPrefix+req.LockKey { + t.Fatalf("lock %q condition %q route %q", req.LockKey, req.ConditionKey, req.RouteKey) + } + clause := req.Condition.GetClauses()[0] + if clause.Kind != filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL || clause.ExtKey != s3_constants.SeaweedFSUploadId || string(clause.ExtValue) != "up1" { + t.Fatalf("condition %v", clause) + } + m := req.Mutations[0] + if m.Type != filer_pb.ObjectMutation_DELETE || m.Directory != uploadsDir || m.Name != "up1" || !m.IsDeleteData || !m.IsRecursive { + t.Fatalf("mutation %v", m) + } + }) + + t.Run("trailing-slash key locks the entry its completion writes as versioning changes", func(t *testing.T) { + svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{}) + for _, tc := range []struct { + versioning string + objectLock bool + lockKey string + }{ + {"", false, "/buckets/b/dir/dir"}, + {s3_constants.VersioningEnabled, false, "/buckets/b/dir/"}, + {s3_constants.VersioningSuspended, false, "/buckets/b/dir/dir"}, + {"", true, "/buckets/b/dir/"}, + } { + bucket := bucketEntry(tc.versioning) + if tc.objectLock { + bucket.Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled) + } + svc.entries["/buckets/b"] = bucket + svc.txnReqs = nil + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "dir/"), lockHeld); err != nil { + t.Fatal(err) + } + req := svc.txnReqs[0] + if req.LockKey != tc.lockKey || req.ConditionKey != tc.lockKey || req.RouteKey != s3_constants.ObjectWriteRouteKeyPrefix+"/buckets/b/dir" { + t.Fatalf("versioning=%q objectLock=%v: lock %q condition %q route %q", tc.versioning, tc.objectLock, req.LockKey, req.ConditionKey, req.RouteKey) + } + } + }) + + t.Run("unreadable bucket deletes nothing", func(t *testing.T) { + svc := &stubFilerSvc{lookupErr: filer_pb.ErrNotFound, listErr: filer_pb.ErrNotFound, txnResp: &filer_pb.ObjectTransactionResponse{}} + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 { + t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs)) + } + }) + + t.Run("upload completed before the transaction keeps its chunks", func(t *testing.T) { + svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}) + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.deleteReqs) != 1 || svc.deleteReqs[0].IsDeleteData || svc.deleteReqs[0].Name != "up1" { + t.Fatalf("deletes %v, want one metadata-only delete of up1", svc.deleteReqs) + } + }) + + t.Run("completed upload goes metadata-only without a transaction", func(t *testing.T) { + svc := &stubFilerSvc{lookupResp: &filer_pb.LookupDirectoryEntryResponse{Entry: completedEntry("up1")}} + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 1 || svc.deleteReqs[0].IsDeleteData { + t.Fatalf("%d transactions, deletes %v", len(svc.txnReqs), svc.deleteReqs) + } + }) + + t.Run("undecidable lookup deletes nothing", func(t *testing.T) { + svc := &stubFilerSvc{lookupErr: errors.New("store unavailable")} + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 { + t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs)) + } + }) + + t.Run("lost lock deletes nothing", func(t *testing.T) { + svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{}) + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), func() bool { return false }); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 0 { + t.Fatalf("%d transactions, %d deletes", len(svc.txnReqs), len(svc.deleteReqs)) + } + }) + + t.Run("transaction error fails the purge", func(t *testing.T) { + svc := openUploadSvc("", &filer_pb.ObjectTransactionResponse{Error: "boom"}) + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, uploadEntry("up1", "obj"), lockHeld); err == nil { + t.Fatal("want an error") + } + if len(svc.deleteReqs) != 0 { + t.Fatalf("deletes %v", svc.deleteReqs) + } + }) + + t.Run("upload without an object key frees its chunks directly", func(t *testing.T) { + svc := &stubFilerSvc{} + if err := c.removeUploadDir(stubFilerClient{client: svc}, io.Discard, "/buckets/b", uploadsDir, &filer_pb.Entry{Name: "up1"}, lockHeld); err != nil { + t.Fatal(err) + } + if len(svc.txnReqs) != 0 || len(svc.deleteReqs) != 1 || !svc.deleteReqs[0].IsDeleteData { + t.Fatalf("%d transactions, deletes %v", len(svc.txnReqs), svc.deleteReqs) + } + }) +}