[#427] layer: Split FrostFS ReadObject to separate methods

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>

api/handler/encryption_test.go

@@ -37,7 +37,7 @@ func TestSimpleGetEncrypted(t *testing.T) {
 
 	objInfo, err := tc.Layer().GetObjectInfo(tc.Context(), &layer.HeadObjectParams{BktInfo: bktInfo, Object: objName})
 	require.NoError(t, err)
-	obj, err := tc.MockedPool().ReadObject(tc.Context(), layer.PrmObjectRead{Container: bktInfo.CID, Object: objInfo.ID})
+	obj, err := tc.MockedPool().GetObject(tc.Context(), layer.PrmObjectGet{Container: bktInfo.CID, Object: objInfo.ID})
 	require.NoError(t, err)
 	encryptedContent, err := io.ReadAll(obj.Payload)
 	require.NoError(t, err)

api/layer/frostfs.go

@@ -78,8 +78,32 @@ type PrmAuth struct {
 	PrivateKey *ecdsa.PrivateKey
 }
 
-// PrmObjectRead groups parameters of FrostFS.ReadObject operation.
+// PrmObjectHead groups parameters of FrostFS.HeadObject operation.
-type PrmObjectRead struct {
+type PrmObjectHead struct {
+	// Authentication parameters.
+	PrmAuth
+
+	// Container to read the object header from.
+	Container cid.ID
+
+	// ID of the object for which to read the header.
+	Object oid.ID
+}
+
+// PrmObjectGet groups parameters of FrostFS.GetObject operation.
+type PrmObjectGet struct {
+	// Authentication parameters.
+	PrmAuth
+
+	// Container to read the object header from.
+	Container cid.ID
+
+	// ID of the object for which to read the header.
+	Object oid.ID
+}
+
+// PrmObjectRange groups parameters of FrostFS.RangeObject operation.
+type PrmObjectRange struct {
 	// Authentication parameters.
 	PrmAuth
 
@@ -89,20 +113,14 @@ type PrmObjectRead struct {
 	// ID of the object for which to read the header.
 	Object oid.ID
 
-	// Flag to read object header.
-	WithHeader bool
-
-	// Flag to read object payload. False overlaps payload range.
-	WithPayload bool
-
 	// Offset-length range of the object payload to be read.
 	PayloadRange [2]uint64
 }
 
-// ObjectPart represents partially read FrostFS object.
+// Object represents full read FrostFS object.
-type ObjectPart struct {
+type Object struct {
-	// Object header with optional in-memory payload part.
+	// Object header (doesn't contain payload).
-	Head *object.Object
+	Header object.Object
 
 	// Object payload part encapsulated in io.Reader primitive.
 	// Returns ErrAccessDenied on read access violation.
@@ -213,13 +231,15 @@ type FrostFS interface {
 	// It returns any error encountered which prevented the removal request from being sent.
 	DeleteContainer(context.Context, cid.ID, *session.Container) error
 
-	// ReadObject reads a part of the object from the FrostFS container by identifier.
+	// HeadObject reads an info of the object from the FrostFS container by identifier.
-	// Exact part is returned according to the parameters:
-	//   * with header only: empty payload (both in-mem and reader parts are nil);
-	//   * with payload only: header is nil (zero range means full payload);
-	//   * with header and payload: full in-mem object, payload reader is nil.
 	//
-	// WithHeader or WithPayload is true. Range length is positive if offset is positive.
+	// It returns ErrAccessDenied on read access violation.
+	//
+	// It returns exactly one non-nil value. It returns any error encountered which
+	// prevented the object header from being read.
+	HeadObject(ctx context.Context, prm PrmObjectHead) (*object.Object, error)
+
+	// GetObject reads an object from the FrostFS container by identifier.
 	//
 	// Payload reader should be closed if it is no longer needed.
 	//
@@ -227,7 +247,17 @@ type FrostFS interface {
 	//
 	// It returns exactly one non-nil value. It returns any error encountered which
 	// prevented the object header from being read.
-	ReadObject(context.Context, PrmObjectRead) (*ObjectPart, error)
+	GetObject(ctx context.Context, prm PrmObjectGet) (*Object, error)
+
+	// RangeObject reads a part of object from the FrostFS container by identifier.
+	//
+	// Payload reader should be closed if it is no longer needed.
+	//
+	// It returns ErrAccessDenied on read access violation.
+	//
+	// It returns exactly one non-nil value. It returns any error encountered which
+	// prevented the object header from being read.
+	RangeObject(ctx context.Context, prm PrmObjectRange) (io.ReadCloser, error)
 
 	// CreateObject creates and saves a parameterized object in the FrostFS container.
 	// It sets 'Timestamp' attribute to the current time.

api/layer/frostfs_mock.go

@@ -204,7 +204,56 @@ func (t *TestFrostFS) UserContainers(context.Context, PrmUserContainers) ([]cid.
 	return res, nil
 }
 
-func (t *TestFrostFS) ReadObject(ctx context.Context, prm PrmObjectRead) (*ObjectPart, error) {
+func (t *TestFrostFS) HeadObject(ctx context.Context, prm PrmObjectHead) (*object.Object, error) {
+	var addr oid.Address
+	addr.SetContainer(prm.Container)
+	addr.SetObject(prm.Object)
+
+	sAddr := addr.EncodeToString()
+
+	if err := t.objectErrors[sAddr]; err != nil {
+		return nil, err
+	}
+
+	if obj, ok := t.objects[sAddr]; ok {
+		owner := getBearerOwner(ctx)
+		if !t.checkAccess(prm.Container, owner) {
+			return nil, ErrAccessDenied
+		}
+
+		return obj, nil
+	}
+
+	return nil, fmt.Errorf("%w: %s", &apistatus.ObjectNotFound{}, addr)
+}
+
+func (t *TestFrostFS) GetObject(ctx context.Context, prm PrmObjectGet) (*Object, error) {
+	var addr oid.Address
+	addr.SetContainer(prm.Container)
+	addr.SetObject(prm.Object)
+
+	sAddr := addr.EncodeToString()
+
+	if err := t.objectErrors[sAddr]; err != nil {
+		return nil, err
+	}
+
+	if obj, ok := t.objects[sAddr]; ok {
+		owner := getBearerOwner(ctx)
+		if !t.checkAccess(prm.Container, owner) {
+			return nil, ErrAccessDenied
+		}
+
+		return &Object{
+			Header:  *obj,
+			Payload: io.NopCloser(bytes.NewReader(obj.Payload())),
+		}, nil
+	}
+
+	return nil, fmt.Errorf("%w: %s", &apistatus.ObjectNotFound{}, addr)
+}
+
+func (t *TestFrostFS) RangeObject(ctx context.Context, prm PrmObjectRange) (io.ReadCloser, error) {
 	var addr oid.Address
 	addr.SetContainer(prm.Container)
 	addr.SetObject(prm.Object)
@@ -223,15 +272,10 @@ func (t *TestFrostFS) ReadObject(ctx context.Context, prm PrmObjectRead) (*Objec
 
 		payload := obj.Payload()
 
-		if prm.PayloadRange[0]+prm.PayloadRange[1] > 0 {
+		off := prm.PayloadRange[0]
-			off := prm.PayloadRange[0]
+		payload = payload[off : off+prm.PayloadRange[1]]
-			payload = payload[off : off+prm.PayloadRange[1]]
-		}
 
-		return &ObjectPart{
+		return io.NopCloser(bytes.NewReader(payload)), nil
-			Head:    obj,
-			Payload: io.NopCloser(bytes.NewReader(payload)),
-		}, nil
 	}
 
 	return nil, fmt.Errorf("%w: %s", &apistatus.ObjectNotFound{}, addr)

api/layer/layer.go

@@ -676,7 +676,7 @@ func (n *Layer) removeCombinedObject(ctx context.Context, bkt *data.BucketInfo,
 	}
 
 	var parts []*data.PartInfo
-	if err = json.Unmarshal(combinedObj.Payload(), &parts); err != nil {
+	if err = json.NewDecoder(combinedObj.Payload).Decode(&parts); err != nil {
 		return fmt.Errorf("unmarshal combined object parts: %w", err)
 	}
 

api/layer/object.go

@@ -68,20 +68,14 @@ func newAddress(cnr cid.ID, obj oid.ID) oid.Address {
 
 // objectHead returns all object's headers.
 func (n *Layer) objectHead(ctx context.Context, bktInfo *data.BucketInfo, idObj oid.ID) (*object.Object, error) {
-	prm := PrmObjectRead{
+	prm := PrmObjectHead{
-		Container:  bktInfo.CID,
+		Container: bktInfo.CID,
-		Object:     idObj,
+		Object:    idObj,
-		WithHeader: true,
 	}
 
 	n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner)
 
-	res, err := n.frostFS.ReadObject(ctx, prm)
+	return n.frostFS.HeadObject(ctx, prm)
-	if err != nil {
-		return nil, err
-	}
-
-	return res.Head, nil
 }
 
 func (n *Layer) initObjectPayloadReader(ctx context.Context, p getParams) (io.Reader, error) {
@@ -100,7 +94,7 @@ func (n *Layer) initObjectPayloadReader(ctx context.Context, p getParams) (io.Re
 	}
 
 	var parts []*data.PartInfo
-	if err = json.Unmarshal(combinedObj.Payload(), &parts); err != nil {
+	if err = json.NewDecoder(combinedObj.Payload).Decode(&parts); err != nil {
 		return nil, fmt.Errorf("unmarshal combined object parts: %w", err)
 	}
 
@@ -132,16 +126,27 @@ func (n *Layer) initObjectPayloadReader(ctx context.Context, p getParams) (io.Re
 // initializes payload reader of the FrostFS object.
 // Zero range corresponds to full payload (panics if only offset is set).
 func (n *Layer) initFrostFSObjectPayloadReader(ctx context.Context, p getFrostFSParams) (io.Reader, error) {
-	prm := PrmObjectRead{
+	var prmAuth PrmAuth
-		Container:    p.bktInfo.CID,
+	n.prepareAuthParameters(ctx, &prmAuth, p.bktInfo.Owner)
-		Object:       p.oid,
+
-		WithPayload:  true,
+	if p.off+p.ln != 0 {
-		PayloadRange: [2]uint64{p.off, p.ln},
+		prm := PrmObjectRange{
+			PrmAuth:      prmAuth,
+			Container:    p.bktInfo.CID,
+			Object:       p.oid,
+			PayloadRange: [2]uint64{p.off, p.ln},
+		}
+
+		return n.frostFS.RangeObject(ctx, prm)
 	}
 
-	n.prepareAuthParameters(ctx, &prm.PrmAuth, p.bktInfo.Owner)
+	prm := PrmObjectGet{
+		PrmAuth:   prmAuth,
+		Container: p.bktInfo.CID,
+		Object:    p.oid,
+	}
 
-	res, err := n.frostFS.ReadObject(ctx, prm)
+	res, err := n.frostFS.GetObject(ctx, prm)
 	if err != nil {
 		return nil, err
 	}
@@ -149,25 +154,6 @@ func (n *Layer) initFrostFSObjectPayloadReader(ctx context.Context, p getFrostFS
 	return res.Payload, nil
 }
 
-// objectGet returns an object with payload in the object.
-func (n *Layer) objectGet(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) (*object.Object, error) {
-	prm := PrmObjectRead{
-		Container:   bktInfo.CID,
-		Object:      objID,
-		WithHeader:  true,
-		WithPayload: true,
-	}
-
-	n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner)
-
-	res, err := n.frostFS.ReadObject(ctx, prm)
-	if err != nil {
-		return nil, err
-	}
-
-	return res.Head, nil
-}
-
 // MimeByFilePath detect mime type by file path extension.
 func MimeByFilePath(path string) string {
 	ext := filepath.Ext(path)
@@ -177,6 +163,18 @@ func MimeByFilePath(path string) string {
 	return mime.TypeByExtension(ext)
 }
 
+// objectGet returns an object with payload in the object.
+func (n *Layer) objectGet(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) (*Object, error) {
+	prm := PrmObjectGet{
+		Container: bktInfo.CID,
+		Object:    objID,
+	}
+
+	n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner)
+
+	return n.frostFS.GetObject(ctx, prm)
+}
+
 func encryptionReader(r io.Reader, size uint64, key []byte) (io.Reader, uint64, error) {
 	encSize, err := sio.EncryptedSize(size)
 	if err != nil {

api/layer/system_object.go

@@ -175,8 +175,7 @@ func (n *Layer) getCORS(ctx context.Context, bkt *data.BucketInfo) (*data.CORSCo
 	}
 
 	cors := &data.CORSConfiguration{}
-
+	if err = xml.NewDecoder(obj.Payload).Decode(&cors); err != nil {
-	if err = xml.Unmarshal(obj.Payload(), &cors); err != nil {
 		return nil, fmt.Errorf("unmarshal cors: %w", err)
 	}
 

cmd/s3-authmate/modules/utils.go

@@ -50,7 +50,7 @@ func createFrostFS(ctx context.Context, log *zap.Logger, cfg PoolConfig) (*frost
 		return nil, fmt.Errorf("dial pool: %w", err)
 	}
 
-	return frostfs.NewAuthmateFrostFS(frostfs.NewFrostFS(p, cfg.Key)), nil
+	return frostfs.NewAuthmateFrostFS(frostfs.NewFrostFS(p, cfg.Key), log), nil
 }
 
 func parsePolicies(val string) (authmate.ContainerPolicies, error) {

cmd/s3-gw/app.go

@@ -121,7 +121,7 @@ func newApp(ctx context.Context, log *Logger, v *viper.Viper) *App {
 	objPool, treePool, key := getPools(ctx, log.logger, v)
 
 	cfg := tokens.Config{
-		FrostFS:                     frostfs.NewAuthmateFrostFS(frostfs.NewFrostFS(objPool, key)),
+		FrostFS:                     frostfs.NewAuthmateFrostFS(frostfs.NewFrostFS(objPool, key), log.logger),
 		Key:                         key,
 		CacheConfig:                 getAccessBoxCacheConfig(v, log.logger),
 		RemovingCheckAfterDurations: fetchRemovingCheckInterval(v, log.logger),

creds/tokens/credentials.go

@@ -92,6 +92,7 @@ type FrostFS interface {
 	//
 	// It returns exactly one non-nil value. It returns any error encountered which
 	// prevented the object payload from being read.
+	// Object must contain full payload.
 	GetCredsObject(context.Context, oid.Address) (*object.Object, error)
 }
 

internal/frostfs/authmate.go

@@ -4,11 +4,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"strconv"
 	"time"
 
 	objectv2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
+	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware"
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/authmate"
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/tokens"
 	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/crdt"
@@ -16,6 +18,7 @@ import (
 	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
 	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
+	"go.uber.org/zap"
 )
 
 const (
@@ -25,11 +28,12 @@ const (
 // AuthmateFrostFS is a mediator which implements authmate.FrostFS through pool.Pool.
 type AuthmateFrostFS struct {
 	frostFS layer.FrostFS
+	log     *zap.Logger
 }
 
 // NewAuthmateFrostFS creates new AuthmateFrostFS using provided pool.Pool.
-func NewAuthmateFrostFS(frostFS layer.FrostFS) *AuthmateFrostFS {
+func NewAuthmateFrostFS(frostFS layer.FrostFS, log *zap.Logger) *AuthmateFrostFS {
-	return &AuthmateFrostFS{frostFS: frostFS}
+	return &AuthmateFrostFS{frostFS: frostFS, log: log}
 }
 
 // ContainerExists implements authmate.FrostFS interface method.
@@ -79,17 +83,27 @@ func (x *AuthmateFrostFS) GetCredsObject(ctx context.Context, addr oid.Address)
 		credObjID = last.ObjID
 	}
 
-	res, err := x.frostFS.ReadObject(ctx, layer.PrmObjectRead{
+	res, err := x.frostFS.GetObject(ctx, layer.PrmObjectGet{
-		Container:   addr.Container(),
+		Container: addr.Container(),
-		Object:      credObjID,
+		Object:    credObjID,
-		WithPayload: true,
-		WithHeader:  true,
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	return res.Head, err
+	defer func() {
+		if closeErr := res.Payload.Close(); closeErr != nil {
+			x.reqLogger(ctx).Warn("close creds object payload", zap.Error(closeErr))
+		}
+	}()
+
+	data, err := io.ReadAll(res.Payload)
+	if err != nil {
+		return nil, err
+	}
+	res.Header.SetPayload(data)
+
+	return &res.Header, err
 }
 
 // CreateObject implements authmate.FrostFS interface method.
@@ -143,21 +157,28 @@ func (x *AuthmateFrostFS) getCredVersions(ctx context.Context, addr oid.Address)
 	versions := crdt.NewObjectVersions(objCredSystemName)
 
 	for _, id := range credVersions {
-		objVersion, err := x.frostFS.ReadObject(ctx, layer.PrmObjectRead{
+		objVersion, err := x.frostFS.HeadObject(ctx, layer.PrmObjectHead{
-			Container:  addr.Container(),
+			Container: addr.Container(),
-			Object:     id,
+			Object:    id,
-			WithHeader: true,
 		})
 		if err != nil {
 			return nil, fmt.Errorf("head crdt access box '%s': %w", id.EncodeToString(), err)
 		}
 
-		versions.AppendVersion(crdt.NewObjectVersion(objVersion.Head))
+		versions.AppendVersion(crdt.NewObjectVersion(objVersion))
 	}
 
 	return versions, nil
 }
 
+func (x *AuthmateFrostFS) reqLogger(ctx context.Context) *zap.Logger {
+	reqLogger := middleware.GetReqLog(ctx)
+	if reqLogger != nil {
+		return reqLogger
+	}
+	return x.log
+}
+
 func credVersionSysName(cnrID cid.ID, objID oid.ID) string {
 	return cnrID.EncodeToString() + "0" + objID.EncodeToString()
 }

internal/frostfs/authmate_test.go

@@ -14,6 +14,7 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
 )
 
 func TestGetCredsObject(t *testing.T) {
@@ -35,7 +36,7 @@ func TestGetCredsObject(t *testing.T) {
 		},
 	}})
 
-	frostfs := NewAuthmateFrostFS(layer.NewTestFrostFS(key))
+	frostfs := NewAuthmateFrostFS(layer.NewTestFrostFS(key), zaptest.NewLogger(t))
 
 	cid, err := frostfs.CreateContainer(ctx, authmate.PrmContainerCreate{
 		FriendlyName: bktName,

internal/frostfs/frostfs.go

@@ -255,8 +255,31 @@ func (x payloadReader) Read(p []byte) (int, error) {
 	return n, handleObjectError("read payload", err)
 }
 
-// ReadObject implements frostfs.FrostFS interface method.
+// HeadObject implements frostfs.FrostFS interface method.
-func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*layer.ObjectPart, error) {
+func (x *FrostFS) HeadObject(ctx context.Context, prm layer.PrmObjectHead) (*object.Object, error) {
+	var addr oid.Address
+	addr.SetContainer(prm.Container)
+	addr.SetObject(prm.Object)
+
+	var prmHead pool.PrmObjectHead
+	prmHead.SetAddress(addr)
+
+	if prm.BearerToken != nil {
+		prmHead.UseBearer(*prm.BearerToken)
+	} else {
+		prmHead.UseKey(prm.PrivateKey)
+	}
+
+	res, err := x.pool.HeadObject(ctx, prmHead)
+	if err != nil {
+		return nil, handleObjectError("read object header via connection pool", err)
+	}
+
+	return &res, nil
+}
+
+// GetObject implements frostfs.FrostFS interface method.
+func (x *FrostFS) GetObject(ctx context.Context, prm layer.PrmObjectGet) (*layer.Object, error) {
 	var addr oid.Address
 	addr.SetContainer(prm.Container)
 	addr.SetObject(prm.Object)
@@ -270,55 +293,23 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
 		prmGet.UseKey(prm.PrivateKey)
 	}
 
-	if prm.WithHeader {
+	res, err := x.pool.GetObject(ctx, prmGet)
-		if prm.WithPayload {
+	if err != nil {
-			res, err := x.pool.GetObject(ctx, prmGet)
+		return nil, handleObjectError("init full object reading via connection pool", err)
-			if err != nil {
-				return nil, handleObjectError("init full object reading via connection pool", err)
-			}
-
-			defer res.Payload.Close()
-
-			payload, err := io.ReadAll(res.Payload)
-			if err != nil {
-				return nil, handleObjectError("read full object payload", err)
-			}
-
-			res.Header.SetPayload(payload)
-
-			return &layer.ObjectPart{
-				Head: &res.Header,
-			}, nil
-		}
-
-		var prmHead pool.PrmObjectHead
-		prmHead.SetAddress(addr)
-
-		if prm.BearerToken != nil {
-			prmHead.UseBearer(*prm.BearerToken)
-		} else {
-			prmHead.UseKey(prm.PrivateKey)
-		}
-
-		hdr, err := x.pool.HeadObject(ctx, prmHead)
-		if err != nil {
-			return nil, handleObjectError("read object header via connection pool", err)
-		}
-
-		return &layer.ObjectPart{
-			Head: &hdr,
-		}, nil
-	} else if prm.PayloadRange[0]+prm.PayloadRange[1] == 0 {
-		res, err := x.pool.GetObject(ctx, prmGet)
-		if err != nil {
-			return nil, handleObjectError("init full payload range reading via connection pool", err)
-		}
-
-		return &layer.ObjectPart{
-			Payload: res.Payload,
-		}, nil
 	}
 
+	return &layer.Object{
+		Header:  res.Header,
+		Payload: res.Payload,
+	}, nil
+}
+
+// RangeObject implements frostfs.FrostFS interface method.
+func (x *FrostFS) RangeObject(ctx context.Context, prm layer.PrmObjectRange) (io.ReadCloser, error) {
+	var addr oid.Address
+	addr.SetContainer(prm.Container)
+	addr.SetObject(prm.Object)
+
 	var prmRange pool.PrmObjectRange
 	prmRange.SetAddress(addr)
 	prmRange.SetOffset(prm.PayloadRange[0])
@@ -335,9 +326,7 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
 		return nil, handleObjectError("init payload range reading via connection pool", err)
 	}
 
-	return &layer.ObjectPart{
+	return payloadReader{&res}, nil
-		Payload: payloadReader{&res},
-	}, nil
 }
 
 // DeleteObject implements frostfs.FrostFS interface method.