diff --git a/pkg/registry/apps/correlations/legacy_storage.go b/pkg/registry/apps/correlations/legacy_storage.go index 733ad4ec330..14b085fb660 100644 --- a/pkg/registry/apps/correlations/legacy_storage.go +++ b/pkg/registry/apps/correlations/legacy_storage.go @@ -2,7 +2,9 @@ package correlations import ( "context" + b64 "encoding/base64" "fmt" + "strconv" "strings" "k8s.io/apimachinery/pkg/apis/meta/internalversion" @@ -78,27 +80,47 @@ func (s *legacyStorage) List(ctx context.Context, options *internalversion.ListO } } - if options.Continue != "" { - return nil, fmt.Errorf("paging not yet supported") + page := int64(0) + limit := int64(100000) + if options != nil { + if options.Limit > 0 { + limit = options.Limit + } + if options.Continue != "" { + token, err := decodeContinueToken(options.Continue) + if err != nil { + return nil, err + } + if token.Limit != limit { + return nil, fmt.Errorf("continue token limit does not match the previous request") + } + page = token.Page + } } rsp, err := s.service.GetCorrelations(ctx, correlations.GetCorrelationsQuery{ OrgId: orgID, - Limit: 1000, + Limit: limit + 1, + Page: page, SourceUIDs: uids, }) if err != nil { return nil, err } list := &correlationsV0.CorrelationList{ - Items: make([]correlationsV0.Correlation, len(rsp.Correlations)), + Items: make([]correlationsV0.Correlation, 0, len(rsp.Correlations)), } for i, orig := range rsp.Correlations { + if i >= int(limit) { + list.Continue = encodeContinueToken(page+1, limit) + break + } + c, err := correlations.ToResource(orig, s.namespacer) if err != nil { return nil, err } - list.Items[i] = *c + list.Items = append(list.Items, *c) } return list, nil } @@ -193,3 +215,33 @@ func (s *legacyStorage) Delete(ctx context.Context, name string, deleteValidatio func (s *legacyStorage) DeleteCollection(ctx context.Context, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions, listOptions *internalversion.ListOptions) (runtime.Object, error) { return nil, fmt.Errorf("DeleteCollection for shorturl not implemented") } + +type continueToken struct { + Page int64 + Limit int64 +} + +func encodeContinueToken(page, limit int64) string { + data := fmt.Sprintf("%d/%d", page, limit) + return b64.StdEncoding.EncodeToString([]byte(data)) // use base64 so it is not treated like query params +} + +func decodeContinueToken(s string) (token continueToken, err error) { + decoded, err := b64.StdEncoding.DecodeString(s) + if err != nil { + return token, fmt.Errorf("invalid continue token") + } + parts := strings.Split(string(decoded), "/") + if len(parts) != 2 { + return token, fmt.Errorf("invalid continue token") + } + token.Page, err = strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return token, fmt.Errorf("invalid continue token (page)") + } + token.Limit, err = strconv.ParseInt(parts[1], 10, 64) + if err != nil { + return token, fmt.Errorf("invalid continue token") + } + return token, nil +}