the collection resource support withRemainingCount and withContinue

Signed-off-by: Iceber Gu <wei.cai-nat@daocloud.io>

pkg/apiserver/registry/clusterpedia/collectionresources/rest.go

@@ -12,7 +12,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/duration"
+	genericfeatures "k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/registry/rest"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog/v2"
 
 	internal "github.com/clusterpedia-io/api/clusterpedia"
@@ -104,9 +106,11 @@ func (s *REST) Get(ctx context.Context, name string, _ *metav1.GetOptions) (runt
 		}
 	}
 
-	// collection resources don't support with remaining count
-	// ignore opts.WithRemainingCount
-	opts.WithRemainingCount = nil
+	if opts.WithRemainingCount == nil {
+		if enabled := utilfeature.DefaultFeatureGate.Enabled(genericfeatures.RemainingItemCount); enabled {
+			opts.WithRemainingCount = &enabled
+		}
+	}
 
 	storage, ok := s.storages[name]
 	if !ok {

pkg/storage/internalstorage/collectionresource_storage.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"sort"
+	"strconv"
 	"strings"
 
 	"gorm.io/gorm"
@@ -98,30 +99,34 @@ func (s *CollectionResourceStorage) Get(ctx context.Context, opts *internal.List
 	if err != nil {
 		return nil, err
 	}
-	_, query, err = applyListOptionsToCollectionResourceQuery(query, opts)
+	offset, amount, query, err := applyListOptionsToCollectionResourceQuery(query, opts)
 	if err != nil {
 		return nil, err
 	}
 
 	if err := list.From(query); err != nil {
 		return nil, InterpretDBError(s.collectionResource.Name, err)
 	}
+	items := list.Items()
+	collection := &internal.CollectionResource{
+		TypeMeta:   s.collectionResource.TypeMeta,
+		ObjectMeta: s.collectionResource.ObjectMeta,
+		Items:      make([]runtime.Object, 0, len(items)),
+	}
 
 	gvrs := make(map[schema.GroupVersionResource]struct{})
-	types := []internal.CollectionResourceType{}
-	objs := make([]runtime.Object, 0)
-	for _, resource := range list.Items() {
+	for _, resource := range items {
 		obj, err := resource.ConvertToUnstructured()
 		if err != nil {
 			return nil, err
 		}
-		objs = append(objs, obj)
+		collection.Items = append(collection.Items, obj)
 
 		if resourceType := resource.GetResourceType(); !resourceType.Empty() {
 			gvr := resourceType.GroupVersionResource()
 			if _, ok := gvrs[gvr]; !ok {
 				gvrs[gvr] = struct{}{}
-				types = append(types, internal.CollectionResourceType{
+				collection.ResourceTypes = append(collection.ResourceTypes, internal.CollectionResourceType{
 					Group:    resourceType.Group,
 					Resource: resourceType.Resource,
 					Version:  resourceType.Version,
@@ -130,14 +135,22 @@ func (s *CollectionResourceStorage) Get(ctx context.Context, opts *internal.List
 			}
 		}
 	}
-	sortCollectionResourceTypes(types)
+	sortCollectionResourceTypes(collection.ResourceTypes)
+
+	if opts.WithContinue != nil && *opts.WithContinue {
+		if int64(len(items)) == opts.Limit {
+			collection.Continue = strconv.FormatInt(offset+opts.Limit, 10)
+		}
+	}
+
+	if amount != nil {
+		// When offset is too large, the data in the response is empty and the remaining count is negative.
+		// This ensures that `amount = offset + len(objects) + remain`
+		remain := *amount - offset - int64(len(items))
+		collection.RemainingItemCount = &remain
+	}
 
-	return &internal.CollectionResource{
-		TypeMeta:      s.collectionResource.TypeMeta,
-		ObjectMeta:    s.collectionResource.ObjectMeta,
-		ResourceTypes: types,
-		Items:         objs,
-	}, nil
+	return collection, nil
 }
 
 func resolveGVRsFromURLQuery(query url.Values) (gvrs []schema.GroupVersionResource, err error) {
@@ -230,8 +243,7 @@ func parseGroupVersionResource(gvr string) (schema.GroupVersionResource, error)
 	return schema.GroupVersionResource{}, fmt.Errorf("unexpected GroupVersionResource string: %v, expect <group>/<resource> or <group>/<version>/<resource>", gvr)
 }
 
-// TODO(iceber): support with remaining count and continue
-func applyListOptionsToCollectionResourceQuery(query *gorm.DB, opts *internal.ListOptions) (int64, *gorm.DB, error) {
+func applyListOptionsToCollectionResourceQuery(query *gorm.DB, opts *internal.ListOptions) (int64, *int64, *gorm.DB, error) {
 	return applyListOptionsToQuery(query, opts, nil)
 }
 

pkg/storage/internalstorage/resource_storage.go

@@ -273,25 +273,16 @@ func (s *ResourceStorage) List(ctx context.Context, listObject runtime.Object, o
 }
 
 func applyListOptionsToResourceQuery(db *gorm.DB, query *gorm.DB, opts *internal.ListOptions) (int64, *int64, *gorm.DB, error) {
-	var amount *int64
 	applyFn := func(query *gorm.DB, opts *internal.ListOptions) (*gorm.DB, error) {
 		query, err := applyOwnerToResourceQuery(db, query, opts)
 		if err != nil {
 			return nil, err
 		}
 
-		if opts.WithRemainingCount != nil && *opts.WithRemainingCount {
-			amount = new(int64)
-			query = query.Count(amount)
-		}
 		return query, nil
 	}
 
-	offset, query, err := applyListOptionsToQuery(query, opts, applyFn)
-	if err != nil {
-		return 0, nil, nil, err
-	}
-	return offset, amount, query, nil
+	return applyListOptionsToQuery(query, opts, applyFn)
 }
 
 func applyOwnerToResourceQuery(db *gorm.DB, query *gorm.DB, opts *internal.ListOptions) (*gorm.DB, error) {

pkg/storage/internalstorage/util.go

@@ -27,7 +27,7 @@ var (
 	supportedOrderByFields = sets.NewString("cluster", "namespace", "name", "created_at", "resource_version")
 )
 
-func applyListOptionsToQuery(query *gorm.DB, opts *internal.ListOptions, applyFn func(query *gorm.DB, opts *internal.ListOptions) (*gorm.DB, error)) (int64, *gorm.DB, error) {
+func applyListOptionsToQuery(query *gorm.DB, opts *internal.ListOptions, applyFn func(query *gorm.DB, opts *internal.ListOptions) (*gorm.DB, error)) (int64, *int64, *gorm.DB, error) {
 	switch len(opts.ClusterNames) {
 	case 0:
 	case 1:
@@ -126,7 +126,7 @@ func applyListOptionsToQuery(query *gorm.DB, opts *internal.ListOptions, applyFn
 				}
 
 				if len(fieldErrors) != 0 {
-					return 0, nil, apierrors.NewInvalid(schema.GroupKind{Group: internal.GroupName, Kind: "ListOptions"}, "fieldSelector", fieldErrors)
+					return 0, nil, nil, apierrors.NewInvalid(schema.GroupKind{Group: internal.GroupName, Kind: "ListOptions"}, "fieldSelector", fieldErrors)
 				}
 
 				values := requirement.Values().List()
@@ -156,10 +156,16 @@ func applyListOptionsToQuery(query *gorm.DB, opts *internal.ListOptions, applyFn
 		var err error
 		query, err = applyFn(query, opts)
 		if err != nil {
-			return 0, nil, err
+			return 0, nil, nil, err
 		}
 	}
 
+	var amount *int64
+	if opts.WithRemainingCount != nil && *opts.WithRemainingCount {
+		amount = new(int64)
+		query = query.Count(amount)
+	}
+
 	// Due to performance reasons, the default order by is not set.
 	// https://github.com/clusterpedia-io/clusterpedia/pull/44
 	for _, orderby := range opts.OrderBy {
@@ -186,5 +192,5 @@ func applyListOptionsToQuery(query *gorm.DB, opts *internal.ListOptions, applyFn
 	if err == nil {
 		query = query.Offset(offset)
 	}
-	return int64(offset), query, nil
+	return int64(offset), amount, query, nil
 }

pkg/storage/internalstorage/util_test.go

@@ -24,7 +24,7 @@ func testApplyListOptionsToQuery(t *testing.T, name string, options *internal.Li
 	t.Run(fmt.Sprintf("%s postgres", name), func(t *testing.T) {
 		postgreSQL, err := toSQL(postgresDB, options,
 			func(query *gorm.DB, options *internal.ListOptions) (*gorm.DB, error) {
-				_, query, err := applyListOptionsToQuery(query, options, nil)
+				_, _, query, err := applyListOptionsToQuery(query, options, nil)
 				return query, err
 			},
 		)
@@ -39,7 +39,7 @@ func testApplyListOptionsToQuery(t *testing.T, name string, options *internal.Li
 		t.Run(fmt.Sprintf("%s mysql-%s", name, version), func(t *testing.T) {
 			mysqlSQL, err := toSQL(mysqlDBs[version], options,
 				func(query *gorm.DB, options *internal.ListOptions) (*gorm.DB, error) {
-					_, query, err := applyListOptionsToQuery(query, options, nil)
+					_, _, query, err := applyListOptionsToQuery(query, options, nil)
 					return query, err
 				},
 			)

staging/src/github.com/clusterpedia-io/api/clusterpedia/types.go

@@ -82,6 +82,9 @@ type CollectionResource struct {
 
 	ResourceTypes []CollectionResourceType
 	Items         []runtime.Object
+
+	Continue           string
+	RemainingItemCount *int64
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

staging/src/github.com/clusterpedia-io/api/clusterpedia/v1beta1/types.go

@@ -75,6 +75,12 @@ type CollectionResource struct {
 
 	// +optional
 	Items []runtime.RawExtension `json:"items,omitempty"`
+
+	// +optional
+	Continue string `json:"continue,omitempty"`
+
+	// +optional
+	RemainingItemCount *int64 `json:"remainintItemCount,omitempty"`
 }
 
 type CollectionResourceType struct {