seaweedfs/weed/operation/delete_content.go

159 lines
4.1 KiB
Go
Raw Normal View History

2012-09-26 10:27:10 +00:00
package operation
import (
2018-10-14 07:30:20 +00:00
"context"
2014-04-15 16:09:40 +00:00
"errors"
2018-10-14 07:30:20 +00:00
"fmt"
"github.com/chrislusf/seaweedfs/weed/pb"
"google.golang.org/grpc"
2018-10-14 07:30:20 +00:00
"net/http"
2014-04-15 16:09:40 +00:00
"strings"
"sync"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
2012-09-26 10:27:10 +00:00
)
2014-04-15 16:09:40 +00:00
type DeleteResult struct {
Fid string `json:"fid"`
Size int `json:"size"`
Status int `json:"status"`
Error string `json:"error,omitempty"`
2014-04-15 16:09:40 +00:00
}
func ParseFileId(fid string) (vid string, key_cookie string, err error) {
commaIndex := strings.Index(fid, ",")
if commaIndex <= 0 {
return "", "", errors.New("Wrong fid format.")
}
return fid[:commaIndex], fid[commaIndex+1:], nil
}
// DeleteFiles batch deletes a list of fileIds
func DeleteFiles(masterFn GetMasterFn, usePublicUrl bool, grpcDialOption grpc.DialOption, fileIds []string) ([]*volume_server_pb.DeleteResult, error) {
2021-08-13 04:40:33 +00:00
lookupFunc := func(vids []string) (results map[string]*LookupResult, err error) {
results, err = LookupVolumeIds(masterFn, grpcDialOption, vids)
2020-03-18 18:16:45 +00:00
if err == nil && usePublicUrl {
for _, result := range results {
for _, loc := range result.Locations {
loc.Url = loc.PublicUrl
}
}
}
return
2018-11-21 04:56:28 +00:00
}
2019-02-18 20:11:52 +00:00
return DeleteFilesWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc)
2018-11-21 04:56:28 +00:00
}
2021-08-13 04:40:33 +00:00
func DeleteFilesWithLookupVolumeId(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func(vid []string) (map[string]*LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
2018-11-21 04:56:28 +00:00
var ret []*volume_server_pb.DeleteResult
2014-04-15 16:09:40 +00:00
vid_to_fileIds := make(map[string][]string)
var vids []string
for _, fileId := range fileIds {
vid, _, err := ParseFileId(fileId)
if err != nil {
ret = append(ret, &volume_server_pb.DeleteResult{
2019-02-15 08:09:19 +00:00
FileId: fileId,
Status: http.StatusBadRequest,
Error: err.Error()},
)
2014-04-15 16:09:40 +00:00
continue
}
if _, ok := vid_to_fileIds[vid]; !ok {
vid_to_fileIds[vid] = make([]string, 0)
vids = append(vids, vid)
}
vid_to_fileIds[vid] = append(vid_to_fileIds[vid], fileId)
}
2018-11-21 04:56:28 +00:00
lookupResults, err := lookupFunc(vids)
2014-04-15 16:09:40 +00:00
if err != nil {
return ret, err
}
server_to_fileIds := make(map[pb.ServerAddress][]string)
2014-04-15 16:09:40 +00:00
for vid, result := range lookupResults {
if result.Error != "" {
ret = append(ret, &volume_server_pb.DeleteResult{
FileId: vid,
Status: http.StatusBadRequest,
Error: err.Error()},
)
2014-04-15 16:09:40 +00:00
continue
}
for _, location := range result.Locations {
serverAddress := location.ServerAddress()
if _, ok := server_to_fileIds[serverAddress]; !ok {
server_to_fileIds[serverAddress] = make([]string, 0)
2014-04-15 16:09:40 +00:00
}
server_to_fileIds[serverAddress] = append(
server_to_fileIds[serverAddress], vid_to_fileIds[vid]...)
2014-04-15 16:09:40 +00:00
}
}
resultChan := make(chan []*volume_server_pb.DeleteResult, len(server_to_fileIds))
2014-04-15 16:09:40 +00:00
var wg sync.WaitGroup
for server, fidList := range server_to_fileIds {
wg.Add(1)
go func(server pb.ServerAddress, fidList []string) {
2014-04-15 16:09:40 +00:00
defer wg.Done()
2021-06-13 23:16:11 +00:00
if deleteResults, deleteErr := DeleteFilesAtOneVolumeServer(server, grpcDialOption, fidList, false); deleteErr != nil {
err = deleteErr
} else if deleteResults != nil {
resultChan <- deleteResults
2014-04-15 16:09:40 +00:00
}
2014-04-15 16:09:40 +00:00
}(server, fidList)
}
wg.Wait()
close(resultChan)
for result := range resultChan {
ret = append(ret, result...)
}
return ret, err
}
// DeleteFilesAtOneVolumeServer deletes a list of files that is on one volume server via gRpc
func DeleteFilesAtOneVolumeServer(volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fileIds []string, includeCookie bool) (ret []*volume_server_pb.DeleteResult, err error) {
err = WithVolumeServerClient(false, volumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
req := &volume_server_pb.BatchDeleteRequest{
2020-03-25 09:21:15 +00:00
FileIds: fileIds,
SkipCookieCheck: !includeCookie,
}
resp, err := volumeServerClient.BatchDelete(context.Background(), req)
2018-11-18 19:51:38 +00:00
// fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)
if err != nil {
return err
}
ret = append(ret, resp.Results...)
return nil
})
if err != nil {
return
}
for _, result := range ret {
2019-01-06 03:52:38 +00:00
if result.Error != "" && result.Error != "not found" {
return nil, fmt.Errorf("delete fileId %s: %v", result.FileId, result.Error)
}
}
return
2014-04-15 16:09:40 +00:00
}