frostfs-node/cmd/frostfs-cli/modules/object/util.go
Dmitrii Stepanov 654384990c [#1120] cli: Fix object nodes for linking objects
Do not use linking objects to get placement for complex object.
Linking objects should be stored on all container nodes, also they are not required.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2024-05-08 15:23:57 +03:00

508 lines
16 KiB
Go

package object
import (
"crypto/ecdsa"
"errors"
"fmt"
"os"
"strings"
internal "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-cli/internal/client"
"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-cli/internal/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-cli/internal/commonflags"
sessionCli "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-cli/modules/session"
commonCmd "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/internal/common"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
"github.com/spf13/cobra"
"github.com/spf13/viper"
)
const (
bearerTokenFlag = "bearer"
rawFlag = "raw"
rawFlagDesc = "Set raw request option"
fileFlag = "file"
binaryFlag = "binary"
)
type RPCParameters interface {
SetBearerToken(prm *bearer.Token)
SetTTL(uint32)
SetXHeaders([]string)
}
// InitBearer adds bearer token flag to a command.
func InitBearer(cmd *cobra.Command) {
flags := cmd.Flags()
flags.String(bearerTokenFlag, "", "File with signed JSON or binary encoded bearer token")
}
// Prepare prepares object-related parameters for a command.
func Prepare(cmd *cobra.Command, prms ...RPCParameters) {
ttl := viper.GetUint32(commonflags.TTL)
common.PrintVerbose(cmd, "TTL: %d", ttl)
for i := range prms {
btok := common.ReadBearerToken(cmd, bearerTokenFlag)
prms[i].SetBearerToken(btok)
prms[i].SetTTL(ttl)
prms[i].SetXHeaders(parseXHeaders(cmd))
}
}
func parseXHeaders(cmd *cobra.Command) []string {
xHeaders, _ := cmd.Flags().GetStringSlice(commonflags.XHeadersKey)
xs := make([]string, 0, 2*len(xHeaders))
for i := range xHeaders {
k, v, found := strings.Cut(xHeaders[i], "=")
if !found {
panic(fmt.Errorf("invalid X-Header format: %s", xHeaders[i]))
}
xs = append(xs, k, v)
}
return xs
}
func readObjectAddress(cmd *cobra.Command, cnr *cid.ID, obj *oid.ID) oid.Address {
readCID(cmd, cnr)
readOID(cmd, obj)
var addr oid.Address
addr.SetContainer(*cnr)
addr.SetObject(*obj)
return addr
}
func readObjectAddressBin(cmd *cobra.Command, cnr *cid.ID, obj *oid.ID, filename string) oid.Address {
buf, err := os.ReadFile(filename)
commonCmd.ExitOnErr(cmd, "unable to read given file: %w", err)
objTemp := objectSDK.New()
commonCmd.ExitOnErr(cmd, "can't unmarshal object from given file: %w", objTemp.Unmarshal(buf))
var addr oid.Address
*cnr, _ = objTemp.ContainerID()
*obj, _ = objTemp.ID()
addr.SetContainer(*cnr)
addr.SetObject(*obj)
return addr
}
func readCID(cmd *cobra.Command, id *cid.ID) {
err := id.DecodeString(cmd.Flag(commonflags.CIDFlag).Value.String())
commonCmd.ExitOnErr(cmd, "decode container ID string: %w", err)
}
func readOID(cmd *cobra.Command, id *oid.ID) {
err := id.DecodeString(cmd.Flag(commonflags.OIDFlag).Value.String())
commonCmd.ExitOnErr(cmd, "decode object ID string: %w", err)
}
// SessionPrm is a common interface of object operation's input which supports
// sessions.
type SessionPrm interface {
SetSessionToken(*session.Object)
SetClient(*client.Client)
}
// forwards all parameters to _readVerifiedSession and object as nil.
func readSessionGlobal(cmd *cobra.Command, dst SessionPrm, key *ecdsa.PrivateKey, cnr cid.ID) {
_readVerifiedSession(cmd, dst, key, cnr, nil)
}
// forwards all parameters to _readVerifiedSession.
func readSession(cmd *cobra.Command, dst SessionPrm, key *ecdsa.PrivateKey, cnr cid.ID, obj oid.ID) {
_readVerifiedSession(cmd, dst, key, cnr, &obj)
}
// decodes session.Object from the file by path specified in the
// commonflags.SessionToken flag. Returns nil if flag is not set.
func getSession(cmd *cobra.Command) *session.Object {
common.PrintVerbose(cmd, "Trying to read session from the file...")
path, _ := cmd.Flags().GetString(commonflags.SessionToken)
if path == "" {
common.PrintVerbose(cmd, "File with session token is not provided.")
return nil
}
common.PrintVerbose(cmd, "Reading session from the file [%s]...", path)
var tok session.Object
err := common.ReadBinaryOrJSON(cmd, &tok, path)
commonCmd.ExitOnErr(cmd, "read session: %v", err)
return &tok
}
// decodes object session from JSON file from commonflags.SessionToken command
// flag if it is provided, and writes resulting session into the provided SessionPrm.
// Returns flag presence. Checks:
//
// - if session verb corresponds to given SessionPrm according to its type
// - relation to the given container
// - relation to the given object if non-nil
// - relation to the given private key used within the command
// - session signature
//
// SessionPrm MUST be one of:
//
// *internal.GetObjectPrm
// *internal.HeadObjectPrm
// *internal.SearchObjectsPrm
// *internal.PayloadRangePrm
// *internal.HashPayloadRangesPrm
func _readVerifiedSession(cmd *cobra.Command, dst SessionPrm, key *ecdsa.PrivateKey, cnr cid.ID, obj *oid.ID) {
var cmdVerb session.ObjectVerb
switch dst.(type) {
default:
panic(fmt.Sprintf("unsupported op parameters %T", dst))
case *internal.GetObjectPrm:
cmdVerb = session.VerbObjectGet
case *internal.HeadObjectPrm:
cmdVerb = session.VerbObjectHead
case *internal.SearchObjectsPrm:
cmdVerb = session.VerbObjectSearch
case *internal.PayloadRangePrm:
cmdVerb = session.VerbObjectRange
case *internal.HashPayloadRangesPrm:
cmdVerb = session.VerbObjectRangeHash
}
tok := getSession(cmd)
if tok == nil {
return
}
common.PrintVerbose(cmd, "Checking session correctness...")
switch false {
case tok.AssertContainer(cnr):
commonCmd.ExitOnErr(cmd, "", errors.New("unrelated container in the session"))
case obj == nil || tok.AssertObject(*obj):
commonCmd.ExitOnErr(cmd, "", errors.New("unrelated object in the session"))
case tok.AssertVerb(cmdVerb):
commonCmd.ExitOnErr(cmd, "", errors.New("wrong verb of the session"))
case tok.AssertAuthKey((*frostfsecdsa.PublicKey)(&key.PublicKey)):
commonCmd.ExitOnErr(cmd, "", errors.New("unrelated key in the session"))
case tok.VerifySignature():
commonCmd.ExitOnErr(cmd, "", errors.New("invalid signature of the session data"))
}
common.PrintVerbose(cmd, "Session is correct.")
dst.SetSessionToken(tok)
}
// ReadOrOpenSession opens client connection and calls ReadOrOpenSessionViaClient with it.
func ReadOrOpenSession(cmd *cobra.Command, dst SessionPrm, key *ecdsa.PrivateKey, cnr cid.ID, obj *oid.ID) {
cli := internal.GetSDKClientByFlag(cmd, key, commonflags.RPC)
ReadOrOpenSessionViaClient(cmd, dst, cli, key, cnr, obj)
}
// ReadOrOpenSessionViaClient tries to read session from the file specified in
// commonflags.SessionToken flag, finalizes structures of the decoded token
// and write the result into provided SessionPrm. If file is missing,
// ReadOrOpenSessionViaClient calls OpenSessionViaClient.
func ReadOrOpenSessionViaClient(cmd *cobra.Command, dst SessionPrm, cli *client.Client, key *ecdsa.PrivateKey, cnr cid.ID, obj *oid.ID) {
tok := getSession(cmd)
if tok == nil {
OpenSessionViaClient(cmd, dst, cli, key, cnr, obj)
return
}
var objs []oid.ID
if obj != nil {
objs = []oid.ID{*obj}
if _, ok := dst.(*internal.DeleteObjectPrm); ok {
common.PrintVerbose(cmd, "Collecting relatives of the removal object...")
objs = append(objs, collectObjectRelatives(cmd, cli, cnr, *obj)...)
}
}
finalizeSession(cmd, dst, tok, key, cnr, objs...)
dst.SetClient(cli)
}
// OpenSession opens client connection and calls OpenSessionViaClient with it.
func OpenSession(cmd *cobra.Command, dst SessionPrm, key *ecdsa.PrivateKey, cnr cid.ID, obj *oid.ID) {
cli := internal.GetSDKClientByFlag(cmd, key, commonflags.RPC)
OpenSessionViaClient(cmd, dst, cli, key, cnr, obj)
}
// OpenSessionViaClient opens object session with the remote node, finalizes
// structure of the session token and writes the result into the provided
// SessionPrm. Also writes provided client connection to the SessionPrm.
//
// SessionPrm MUST be one of:
//
// *internal.PutObjectPrm
// *internal.DeleteObjectPrm
//
// If provided SessionPrm is of type internal.DeleteObjectPrm, OpenSessionViaClient
// spreads the session to all object's relatives.
func OpenSessionViaClient(cmd *cobra.Command, dst SessionPrm, cli *client.Client, key *ecdsa.PrivateKey, cnr cid.ID, obj *oid.ID) {
var objs []oid.ID
if obj != nil {
if _, ok := dst.(*internal.DeleteObjectPrm); ok {
common.PrintVerbose(cmd, "Collecting relatives of the removal object...")
rels := collectObjectRelatives(cmd, cli, cnr, *obj)
if len(rels) == 0 {
objs = []oid.ID{*obj}
} else {
objs = append(rels, *obj)
}
}
}
var tok session.Object
const sessionLifetime = 10 // in FrostFS epochs
common.PrintVerbose(cmd, "Opening remote session with the node...")
err := sessionCli.CreateSession(cmd.Context(), &tok, cli, sessionLifetime)
commonCmd.ExitOnErr(cmd, "open remote session: %w", err)
common.PrintVerbose(cmd, "Session successfully opened.")
finalizeSession(cmd, dst, &tok, key, cnr, objs...)
dst.SetClient(cli)
}
// specifies session verb, binds the session to the given container and limits
// the session by the given objects (if specified). After all data is written,
// signs session using provided private key and writes the session into the
// given SessionPrm.
//
// SessionPrm MUST be one of:
//
// *internal.PutObjectPrm
// *internal.DeleteObjectPrm
func finalizeSession(cmd *cobra.Command, dst SessionPrm, tok *session.Object, key *ecdsa.PrivateKey, cnr cid.ID, objs ...oid.ID) {
common.PrintVerbose(cmd, "Finalizing session token...")
switch dst.(type) {
default:
panic(fmt.Sprintf("unsupported op parameters %T", dst))
case *internal.PutObjectPrm:
common.PrintVerbose(cmd, "Binding session to object PUT...")
tok.ForVerb(session.VerbObjectPut)
case *internal.DeleteObjectPrm:
common.PrintVerbose(cmd, "Binding session to object DELETE...")
tok.ForVerb(session.VerbObjectDelete)
}
common.PrintVerbose(cmd, "Binding session to container %s...", cnr)
tok.BindContainer(cnr)
if len(objs) > 0 {
common.PrintVerbose(cmd, "Limiting session by the objects %v...", objs)
tok.LimitByObjects(objs...)
}
common.PrintVerbose(cmd, "Signing session...")
err := tok.Sign(*key)
commonCmd.ExitOnErr(cmd, "sign session: %w", err)
common.PrintVerbose(cmd, "Session token successfully formed and attached to the request.")
dst.SetSessionToken(tok)
}
// calls commonflags.InitSession with "object <verb>" name.
func initFlagSession(cmd *cobra.Command, verb string) {
commonflags.InitSession(cmd, "object "+verb)
}
// collects and returns all relatives of the given object stored in the specified
// container. Empty result without an error means lack of relationship in the
// container.
//
// The object itself is not included in the result.
func collectObjectRelatives(cmd *cobra.Command, cli *client.Client, cnr cid.ID, obj oid.ID) []oid.ID {
common.PrintVerbose(cmd, "Fetching raw object header...")
// request raw header first
var addrObj oid.Address
addrObj.SetContainer(cnr)
addrObj.SetObject(obj)
var prmHead internal.HeadObjectPrm
prmHead.SetClient(cli)
prmHead.SetAddress(addrObj)
prmHead.SetRawFlag(true)
Prepare(cmd, &prmHead)
_, err := internal.HeadObject(cmd.Context(), prmHead)
var errSplit *objectSDK.SplitInfoError
switch {
default:
commonCmd.ExitOnErr(cmd, "failed to get raw object header: %w", err)
case err == nil:
common.PrintVerbose(cmd, "Raw header received - object is singular.")
return nil
case errors.As(err, &errSplit):
common.PrintVerbose(cmd, "Split information received - object is virtual.")
}
splitInfo := errSplit.SplitInfo()
if members, ok := tryGetSplitMembersByLinkingObject(cmd, splitInfo, prmHead, cnr, true); ok {
return members
}
if members, ok := tryGetSplitMembersBySplitID(cmd, splitInfo, cli, cnr); ok {
return members
}
return tryRestoreChainInReverse(cmd, splitInfo, prmHead, cli, cnr, obj)
}
func tryGetSplitMembersByLinkingObject(cmd *cobra.Command, splitInfo *objectSDK.SplitInfo, prmHead internal.HeadObjectPrm, cnr cid.ID, withLinking bool) ([]oid.ID, bool) {
// collect split chain by the descending ease of operations (ease is evaluated heuristically).
// If any approach fails, we don't try the next since we assume that it will fail too.
if idLinking, ok := splitInfo.Link(); ok {
common.PrintVerbose(cmd, "Collecting split members using linking object %s...", idLinking)
var addrObj oid.Address
addrObj.SetContainer(cnr)
addrObj.SetObject(idLinking)
prmHead.SetAddress(addrObj)
prmHead.SetRawFlag(false)
// client is already set
res, err := internal.HeadObject(cmd.Context(), prmHead)
if err == nil {
children := res.Header().Children()
common.PrintVerbose(cmd, "Received split members from the linking object: %v", children)
if withLinking {
return append(children, idLinking), true
}
return children, true
}
// linking object is not required for
// object collecting
common.PrintVerbose(cmd, "failed to get linking object's header: %w", err)
}
return nil, false
}
func tryGetSplitMembersBySplitID(cmd *cobra.Command, splitInfo *objectSDK.SplitInfo, cli *client.Client, cnr cid.ID) ([]oid.ID, bool) {
if idSplit := splitInfo.SplitID(); idSplit != nil {
common.PrintVerbose(cmd, "Collecting split members by split ID...")
var query objectSDK.SearchFilters
query.AddSplitIDFilter(objectSDK.MatchStringEqual, idSplit)
var prm internal.SearchObjectsPrm
prm.SetContainerID(cnr)
prm.SetClient(cli)
prm.SetFilters(query)
res, err := internal.SearchObjects(cmd.Context(), prm)
commonCmd.ExitOnErr(cmd, "failed to search objects by split ID: %w", err)
parts := res.IDList()
common.PrintVerbose(cmd, "Found objects by split ID: %v", res.IDList())
return parts, true
}
return nil, false
}
func tryRestoreChainInReverse(cmd *cobra.Command, splitInfo *objectSDK.SplitInfo, prmHead internal.HeadObjectPrm, cli *client.Client, cnr cid.ID, obj oid.ID) []oid.ID {
var addrObj oid.Address
addrObj.SetContainer(cnr)
idMember, ok := splitInfo.LastPart()
if !ok {
commonCmd.ExitOnErr(cmd, "", errors.New("missing any data in received object split information"))
}
common.PrintVerbose(cmd, "Traverse the object split chain in reverse...", idMember)
var res *internal.HeadObjectRes
var err error
chain := []oid.ID{idMember}
chainSet := map[oid.ID]struct{}{idMember: {}}
prmHead.SetRawFlag(false)
// split members are almost definitely singular, but don't get hung up on it
for {
common.PrintVerbose(cmd, "Reading previous element of the split chain member %s...", idMember)
addrObj.SetObject(idMember)
prmHead.SetAddress(addrObj)
res, err = internal.HeadObject(cmd.Context(), prmHead)
commonCmd.ExitOnErr(cmd, "failed to read split chain member's header: %w", err)
idMember, ok = res.Header().PreviousID()
if !ok {
common.PrintVerbose(cmd, "Chain ended.")
break
}
if _, ok = chainSet[idMember]; ok {
commonCmd.ExitOnErr(cmd, "", fmt.Errorf("duplicated member in the split chain %s", idMember))
}
chain = append(chain, idMember)
chainSet[idMember] = struct{}{}
}
common.PrintVerbose(cmd, "Looking for a linking object...")
var query objectSDK.SearchFilters
query.AddParentIDFilter(objectSDK.MatchStringEqual, obj)
var prmSearch internal.SearchObjectsPrm
prmSearch.SetClient(cli)
prmSearch.SetContainerID(cnr)
prmSearch.SetFilters(query)
resSearch, err := internal.SearchObjects(cmd.Context(), prmSearch)
commonCmd.ExitOnErr(cmd, "failed to find object children: %w", err)
list := resSearch.IDList()
for i := range list {
if _, ok = chainSet[list[i]]; !ok {
common.PrintVerbose(cmd, "Found one more related object %s.", list[i])
chain = append(chain, list[i])
}
}
return chain
}