diff --git a/pkg/cache/defaulting_test.go b/pkg/cache/defaulting_test.go index 36cbb4eccb..5882b0828a 100644 --- a/pkg/cache/defaulting_test.go +++ b/pkg/cache/defaulting_test.go @@ -17,14 +17,18 @@ limitations under the License. package cache import ( + "context" + "errors" "reflect" "sync" "testing" + "testing/synctest" "time" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" fuzz "github.com/google/gofuzz" + . "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/fields" @@ -517,3 +521,126 @@ func TestDefaultConfigConsidersAllFields(t *testing.T) { } } } + +// mockCache is a mock implementation of Cache for testing Start() behavior. +type mockCache struct { + startFunc func(ctx context.Context) error + Cache +} + +func (m *mockCache) Start(ctx context.Context) error { + if m.startFunc != nil { + return m.startFunc(ctx) + } + <-ctx.Done() + return nil +} + +func TestMultiNamespaceCacheStart_MultipleErrors(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + + errTest := errors.New("test error") + namespaceToCache := map[string]Cache{ + "ns1": &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + "ns2": &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + "ns3": &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + } + + c := &multiNamespaceCache{ + namespaceToCache: namespaceToCache, + } + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + err := c.Start(ctx) + g.Expect(err).To(HaveOccurred()) + g.Expect(err).To(MatchError(errTest)) + }) +} + +func TestMultiNamespaceCacheStart_ContextCancel(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + + normalCache := &mockCache{ + startFunc: func(ctx context.Context) error { + <-ctx.Done() + return nil + }, + } + + namespaceToCache := map[string]Cache{ + "ns1": normalCache, + "ns2": normalCache, + } + + c := &multiNamespaceCache{ + namespaceToCache: namespaceToCache, + } + + ctx, cancel := context.WithCancel(t.Context()) + cancel() // Cancel immediately + + err := c.Start(ctx) + g.Expect(err).NotTo(HaveOccurred()) + }) +} + +func TestDelegatingByGVKCacheStart_MultipleErrors(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + + errTest := errors.New("test error") + caches := map[schema.GroupVersionKind]Cache{ + {Group: "apps", Version: "v1", Kind: "Deployment"}: &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + {Group: "apps", Version: "v1", Kind: "StatefulSet"}: &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + {Group: "apps", Version: "v1", Kind: "DaemonSet"}: &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + } + + c := &delegatingByGVKCache{ + caches: caches, + defaultCache: &mockCache{startFunc: func(ctx context.Context) error { return errTest }}, + } + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + err := c.Start(ctx) + g.Expect(err).To(HaveOccurred()) + g.Expect(err).To(MatchError(errTest)) + }) +} + +func TestDelegatingByGVKCacheStart_ContextCancel(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + + normalCache := &mockCache{ + startFunc: func(ctx context.Context) error { + <-ctx.Done() + return nil + }, + } + + caches := map[schema.GroupVersionKind]Cache{ + {Group: "apps", Version: "v1", Kind: "Deployment"}: normalCache, + } + + c := &delegatingByGVKCache{ + caches: caches, + defaultCache: normalCache, + } + + ctx, cancel := context.WithCancel(t.Context()) + cancel() // Cancel immediately + + err := c.Start(ctx) + g.Expect(err).NotTo(HaveOccurred()) + }) +} diff --git a/pkg/cache/delegating_by_gvk_cache.go b/pkg/cache/delegating_by_gvk_cache.go index adc5d957a4..faeb55454b 100644 --- a/pkg/cache/delegating_by_gvk_cache.go +++ b/pkg/cache/delegating_by_gvk_cache.go @@ -21,8 +21,8 @@ import ( "maps" "slices" "strings" - "sync" + "golang.org/x/sync/errgroup" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "sigs.k8s.io/controller-runtime/pkg/client" @@ -77,24 +77,15 @@ func (dbt *delegatingByGVKCache) Start(ctx context.Context) error { allCaches := slices.Collect(maps.Values(dbt.caches)) allCaches = append(allCaches, dbt.defaultCache) - wg := &sync.WaitGroup{} - errs := make(chan error) + group, childCtx := errgroup.WithContext(ctx) for idx := range allCaches { cache := allCaches[idx] - wg.Go(func() { - if err := cache.Start(ctx); err != nil { - errs <- err - } + group.Go(func() error { + return cache.Start(childCtx) }) } - select { - case err := <-errs: - return err - case <-ctx.Done(): - wg.Wait() - return nil - } + return ignoreContextCanceled(group.Wait()) } func (dbt *delegatingByGVKCache) WaitForCacheSync(ctx context.Context) bool { diff --git a/pkg/cache/multi_namespace_cache.go b/pkg/cache/multi_namespace_cache.go index 592519c35d..05fdcb8655 100644 --- a/pkg/cache/multi_namespace_cache.go +++ b/pkg/cache/multi_namespace_cache.go @@ -18,10 +18,12 @@ package cache import ( "context" + "errors" "fmt" "strings" "time" + "golang.org/x/sync/errgroup" corev1 "k8s.io/api/core/v1" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -160,31 +162,34 @@ func (c *multiNamespaceCache) GetInformerForKind(ctx context.Context, gvk schema } func (c *multiNamespaceCache) Start(ctx context.Context) error { - errs := make(chan error) - // start global cache + group, childCtx := errgroup.WithContext(ctx) + if c.clusterCache != nil { - go func() { - err := c.clusterCache.Start(ctx) - if err != nil { - errs <- fmt.Errorf("failed to start cluster-scoped cache: %w", err) - } - }() + clusterCache := c.clusterCache + group.Go(func() error { + return clusterCache.Start(childCtx) + }) } - // start namespaced caches for ns, cache := range c.namespaceToCache { - go func(ns string, cache Cache) { - if err := cache.Start(ctx); err != nil { - errs <- fmt.Errorf("failed to start cache for namespace %s: %w", ns, err) + group.Go(func() error { + if err := cache.Start(childCtx); err != nil { + return fmt.Errorf("failed to start cache for namespace %s: %w", ns, err) } - }(ns, cache) + return nil + }) } - select { - case <-ctx.Done(): + + return ignoreContextCanceled(group.Wait()) +} + +// ignoreContextCanceled returns nil if the error is a context.Canceled error, +// otherwise returns the error unchanged. +func ignoreContextCanceled(err error) error { + if errors.Is(err, context.Canceled) { return nil - case err := <-errs: - return err } + return err } func (c *multiNamespaceCache) WaitForCacheSync(ctx context.Context) bool {