delete_content.go 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158
  1. package operation
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/pb"
  7. "google.golang.org/grpc"
  8. "net/http"
  9. "strings"
  10. "sync"
  11. "github.com/seaweedfs/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. // DeleteFileIds batch deletes a list of fileIds
  27. func DeleteFileIds(masterFn GetMasterFn, usePublicUrl bool, grpcDialOption grpc.DialOption, fileIds []string) ([]*volume_server_pb.DeleteResult, error) {
  28. lookupFunc := func(vids []string) (results map[string]*LookupResult, err error) {
  29. results, err = LookupVolumeIds(masterFn, grpcDialOption, vids)
  30. if err == nil && usePublicUrl {
  31. for _, result := range results {
  32. for _, loc := range result.Locations {
  33. loc.Url = loc.PublicUrl
  34. }
  35. }
  36. }
  37. return
  38. }
  39. return DeleteFileIdsWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc)
  40. }
  41. func DeleteFileIdsWithLookupVolumeId(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func(vid []string) (map[string]*LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
  42. var ret []*volume_server_pb.DeleteResult
  43. vid_to_fileIds := make(map[string][]string)
  44. var vids []string
  45. for _, fileId := range fileIds {
  46. vid, _, err := ParseFileId(fileId)
  47. if err != nil {
  48. ret = append(ret, &volume_server_pb.DeleteResult{
  49. FileId: fileId,
  50. Status: http.StatusBadRequest,
  51. Error: err.Error()},
  52. )
  53. continue
  54. }
  55. if _, ok := vid_to_fileIds[vid]; !ok {
  56. vid_to_fileIds[vid] = make([]string, 0)
  57. vids = append(vids, vid)
  58. }
  59. vid_to_fileIds[vid] = append(vid_to_fileIds[vid], fileId)
  60. }
  61. lookupResults, err := lookupFunc(vids)
  62. if err != nil {
  63. return ret, err
  64. }
  65. server_to_fileIds := make(map[pb.ServerAddress][]string)
  66. for vid, result := range lookupResults {
  67. if result.Error != "" {
  68. ret = append(ret, &volume_server_pb.DeleteResult{
  69. FileId: vid,
  70. Status: http.StatusBadRequest,
  71. Error: result.Error},
  72. )
  73. continue
  74. }
  75. for _, location := range result.Locations {
  76. serverAddress := location.ServerAddress()
  77. if _, ok := server_to_fileIds[serverAddress]; !ok {
  78. server_to_fileIds[serverAddress] = make([]string, 0)
  79. }
  80. server_to_fileIds[serverAddress] = append(
  81. server_to_fileIds[serverAddress], vid_to_fileIds[vid]...)
  82. }
  83. }
  84. resultChan := make(chan []*volume_server_pb.DeleteResult, len(server_to_fileIds))
  85. var wg sync.WaitGroup
  86. for server, fidList := range server_to_fileIds {
  87. wg.Add(1)
  88. go func(server pb.ServerAddress, fidList []string) {
  89. defer wg.Done()
  90. if deleteResults, deleteErr := DeleteFileIdsAtOneVolumeServer(server, grpcDialOption, fidList, false); deleteErr != nil {
  91. err = deleteErr
  92. } else if deleteResults != nil {
  93. resultChan <- deleteResults
  94. }
  95. }(server, fidList)
  96. }
  97. wg.Wait()
  98. close(resultChan)
  99. for result := range resultChan {
  100. ret = append(ret, result...)
  101. }
  102. return ret, err
  103. }
  104. // DeleteFileIdsAtOneVolumeServer deletes a list of files that is on one volume server via gRpc
  105. func DeleteFileIdsAtOneVolumeServer(volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fileIds []string, includeCookie bool) (ret []*volume_server_pb.DeleteResult, err error) {
  106. err = WithVolumeServerClient(false, volumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  107. req := &volume_server_pb.BatchDeleteRequest{
  108. FileIds: fileIds,
  109. SkipCookieCheck: !includeCookie,
  110. }
  111. resp, err := volumeServerClient.BatchDelete(context.Background(), req)
  112. // fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)
  113. if err != nil {
  114. return err
  115. }
  116. ret = append(ret, resp.Results...)
  117. return nil
  118. })
  119. if err != nil {
  120. return
  121. }
  122. for _, result := range ret {
  123. if result.Error != "" && result.Error != "not found" {
  124. return nil, fmt.Errorf("delete fileId %s: %v", result.FileId, result.Error)
  125. }
  126. }
  127. return
  128. }