frostfs-node/pkg/local_object_storage/engine/head.go

package engine

import (
	"context"
	"errors"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
)

// HeadPrm groups the parameters of Head operation.
type HeadPrm struct {
	addr oid.Address
	raw  bool
}

// HeadRes groups the resulting values of Head operation.
type HeadRes struct {
	head *objectSDK.Object
}

// WithAddress is a Head option to set the address of the requested object.
//
// Option is required.
func (p *HeadPrm) WithAddress(addr oid.Address) {
	p.addr = addr
}

// WithRaw is a Head option to set raw flag value. If flag is unset, then Head
// returns the header of the virtual object, otherwise it returns SplitInfo of the virtual
// object.
func (p *HeadPrm) WithRaw(raw bool) {
	p.raw = raw
}

// Header returns the requested object header.
//
// Instance has empty payload.
func (r HeadRes) Header() *objectSDK.Object {
	return r.head
}

// Head reads object header from local storage.
//
// Returns any error encountered that
// did not allow to completely read the object header.
//
// Returns an error of type apistatus.ObjectNotFound if the requested object is missing in local storage.
// Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object was inhumed.
//
// Returns an error if executions are blocked (see BlockExecution).
func (e *StorageEngine) Head(ctx context.Context, prm HeadPrm) (res HeadRes, err error) {
	err = e.execIfNotBlocked(func() error {
		res, err = e.head(ctx, prm)
		return err
	})

	return
}

func (e *StorageEngine) head(ctx context.Context, prm HeadPrm) (HeadRes, error) {
	ctx, span := tracing.StartSpanFromContext(ctx, "StorageEngine.head")
	defer span.End()
	if e.metrics != nil {
		defer elapsed("Head", e.metrics.AddMethodDuration)()
	}

	var (
		head     *objectSDK.Object
		siErr    *objectSDK.SplitInfoError
		outSI    *objectSDK.SplitInfo
		eiErr    *objectSDK.ECInfoError
		outEI    *objectSDK.ECInfo
		outError error = new(apistatus.ObjectNotFound)
		shPrm    shard.HeadPrm
	)
	shPrm.SetAddress(prm.addr)
	shPrm.SetRaw(prm.raw)

	e.iterateOverSortedShards(prm.addr, func(_ int, sh hashedShard) (stop bool) {
		shPrm.ShardLooksBad = sh.errorCount.Load() >= e.errorsThreshold
		res, err := sh.Head(ctx, shPrm)
		if err != nil {
			switch {
			case client.IsErrObjectNotFound(err):
				return false // ignore, go to next shard
			case errors.As(err, &siErr):
				if outSI == nil {
					outSI = objectSDK.NewSplitInfo()
				}
				util.MergeSplitInfo(siErr.SplitInfo(), outSI)
				_, withLink := outSI.Link()
				_, withLast := outSI.LastPart()
				// stop iterating over shards if SplitInfo structure is complete
				if withLink && withLast {
					return true
				}
				return false
			case errors.As(err, &eiErr):
				if outEI == nil {
					outEI = objectSDK.NewECInfo()
				}
				util.MergeECInfo(eiErr.ECInfo(), outEI)
				// stop iterating over shards if ECInfo structure is complete
				return len(outEI.Chunks) == int(outEI.Chunks[0].Total)
			case client.IsErrObjectAlreadyRemoved(err):
				outError = err
				return true // stop, return it back
			case shard.IsErrObjectExpired(err):
				// object is found but should not
				// be returned
				outError = new(apistatus.ObjectNotFound)
				return true
			default:
				e.reportShardError(sh, "could not head object from shard", err)
				return false
			}
		}
		head = res.Object()
		return true
	})

	if outSI != nil {
		return HeadRes{}, logicerr.Wrap(objectSDK.NewSplitInfoError(outSI))
	} else if outEI != nil {
		return HeadRes{}, logicerr.Wrap(objectSDK.NewECInfoError(outEI))
	} else if head == nil {
		return HeadRes{}, outError
	}

	return HeadRes{
		head: head,
	}, nil
}

// Head reads object header from local storage by provided address.
func Head(ctx context.Context, storage *StorageEngine, addr oid.Address) (*objectSDK.Object, error) {
	var headPrm HeadPrm
	headPrm.WithAddress(addr)

	res, err := storage.Head(ctx, headPrm)
	if err != nil {
		return nil, err
	}

	return res.Header(), nil
}