TrueCloudLab/frostfs-http-gw
17
1
Fork 18
Code Issues 8 Pull requests Projects Releases 16 Packages 1 Activity Actions

[#142] Fix multipart-objects download
All checks were successful
/ DCO (pull_request) Successful in 43s
Details
/ Vulncheck (pull_request) Successful in 1m24s
Details
/ Builds (pull_request) Successful in 1m45s
Details
/ Lint (pull_request) Successful in 2m53s
Details
/ Tests (pull_request) Successful in 1m37s
Details

Browse source 
Signed-off-by: Nikita Zinkevich <n.zinkevich@yadro.com>
This commit is contained in:
Nikita Zinkevich 2024-09-18 07:35:26 +03:00
parent 843708a558
commit 33dba5d68c
16 changed files with 803 additions and 230 deletions
Download patch file Download diff file Expand all files Collapse all files

11
internal/handler/download.go
View file

@ -11,6 +11,7 @@ import (
"time"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
@ -45,13 +46,13 @@ func (h *Handler) DownloadByAttribute(c *fasthttp.RequestCtx) {
h.byAttribute(c, h.receiveFile)
}
func (h *Handler) search(ctx context.Context, cnrID *cid.ID, key, val string, op object.SearchMatchType) (ResObjectSearch, error) {
func (h *Handler) search(ctx context.Context, cnrID *cid.ID, key, val string, op object.SearchMatchType) (frostfs.ResObjectSearch, error) {
filters := object.NewSearchFilters()
filters.AddRootFilter()
filters.AddFilter(key, val, op)
prm := PrmObjectSearch{
PrmAuth: PrmAuth{
prm := frostfs.PrmObjectSearch{
PrmAuth: frostfs.PrmAuth{
BearerToken: bearerToken(ctx),
},
Container: *cnrID,
@ -153,8 +154,8 @@ func (h *Handler) DownloadZipped(c *fasthttp.RequestCtx) {
}
func (h *Handler) zipObject(ctx context.Context, zipWriter *zip.Writer, addr oid.Address, btoken *bearer.Token, bufZip []byte) error {
prm := PrmObjectGet{
PrmAuth: PrmAuth{
prm := frostfs.PrmObjectGet{
PrmAuth: frostfs.PrmAuth{
BearerToken: btoken,
},
Address: addr,

21
internal/handler/frostfs_mock.go
View file

@ -9,6 +9,7 @@ import (
"io"
"strings"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/checksum"
@ -58,7 +59,7 @@ func (t *TestFrostFS) AllowUserOperation(cnrID cid.ID, userID user.ID, op acl.Op
t.accessList[fmt.Sprintf("%s/%s/%s/%s", cnrID, userID, op, objID)] = true
}
func (t *TestFrostFS) Container(_ context.Context, prm PrmContainer) (*container.Container, error) {
func (t *TestFrostFS) Container(_ context.Context, prm frostfs.PrmContainer) (*container.Container, error) {
for k, v := range t.containers {
if k == prm.ContainerID.EncodeToString() {
return v, nil
@ -85,7 +86,7 @@ func (t *TestFrostFS) retrieveObject(addr oid.Address, btoken *bearer.Token) (*o
owner := t.requestOwner(btoken)
if !t.isAllowed(addr.Container(), owner, acl.OpObjectGet, addr.Object()) {
return nil, ErrAccessDenied
return nil, frostfs.ErrAccessDenied
}
return obj, nil
@ -94,23 +95,23 @@ func (t *TestFrostFS) retrieveObject(addr oid.Address, btoken *bearer.Token) (*o
return nil, fmt.Errorf("%w: %s", &apistatus.ObjectNotFound{}, addr)
}
func (t *TestFrostFS) HeadObject(_ context.Context, prm PrmObjectHead) (*object.Object, error) {
func (t *TestFrostFS) HeadObject(_ context.Context, prm frostfs.PrmObjectHead) (*object.Object, error) {
return t.retrieveObject(prm.Address, prm.BearerToken)
}
func (t *TestFrostFS) GetObject(_ context.Context, prm PrmObjectGet) (*Object, error) {
func (t *TestFrostFS) GetObject(_ context.Context, prm frostfs.PrmObjectGet) (*frostfs.Object, error) {
obj, err := t.retrieveObject(prm.Address, prm.BearerToken)
if err != nil {
return nil, err
}
return &Object{
return &frostfs.Object{
Header: *obj,
Payload: io.NopCloser(bytes.NewReader(obj.Payload())),
}, nil
}
func (t *TestFrostFS) RangeObject(_ context.Context, prm PrmObjectRange) (io.ReadCloser, error) {
func (t *TestFrostFS) RangeObject(_ context.Context, prm frostfs.PrmObjectRange) (io.ReadCloser, error) {
obj, err := t.retrieveObject(prm.Address, prm.BearerToken)
if err != nil {
return nil, err
@ -121,7 +122,7 @@ func (t *TestFrostFS) RangeObject(_ context.Context, prm PrmObjectRange) (io.Rea
return io.NopCloser(bytes.NewReader(payload)), nil
}
func (t *TestFrostFS) CreateObject(_ context.Context, prm PrmObjectCreate) (oid.ID, error) {
func (t *TestFrostFS) CreateObject(_ context.Context, prm frostfs.PrmObjectCreate) (oid.ID, error) {
b := make([]byte, 32)
if _, err := io.ReadFull(rand.Reader, b); err != nil {
return oid.ID{}, err
@ -158,7 +159,7 @@ func (t *TestFrostFS) CreateObject(_ context.Context, prm PrmObjectCreate) (oid.
owner := t.requestOwner(prm.BearerToken)
if !t.isAllowed(cnrID, owner, acl.OpObjectPut, objID) {
return oid.ID{}, ErrAccessDenied
return oid.ID{}, frostfs.ErrAccessDenied
}
addr := newAddress(cnrID, objID)
@ -195,9 +196,9 @@ func (r *resObjectSearchMock) Iterate(f func(oid.ID) bool) error {
func (r *resObjectSearchMock) Close() {}
func (t *TestFrostFS) SearchObjects(_ context.Context, prm PrmObjectSearch) (ResObjectSearch, error) {
func (t *TestFrostFS) SearchObjects(_ context.Context, prm frostfs.PrmObjectSearch) (frostfs.ResObjectSearch, error) {
if !t.isAllowed(prm.Container, t.requestOwner(prm.BearerToken), acl.OpObjectSearch, oid.ID{}) {
return nil, ErrAccessDenied
return nil, frostfs.ErrAccessDenied
}
cidStr := prm.Container.EncodeToString()

124
internal/handler/handler.go
View file

@ -12,10 +12,10 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/handler/middleware"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tree"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
@ -34,110 +34,14 @@ type Config interface {
NamespaceHeader() string
}
// PrmContainer groups parameters of FrostFS.Container operation.
type PrmContainer struct {
// Container identifier.
ContainerID cid.ID
}
// PrmAuth groups authentication parameters for the FrostFS operation.
type PrmAuth struct {
// Bearer token to be used for the operation. Overlaps PrivateKey. Optional.
BearerToken *bearer.Token
}
// PrmObjectHead groups parameters of FrostFS.HeadObject operation.
type PrmObjectHead struct {
// Authentication parameters.
PrmAuth
// Address to read the object header from.
Address oid.Address
}
// PrmObjectGet groups parameters of FrostFS.GetObject operation.
type PrmObjectGet struct {
// Authentication parameters.
PrmAuth
// Address to read the object header from.
Address oid.Address
}
// PrmObjectRange groups parameters of FrostFS.RangeObject operation.
type PrmObjectRange struct {
// Authentication parameters.
PrmAuth
// Address to read the object header from.
Address oid.Address
// Offset-length range of the object payload to be read.
PayloadRange [2]uint64
}
// Object represents FrostFS object.
type Object struct {
// Object header (doesn't contain payload).
Header object.Object
// Object payload part encapsulated in io.Reader primitive.
// Returns ErrAccessDenied on read access violation.
Payload io.ReadCloser
}
// PrmObjectCreate groups parameters of FrostFS.CreateObject operation.
type PrmObjectCreate struct {
// Authentication parameters.
PrmAuth
Object *object.Object
// Object payload encapsulated in io.Reader primitive.
Payload io.Reader
// Enables client side object preparing.
ClientCut bool
// Disables using Tillich-Zémor hash for payload.
WithoutHomomorphicHash bool
// Sets max buffer size to read payload.
BufferMaxSize uint64
}
// PrmObjectSearch groups parameters of FrostFS.sear SearchObjects operation.
type PrmObjectSearch struct {
// Authentication parameters.
PrmAuth
// Container to select the objects from.
Container cid.ID
Filters object.SearchFilters
}
type ResObjectSearch interface {
Read(buf []oid.ID) (int, error)
Iterate(f func(oid.ID) bool) error
Close()
}
var (
// ErrAccessDenied is returned from FrostFS in case of access violation.
ErrAccessDenied = errors.New("access denied")
// ErrGatewayTimeout is returned from FrostFS in case of timeout, deadline exceeded etc.
ErrGatewayTimeout = errors.New("gateway timeout")
)
// FrostFS represents virtual connection to FrostFS network.
type FrostFS interface {
Container(context.Context, PrmContainer) (*container.Container, error)
HeadObject(context.Context, PrmObjectHead) (*object.Object, error)
GetObject(context.Context, PrmObjectGet) (*Object, error)
RangeObject(context.Context, PrmObjectRange) (io.ReadCloser, error)
CreateObject(context.Context, PrmObjectCreate) (oid.ID, error)
SearchObjects(context.Context, PrmObjectSearch) (ResObjectSearch, error)
Container(context.Context, frostfs.PrmContainer) (*container.Container, error)
HeadObject(context.Context, frostfs.PrmObjectHead) (*object.Object, error)
GetObject(context.Context, frostfs.PrmObjectGet) (*frostfs.Object, error)
RangeObject(context.Context, frostfs.PrmObjectRange) (io.ReadCloser, error)
CreateObject(context.Context, frostfs.PrmObjectCreate) (oid.ID, error)
SearchObjects(context.Context, frostfs.PrmObjectSearch) (frostfs.ResObjectSearch, error)
utils.EpochInfoFetcher
}
@ -177,7 +81,7 @@ func New(params *AppParams, config Config, tree *tree.Tree) *Handler {
// byAddress is a wrapper for function (e.g. request.headObject, request.receiveFile) that
// prepares request and object address to it.
func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address, *data.BucketInfo)) {
var (
idCnr, _ = c.UserValue("cid").(string)
idObj, _ = c.UserValue("oid").(string)
@ -203,12 +107,12 @@ func (h *Handler) byAddress(c *fasthttp.RequestCtx, f func(context.Context, requ
addr.SetContainer(bktInfo.CID)
addr.SetObject(*objID)
f(ctx, *h.newRequest(c, log), addr)
f(ctx, *h.newRequest(c, log), addr, bktInfo)
}
// byObjectName is a wrapper for function (e.g. request.headObject, request.receiveFile) that
// prepares request and object address to it.
func (h *Handler) byObjectName(req *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
func (h *Handler) byObjectName(req *fasthttp.RequestCtx, f func(context.Context, request, oid.Address, *data.BucketInfo)) {
var (
bucketname = req.UserValue("cid").(string)
key = req.UserValue("oid").(string)
@ -250,11 +154,11 @@ func (h *Handler) byObjectName(req *fasthttp.RequestCtx, f func(context.Context,
addr.SetContainer(bktInfo.CID)
addr.SetObject(foundOid.OID)
f(ctx, *h.newRequest(req, log), addr)
f(ctx, *h.newRequest(req, log), addr, bktInfo)
}
// byAttribute is a wrapper similar to byAddress.
func (h *Handler) byAttribute(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address)) {
func (h *Handler) byAttribute(c *fasthttp.RequestCtx, f func(context.Context, request, oid.Address, *data.BucketInfo)) {
scid, _ := c.UserValue("cid").(string)
key, _ := c.UserValue("attr_key").(string)
val, _ := c.UserValue("attr_val").(string)
@ -311,7 +215,7 @@ func (h *Handler) byAttribute(c *fasthttp.RequestCtx, f func(context.Context, re
addrObj.SetContainer(bktInfo.CID)
addrObj.SetObject(buf[0])
f(ctx, *h.newRequest(c, log), addrObj)
f(ctx, *h.newRequest(c, log), addrObj, bktInfo)
}
// resolveContainer decode container id, if it's not a valid container id
@ -359,7 +263,7 @@ func (h *Handler) getBucketInfo(ctx context.Context, containerName string, log *
}
func (h *Handler) readContainer(ctx context.Context, cnrID cid.ID) (*data.BucketInfo, error) {
prm := PrmContainer{ContainerID: cnrID}
prm := frostfs.PrmContainer{ContainerID: cnrID}
res, err := h.frostfs.Container(ctx, prm)
if err != nil {
return nil, fmt.Errorf("get frostfs container '%s': %w", cnrID.String(), err)

12
internal/handler/head.go
View file

@ -7,7 +7,9 @@ import (
"strconv"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
@ -24,13 +26,13 @@ const (
hdrContainerID = "X-Container-Id"
)
func (h *Handler) headObject(ctx context.Context, req request, objectAddress oid.Address) {
func (h *Handler) headObject(ctx context.Context, req request, objectAddress oid.Address, _ *data.BucketInfo) {
var start = time.Now()
btoken := bearerToken(ctx)
prm := PrmObjectHead{
PrmAuth: PrmAuth{
prm := frostfs.PrmObjectHead{
PrmAuth: frostfs.PrmAuth{
BearerToken: btoken,
},
Address: objectAddress,
@ -74,8 +76,8 @@ func (h *Handler) headObject(ctx context.Context, req request, objectAddress oid
if len(contentType) == 0 {
contentType, _, err = readContentType(obj.PayloadSize(), func(sz uint64) (io.Reader, error) {
prmRange := PrmObjectRange{
PrmAuth: PrmAuth{
prmRange := frostfs.PrmObjectRange{
PrmAuth: frostfs.PrmAuth{
BearerToken: btoken,
},
Address: objectAddress,

159
internal/handler/multipart.go
View file

@ -1,13 +1,30 @@
package handler
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"strconv"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/handler/multipart"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"github.com/minio/sio"
"go.uber.org/zap"
)
const (
FrostFSSystemMetadataPrefix = "S3-"
AttributeEncryptionAlgorithm = FrostFSSystemMetadataPrefix + "Algorithm"
AttributeHMACSalt = FrostFSSystemMetadataPrefix + "HMAC-Salt"
AttributeHMACKey = FrostFSSystemMetadataPrefix + "HMAC-Key"
AttributeMultipartObjectSize = FrostFSSystemMetadataPrefix + "Multipart-Object-Size"
)
// MultipartFile provides standard ReadCloser interface and also allows one to
// get file name, it's used for multipart uploads.
type MultipartFile interface {
@ -45,3 +62,145 @@ func fetchMultipartFile(l *zap.Logger, r io.Reader, boundary string) (MultipartF
return part, nil
}
}
type getParams struct {
// payload range
off, ln uint64
objInfo *data.ObjectInfo
bktInfo *data.BucketInfo
}
// getPayload returns initial payload if object is not multipart else composes new reader with parts data.
func (h *Handler) getPayload(p getPayloadParams) (io.ReadCloser, uint64, error) {
sizeValue, ok := p.attrs[AttributeMultipartObjectSize]
if !ok {
return p.obj.Payload, p.obj.Header.PayloadSize(), nil
}
cid, ok := p.obj.Header.ContainerID()
if !ok {
return nil, 0, errors.New("no container id set")
}
oid, ok := p.obj.Header.ID()
if !ok {
return nil, 0, errors.New("no object id set")
}
size, err := strconv.ParseUint(sizeValue, 10, 64)
if err != nil {
return nil, 0, err
}
ctx := p.req.RequestCtx
params := getParams{
off: 0,
ln: 0,
objInfo: &data.ObjectInfo{
ID: oid,
CID: cid,
Bucket: p.bktinfo.Name,
Name: p.attrs["FilePath"],
Size: size,
Headers: p.attrs,
},
bktInfo: p.bktinfo,
}
payload, err := h.initMultipartReader(ctx, params)
if err != nil {
return nil, 0, err
}
return io.NopCloser(payload), size, nil
}
func (h *Handler) initMultipartReader(ctx context.Context, p getParams) (io.Reader, error) {
combinedObj, err := h.frostfs.GetObject(ctx, frostfs.PrmObjectGet{
PrmAuth: frostfs.PrmAuth{BearerToken: bearerToken(ctx)},
Address: p.objInfo.Address(),
})
if err != nil {
return nil, fmt.Errorf("get combined object '%s': %w", p.objInfo.ID.EncodeToString(), err)
}
var parts []*data.PartInfo
if err = json.NewDecoder(combinedObj.Payload).Decode(&parts); err != nil {
return nil, fmt.Errorf("unmarshal combined object parts: %w", err)
}
isEncrypted := FormEncryptionInfo(p.objInfo.Headers).Enabled
objParts := make([]frostfs.PartObj, len(parts))
for i, part := range parts {
size := part.Size
if isEncrypted {
if size, err = sio.EncryptedSize(part.Size); err != nil {
return nil, fmt.Errorf("compute encrypted size: %w", err)
}
}
objParts[i] = frostfs.PartObj{
OID: part.OID,
Size: size,
}
}
return frostfs.NewMultiObjectReader(ctx, frostfs.MultiObjectReaderConfig{
Handler: h,
Off: p.off,
Ln: p.ln,
Parts: objParts,
BktInfo: p.bktInfo,
Log: h.log,
})
}
// InitFrostFSObjectPayloadReader initializes payload reader of the FrostFS object.
// Zero range corresponds to full payload (panics if only offset is set).
func (h *Handler) InitFrostFSObjectPayloadReader(ctx context.Context, p frostfs.GetFrostFSParams) (io.Reader, error) {
var prmAuth frostfs.PrmAuth
var addr oid.Address
addr.SetContainer(p.BktInfo.CID)
addr.SetObject(p.Oid)
if p.Off+p.Ln != 0 {
prm := frostfs.PrmObjectRange{
PrmAuth: prmAuth,
Container: p.BktInfo.CID,
Object: p.Oid,
PayloadRange: [2]uint64{p.Off, p.Ln},
Address: addr,
}
return h.frostfs.RangeObject(ctx, prm)
}
prm := frostfs.PrmObjectGet{
PrmAuth: prmAuth,
Container: p.BktInfo.CID,
Object: p.Oid,
Address: addr,
}
res, err := h.frostfs.GetObject(ctx, prm)
if err != nil {
return nil, err
}
return res.Payload, nil
}
// ObjectEncryption stores parsed object encryption headers.
type ObjectEncryption struct {
Enabled bool
Algorithm string
HMACKey string
HMACSalt string
}
func FormEncryptionInfo(headers map[string]string) ObjectEncryption {
algorithm := headers[AttributeEncryptionAlgorithm]
return ObjectEncryption{
Enabled: len(algorithm) > 0,
Algorithm: algorithm,
HMACKey: headers[AttributeHMACKey],
HMACSalt: headers[AttributeHMACSalt],
}
}

104
internal/handler/reader.go
View file

@ -5,11 +5,12 @@ import (
"context"
"io"
"net/http"
"path"
"strconv"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
@ -47,19 +48,24 @@ func readContentType(maxSize uint64, rInit func(uint64) (io.Reader, error)) (str
return http.DetectContentType(buf), buf, err // to not lose io.EOF
}
func (h *Handler) receiveFile(ctx context.Context, req request, objectAddress oid.Address) {
type getPayloadParams struct {
obj *frostfs.Object
req request
bktinfo *data.BucketInfo
attrs map[string]string
}
func (h *Handler) receiveFile(ctx context.Context, req request, objAddress oid.Address, bktInfo *data.BucketInfo) {
var (
err error
dis = "inline"
start = time.Now()
filename string
shouldDownload = req.QueryArgs().GetBool("download")
start = time.Now()
)
prm := PrmObjectGet{
PrmAuth: PrmAuth{
prm := frostfs.PrmObjectGet{
PrmAuth: frostfs.PrmAuth{
BearerToken: bearerToken(ctx),
},
Address: objectAddress,
Address: objAddress,
}
rObj, err := h.frostfs.GetObject(ctx, prm)
@ -70,51 +76,40 @@ func (h *Handler) receiveFile(ctx context.Context, req request, objectAddress oi
// we can't close reader in this function, so how to do it?
if req.Request.URI().QueryArgs().GetBool("download") {
dis = "attachment"
attrs := makeAttributesMap(rObj.Header.Attributes())
req.setAttributes(attrs)
req.setIDs(rObj.Header)
req.setDisposition(shouldDownload, attrs[object.AttributeFileName])
if err = req.setTimestamp(attrs[object.AttributeTimestamp]); err != nil {
req.log.Error(logs.CouldntParseCreationDate,
zap.String("val", attrs[object.AttributeTimestamp]),
zap.Error(err))
response.Error(req.RequestCtx, "failed to convert timestamp: "+err.Error(), fasthttp.StatusInternalServerError)
}
payloadSize := rObj.Header.PayloadSize()
payloadParams := getPayloadParams{
obj: rObj,
req: req,
bktinfo: bktInfo,
attrs: attrs,
}
payload, payloadSize, err := h.getPayload(payloadParams)
if err != nil {
req.handleFrostFSErr(err, start)
return
}
req.Response.Header.Set(fasthttp.HeaderContentLength, strconv.FormatUint(payloadSize, 10))
var contentType string
for _, attr := range rObj.Header.Attributes() {
key := attr.Key()
val := attr.Value()
if !isValidToken(key) || !isValidValue(val) {
continue
}
key = utils.BackwardTransformIfSystem(key)
req.Response.Header.Set(utils.UserAttributeHeaderPrefix+key, val)
switch key {
case object.AttributeFileName:
filename = val
case object.AttributeTimestamp:
value, err := strconv.ParseInt(val, 10, 64)
if err != nil {
req.log.Info(logs.CouldntParseCreationDate,
zap.String("key", key),
zap.String("val", val),
zap.Error(err))
continue
}
req.Response.Header.Set(fasthttp.HeaderLastModified,
time.Unix(value, 0).UTC().Format(http.TimeFormat))
case object.AttributeContentType:
contentType = val
}
}
idsToResponse(&req.Response, &rObj.Header)
contentType := attrs[object.AttributeContentType]
if len(contentType) == 0 {
// determine the Content-Type from the payload head
var payloadHead []byte
contentType, payloadHead, err = readContentType(payloadSize, func(uint64) (io.Reader, error) {
return rObj.Payload, nil
return payload, nil
})
if err != nil && err != io.EOF {
req.log.Error(logs.CouldNotDetectContentTypeFromPayload, zap.Error(err))
@ -126,16 +121,27 @@ func (h *Handler) receiveFile(ctx context.Context, req request, objectAddress oi
var headReader io.Reader = bytes.NewReader(payloadHead)
if err != io.EOF { // otherwise, we've already read full payload
headReader = io.MultiReader(headReader, rObj.Payload)
headReader = io.MultiReader(headReader, payload)
}
// note: we could do with io.Reader, but SetBodyStream below closes body stream
// if it implements io.Closer and that's useful for us.
rObj.Payload = readCloser{headReader, rObj.Payload}
payload = readCloser{headReader, payload}
}
req.SetContentType(contentType)
req.Response.Header.Set(fasthttp.HeaderContentDisposition, dis+"; filename="+path.Base(filename))
req.Response.SetBodyStream(rObj.Payload, int(payloadSize))
req.Response.SetBodyStream(payload, int(payloadSize))
}
func makeAttributesMap(attrs []object.Attribute) map[string]string {
attributes := make(map[string]string)
for _, attr := range attrs {
if !isValidToken(attr.Key()) || !isValidValue(attr.Value()) {
continue
}
key := utils.BackwardTransformIfSystem(attr.Key())
attributes[key] = attr.Value()
}
return attributes
}

5
internal/handler/upload.go
View file

@ -9,6 +9,7 @@ import (
"time"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/service/frostfs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tokens"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
@ -131,8 +132,8 @@ func (h *Handler) Upload(req *fasthttp.RequestCtx) {
obj.SetOwnerID(*h.ownerID)
obj.SetAttributes(attributes...)
prm := PrmObjectCreate{
PrmAuth: PrmAuth{
prm := frostfs.PrmObjectCreate{
PrmAuth: frostfs.PrmAuth{
BearerToken: h.fetchBearerToken(ctx),
},
Object: obj,

42
internal/handler/utils.go
View file

@ -2,14 +2,19 @@ package handler
import (
"context"
"net/http"
"path"
"strconv"
"strings"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/response"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tokens"
"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
"github.com/valyala/fasthttp"
"go.uber.org/zap"
)
@ -31,6 +36,43 @@ func (r *request) handleFrostFSErr(err error, start time.Time) {
response.Error(r.RequestCtx, msg, statusCode)
}
func (r *request) setAttributes(attrs map[string]string) {
for key, val := range attrs {
r.Response.Header.Set(utils.UserAttributeHeaderPrefix+key, val)
}
}
func (r *request) setIDs(obj object.Object) {
objID, _ := obj.ID()
cnrID, _ := obj.ContainerID()
r.Response.Header.Set(hdrObjectID, objID.String())
r.Response.Header.Set(hdrOwnerID, obj.OwnerID().String())
r.Response.Header.Set(hdrContainerID, cnrID.String())
}
func (r *request) setDisposition(shouldDownload bool, filename string) {
const (
inlineDisposition = "inline"
attachmentDisposition = "attachment"
)
dis := inlineDisposition
if shouldDownload {
dis = attachmentDisposition
}
r.Response.Header.Set(fasthttp.HeaderContentDisposition, dis+"; filename="+path.Base(filename))
}
func (r *request) setTimestamp(timestamp string) error {
value, err := strconv.ParseInt(timestamp, 10, 64)
if err != nil {
return err
}
r.Response.Header.Set(fasthttp.HeaderLastModified,
time.Unix(value, 0).UTC().Format(http.TimeFormat))
return nil
}
func bearerToken(ctx context.Context) *bearer.Token {
if tkn, err := tokens.LoadBearerToken(ctx); err == nil {
return tkn