forked from TrueCloudLab/frostfs-sdk-go
305 lines
8 KiB
Go
305 lines
8 KiB
Go
package client
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ecdsa"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/acl"
|
|
v2object "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/object"
|
|
v2refs "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/refs"
|
|
rpcapi "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc/client"
|
|
v2session "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/session"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/signature"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
|
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
|
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"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
|
|
)
|
|
|
|
// PrmObjectSearch groups parameters of ObjectSearch operation.
|
|
type PrmObjectSearch struct {
|
|
XHeaders []string
|
|
|
|
Local bool
|
|
|
|
BearerToken *bearer.Token
|
|
|
|
Session *session.Object
|
|
|
|
ContainerID *cid.ID
|
|
|
|
Key *ecdsa.PrivateKey
|
|
|
|
Filters object.SearchFilters
|
|
}
|
|
|
|
// MarkLocal tells the server to execute the operation locally.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.Local instead.
|
|
func (x *PrmObjectSearch) MarkLocal() {
|
|
x.Local = true
|
|
}
|
|
|
|
// WithinSession specifies session within which the search query must be executed.
|
|
//
|
|
// Creator of the session acquires the authorship of the request.
|
|
// This may affect the execution of an operation (e.g. access control).
|
|
//
|
|
// Must be signed.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.Session instead.
|
|
func (x *PrmObjectSearch) WithinSession(t session.Object) {
|
|
x.Session = &t
|
|
}
|
|
|
|
// WithBearerToken attaches bearer token to be used for the operation.
|
|
//
|
|
// If set, underlying eACL rules will be used in access control.
|
|
//
|
|
// Must be signed.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.BearerToken instead.
|
|
func (x *PrmObjectSearch) WithBearerToken(t bearer.Token) {
|
|
x.BearerToken = &t
|
|
}
|
|
|
|
// WithXHeaders specifies list of extended headers (string key-value pairs)
|
|
// to be attached to the request. Must have an even length.
|
|
//
|
|
// Slice must not be mutated until the operation completes.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.XHeaders instead.
|
|
func (x *PrmObjectSearch) WithXHeaders(hs ...string) {
|
|
x.XHeaders = hs
|
|
}
|
|
|
|
// UseKey specifies private key to sign the requests.
|
|
// If key is not provided, then Client default key is used.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.Key instead.
|
|
func (x *PrmObjectSearch) UseKey(key ecdsa.PrivateKey) {
|
|
x.Key = &key
|
|
}
|
|
|
|
// InContainer specifies the container in which to look for objects.
|
|
// Required parameter.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.ContainerID instead.
|
|
func (x *PrmObjectSearch) InContainer(id cid.ID) {
|
|
x.ContainerID = &id
|
|
}
|
|
|
|
// SetFilters sets filters by which to select objects. All container objects
|
|
// match unset/empty filters.
|
|
//
|
|
// Deprecated: Use PrmObjectSearch.Filters instead.
|
|
func (x *PrmObjectSearch) SetFilters(filters object.SearchFilters) {
|
|
x.Filters = filters
|
|
}
|
|
|
|
// ResObjectSearch groups the final result values of ObjectSearch operation.
|
|
type ResObjectSearch struct {
|
|
statusRes
|
|
}
|
|
|
|
// ObjectListReader is designed to read list of object identifiers from FrostFS system.
|
|
//
|
|
// Must be initialized using Client.ObjectSearch, any other usage is unsafe.
|
|
type ObjectListReader struct {
|
|
client *Client
|
|
cancelCtxStream context.CancelFunc
|
|
err error
|
|
res ResObjectSearch
|
|
stream interface {
|
|
Read(resp *v2object.SearchResponse) error
|
|
}
|
|
tail []v2refs.ObjectID
|
|
}
|
|
|
|
// Read reads another list of the object identifiers. Works similar to
|
|
// io.Reader.Read but copies oid.ID and returns success flag instead of error.
|
|
//
|
|
// Failure reason can be received via Close.
|
|
//
|
|
// Panics if buf has zero length.
|
|
func (x *ObjectListReader) Read(buf []oid.ID) (int, bool) {
|
|
if len(buf) == 0 {
|
|
panic("empty buffer in ObjectListReader.ReadList")
|
|
}
|
|
|
|
read := copyIDBuffers(buf, x.tail)
|
|
x.tail = x.tail[read:]
|
|
|
|
if len(buf) == read {
|
|
return read, true
|
|
}
|
|
|
|
for {
|
|
var resp v2object.SearchResponse
|
|
x.err = x.stream.Read(&resp)
|
|
if x.err != nil {
|
|
return read, false
|
|
}
|
|
|
|
x.res.st, x.err = x.client.processResponse(&resp)
|
|
if x.err != nil || !apistatus.IsSuccessful(x.res.st) {
|
|
return read, false
|
|
}
|
|
|
|
// read new chunk of objects
|
|
ids := resp.GetBody().GetIDList()
|
|
if len(ids) == 0 {
|
|
// just skip empty lists since they are not prohibited by protocol
|
|
continue
|
|
}
|
|
|
|
ln := copyIDBuffers(buf[read:], ids)
|
|
read += ln
|
|
|
|
if read == len(buf) {
|
|
// save the tail
|
|
x.tail = append(x.tail, ids[ln:]...)
|
|
|
|
return read, true
|
|
}
|
|
}
|
|
}
|
|
|
|
func copyIDBuffers(dst []oid.ID, src []v2refs.ObjectID) int {
|
|
var i int
|
|
for ; i < len(dst) && i < len(src); i++ {
|
|
_ = dst[i].ReadFromV2(src[i])
|
|
}
|
|
return i
|
|
}
|
|
|
|
// Iterate iterates over the list of found object identifiers.
|
|
// f can return true to stop iteration earlier.
|
|
//
|
|
// Returns an error if object can't be read.
|
|
func (x *ObjectListReader) Iterate(f func(oid.ID) bool) error {
|
|
buf := make([]oid.ID, 1)
|
|
|
|
for {
|
|
// Do not check first return value because `len(buf) == 1`,
|
|
// so false means nothing was read.
|
|
_, ok := x.Read(buf)
|
|
if !ok {
|
|
res, err := x.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return apistatus.ErrFromStatus(res.Status())
|
|
}
|
|
if f(buf[0]) {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close ends reading list of the matched objects and returns the result of the operation
|
|
// along with the final results. Must be called after using the ObjectListReader.
|
|
//
|
|
// Exactly one return value is non-nil. By default, server status is returned in res structure.
|
|
// Any client's internal or transport errors are returned as Go built-in error.
|
|
// If Client is tuned to resolve FrostFS API statuses, then FrostFS failures
|
|
// codes are returned as error.
|
|
//
|
|
// Return statuses:
|
|
// - global (see Client docs);
|
|
// - *apistatus.ContainerNotFound;
|
|
// - *apistatus.ObjectAccessDenied;
|
|
// - *apistatus.SessionTokenExpired.
|
|
func (x *ObjectListReader) Close() (*ResObjectSearch, error) {
|
|
defer x.cancelCtxStream()
|
|
|
|
if x.err != nil && !errors.Is(x.err, io.EOF) {
|
|
return &x.res, x.err
|
|
}
|
|
|
|
return &x.res, nil
|
|
}
|
|
|
|
func (x *PrmObjectSearch) buildRequest(c *Client) (*v2object.SearchRequest, error) {
|
|
if x.ContainerID == nil {
|
|
return nil, errorMissingContainer
|
|
}
|
|
|
|
if len(x.XHeaders)%2 != 0 {
|
|
return nil, errorInvalidXHeaders
|
|
}
|
|
|
|
meta := new(v2session.RequestMetaHeader)
|
|
writeXHeadersToMeta(x.XHeaders, meta)
|
|
|
|
if x.BearerToken != nil {
|
|
v2BearerToken := new(acl.BearerToken)
|
|
x.BearerToken.WriteToV2(v2BearerToken)
|
|
meta.SetBearerToken(v2BearerToken)
|
|
}
|
|
|
|
if x.Session != nil {
|
|
v2SessionToken := new(v2session.Token)
|
|
x.Session.WriteToV2(v2SessionToken)
|
|
meta.SetSessionToken(v2SessionToken)
|
|
}
|
|
|
|
if x.Local {
|
|
meta.SetTTL(1)
|
|
}
|
|
cnrV2 := new(v2refs.ContainerID)
|
|
x.ContainerID.WriteToV2(cnrV2)
|
|
|
|
body := new(v2object.SearchRequestBody)
|
|
body.SetVersion(1)
|
|
body.SetContainerID(cnrV2)
|
|
body.SetFilters(x.Filters.ToV2())
|
|
|
|
req := new(v2object.SearchRequest)
|
|
req.SetBody(body)
|
|
c.prepareRequest(req, meta)
|
|
|
|
return req, nil
|
|
}
|
|
|
|
// ObjectSearchInit initiates object selection through a remote server using FrostFS API protocol.
|
|
//
|
|
// The call only opens the transmission channel, explicit fetching of matched objects
|
|
// is done using the ObjectListReader. Exactly one return value is non-nil.
|
|
// Resulting reader must be finally closed.
|
|
//
|
|
// Returns an error if parameters are set incorrectly (see PrmObjectSearch docs).
|
|
// Context is required and must not be nil. It is used for network communication.
|
|
func (c *Client) ObjectSearchInit(ctx context.Context, prm PrmObjectSearch) (*ObjectListReader, error) {
|
|
req, err := prm.buildRequest(c)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
key := prm.Key
|
|
if key == nil {
|
|
key = &c.prm.Key
|
|
}
|
|
|
|
err = signature.SignServiceMessage(key, req)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("sign request: %w", err)
|
|
}
|
|
|
|
var r ObjectListReader
|
|
ctx, r.cancelCtxStream = context.WithCancel(ctx)
|
|
|
|
r.stream, err = rpcapi.SearchObjects(&c.c, req, client.WithContext(ctx))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open stream: %w", err)
|
|
}
|
|
r.client = c
|
|
|
|
return &r, nil
|
|
}
|