Skip to content

fix: stop custom resource reflectors on context cancellation #2883

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
colega wants to merge 2 commits into from
+23 −4
Closed

fix: stop custom resource reflectors on context cancellation #2883

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6863cff
fix: stop custom resource reflectors on context cancellation
colega Mar 6, 2026
6efdcb4
fix: Builder owns its context lifecycle to avoid data races
colega Mar 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 23 additions & 4 deletions internal/store/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ var _ ksmtypes.BuilderInterface = &Builder{}
type Builder struct {
kubeClient clientset.Interface
ctx context.Context
cancel func()
familyGeneratorFilter generator.FamilyGeneratorFilter
customResourceClients map[string]interface{}
listWatchMetrics *watch.ListWatchMetrics
Expand All @@ -91,7 +92,8 @@ type Builder struct {

// NewBuilder returns a new builder.
func NewBuilder() *Builder {
b := &Builder{}
ctx, cancel := context.WithCancel(context.Background())
b := &Builder{ctx: ctx, cancel: cancel}
return b
}

Expand Down Expand Up @@ -149,9 +151,17 @@ func (b *Builder) WithSharding(shard int32, totalShards int) {
b.shardingMetrics.Total.Set(float64(totalShards))
}

// WithContext sets the ctx property of a Builder.
// WithContext will observe the cancellations of the provided context
// to cancel the internal b.ctx.
func (b *Builder) WithContext(ctx context.Context) {
b.ctx = ctx
// WithContext might be called concurrently with startReflector.
// In order to avoid the data race, and also to notify the reflectors that
// are already running with the context, we don't replace b.ctx,
// but just subscribe to the cancellations of the provided context.
go func() {
<-ctx.Done()
b.cancel()
}()
}
Comment on lines +154 to 165
Copy link

Copilot AI Mar 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

WithContext now cancels a single internal context (created in NewBuilder) when the provided ctx is canceled, but that internal context is never replaced/reset. Since MetricsHandler.BuildWriters cancels the previous ctx on every rebuild, the second BuildWriters call will cancel b.ctx permanently, causing all subsequently started reflectors (including non-CR ones using b.ctx.Done()) to stop immediately and the cache to stop updating. To preserve rebuild behavior, WithContext should continue to replace the builder context (or atomically swap to a new derived context/cancel func) rather than subscribing old contexts to cancel a long-lived internal context.

Copilot uses AI. Check for mistakes.

// WithKubeClient sets the kubeClient property of a Builder.
Expand Down Expand Up @@ -621,7 +631,16 @@ func (b *Builder) startReflector(
instrumentedListWatch := watch.NewInstrumentedListerWatcher(listWatcher, b.listWatchMetrics, reflect.TypeOf(expectedType).String(), useAPIServerCache, objectLimit)
reflector := cache.NewReflectorWithOptions(sharding.NewShardedListWatch(b.shard, b.totalShards, instrumentedListWatch), expectedType, store, cache.ReflectorOptions{ResyncPeriod: 0})
if cr, ok := expectedType.(*unstructured.Unstructured); ok {
go reflector.Run((*b.GVKToReflectorStopChanMap)[cr.GroupVersionKind().String()])
stopCh := make(chan struct{})
crStopCh := (*b.GVKToReflectorStopChanMap)[cr.GroupVersionKind().String()]
go func(builderContext context.Context) {
select {
case <-builderContext.Done():
case <-crStopCh:
}
close(stopCh)
}(b.ctx)
Comment on lines +634 to +642
Copy link

Copilot AI Mar 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new stop-signal merge for custom resource reflectors is a concurrency/cancellation fix but isn't covered by tests in internal/store/builder_test.go. Adding a focused unit/integration test that exercises a CR reflector stopping on context cancellation (without closing the per-GVK stop channel) would help prevent regressions like goroutine leaks or reflectors not stopping on rebuild.

Suggested change
stopCh := make(chan struct{})
crStopCh := (*b.GVKToReflectorStopChanMap)[cr.GroupVersionKind().String()]
go func(builderContext context.Context) {
select {
case <-builderContext.Done():
case <-crStopCh:
}
close(stopCh)
}(b.ctx)
// For custom resources, prefer a per-GVK stop channel, but fall back to the
// builder context if the map or entry is not available.
if b.GVKToReflectorStopChanMap == nil {
go reflector.Run(b.ctx.Done())
return
}
gvk := cr.GroupVersionKind().String()
crStopCh, found := (*b.GVKToReflectorStopChanMap)[gvk]
if !found || crStopCh == nil {
go reflector.Run(b.ctx.Done())
return
}
stopCh := make(chan struct{})
go func(builderContext context.Context, crStopCh <-chan struct{}, stopCh chan struct{}) {
defer close(stopCh)
select {
case <-builderContext.Done():
case <-crStopCh:
}
}(b.ctx, crStopCh, stopCh)

Copilot uses AI. Check for mistakes.
go reflector.Run(stopCh)
} else {
go reflector.Run(b.ctx.Done())
}
Expand Down
Loading