package client import ( "context" "crypto/ecdsa" "errors" "fmt" "io" "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/acl" v2object "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object" v2refs "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/refs" rpcapi "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc" "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client" v2session "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session" "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/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 nil, 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 }