mirror of
https://github.com/fluxcd/flux2.git
synced 2026-02-19 22:21:46 +00:00
Fix event listing ignoring pagination token
Signed-off-by: Matheus Pimenta <matheuscscp@gmail.com>
This commit is contained in:
parent
99f182be06
commit
4eddf80724
2 changed files with 109 additions and 2 deletions
|
|
@ -196,11 +196,14 @@ func getRows(ctx context.Context, kubeclient client.Client, clientListOpts []cli
|
||||||
|
|
||||||
func addEventsToList(ctx context.Context, kubeclient client.Client, el *corev1.EventList, clientListOpts []client.ListOption) error {
|
func addEventsToList(ctx context.Context, kubeclient client.Client, el *corev1.EventList, clientListOpts []client.ListOption) error {
|
||||||
listOpts := &metav1.ListOptions{}
|
listOpts := &metav1.ListOptions{}
|
||||||
clientListOpts = append(clientListOpts, client.Limit(cmdutil.DefaultChunkSize))
|
|
||||||
err := runtimeresource.FollowContinue(listOpts,
|
err := runtimeresource.FollowContinue(listOpts,
|
||||||
func(options metav1.ListOptions) (runtime.Object, error) {
|
func(options metav1.ListOptions) (runtime.Object, error) {
|
||||||
newEvents := &corev1.EventList{}
|
newEvents := &corev1.EventList{}
|
||||||
if err := kubeclient.List(ctx, newEvents, clientListOpts...); err != nil {
|
opts := append(clientListOpts, client.Limit(cmdutil.DefaultChunkSize))
|
||||||
|
if options.Continue != "" {
|
||||||
|
opts = append(opts, client.Continue(options.Continue))
|
||||||
|
}
|
||||||
|
if err := kubeclient.List(ctx, newEvents, opts...); err != nil {
|
||||||
return nil, fmt.Errorf("error getting events: %w", err)
|
return nil, fmt.Errorf("error getting events: %w", err)
|
||||||
}
|
}
|
||||||
el.Items = append(el.Items, newEvents.Items...)
|
el.Items = append(el.Items, newEvents.Items...)
|
||||||
|
|
|
||||||
|
|
@ -20,11 +20,13 @@ package main
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
. "github.com/onsi/gomega"
|
. "github.com/onsi/gomega"
|
||||||
corev1 "k8s.io/api/core/v1"
|
corev1 "k8s.io/api/core/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/api/meta"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
"k8s.io/apimachinery/pkg/fields"
|
"k8s.io/apimachinery/pkg/fields"
|
||||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
|
|
@ -419,6 +421,108 @@ func createEvent(obj client.Object, eventType, msg, reason string) corev1.Event
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// paginatedClient wraps a client.Client and simulates real Kubernetes API
|
||||||
|
// pagination by splitting List results into pages of pageSize items,
|
||||||
|
// using the ListMeta.Continue token.
|
||||||
|
type paginatedClient struct {
|
||||||
|
client.Client
|
||||||
|
pageSize int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *paginatedClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
|
||||||
|
listOpts := &client.ListOptions{}
|
||||||
|
listOpts.ApplyOptions(opts)
|
||||||
|
|
||||||
|
// Fetch all results from the underlying client (without Limit/Continue).
|
||||||
|
stripped := make([]client.ListOption, 0, len(opts))
|
||||||
|
for _, o := range opts {
|
||||||
|
if _, ok := o.(client.Limit); ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := o.(client.Continue); ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
stripped = append(stripped, o)
|
||||||
|
}
|
||||||
|
if err := c.Client.List(ctx, list, stripped...); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
items, err := meta.ExtractList(list)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Determine the page window based on the Continue token.
|
||||||
|
start := 0
|
||||||
|
if listOpts.Continue != "" {
|
||||||
|
n, err := strconv.Atoi(listOpts.Continue)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("invalid continue token: %w", err)
|
||||||
|
}
|
||||||
|
start = n
|
||||||
|
}
|
||||||
|
if start > len(items) {
|
||||||
|
start = len(items)
|
||||||
|
}
|
||||||
|
|
||||||
|
end := start + c.pageSize
|
||||||
|
if end > len(items) {
|
||||||
|
end = len(items)
|
||||||
|
}
|
||||||
|
|
||||||
|
page := items[start:end]
|
||||||
|
if err := meta.SetList(list, page); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Set the Continue token when there are more pages.
|
||||||
|
listAccessor, err := meta.ListAccessor(list)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if end < len(items) {
|
||||||
|
listAccessor.SetContinue(strconv.Itoa(end))
|
||||||
|
} else {
|
||||||
|
listAccessor.SetContinue("")
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_addEventsToList_pagination(t *testing.T) {
|
||||||
|
g := NewWithT(t)
|
||||||
|
objs, err := ssautil.ReadObjects(strings.NewReader(objects))
|
||||||
|
g.Expect(err).To(Not(HaveOccurred()))
|
||||||
|
|
||||||
|
builder := fake.NewClientBuilder().WithScheme(utils.NewScheme())
|
||||||
|
for _, obj := range objs {
|
||||||
|
builder = builder.WithObjects(obj)
|
||||||
|
}
|
||||||
|
|
||||||
|
eventList := &corev1.EventList{}
|
||||||
|
for _, obj := range objs {
|
||||||
|
infoEvent := createEvent(obj, eventv1.EventSeverityInfo, "Info Message", "Info Reason")
|
||||||
|
warningEvent := createEvent(obj, eventv1.EventSeverityError, "Error Message", "Error Reason")
|
||||||
|
eventList.Items = append(eventList.Items, infoEvent, warningEvent)
|
||||||
|
}
|
||||||
|
builder = builder.WithLists(eventList)
|
||||||
|
c := builder.Build()
|
||||||
|
|
||||||
|
totalEvents := len(eventList.Items)
|
||||||
|
g.Expect(totalEvents).To(BeNumerically(">", 2), "need more than 2 events to test pagination")
|
||||||
|
|
||||||
|
// Wrap the client to paginate at 2 items per page, forcing multiple
|
||||||
|
// round-trips through FollowContinue.
|
||||||
|
pc := &paginatedClient{Client: c, pageSize: 2}
|
||||||
|
|
||||||
|
el := &corev1.EventList{}
|
||||||
|
err = addEventsToList(context.Background(), pc, el, nil)
|
||||||
|
g.Expect(err).To(Not(HaveOccurred()))
|
||||||
|
g.Expect(el.Items).To(HaveLen(totalEvents),
|
||||||
|
"addEventsToList should collect all events across paginated responses")
|
||||||
|
}
|
||||||
|
|
||||||
func kindNameIndexer(obj client.Object) []string {
|
func kindNameIndexer(obj client.Object) []string {
|
||||||
e, ok := obj.(*corev1.Event)
|
e, ok := obj.(*corev1.Event)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue