mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-12-04 02:39:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2900 from kmlebedev/fix_cleanupUploads

Browse Source 
avoid breaking loop in cleanupUploads if error is empty
This commit is contained in:
Chris Lu 2022-04-12 02:03:31 -07:00 committed by GitHub
commit dd13764a5c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
weed/shell/command_s3_clean_uploads.go
View File

@ -3,6 +3,7 @@ package shell
import ( import (
"flag" "flag"
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
"io" "io"
"math" "math"
@ -39,6 +40,8 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer
return nil return nil
} }
signingKey := util.GetViper().GetString("jwt.signing.key")
var filerBucketsPath string var filerBucketsPath string
filerBucketsPath, err = readFilerBucketsPath(commandEnv) filerBucketsPath, err = readFilerBucketsPath(commandEnv)
if err != nil { if err != nil {
@ -55,14 +58,16 @@ func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer
} }
for _, bucket := range buckets { for _, bucket := range buckets {
c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket, *uploadedTimeAgo) if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket, *uploadedTimeAgo, signingKey); err != nil {
fmt.Fprintf(writer, fmt.Sprintf("failed cleanup uploads for backet %s: %v", bucket, err))
}
} }
return err return err
} }
func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration) error { func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, signingKey string) error {
uploadsDir := filerBucketsPath + "/" + bucket + "/.uploads" uploadsDir := filerBucketsPath + "/" + bucket + "/.uploads"
var staleUploads []string var staleUploads []string
now := time.Now() now := time.Now()
@ -77,12 +82,17 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io
return fmt.Errorf("list uploads under %v: %v", uploadsDir, err) return fmt.Errorf("list uploads under %v: %v", uploadsDir, err)
} }
var encodedJwt security.EncodedJwt
if signingKey != "" {
encodedJwt = security.GenJwtForFilerServer(security.SigningKey(signingKey), 15*60)
}
for _, staleUpload := range staleUploads { for _, staleUpload := range staleUploads {
deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload) deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload)
fmt.Fprintf(writer, "purge %s\n", deleteUrl) fmt.Fprintf(writer, "purge %s\n", deleteUrl)
err = util.Delete(deleteUrl, "") err = util.Delete(deleteUrl, string(encodedJwt))
if err != nil { if err != nil && err.Error() != "" {
return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload, err) return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload, err)
} }
} }