delete_content.go 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. package operation
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "net/http"
  7. "strings"
  8. "sync"
  9. "google.golang.org/grpc"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  12. )
  13. type DeleteResult struct {
  14. Fid string `json:"fid"`
  15. Size int `json:"size"`
  16. Status int `json:"status"`
  17. Error string `json:"error,omitempty"`
  18. }
  19. func ParseFileId(fid string) (vid string, key_cookie string, err error) {
  20. commaIndex := strings.Index(fid, ",")
  21. if commaIndex <= 0 {
  22. return "", "", errors.New("Wrong fid format.")
  23. }
  24. return fid[:commaIndex], fid[commaIndex+1:], nil
  25. }
  26. // DeleteFiles batch deletes a list of fileIds
  27. func DeleteFiles(master string, grpcDialOption grpc.DialOption, fileIds []string) ([]*volume_server_pb.DeleteResult, error) {
  28. lookupFunc := func(vids []string) (map[string]LookupResult, error) {
  29. return LookupVolumeIds(master, grpcDialOption, vids)
  30. }
  31. return DeleteFilesWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc)
  32. }
  33. func DeleteFilesWithLookupVolumeId(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func(vid []string) (map[string]LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
  34. var ret []*volume_server_pb.DeleteResult
  35. vid_to_fileIds := make(map[string][]string)
  36. var vids []string
  37. for _, fileId := range fileIds {
  38. vid, _, err := ParseFileId(fileId)
  39. if err != nil {
  40. ret = append(ret, &volume_server_pb.DeleteResult{
  41. FileId: fileId,
  42. Status: http.StatusBadRequest,
  43. Error: err.Error()},
  44. )
  45. continue
  46. }
  47. if _, ok := vid_to_fileIds[vid]; !ok {
  48. vid_to_fileIds[vid] = make([]string, 0)
  49. vids = append(vids, vid)
  50. }
  51. vid_to_fileIds[vid] = append(vid_to_fileIds[vid], fileId)
  52. }
  53. lookupResults, err := lookupFunc(vids)
  54. if err != nil {
  55. return ret, err
  56. }
  57. server_to_fileIds := make(map[string][]string)
  58. for vid, result := range lookupResults {
  59. if result.Error != "" {
  60. ret = append(ret, &volume_server_pb.DeleteResult{
  61. FileId: vid,
  62. Status: http.StatusBadRequest,
  63. Error: err.Error()},
  64. )
  65. continue
  66. }
  67. for _, location := range result.Locations {
  68. if _, ok := server_to_fileIds[location.Url]; !ok {
  69. server_to_fileIds[location.Url] = make([]string, 0)
  70. }
  71. server_to_fileIds[location.Url] = append(
  72. server_to_fileIds[location.Url], vid_to_fileIds[vid]...)
  73. }
  74. }
  75. resultChan := make(chan []*volume_server_pb.DeleteResult, len(server_to_fileIds))
  76. var wg sync.WaitGroup
  77. for server, fidList := range server_to_fileIds {
  78. wg.Add(1)
  79. go func(server string, fidList []string) {
  80. defer wg.Done()
  81. if deleteResults, deleteErr := DeleteFilesAtOneVolumeServer(server, grpcDialOption, fidList); deleteErr != nil {
  82. err = deleteErr
  83. } else if deleteResults != nil {
  84. resultChan <- deleteResults
  85. }
  86. }(server, fidList)
  87. }
  88. wg.Wait()
  89. close(resultChan)
  90. for result := range resultChan {
  91. ret = append(ret, result...)
  92. }
  93. glog.V(1).Infof("deleted %d items", len(ret))
  94. return ret, err
  95. }
  96. // DeleteFilesAtOneVolumeServer deletes a list of files that is on one volume server via gRpc
  97. func DeleteFilesAtOneVolumeServer(volumeServer string, grpcDialOption grpc.DialOption, fileIds []string) (ret []*volume_server_pb.DeleteResult, err error) {
  98. err = WithVolumeServerClient(volumeServer, grpcDialOption, func(ctx context.Context, volumeServerClient volume_server_pb.VolumeServerClient) error {
  99. req := &volume_server_pb.BatchDeleteRequest{
  100. FileIds: fileIds,
  101. }
  102. resp, err := volumeServerClient.BatchDelete(context.Background(), req)
  103. // fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)
  104. if err != nil {
  105. return err
  106. }
  107. ret = append(ret, resp.Results...)
  108. return nil
  109. })
  110. if err != nil {
  111. return
  112. }
  113. for _, result := range ret {
  114. if result.Error != "" && result.Error != "not found" {
  115. return nil, fmt.Errorf("delete fileId %s: %v", result.FileId, result.Error)
  116. }
  117. }
  118. return
  119. }