Paginate targets in cache (#5101)
* Update Go API to support pagination * Implement pagination of targets in cache (cherry picked from commit cd25fad42f325fc050b7ea3bec3336780d2b5e16)
This commit is contained in:
1 parent
2a87dcad48
commit
3a4ee032d9
62 files changed
+3875
-1533
No files matched your search
@@ -21,13 +21,15 @@ import (
|
||||
type Option func(*options)
|
||||
|
||||
type options struct {
|
||||
postMap map[string]any
|
||||
queryMap map[string]string
|
||||
withAutomaticVersioning bool
|
||||
withSkipCurlOutput bool
|
||||
withFilter string
|
||||
withListToken string
|
||||
withRecursive bool
|
||||
postMap map[string]any
|
||||
queryMap map[string]string
|
||||
withAutomaticVersioning bool
|
||||
withSkipCurlOutput bool
|
||||
withFilter string
|
||||
withListToken string
|
||||
withClientDirectedPagination bool
|
||||
withPageSize uint32
|
||||
withRecursive bool
|
||||
}
|
||||
|
||||
func getDefaultOptions() options {
|
||||
@@ -57,6 +59,9 @@ func getOpts(opt ...Option) (options, []api.Option) {
|
||||
if opts.withRecursive {
|
||||
opts.queryMap["recursive"] = strconv.FormatBool(opts.withRecursive)
|
||||
}
|
||||
if opts.withPageSize != 0 {
|
||||
opts.queryMap["page_size"] = strconv.FormatUint(uint64(opts.withPageSize), 10)
|
||||
}
|
||||
return opts, apiOpts
|
||||
}
|
||||
|
||||
@@ -95,6 +100,21 @@ func WithFilter(filter string) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// WithClientDirectedPagination tells the List function to return only the first
|
||||
// page, if more pages are available
|
||||
func WithClientDirectedPagination(with bool) Option {
|
||||
return func(o *options) {
|
||||
o.withClientDirectedPagination = with
|
||||
}
|
||||
}
|
||||
|
||||
// WithPageSize controls the size of pages used during List
|
||||
func WithPageSize(with uint32) Option {
|
||||
return func(o *options) {
|
||||
o.withPageSize = with
|
||||
}
|
||||
}
|
||||
|
||||
// WithRecursive tells the API to use recursion for listing operations on this
|
||||
// resource
|
||||
func WithRecursive(recurse bool) Option {
|
||||
|
||||
+8
-103
@@ -6,11 +6,9 @@ package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/api"
|
||||
@@ -78,6 +76,13 @@ type WorkerListResult struct {
|
||||
ListToken string `json:"list_token,omitempty"`
|
||||
ResponseType string `json:"response_type,omitempty"`
|
||||
Response *api.Response
|
||||
|
||||
// The following fields are used for cached information when client-directed
|
||||
// pagination is used.
|
||||
recursive bool
|
||||
pageSize uint32
|
||||
scopeId string
|
||||
allRemovedIds []string
|
||||
}
|
||||
|
||||
func (n WorkerListResult) GetItems() []*Worker {
|
||||
@@ -399,109 +404,9 @@ func (c *Client) List(ctx context.Context, scopeId string, opt ...Option) (*Work
|
||||
return nil, apiErr
|
||||
}
|
||||
target.Response = resp
|
||||
if target.ResponseType == "complete" || target.ResponseType == "" {
|
||||
return target, nil
|
||||
}
|
||||
// If there are more results, automatically fetch the rest of the results.
|
||||
// idToIndex keeps a map from the ID of an item to its index in target.Items.
|
||||
// This is used to update updated items in-place and remove deleted items
|
||||
// from the result after pagination is done.
|
||||
idToIndex := map[string]int{}
|
||||
for i, item := range target.Items {
|
||||
idToIndex[item.Id] = i
|
||||
}
|
||||
for {
|
||||
req, err := c.client.NewRequest(ctx, "GET", "workers", nil, apiOpts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error creating List request: %w", err)
|
||||
}
|
||||
|
||||
opts.queryMap["list_token"] = target.ListToken
|
||||
if len(opts.queryMap) > 0 {
|
||||
q := url.Values{}
|
||||
for k, v := range opts.queryMap {
|
||||
q.Add(k, v)
|
||||
}
|
||||
req.URL.RawQuery = q.Encode()
|
||||
}
|
||||
|
||||
resp, err := c.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error performing client request during List call: %w", err)
|
||||
}
|
||||
|
||||
page := new(WorkerListResult)
|
||||
apiErr, err := resp.Decode(page)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error decoding List response: %w", err)
|
||||
}
|
||||
if apiErr != nil {
|
||||
return nil, apiErr
|
||||
}
|
||||
for _, item := range page.Items {
|
||||
if i, ok := idToIndex[item.Id]; ok {
|
||||
// Item has already been seen at index i, update in-place
|
||||
target.Items[i] = item
|
||||
} else {
|
||||
target.Items = append(target.Items, item)
|
||||
idToIndex[item.Id] = len(target.Items) - 1
|
||||
}
|
||||
}
|
||||
// RemovedIds contain any Worker that were deleted since the last response.
|
||||
target.RemovedIds = append(target.RemovedIds, page.RemovedIds...)
|
||||
target.EstItemCount = page.EstItemCount
|
||||
target.ListToken = page.ListToken
|
||||
target.ResponseType = page.ResponseType
|
||||
target.Response = resp
|
||||
if target.ResponseType == "complete" {
|
||||
break
|
||||
}
|
||||
}
|
||||
// For now, removedIds will only be populated if this pagination cycle was the result of a
|
||||
// "refresh" operation (i.e., the caller provided a list token option to this call).
|
||||
//
|
||||
// Sort to make response deterministic
|
||||
slices.Sort(target.RemovedIds)
|
||||
// Remove any duplicates
|
||||
target.RemovedIds = slices.Compact(target.RemovedIds)
|
||||
// Remove items that were deleted since the end of the last iteration.
|
||||
// If a Worker has been updated and subsequently removed, we don't want
|
||||
// it to appear both in the Items and RemovedIds, so we remove it from the Items.
|
||||
for _, removedId := range target.RemovedIds {
|
||||
if i, ok := idToIndex[removedId]; ok {
|
||||
// Remove the item at index i without preserving order
|
||||
// https://github.com/golang/go/wiki/SliceTricks#delete-without-preserving-order
|
||||
target.Items[i] = target.Items[len(target.Items)-1]
|
||||
target.Items = target.Items[:len(target.Items)-1]
|
||||
// Update the index of the previously last element
|
||||
idToIndex[target.Items[i].Id] = i
|
||||
}
|
||||
}
|
||||
// Since we paginated to the end, we can avoid confusion
|
||||
// for the user by setting the estimated item count to the
|
||||
// length of the items slice. If we don't set this here, it
|
||||
// will equal the value returned in the last response, which is
|
||||
// often much smaller than the total number returned.
|
||||
target.EstItemCount = uint(len(target.Items))
|
||||
// Sort the results again since in-place updates and deletes
|
||||
// may have shuffled items. We sort by created time descending
|
||||
// (most recently created first), same as the API.
|
||||
slices.SortFunc(target.Items, func(i, j *Worker) int {
|
||||
return j.CreatedTime.Compare(i.CreatedTime)
|
||||
})
|
||||
// Finally, since we made at least 2 requests to the server to fulfill this
|
||||
// function call, resp.Body and resp.Map will only contain the most recent response.
|
||||
// Overwrite them with the true response.
|
||||
target.Response.Body.Reset()
|
||||
if err := json.NewEncoder(target.Response.Body).Encode(target); err != nil {
|
||||
return nil, fmt.Errorf("error encoding final JSON list response: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(target.Response.Body.Bytes(), &target.Response.Map); err != nil {
|
||||
return nil, fmt.Errorf("error encoding final map list response: %w", err)
|
||||
}
|
||||
// Note: the HTTP response body is consumed by resp.Decode in the loop,
|
||||
// so it doesn't need to be updated (it will always be, and has always been, empty).
|
||||
return target, nil
|
||||
|
||||
}
|
||||
|
||||
func (c *Client) AddWorkerTags(ctx context.Context, id string, version uint32, apiTags map[string][]string, opt ...Option) (*WorkerUpdateResult, error) {
|
||||
|
||||
Reference in new issue
Block a user