command_s3_clean_uploads.go 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103
  1. package shell
  2. import (
  3. "flag"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  6. "github.com/seaweedfs/seaweedfs/weed/security"
  7. "github.com/seaweedfs/seaweedfs/weed/util"
  8. "io"
  9. "math"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  12. )
  13. func init() {
  14. Commands = append(Commands, &commandS3CleanUploads{})
  15. }
  16. type commandS3CleanUploads struct {
  17. }
  18. func (c *commandS3CleanUploads) Name() string {
  19. return "s3.clean.uploads"
  20. }
  21. func (c *commandS3CleanUploads) Help() string {
  22. return `clean up stale multipart uploads
  23. Example:
  24. s3.clean.uploads -timeAgo 1.5h
  25. `
  26. }
  27. func (c *commandS3CleanUploads) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  28. bucketCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  29. uploadedTimeAgo := bucketCommand.Duration("timeAgo", 24*time.Hour, "created time before now. \"1.5h\" or \"2h45m\". Valid time units are \"m\", \"h\"")
  30. if err = bucketCommand.Parse(args); err != nil {
  31. return nil
  32. }
  33. signingKey := util.GetViper().GetString("jwt.signing.key")
  34. var filerBucketsPath string
  35. filerBucketsPath, err = readFilerBucketsPath(commandEnv)
  36. if err != nil {
  37. return fmt.Errorf("read buckets: %v", err)
  38. }
  39. var buckets []string
  40. err = filer_pb.List(commandEnv, filerBucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error {
  41. buckets = append(buckets, entry.Name)
  42. return nil
  43. }, "", false, math.MaxUint32)
  44. if err != nil {
  45. return fmt.Errorf("list buckets under %v: %v", filerBucketsPath, err)
  46. }
  47. for _, bucket := range buckets {
  48. if err := c.cleanupUploads(commandEnv, writer, filerBucketsPath, bucket, *uploadedTimeAgo, signingKey); err != nil {
  49. fmt.Fprintf(writer, fmt.Sprintf("failed cleanup uploads for bucket %s: %v", bucket, err))
  50. }
  51. }
  52. return err
  53. }
  54. func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io.Writer, filerBucketsPath string, bucket string, timeAgo time.Duration, signingKey string) error {
  55. uploadsDir := filerBucketsPath + "/" + bucket + "/" + s3_constants.MultipartUploadsFolder
  56. var staleUploads []string
  57. now := time.Now()
  58. err := filer_pb.List(commandEnv, uploadsDir, "", func(entry *filer_pb.Entry, isLast bool) error {
  59. ctime := time.Unix(entry.Attributes.Crtime, 0)
  60. if ctime.Add(timeAgo).Before(now) {
  61. staleUploads = append(staleUploads, entry.Name)
  62. }
  63. return nil
  64. }, "", false, math.MaxUint32)
  65. if err != nil {
  66. return fmt.Errorf("list uploads under %v: %v", uploadsDir, err)
  67. }
  68. var encodedJwt security.EncodedJwt
  69. if signingKey != "" {
  70. encodedJwt = security.GenJwtForFilerServer(security.SigningKey(signingKey), 15*60)
  71. }
  72. for _, staleUpload := range staleUploads {
  73. deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload)
  74. fmt.Fprintf(writer, "purge %s\n", deleteUrl)
  75. err = util.Delete(deleteUrl, string(encodedJwt))
  76. if err != nil && err.Error() != "" {
  77. return fmt.Errorf("purge %s/%s: %v", uploadsDir, staleUpload, err)
  78. }
  79. }
  80. return nil
  81. }