Skip to content

Commit

Permalink
internal/native: add Search operation to extension
Browse files Browse the repository at this point in the history
Simple operation, supports generic filters and relative metrics (seconds per
single object in results).

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
  • Loading branch information
carpawell committed Aug 9, 2024
1 parent 0d3cabc commit d836981
Show file tree
Hide file tree
Showing 2 changed files with 60 additions and 0 deletions.
57 changes: 57 additions & 0 deletions internal/native/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,63 @@ func (c *Client) PutContainer(params map[string]string) PutContainerResponse {
return PutContainerResponse{Success: true, ContainerID: contID.EncodeToString()}
}

type Filter struct {
Key string
Operation string
Value string
}

// Search searches for the objects in container that satisfies provided filters.
// Returns number of found objects.
func (c *Client) Search(cnrString string, filtersJS []Filter) (int, error) {
var cID cid.ID
err := cID.DecodeString(cnrString)
if err != nil {
return 0, fmt.Errorf("reading container ID: %w", err)
}

var op object.SearchMatchType
var filters object.SearchFilters
for _, flt := range filtersJS {
if !op.DecodeString(flt.Operation) {
return 0, fmt.Errorf("unknown filter operation: %s", flt.Operation)
}

filters.AddFilter(flt.Key, flt.Value, op)
}

var prm client.PrmObjectSearch
prm.SetFilters(filters)

start := time.Now()

r, err := c.cli.ObjectSearchInit(c.vu.Context(), cID, c.signer, prm)
if err != nil {
return 0, fmt.Errorf("search stream initialization: %w", err)
}
defer func() {
_ = r.Close()
}()

var objsNum int
err = r.Iterate(func(_ oid.ID) bool {
objsNum++
return false
})
if err != nil {
return 0, fmt.Errorf("reading search results: %w", err)
}

var relativeTime time.Duration
if objsNum > 0 {
relativeTime = time.Since(start) / time.Duration(objsNum)
}

stats.Report(c.vu, objSearchDurationRelative, metrics.D(relativeTime))

return objsNum, nil
}

func (c *Client) Onsite(containerID string, payload goja.ArrayBuffer) PreparedObject {
maxObjectSize, epoch, hhDisabled, err := parseNetworkInfo(c.vu.Context(), c.cli)
if err != nil {
Expand Down
3 changes: 3 additions & 0 deletions internal/native/native.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ var (
objGetTotal, objGetFails, objGetDuration *metrics.Metric
objDeleteTotal, objDeleteFails, objDeleteDuration *metrics.Metric
cnrPutTotal, cnrPutFails, cnrPutDuration *metrics.Metric
objSearchDurationRelative *metrics.Metric
)

func init() {
Expand Down Expand Up @@ -136,6 +137,8 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
cnrPutFails, _ = registry.NewMetric("neofs_cnr_put_fails", metrics.Counter)
cnrPutDuration, _ = registry.NewMetric("neofs_cnr_put_duration", metrics.Trend, metrics.Time)

objSearchDurationRelative, _ = registry.NewMetric("neofs_search_duration_relative", metrics.Trend, metrics.Time)

return &Client{
vu: n.vu,
signer: signer,
Expand Down

0 comments on commit d836981

Please sign in to comment.