agent: improve log subscription lifecycle and event handling - #3259
Open
thaJeztah wants to merge 9 commits into
Open
agent: improve log subscription lifecycle and event handling#3259thaJeztah wants to merge 9 commits into
thaJeztah wants to merge 9 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3259 +/- ##
==========================================
- Coverage 14.73% 14.70% -0.03%
==========================================
Files 200 200
Lines 93077 93033 -44
==========================================
- Hits 13712 13679 -33
+ Misses 78019 78013 -6
+ Partials 1346 1341 -5 🚀 New features to boost your workflow:
|
This comment was marked as outdated.
This comment was marked as outdated.
thaJeztah
force-pushed
the
better_watch
branch
5 times, most recently
from
July 28, 2026 20:56
f04f45c to
041e714
Compare
Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
- return early for closed event channels - use early "continue" to reduce nesting Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Collecting the initial set of matching task managers only reads from w.taskManagers; no shared state is modified. Use RLock instead of Lock to allow concurrent readers while still protecting iteration over the map. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Avoid a panic if subscription.Options is nil; the function already had guards in place for the "Follow" option, but lacked guards in code before that. While updating, also update the debug-logs to structured logs, and include the Follow option. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
The [ControllerLogs contract][1] requires `Logs` to return when its context is cancelled, and [`taskManager.Logs`][2] passes the subscription context through to the controller implementation. Both the [swarmd implementation][3] and the [dockerd implementation][4] honour this contract by propagating the context through their log handling and cancellation paths. Wait for the active log streams directly instead of starting a separate goroutine and channel solely to make `WaitGroup.Wait` selectable. Previously, `Subscribe` could return as soon as the context was cancelled, closing the publisher before all active log streams had finished handling cancellation. It could also leave behind the goroutine waiting in `WaitGroup.Wait`, along with any `Logs` goroutines that had not yet returned. Waiting directly ensures `Subscribe` does not return until all log goroutines it started have exited. If an implementation violates the `ControllerLogs` contract and fails to return on cancellation, the subscription now remains blocked instead of silently abandoning those goroutines. [1]: https://github.com/moby/swarmkit/blob/v2.1.2/agent/exec/controller.go#L47-L54 [2]: https://github.com/moby/swarmkit/blob/v2.1.2/agent/task.go#L65-L75 [3]: https://github.com/moby/swarmkit/blob/v2.1.2/swarmd/dockerexec/controller.go#L460-L537 [4]: https://github.com/moby/moby/blob/docker-v29.6.2/daemon/cluster/executor/container/controller.go#L505-L592 Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Use the same WaitGroup for log streams started while handling follow-mode task events. This ensures Subscribe waits for all log streams it starts before returning. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Collect the initial set of matching task managers while holding the worker's read lock, then release the lock before starting their log streams. This reduces the time the worker lock is held and avoids starting goroutines while holding the lock. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
…shot Register the task event watcher before collecting the initial set of matching task managers for follow-mode subscriptions. This narrows the window in which newly created tasks could otherwise be missed between taking the snapshot and starting the watcher. Use CallbackWatchContext with a matcher to receive only matching task events, allowing the event loop to rely on the watcher for filtering and context cancellation. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
Track task IDs for which a subscription has already started a log stream. A task may be present in the initial snapshot and also be delivered by the watcher after it is registered. Use a shared helper for both paths to ensure that only one log stream is started for each task. Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
There was a problem hiding this comment.
Pull request overview
Improves the agent-side log subscription handling to better align with the ControllerLogs cancellation contract, reduce locking scope, and avoid duplicate log streams in follow mode.
Changes:
- Refactors
worker.Subscribeto snapshot matching tasks underRLock, start log streams once per task, and (in follow mode) rely on a filtered task event watcher. - Ensures
Subscribewaits for all started log streams to exit before returning, and avoids nil dereferences whensubscription.Optionsis absent. - Simplifies
Agent.Publishervariable scoping and makes the ignoredCloseAndRecvreturn values explicit.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| agent/worker.go | Reworks log subscription lifecycle: options handling, task snapshot + follow-mode watcher, deduped log stream starts, and WaitGroup-based shutdown ordering. |
| agent/agent.go | Minor cleanup in Agent.Publisher scoping and explicit discard of CloseAndRecv results. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+625
to
+633
| var ch <-chan events.Event | ||
| if options.Follow { | ||
| // Start watching before collecting the current task managers so that | ||
| // tasks added while taking the snapshot are queued for processing. | ||
| ch = w.taskevents.CallbackWatchContext(ctx, events.MatcherFunc(func(v events.Event) bool { | ||
| task, ok := v.(*api.Task) | ||
| return ok && match(task) | ||
| })) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
agent: Agent.Publisher: scope variables
agent: worker.Subscribe: use waitgroup.Go
agent: worker.Subscribe: simplify subscription event loop
agent: worker.Subscribe: use read-lock
Collecting the initial set of matching task managers only reads from
w.taskManagers; no shared state is modified. Use RLock instead of
Lock to allow concurrent readers while still protecting iteration over
the map.
agent: worker.Subscribe: avoid nil dereference
Avoid a panic if subscription.Options is nil; the function already
had guards in place for the "Follow" option, but lacked guards in
code before that.
While updating, also update the debug-logs to structured logs, and
include the Follow option.
agent: worker.Subscribe: wait for log streams to finish
The ControllerLogs contract requires
Logsto return when itscontext is cancelled, and
taskManager.Logspasses thesubscription context through to the controller implementation.
Both the swarmd implementation and the dockerd implementation
honour this contract by propagating the context through their log
handling and cancellation paths.
Wait for the active log streams directly instead of starting a separate
goroutine and channel solely to make
WaitGroup.Waitselectable.Previously,
Subscribecould return as soon as the context wascancelled, closing the publisher before all active log streams had
finished handling cancellation. It could also leave behind the goroutine
waiting in
WaitGroup.Wait, along with anyLogsgoroutines that hadnot yet returned.
Waiting directly ensures
Subscribedoes not return until all loggoroutines it started have exited. If an implementation violates the
ControllerLogscontract and fails to return on cancellation, thesubscription now remains blocked instead of silently abandoning those
goroutines.
agent: worker.Subscribe: follow-mode: wait for log streams to finish
Use the same WaitGroup for log streams started while handling follow-mode
task events. This ensures Subscribe waits for all log streams it starts
before returning.
agent: worker.Subscribe:: reduce lock scope when starting log streams
Collect the initial set of matching task managers while holding the
worker's read lock, then release the lock before starting their log
streams.
This reduces the time the worker lock is held and avoids starting
goroutines while holding the lock.
agent: worker.Subscribe: register log watcher before taking task snapshot
Register the task event watcher before collecting the initial set of
matching task managers for follow-mode subscriptions.
This narrows the window in which newly created tasks could otherwise be
missed between taking the snapshot and starting the watcher.
Use CallbackWatchContext with a matcher to receive only matching task
events, allowing the event loop to rely on the watcher for filtering and
context cancellation.
agent: worker.Subscribe: start at most one log stream per task
Track task IDs for which a subscription has already started a log
stream.
A task may be present in the initial snapshot and also be delivered by
the watcher after it is registered. Use a shared helper for both paths
to ensure that only one log stream is started for each task.
- How to test it
- Description for the changelog