Skip to content

Commit b0fffe4

Browse files
authored
fix: resolve deadlock (#539)
Signed-off-by: Soumya Ghosh Dastidar <[email protected]>
1 parent 187312f commit b0fffe4

File tree

1 file changed

+10
-6
lines changed

1 file changed

+10
-6
lines changed

pkg/cache/cluster.go

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -490,7 +490,7 @@ func (c *clusterCache) startMissingWatches() error {
490490
c.apisMeta[api.GroupKind] = &apiMeta{namespaced: api.Meta.Namespaced, watchCancel: cancel}
491491

492492
err := c.processApi(client, api, func(resClient dynamic.ResourceInterface, ns string) error {
493-
resourceVersion, err := c.loadInitialState(ctx, api, resClient, ns)
493+
resourceVersion, err := c.loadInitialState(ctx, api, resClient, ns, false) // don't lock here, we are already in a lock before startMissingWatches is called inside watchEvents
494494
if err != nil && c.isRestrictedResource(err) {
495495
keep := false
496496
if c.respectRBAC == RespectRbacStrict {
@@ -566,7 +566,7 @@ func (c *clusterCache) listResources(ctx context.Context, resClient dynamic.Reso
566566
return resourceVersion, callback(listPager)
567567
}
568568

569-
func (c *clusterCache) loadInitialState(ctx context.Context, api kube.APIResourceInfo, resClient dynamic.ResourceInterface, ns string) (string, error) {
569+
func (c *clusterCache) loadInitialState(ctx context.Context, api kube.APIResourceInfo, resClient dynamic.ResourceInterface, ns string, lock bool) (string, error) {
570570
return c.listResources(ctx, resClient, func(listPager *pager.ListPager) error {
571571
var items []*Resource
572572
err := listPager.EachListItem(ctx, metav1.ListOptions{}, func(obj runtime.Object) error {
@@ -581,11 +581,15 @@ func (c *clusterCache) loadInitialState(ctx context.Context, api kube.APIResourc
581581
if err != nil {
582582
return fmt.Errorf("failed to load initial state of resource %s: %w", api.GroupKind.String(), err)
583583
}
584-
585-
return runSynced(&c.lock, func() error {
584+
if lock {
585+
return runSynced(&c.lock, func() error {
586+
c.replaceResourceCache(api.GroupKind, items, ns)
587+
return nil
588+
})
589+
} else {
586590
c.replaceResourceCache(api.GroupKind, items, ns)
587591
return nil
588-
})
592+
}
589593
})
590594
}
591595

@@ -599,7 +603,7 @@ func (c *clusterCache) watchEvents(ctx context.Context, api kube.APIResourceInfo
599603

600604
// load API initial state if no resource version provided
601605
if resourceVersion == "" {
602-
resourceVersion, err = c.loadInitialState(ctx, api, resClient, ns)
606+
resourceVersion, err = c.loadInitialState(ctx, api, resClient, ns, true)
603607
if err != nil {
604608
return err
605609
}

0 commit comments

Comments
 (0)