diff --git a/pkg/storage/driver/cfgmaps.go b/pkg/storage/driver/cfgmaps.go index 5af432d8a..7d1c53194 100644 --- a/pkg/storage/driver/cfgmaps.go +++ b/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 { diff --git a/pkg/storage/driver/cfgmaps_test.go b/pkg/storage/driver/cfgmaps_test.go index 8beb45547..353ff39ed 100644 --- a/pkg/storage/driver/cfgmaps_test.go +++ b/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) { diff --git a/pkg/storage/driver/driver.go b/pkg/storage/driver/driver.go index 6efd1dbaa..0a2f02af6 100644 --- a/pkg/storage/driver/driver.go +++ b/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 diff --git a/pkg/storage/driver/memory.go b/pkg/storage/driver/memory.go index 7ea4a014a..fdc5e6dec 100644 --- a/pkg/storage/driver/memory.go +++ b/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()) diff --git a/pkg/storage/driver/memory_test.go b/pkg/storage/driver/memory_test.go index 329b82b2f..61e4e1429 100644 --- a/pkg/storage/driver/memory_test.go +++ b/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) diff --git a/pkg/storage/driver/mock_test.go b/pkg/storage/driver/mock_test.go index e62b02f43..c426c8937 100644 --- a/pkg/storage/driver/mock_test.go +++ b/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 } diff --git a/pkg/storage/driver/secrets.go b/pkg/storage/driver/secrets.go index 67315b2d1..b0f8cc851 100644 --- a/pkg/storage/driver/secrets.go +++ b/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 }) } diff --git a/pkg/storage/driver/secrets_test.go b/pkg/storage/driver/secrets_test.go index a70c55793..24e989654 100644 --- a/pkg/storage/driver/secrets_test.go +++ b/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) } diff --git a/pkg/storage/driver/sql.go b/pkg/storage/driver/sql.go index b6ea3916d..bd099fb0c 100644 --- a/pkg/storage/driver/sql.go +++ b/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) diff --git a/pkg/storage/storage_test.go b/pkg/storage/storage_test.go index 5b2a3bba5..63bd2431f 100644 --- a/pkg/storage/storage_test.go +++ b/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() }