Skip to content

agent: improve log subscription lifecycle and event handling - #3259

Open
thaJeztah wants to merge 9 commits into
moby:masterfrom
thaJeztah:better_watch
Open

agent: improve log subscription lifecycle and event handling#3259
thaJeztah wants to merge 9 commits into
moby:masterfrom
thaJeztah:better_watch

Conversation

@thaJeztah

@thaJeztah thaJeztah commented Jul 26, 2026

Copy link
Copy Markdown
Member

agent: Agent.Publisher: scope variables

agent: worker.Subscribe: use waitgroup.Go

agent: worker.Subscribe: simplify subscription event loop

  • return early for closed event channels
  • use early "continue" to reduce nesting

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 Logs to return when its
context is cancelled, and taskManager.Logs passes the
subscription 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.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.

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

@codecov-commenter

codecov-commenter commented Jul 26, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 0% with 44 lines in your changes missing coverage. Please review.
✅ Project coverage is 14.70%. Comparing base (6e9e7b8) to head (f71c011).
⚠️ Report is 56 commits behind head on master.

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:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@thaJeztah

This comment was marked as outdated.

@thaJeztah
thaJeztah force-pushed the better_watch branch 5 times, most recently from f04f45c to 041e714 Compare July 28, 2026 20:56
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>
@thaJeztah
thaJeztah marked this pull request as ready for review July 29, 2026 15:07
@thaJeztah
thaJeztah requested a review from Copilot July 29, 2026 15:07

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

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.Subscribe to snapshot matching tasks under RLock, start log streams once per task, and (in follow mode) rely on a filtered task event watcher.
  • Ensures Subscribe waits for all started log streams to exit before returning, and avoids nil dereferences when subscription.Options is absent.
  • Simplifies Agent.Publisher variable scoping and makes the ignored CloseAndRecv return 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 thread agent/worker.go
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)
}))
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants