gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

batch delete EC needles

Browse source 
fix https://github.com/seaweedfs/seaweedfs/issues/4107
This commit is contained in:
chrislu 2023-01-03 22:05:26 -08:00
parent 8aec430df7
commit 7bdae5172e
1 changed files with 46 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
weed/server/volume_grpc_batch_delete.go
View file

@ -28,6 +28,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
n := new(needle.Needle) n := new(needle.Needle)
volumeId, _ := needle.NewVolumeId(vid) volumeId, _ := needle.NewVolumeId(vid)
ecVolume, isEcVolume := vs.store.FindEcVolume(volumeId)
if req.SkipCookieCheck { if req.SkipCookieCheck {
n.Id, _, err = needle.ParseNeedleIdCookie(id_cookie) n.Id, _, err = needle.ParseNeedleIdCookie(id_cookie)
if err != nil { if err != nil {
@ -40,6 +41,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
} else { } else {
n.ParsePath(id_cookie) n.ParsePath(id_cookie)
cookie := n.Cookie cookie := n.Cookie
if !isEcVolume {
if _, err := vs.store.ReadVolumeNeedle(volumeId, n, nil, nil); err != nil { if _, err := vs.store.ReadVolumeNeedle(volumeId, n, nil, nil); err != nil {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid, FileId: fid,
@ -48,6 +50,16 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
}) })
continue continue
} }
} else {
if _, err := vs.store.ReadEcShardNeedle(volumeId, n, nil); err != nil {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid,
Status: http.StatusNotFound,
Error: err.Error(),
})
continue
}
}
if n.Cookie != cookie { if n.Cookie != cookie {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid, FileId: fid,
@ -68,6 +80,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
} }
n.LastModified = now n.LastModified = now
if !isEcVolume {
if size, err := vs.store.DeleteVolumeNeedle(volumeId, n); err != nil { if size, err := vs.store.DeleteVolumeNeedle(volumeId, n); err != nil {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid, FileId: fid,
@ -81,6 +94,21 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
Size: uint32(size)}, Size: uint32(size)},
) )
} }
} else {
if size, err := vs.store.DeleteEcShardNeedle(ecVolume, n, n.Cookie); err != nil {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid,
Status: http.StatusInternalServerError,
Error: err.Error()},
)
} else {
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid,
Status: http.StatusAccepted,
Size: uint32(size)},
)
}
}
} }
return resp, nil return resp, nil