Skip to content
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

Jump to bottom

🐛 Prevent source.Channel from shutting down immediately #1345

Merged
merged 1 commit into from Jan 19, 2021
Merged

🐛 Prevent source.Channel from shutting down immediately #1345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
26 changes: 15 additions & 11 deletions pkg/internal/controller/controller.go
View file
Open in desktop
Expand Up @@ -163,9 +163,7 @@ func (c *Controller) Start(ctx context.Context) error {
for _, watch := range c.startWatches {
c.Log.Info("Starting EventSource", "source", watch.src)

watchStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
defer cancel()
if err := watch.src.Start(watchStartCtx, watch.handler, c.Queue, watch.predicates...); err != nil {
if err := watch.src.Start(ctx, watch.handler, c.Queue, watch.predicates...); err != nil {
Copy link
Member

Choose a reason for hiding this comment

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

Unfortunately, what we do today is the correct behaviour for source.Kind, it has a different interpretation of the purpose of its ctx argument. Let's merge this as source.Channel doesn't work at all currently, I will open a n issue so we can clarify our interface definitions and update one or the other accordingly

Copy link
Member

Choose a reason for hiding this comment

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

(This change will break the waitForSync functionality I believe)

return err
}
}
Expand All @@ -179,15 +177,21 @@ func (c *Controller) Start(ctx context.Context) error {
continue
}

// use a context with timeout for launching sources and syncing caches.
sourceStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
defer cancel()
if err := func() error {
// use a context with timeout for launching sources and syncing caches.
sourceStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
defer cancel()

// WaitForSync waits for a definitive timeout, and returns if there
// is an error or a timeout
if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
err := fmt.Errorf("failed to wait for %s caches to sync: %w", c.Name, err)
c.Log.Error(err, "Could not wait for Cache to sync")
return err
}

// WaitForSync waits for a definitive timeout, and returns if there
// is an error or a timeout
if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
err := fmt.Errorf("failed to wait for %s caches to sync: %w", c.Name, err)
c.Log.Error(err, "Could not wait for Cache to sync")
return nil
}(); err != nil {
return err
}
}
Expand Down
6 changes: 3 additions & 3 deletions pkg/source/source.go
View file
Open in desktop
Expand Up @@ -215,7 +215,10 @@ func (cs *Channel) Start(
}

dst := make(chan event.GenericEvent, cs.DestBufferSize)

cs.destLock.Lock()
cs.dest = append(cs.dest, dst)
cs.destLock.Unlock()

cs.once.Do(func() {
// Distribute GenericEvents to all EventHandler / Queue pairs Watching this source
Expand All @@ -238,9 +241,6 @@ func (cs *Channel) Start(
}
}()

cs.destLock.Lock()
defer cs.destLock.Unlock()

return nil
}

Expand Down