-
Notifications
You must be signed in to change notification settings - Fork 342
/
deleteobject.go
65 lines (57 loc) · 2.27 KB
/
deleteobject.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
package operations
import (
"errors"
"net/http"
"github.com/treeverse/lakefs/pkg/auth"
"github.com/treeverse/lakefs/pkg/block"
"github.com/treeverse/lakefs/pkg/catalog"
gatewayerrors "github.com/treeverse/lakefs/pkg/gateway/errors"
"github.com/treeverse/lakefs/pkg/graveler"
"github.com/treeverse/lakefs/pkg/logging"
"github.com/treeverse/lakefs/pkg/permissions"
)
type DeleteObject struct{}
func (controller *DeleteObject) RequiredPermissions(_ *http.Request, repoID, _, path string) (auth.PermissionNode, error) {
return &auth.OnePermission{
Action: permissions.DeleteObjectAction,
Resource: permissions.ObjectArn(repoID, path),
}, nil
}
func (controller *DeleteObject) HandleAbortMultipartUpload(w http.ResponseWriter, req *http.Request, o *PathOperation) {
o.Incr("abort_mpu")
query := req.URL.Query()
uploadID := query.Get(QueryParamUploadID)
req = req.WithContext(logging.AddFields(req.Context(), logging.Fields{logging.UploadIDFieldKey: uploadID}))
err := o.BlockStore.AbortMultiPartUpload(req.Context(), block.ObjectPointer{StorageNamespace: o.Repository.StorageNamespace, Identifier: o.Path}, uploadID)
if err != nil {
o.Log(req).WithError(err).Error("could not abort multipart upload")
_ = o.EncodeError(w, req, gatewayerrors.Codes.ToAPIErr(gatewayerrors.ErrInternalError))
return
}
// done.
w.WriteHeader(http.StatusNoContent)
}
func (controller *DeleteObject) Handle(w http.ResponseWriter, req *http.Request, o *PathOperation) {
query := req.URL.Query()
_, hasUploadID := query[QueryParamUploadID]
if hasUploadID {
controller.HandleAbortMultipartUpload(w, req, o)
return
}
o.Incr("delete_object")
lg := o.Log(req).WithField("key", o.Path)
err := o.Catalog.DeleteEntry(req.Context(), o.Repository.Name, o.Reference, o.Path)
switch {
case errors.Is(err, catalog.ErrNotFound):
lg.WithError(err).Debug("could not delete object, it doesn't exist")
case errors.Is(err, graveler.ErrWriteToProtectedBranch):
_ = o.EncodeError(w, req, gatewayerrors.Codes.ToAPIErr(gatewayerrors.ErrWriteToProtectedBranch))
case err != nil:
lg.WithError(err).Error("could not delete object")
_ = o.EncodeError(w, req, gatewayerrors.Codes.ToAPIErr(gatewayerrors.ErrInternalError))
return
default:
lg.Debug("object set for deletion")
}
w.WriteHeader(http.StatusNoContent)
}