404 lines
13 KiB
Go
404 lines
13 KiB
Go
// Copyright (C) 2019 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package metainfo
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/zeebo/errs"
|
|
"go.uber.org/zap"
|
|
|
|
"storj.io/common/pb"
|
|
"storj.io/common/storj"
|
|
)
|
|
|
|
// Batch handle requests sent in batch.
|
|
func (endpoint *Endpoint) Batch(ctx context.Context, req *pb.BatchRequest) (resp *pb.BatchResponse, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
resp = &pb.BatchResponse{}
|
|
|
|
resp.Responses = make([]*pb.BatchResponseItem, 0, len(req.Requests))
|
|
|
|
var lastStreamID storj.StreamID
|
|
var lastSegmentID storj.SegmentID
|
|
var prevSegmentReq *pb.BatchRequestItem
|
|
for i, request := range req.Requests {
|
|
switch singleRequest := request.Request.(type) {
|
|
// BUCKET
|
|
case *pb.BatchRequestItem_BucketCreate:
|
|
singleRequest.BucketCreate.Header = req.Header
|
|
response, err := endpoint.CreateBucket(ctx, singleRequest.BucketCreate)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketCreate{
|
|
BucketCreate: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketGet:
|
|
singleRequest.BucketGet.Header = req.Header
|
|
response, err := endpoint.GetBucket(ctx, singleRequest.BucketGet)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketGet{
|
|
BucketGet: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketGetLocation:
|
|
singleRequest.BucketGetLocation.Header = req.Header
|
|
response, err := endpoint.GetBucketLocation(ctx, singleRequest.BucketGetLocation)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketGetLocation{
|
|
BucketGetLocation: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketGetVersioning:
|
|
singleRequest.BucketGetVersioning.Header = req.Header
|
|
response, err := endpoint.GetBucketVersioning(ctx, singleRequest.BucketGetVersioning)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketGetVersioning{
|
|
BucketGetVersioning: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketSetVersioning:
|
|
singleRequest.BucketSetVersioning.Header = req.Header
|
|
response, err := endpoint.SetBucketVersioning(ctx, singleRequest.BucketSetVersioning)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketSetVersioning{
|
|
BucketSetVersioning: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketDelete:
|
|
singleRequest.BucketDelete.Header = req.Header
|
|
response, err := endpoint.DeleteBucket(ctx, singleRequest.BucketDelete)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketDelete{
|
|
BucketDelete: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_BucketList:
|
|
singleRequest.BucketList.Header = req.Header
|
|
response, err := endpoint.ListBuckets(ctx, singleRequest.BucketList)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_BucketList{
|
|
BucketList: response,
|
|
},
|
|
})
|
|
|
|
// OBJECT
|
|
case *pb.BatchRequestItem_ObjectBegin:
|
|
singleRequest.ObjectBegin.Header = req.Header
|
|
response, err := endpoint.BeginObject(ctx, singleRequest.ObjectBegin)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectBegin{
|
|
ObjectBegin: response,
|
|
},
|
|
})
|
|
if response != nil {
|
|
lastStreamID = response.StreamId
|
|
}
|
|
case *pb.BatchRequestItem_ObjectCommit:
|
|
singleRequest.ObjectCommit.Header = req.Header
|
|
|
|
if singleRequest.ObjectCommit.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.ObjectCommit.StreamId = lastStreamID
|
|
}
|
|
|
|
var response *pb.ObjectCommitResponse
|
|
var err error
|
|
switch {
|
|
case prevSegmentReq.GetSegmentMakeInline() != nil:
|
|
segmentResp, segmentErr := endpoint.MakeInlineSegment(ctx, prevSegmentReq.GetSegmentMakeInline())
|
|
prevSegmentReq = nil
|
|
if segmentErr != nil {
|
|
return resp, segmentErr
|
|
}
|
|
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentMakeInline{
|
|
SegmentMakeInline: segmentResp,
|
|
},
|
|
})
|
|
response, err = endpoint.CommitObject(ctx, singleRequest.ObjectCommit)
|
|
case prevSegmentReq.GetSegmentCommit() != nil:
|
|
segmentResp, segmentErr := endpoint.CommitSegment(ctx, prevSegmentReq.GetSegmentCommit())
|
|
prevSegmentReq = nil
|
|
if segmentErr != nil {
|
|
return resp, segmentErr
|
|
}
|
|
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentCommit{
|
|
SegmentCommit: segmentResp,
|
|
},
|
|
})
|
|
response, err = endpoint.CommitObject(ctx, singleRequest.ObjectCommit)
|
|
default:
|
|
response, err = endpoint.CommitObject(ctx, singleRequest.ObjectCommit)
|
|
}
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectCommit{
|
|
ObjectCommit: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_ObjectGet:
|
|
singleRequest.ObjectGet.Header = req.Header
|
|
response, err := endpoint.GetObject(ctx, singleRequest.ObjectGet)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectGet{
|
|
ObjectGet: response,
|
|
},
|
|
})
|
|
if response != nil && response.Object != nil {
|
|
lastStreamID = response.Object.StreamId
|
|
}
|
|
case *pb.BatchRequestItem_ObjectList:
|
|
singleRequest.ObjectList.Header = req.Header
|
|
response, err := endpoint.ListObjects(ctx, singleRequest.ObjectList)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectList{
|
|
ObjectList: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_ObjectBeginDelete:
|
|
singleRequest.ObjectBeginDelete.Header = req.Header
|
|
response, err := endpoint.BeginDeleteObject(ctx, singleRequest.ObjectBeginDelete)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectBeginDelete{
|
|
ObjectBeginDelete: response,
|
|
},
|
|
})
|
|
if response != nil {
|
|
lastStreamID = response.StreamId
|
|
}
|
|
case *pb.BatchRequestItem_ObjectFinishDelete:
|
|
singleRequest.ObjectFinishDelete.Header = req.Header
|
|
|
|
if singleRequest.ObjectFinishDelete.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.ObjectFinishDelete.StreamId = lastStreamID
|
|
}
|
|
|
|
response, err := endpoint.FinishDeleteObject(ctx, singleRequest.ObjectFinishDelete)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_ObjectFinishDelete{
|
|
ObjectFinishDelete: response,
|
|
},
|
|
})
|
|
// SEGMENT
|
|
case *pb.BatchRequestItem_SegmentBegin:
|
|
singleRequest.SegmentBegin.Header = req.Header
|
|
|
|
justCreatedObject := false
|
|
if singleRequest.SegmentBegin.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.SegmentBegin.StreamId = lastStreamID
|
|
justCreatedObject = true
|
|
}
|
|
|
|
response, err := endpoint.beginSegment(ctx, singleRequest.SegmentBegin, justCreatedObject)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentBegin{
|
|
SegmentBegin: response,
|
|
},
|
|
})
|
|
if response != nil {
|
|
lastSegmentID = response.SegmentId
|
|
}
|
|
case *pb.BatchRequestItem_SegmentCommit:
|
|
singleRequest.SegmentCommit.Header = req.Header
|
|
|
|
if singleRequest.SegmentCommit.SegmentId.IsZero() && !lastSegmentID.IsZero() {
|
|
singleRequest.SegmentCommit.SegmentId = lastSegmentID
|
|
}
|
|
|
|
segmentID, err := endpoint.unmarshalSatSegmentID(ctx, singleRequest.SegmentCommit.SegmentId)
|
|
if err != nil {
|
|
endpoint.log.Error("unable to unmarshal segment id", zap.Error(err))
|
|
} else if endpoint.shouldCombine(segmentID.Index, i, req.Requests) {
|
|
prevSegmentReq = request
|
|
continue
|
|
}
|
|
|
|
response, err := endpoint.CommitSegment(ctx, singleRequest.SegmentCommit)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentCommit{
|
|
SegmentCommit: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_SegmentList:
|
|
singleRequest.SegmentList.Header = req.Header
|
|
|
|
if singleRequest.SegmentList.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.SegmentList.StreamId = lastStreamID
|
|
}
|
|
|
|
response, err := endpoint.ListSegments(ctx, singleRequest.SegmentList)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentList{
|
|
SegmentList: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_SegmentMakeInline:
|
|
singleRequest.SegmentMakeInline.Header = req.Header
|
|
|
|
if singleRequest.SegmentMakeInline.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.SegmentMakeInline.StreamId = lastStreamID
|
|
}
|
|
|
|
if endpoint.shouldCombine(singleRequest.SegmentMakeInline.Position.Index, i, req.Requests) {
|
|
prevSegmentReq = request
|
|
continue
|
|
}
|
|
|
|
response, err := endpoint.MakeInlineSegment(ctx, singleRequest.SegmentMakeInline)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentMakeInline{
|
|
SegmentMakeInline: response,
|
|
},
|
|
})
|
|
case *pb.BatchRequestItem_SegmentDownload:
|
|
singleRequest.SegmentDownload.Header = req.Header
|
|
|
|
if singleRequest.SegmentDownload.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.SegmentDownload.StreamId = lastStreamID
|
|
}
|
|
|
|
response, err := endpoint.DownloadSegment(ctx, singleRequest.SegmentDownload)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentDownload{
|
|
SegmentDownload: response,
|
|
},
|
|
})
|
|
if response != nil {
|
|
lastSegmentID = response.SegmentId
|
|
}
|
|
case *pb.BatchRequestItem_SegmentBeginDelete:
|
|
singleRequest.SegmentBeginDelete.Header = req.Header
|
|
|
|
if singleRequest.SegmentBeginDelete.StreamId.IsZero() && !lastStreamID.IsZero() {
|
|
singleRequest.SegmentBeginDelete.StreamId = lastStreamID
|
|
}
|
|
|
|
response, err := endpoint.BeginDeleteSegment(ctx, singleRequest.SegmentBeginDelete)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentBeginDelete{
|
|
SegmentBeginDelete: response,
|
|
},
|
|
})
|
|
if response != nil {
|
|
lastSegmentID = response.SegmentId
|
|
}
|
|
case *pb.BatchRequestItem_SegmentFinishDelete:
|
|
singleRequest.SegmentFinishDelete.Header = req.Header
|
|
|
|
if singleRequest.SegmentFinishDelete.SegmentId.IsZero() && !lastSegmentID.IsZero() {
|
|
singleRequest.SegmentFinishDelete.SegmentId = lastSegmentID
|
|
}
|
|
|
|
response, err := endpoint.FinishDeleteSegment(ctx, singleRequest.SegmentFinishDelete)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_SegmentFinishDelete{
|
|
SegmentFinishDelete: response,
|
|
},
|
|
})
|
|
|
|
// Revoke API key.
|
|
case *pb.BatchRequestItem_RevokeApiKey:
|
|
singleRequest.RevokeApiKey.Header = req.Header
|
|
response, err := endpoint.RevokeAPIKey(ctx, singleRequest.RevokeApiKey)
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
resp.Responses = append(resp.Responses, &pb.BatchResponseItem{
|
|
Response: &pb.BatchResponseItem_RevokeApiKey{
|
|
RevokeApiKey: response,
|
|
},
|
|
})
|
|
default:
|
|
return nil, errs.New("unsupported request type")
|
|
}
|
|
}
|
|
|
|
return resp, nil
|
|
}
|
|
|
|
// shouldCombine returns true if we are able to combine current request with next one. Main case is
|
|
// combining CommitSegment/MakeInlineSegment with ObjectCommmit.
|
|
//
|
|
// This method has a workaround for a bug in uplink where ObjectCommit was batched with
|
|
// segment N-2 instead of N-1 if list segment was inline segment. We are checking that
|
|
// current request segment index is last one before 'l' segment and we are not combining otherwise.
|
|
func (endpoint *Endpoint) shouldCombine(segmentIndex int32, reqIndex int, requests []*pb.BatchRequestItem) bool {
|
|
if reqIndex < len(requests)-1 && requests[reqIndex+1].GetObjectCommit() != nil {
|
|
objCommitReq := requests[reqIndex+1].GetObjectCommit()
|
|
|
|
streamMeta := pb.StreamMeta{}
|
|
err := pb.Unmarshal(objCommitReq.EncryptedMetadata, &streamMeta)
|
|
if err != nil {
|
|
endpoint.log.Error("unable to unmarshal stream meta", zap.Error(err))
|
|
return false
|
|
}
|
|
|
|
return int64(segmentIndex) != streamMeta.NumberOfSegments-2
|
|
}
|
|
return false
|
|
}
|