implement all storage queryor

Signed-off-by: huangjunwei <huangjunwei@youmi.net>

pkg/storage/driver/cfgmaps.go

@@ -89,11 +89,51 @@ func (cfgmaps *ConfigMaps) Get(key string) (release.Releaser, error) {
 	return r, nil
 }
 
+// listPages is a common method to list release with pagination
+func (cfgmaps *ConfigMaps) listPages(f func(page []release.Releaser, lastPage bool) (end bool), opts metav1.ListOptions, filter func(release.Releaser) bool) (err error) {
+	if opts.Limit == 0 {
+		opts.Limit = DefaultPaginationLimit
+	}
+
+Loop:
+	for {
+		list, err := cfgmaps.impl.List(context.Background(), opts)
+		if nil != err {
+			return err
+		}
+		opts.Continue = list.Continue
+
+		var results []release.Releaser
+
+		// iterate over the configmaps object list
+		// and decode each release
+		for _, item := range list.Items {
+			rls, err := decodeRelease(string(item.Data["release"]))
+			if err != nil {
+				cfgmaps.Logger().Debug("failed to decode release", "item", item, slog.Any("error", err))
+				continue
+			}
+
+			rls.Labels = item.Labels
+
+			if filter(rls) {
+				results = append(results, rls)
+			}
+		}
+
+		if f(results, list.Continue != "") {
+			break Loop
+		}
+	}
+
+	return nil
+}
+
 // List fetches all releases and returns the list releases such
 // that filter(release) == true. An error is returned if the
 // configmap fails to retrieve the releases.
 func (cfgmaps *ConfigMaps) List(filter func(release.Releaser) bool) ([]release.Releaser, error) {
-	lsel := kblabels.Set{"owner": "helm"}.AsSelector()
+	lsel := kblabels.Set{"owner": owner}.AsSelector()
 	opts := metav1.ListOptions{LabelSelector: lsel.String()}
 
 	list, err := cfgmaps.impl.List(context.Background(), opts)
@@ -122,6 +162,14 @@ func (cfgmaps *ConfigMaps) List(filter func(release.Releaser) bool) ([]release.R
 	return results, nil
 }
 
+// ListPages same as List, but with pagination
+func (cfgmaps *ConfigMaps) ListPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, filter func(release.Releaser) bool) error {
+	lsel := kblabels.Set{"owner": owner}.AsSelector()
+	opts := metav1.ListOptions{Limit: limit, LabelSelector: lsel.String()}
+
+	return cfgmaps.listPages(f, opts, filter)
+}
+
 // Query fetches all releases that match the provided map of labels.
 // An error is returned if the configmap fails to retrieve the releases.
 func (cfgmaps *ConfigMaps) Query(labels map[string]string) ([]release.Releaser, error) {
@@ -158,6 +206,21 @@ func (cfgmaps *ConfigMaps) Query(labels map[string]string) ([]release.Releaser,
 	return results, nil
 }
 
+// QueryPages same as Query, but with pagination
+func (cfgmaps *ConfigMaps) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, labels map[string]string) error {
+	ls := kblabels.Set{}
+	for k, v := range labels {
+		if errs := validation.IsValidLabelValue(v); len(errs) != 0 {
+			return fmt.Errorf("invalid label value: %q: %s", v, strings.Join(errs, "; "))
+		}
+		ls[k] = v
+	}
+
+	opts := metav1.ListOptions{Limit: limit, LabelSelector: ls.AsSelector().String()}
+
+	return cfgmaps.listPages(f, opts, func(release.Releaser) bool { return true })
+}
+
 // Create creates a new ConfigMap holding the release. If the
 // ConfigMap already exists, ErrReleaseExists is returned.
 func (cfgmaps *ConfigMaps) Create(key string, rls release.Releaser) error {
@@ -252,8 +315,6 @@ func (cfgmaps *ConfigMaps) Delete(key string) (rls release.Releaser, err error)
 //	"owner"          - owner of the configmap, currently "helm".
 //	"name"           - name of the release.
 func newConfigMapsObject(key string, rls *rspb.Release, lbs labels) (*v1.ConfigMap, error) {
-	const owner = "helm"
-
 	// encode the release
 	s, err := encodeRelease(rls)
 	if err != nil {

pkg/storage/driver/cfgmaps_test.go

@@ -111,6 +111,20 @@ func TestConfigMapList(t *testing.T) {
 		releaseStub("key-6", 1, "default", common.StatusSuperseded),
 	}...)
 
+	// list releases with pagination
+	err := cfgmaps.ListPages(func(page []release.Releaser, lastPage bool) (end bool) {
+		// check
+		if len(page) != 2 {
+			t.Errorf("Expected 2 cfgmaps, got %d:\n%v\n", len(page), page)
+		}
+
+		return !lastPage
+	}, 2, func(rel release.Releaser) bool { return true })
+	// check
+	if err != nil {
+		t.Errorf("Failed to list cfgmaps: %s", err)
+	}
+
 	// list all deleted releases
 	del, err := cfgmaps.List(func(rel release.Releaser) bool {
 		rls := convertReleaserToV1(t, rel)
@@ -183,6 +197,18 @@ func TestConfigMapQuery(t *testing.T) {
 	if err != ErrReleaseNotFound {
 		t.Errorf("Expected {%v}, got {%v}", ErrReleaseNotFound, err)
 	}
+
+	// query cfgmaps with pagination
+	err = cfgmaps.QueryPages(func(page []release.Releaser, lastPage bool) (err bool) {
+		if len(page) != 2 {
+			t.Fatalf("Expected 2 results, actual %d", len(page))
+		}
+
+		return !lastPage
+	}, 2, map[string]string{"status": "deployed"})
+	if err != nil {
+		t.Fatalf("Failed to query: %s", err)
+	}
 }
 
 func TestConfigMapCreate(t *testing.T) {

pkg/storage/driver/driver.go

@@ -24,6 +24,13 @@ import (
 	rspb "helm.sh/helm/v4/pkg/release/v1"
 )
 
+const (
+	// owner is owner of the secret
+	owner = "helm"
+	// DefaultPaginationLimit .
+	DefaultPaginationLimit = 50
+)
+
 var (
 	// ErrReleaseNotFound indicates that a release is not found.
 	ErrReleaseNotFound = errors.New("release: not found")
@@ -89,7 +96,9 @@ type Deletor interface {
 type Queryor interface {
 	Get(key string) (release.Releaser, error)
 	List(filter func(release.Releaser) bool) ([]release.Releaser, error)
+	ListPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, filter func(release.Releaser) bool) error
 	Query(labels map[string]string) ([]release.Releaser, error)
+	QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, labels map[string]string) error
 }
 
 // Driver is the interface composed of Creator, Updator, Deletor, and Queryor

pkg/storage/driver/memory.go

@@ -17,6 +17,7 @@ limitations under the License.
 package driver
 
 import (
+	"fmt"
 	"log/slog"
 	"strconv"
 	"strings"
@@ -114,6 +115,72 @@ func (mem *Memory) List(filter func(release.Releaser) bool) ([]release.Releaser,
 	return ls, nil
 }
 
+// ListPages is a common method to list release with pagination
+func (mem *Memory) ListPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, filter func(release.Releaser) bool) (err error) {
+	defer unlock(mem.rlock())
+
+	if limit == 0 {
+		limit = DefaultPaginationLimit
+	}
+
+	ls := make([]release.Releaser, 0, int(limit))
+	if mem.namespace != "" {
+		for _, recs := range mem.cache[mem.namespace] {
+			recs.Iter(func(i int, rec *record) bool {
+				if filter(rec.rls) {
+					ls = append(ls, rec.rls)
+				}
+
+				if int64(len(ls)) >= limit {
+					end := f(ls, i == len(recs))
+					ls = make([]release.Releaser, 0, int(limit))
+
+					return !end
+				}
+
+				return true
+			})
+
+			if len(ls) > 0 {
+				_ = f(ls, true)
+			}
+		}
+
+		return nil
+	}
+
+	var total int
+	for _, releases := range mem.cache {
+		total += len(releases)
+	}
+
+	var index int
+	for namespace := range mem.cache {
+		for _, recs := range mem.cache[namespace] {
+			recs.Iter(func(_ int, rec *record) bool {
+				index++
+				if filter(rec.rls) {
+					ls = append(ls, rec.rls)
+				}
+
+				if int64(len(ls)) >= limit {
+					end := f(ls, index == total)
+					ls = make([]release.Releaser, 0, int(limit))
+
+					return !end
+				}
+
+				return true
+			})
+
+		}
+	}
+
+	_ = f(ls, true)
+
+	return nil
+}
+
 // Query returns the set of releases that match the provided set of labels
 func (mem *Memory) Query(keyvals map[string]string) ([]release.Releaser, error) {
 	defer unlock(mem.rlock())
@@ -155,6 +222,83 @@ func (mem *Memory) Query(keyvals map[string]string) ([]release.Releaser, error)
 	return ls, nil
 }
 
+// QueryPages same as Query, but with pagination
+func (mem *Memory) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, keyvals map[string]string) error {
+	defer unlock(mem.rlock())
+
+	if limit == 0 {
+		limit = DefaultPaginationLimit
+	}
+
+	var lbs labels
+
+	lbs.init()
+	lbs.fromMap(keyvals)
+
+	fmt.Printf("namespace: %s\n", mem.namespace)
+
+	ls := make([]release.Releaser, 0, int(limit))
+	if mem.namespace != "" {
+		for _, recs := range mem.cache[mem.namespace] {
+			recs.Iter(func(i int, rec *record) bool {
+				if rec.lbs.match(lbs) {
+					ls = append(ls, rec.rls)
+				}
+
+				if int64(len(ls)) >= limit {
+					end := f(ls, i == len(recs))
+					ls = make([]release.Releaser, 0, int(limit))
+
+					return !end
+				}
+
+				return true
+			})
+
+			if len(ls) > 0 {
+				_ = f(ls, true)
+			}
+		}
+
+		return nil
+	}
+
+	var total int
+	for _, releases := range mem.cache {
+		for _, recs := range releases {
+			total += len(recs)
+		}
+	}
+
+	var index int
+	for _, releases := range mem.cache {
+		for _, recs := range releases {
+			recs.Iter(func(_ int, rec *record) bool {
+				index++
+				if rec.lbs.match(lbs) {
+					ls = append(ls, rec.rls)
+				}
+
+				if int64(len(ls)) >= limit {
+					end := f(ls, index == total)
+					ls = make([]release.Releaser, 0, int(limit))
+
+					return !end
+				}
+
+				return true
+			})
+
+		}
+	}
+
+	if len(ls) > 0 {
+		_ = f(ls, true)
+	}
+
+	return nil
+}
+
 // Create creates a new release or returns ErrReleaseExists.
 func (mem *Memory) Create(key string, rel release.Releaser) error {
 	defer unlock(mem.wlock())

pkg/storage/driver/memory_test.go

@@ -107,6 +107,20 @@ func TestMemoryList(t *testing.T) {
 	ts := tsFixtureMemory(t)
 	ts.SetNamespace("default")
 
+	// list releases with pagination
+	err := ts.ListPages(func(page []release.Releaser, lastPage bool) (end bool) {
+		// check
+		if len(page) != 2 {
+			t.Errorf("Expected 2 releases, got %d:\n%v\n", len(page), page)
+		}
+
+		return !lastPage
+	}, 2, func(rel release.Releaser) bool { return true })
+	// check
+	if err != nil {
+		t.Errorf("Failed to list releases: %s", err)
+	}
+
 	// list all deployed releases
 	dpl, err := ts.List(func(rel release.Releaser) bool {
 		rls := convertReleaserToV1(t, rel)
@@ -255,6 +269,19 @@ func TestMemoryDelete(t *testing.T) {
 
 	ts := tsFixtureMemory(t)
 	ts.SetNamespace("")
+
+	// query cfgmaps with pagination
+	err := ts.QueryPages(func(page []release.Releaser, lastPage bool) (err bool) {
+		if len(page) != 1 {
+			t.Fatalf("Expected 1 results, actual %d", len(page))
+		}
+
+		return !lastPage
+	}, 1, map[string]string{"status": "deployed"})
+	if err != nil {
+		t.Fatalf("Failed to query: %s", err)
+	}
+
 	start, err := ts.Query(map[string]string{"status": "deployed"})
 	if err != nil {
 		t.Errorf("Query failed: %s", err)

pkg/storage/driver/mock_test.go

@@ -19,6 +19,7 @@ package driver // import "helm.sh/helm/v4/pkg/storage/driver"
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"testing"
 
 	sqlmock "github.com/DATA-DOG/go-sqlmock"
@@ -133,11 +134,25 @@ func (mock *MockConfigMapsInterface) List(_ context.Context, opts metav1.ListOpt
 		return nil, err
 	}
 
+	objects := make([]*v1.ConfigMap, 0, len(mock.objects))
 	for _, cfgmap := range mock.objects {
 		if labelSelector.Matches(kblabels.Set(cfgmap.Labels)) {
-			list.Items = append(list.Items, *cfgmap)
+			objects = append(objects, cfgmap)
 		}
 	}
+
+	if opts.Continue != "" {
+		i, _ := strconv.ParseInt(opts.Continue, 10, 64)
+		objects = objects[int(i):]
+	}
+
+	if opts.Limit > 0 && opts.Limit <= int64(len(objects)) {
+		objects = objects[:int(opts.Limit)]
+	}
+
+	for _, cfgmap := range objects {
+		list.Items = append(list.Items, *cfgmap)
+	}
 	return &list, nil
 }
 
@@ -221,11 +236,25 @@ func (mock *MockSecretsInterface) List(_ context.Context, opts metav1.ListOption
 		return nil, err
 	}
 
+	objects := make([]*v1.Secret, 0, len(mock.objects))
 	for _, secret := range mock.objects {
 		if labelSelector.Matches(kblabels.Set(secret.Labels)) {
-			list.Items = append(list.Items, *secret)
+			objects = append(objects, secret)
 		}
 	}
+
+	if opts.Continue != "" {
+		i, _ := strconv.ParseInt(opts.Continue, 10, 64)
+		objects = objects[int(i):]
+	}
+
+	if opts.Limit > 0 && opts.Limit <= int64(len(objects)) {
+		objects = objects[:int(opts.Limit)]
+	}
+
+	for _, secret := range objects {
+		list.Items = append(list.Items, *secret)
+	}
 	return &list, nil
 }
 

pkg/storage/driver/secrets.go

@@ -38,14 +38,8 @@ import (
 
 var _ Driver = (*Secrets)(nil)
 
-const (
-	// SecretsDriverName is the string name of the driver.
-	SecretsDriverName = "Secret"
-	// owner is owner of the secret
-	owner = "helm"
-	// DefaultPaginationLimit .
-	DefaultPaginationLimit = 50
-)
+// SecretsDriverName is the string name of the driver.
+const SecretsDriverName = "Secret"
 
 // Secrets is a wrapper around an implementation of a kubernetes
 // SecretsInterface.
@@ -90,7 +84,7 @@ func (secrets *Secrets) Get(key string) (release.Releaser, error) {
 	return r, nil
 }
 
-// listPages common methods to list release pagination
+// listPages is a common method to list release with pagination
 func (secrets *Secrets) listPages(f func(page []release.Releaser, lastPage bool) (end bool), opts metav1.ListOptions, filter func(release.Releaser) bool) (err error) {
 	if opts.Limit == 0 {
 		opts.Limit = DefaultPaginationLimit
@@ -206,7 +200,7 @@ func (secrets *Secrets) Query(labels map[string]string) ([]release.Releaser, err
 }
 
 // QueryPages same as Query, but with pagination
-func (secrets *Secrets) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), labels map[string]string) error {
+func (secrets *Secrets) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, labels map[string]string) error {
 	ls := kblabels.Set{}
 	for k, v := range labels {
 		if errs := validation.IsValidLabelValue(v); len(errs) != 0 {
@@ -215,7 +209,7 @@ func (secrets *Secrets) QueryPages(f func(page []release.Releaser, lastPage bool
 		ls[k] = v
 	}
 
-	opts := metav1.ListOptions{LabelSelector: ls.AsSelector().String()}
+	opts := metav1.ListOptions{Limit: limit, LabelSelector: ls.AsSelector().String()}
 
 	return secrets.listPages(f, opts, func(release.Releaser) bool { return true })
 }

pkg/storage/driver/secrets_test.go

@@ -99,15 +99,15 @@ func TestSecretList(t *testing.T) {
 	// list releases with pagination
 	err := secrets.ListPages(func(page []release.Releaser, lastPage bool) (end bool) {
 		// check
-		if len(page) != 6 {
-			t.Errorf("Expected 6 secrets, got %d:\n%v\n", len(page), page)
+		if len(page) != 2 {
+			t.Errorf("Expected 2 secrets, got %d:\n%v\n", len(page), page)
 		}
 
 		return !lastPage
-	}, func(rel release.Releaser) bool { return true })
+	}, 2, func(rel release.Releaser) bool { return true })
 	// check
 	if err != nil {
-		t.Errorf("Failed to list deleted: %s", err)
+		t.Errorf("Failed to list secrets: %s", err)
 	}
 
 	// list all deleted releases
@@ -190,7 +190,7 @@ func TestSecretQuery(t *testing.T) {
 		}
 
 		return !lastPage
-	}, map[string]string{"status": "deployed"})
+	}, 2, map[string]string{"status": "deployed"})
 	if err != nil {
 		t.Fatalf("Failed to query: %s", err)
 	}

pkg/storage/driver/sql.go

@@ -383,6 +383,62 @@ func (s *SQL) List(filter func(release.Releaser) bool) ([]release.Releaser, erro
 	return releases, nil
 }
 
+// ListPages is a common method to list release with pagination
+func (s *SQL) ListPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, filter func(release.Releaser) bool) (err error) {
+	if limit == 0 {
+		limit = DefaultPaginationLimit
+	}
+
+	sb := s.statementBuilder.
+		Select(sqlReleaseTableKeyColumn, sqlReleaseTableNamespaceColumn, sqlReleaseTableBodyColumn).
+		From(sqlReleaseTableName).
+		Where(sq.Eq{sqlReleaseTableOwnerColumn: sqlReleaseDefaultOwner})
+
+	// If a namespace was specified, we only list releases from that namespace
+	if s.namespace != "" {
+		sb = sb.Where(sq.Eq{sqlReleaseTableNamespaceColumn: s.namespace})
+	}
+
+	for {
+		query, args, err := sb.Limit(uint64(limit)).ToSql()
+		if err != nil {
+			s.Logger().Debug("failed to build query", slog.Any("error", err))
+			return err
+		}
+
+		var records = []SQLReleaseWrapper{}
+		if err := s.db.Select(&records, query, args...); err != nil {
+			s.Logger().Debug("failed to list", slog.Any("error", err))
+			return err
+		}
+
+		var releases []release.Releaser
+		for _, record := range records {
+			release, err := decodeRelease(record.Body)
+			if err != nil {
+				s.Logger().Debug("failed to decode release", "record", record, slog.Any("error", err))
+				continue
+			}
+
+			if release.Labels, err = s.getReleaseCustomLabels(record.Key, record.Namespace); err != nil {
+				s.Logger().Debug("failed to get release custom labels", "namespace", record.Namespace, "key", record.Key, slog.Any("error", err))
+				return err
+			}
+			maps.Copy(release.Labels, getReleaseSystemLabels(release))
+
+			if filter(release) {
+				releases = append(releases, release)
+			}
+		}
+
+		if f(releases, int64(len(records)) < limit) {
+			break
+		}
+	}
+
+	return nil
+}
+
 // Query returns the set of releases that match the provided set of labels.
 func (s *SQL) Query(labels map[string]string) ([]release.Releaser, error) {
 	sb := s.statementBuilder.
@@ -448,6 +504,73 @@ func (s *SQL) Query(labels map[string]string) ([]release.Releaser, error) {
 	return releases, nil
 }
 
+// QueryPages same as Query, but with pagination
+func (s *SQL) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, labels map[string]string) error {
+	sb := s.statementBuilder.
+		Select(sqlReleaseTableKeyColumn, sqlReleaseTableNamespaceColumn, sqlReleaseTableBodyColumn).
+		From(sqlReleaseTableName)
+
+	keys := make([]string, 0, len(labels))
+	for key := range labels {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		if _, ok := labelMap[key]; ok {
+			sb = sb.Where(sq.Eq{key: labels[key]})
+		} else {
+			s.Logger().Debug("unknown label", "key", key)
+			return fmt.Errorf("unknown label %s", key)
+		}
+	}
+
+	// If a namespace was specified, we only list releases from that namespace
+	if s.namespace != "" {
+		sb = sb.Where(sq.Eq{sqlReleaseTableNamespaceColumn: s.namespace})
+	}
+
+	for {
+		// Build our query
+		query, args, err := sb.Limit(uint64(limit)).ToSql()
+		if err != nil {
+			s.Logger().Debug("failed to build query", slog.Any("error", err))
+			return err
+		}
+
+		var records = []SQLReleaseWrapper{}
+		if err := s.db.Select(&records, query, args...); err != nil {
+			s.Logger().Debug("failed to query with labels", slog.Any("error", err))
+			return err
+		}
+
+		if len(records) == 0 {
+			return ErrReleaseNotFound
+		}
+
+		var releases []release.Releaser
+		for _, record := range records {
+			release, err := decodeRelease(record.Body)
+			if err != nil {
+				s.Logger().Debug("failed to decode release", "record", record, slog.Any("error", err))
+				continue
+			}
+
+			if release.Labels, err = s.getReleaseCustomLabels(record.Key, record.Namespace); err != nil {
+				s.Logger().Debug("failed to get release custom labels", "namespace", record.Namespace, "key", record.Key, slog.Any("error", err))
+				return err
+			}
+
+			releases = append(releases, release)
+		}
+
+		if f(releases, int64(len(records)) < limit) {
+			break
+		}
+	}
+
+	return nil
+}
+
 // Create creates a new release.
 func (s *SQL) Create(key string, rel release.Releaser) error {
 	rls, err := releaserToV1Release(rel)

pkg/storage/storage_test.go

@@ -314,9 +314,15 @@ func (d *MaxHistoryMockDriver) Get(key string) (release.Releaser, error) {
 func (d *MaxHistoryMockDriver) List(filter func(release.Releaser) bool) ([]release.Releaser, error) {
 	return d.Driver.List(filter)
 }
+func (d *MaxHistoryMockDriver) ListPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, filter func(release.Releaser) bool) error {
+	return d.Driver.ListPages(f, limit, filter)
+}
 func (d *MaxHistoryMockDriver) Query(labels map[string]string) ([]release.Releaser, error) {
 	return d.Driver.Query(labels)
 }
+func (d *MaxHistoryMockDriver) QueryPages(f func(page []release.Releaser, lastPage bool) (end bool), limit int64, labels map[string]string) error {
+	return d.Driver.QueryPages(f, limit, labels)
+}
 func (d *MaxHistoryMockDriver) Name() string {
 	return d.Driver.Name()
 }