From b69498124ed34ae70079da066f2d1d09efec2489 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 12:51:55 +0200 Subject: [PATCH 1/9] agent: Agent.Publisher: scope variables Signed-off-by: Sebastiaan van Stijn --- agent/agent.go | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index 9acec60bfe..d0b02fdebb 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -582,12 +582,9 @@ func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogP // These should only be best effort and really just buffer until a session is // ready. Ideally, they would use a separate connection completely. - var ( - err error - publisher api.LogBroker_PublishLogsClient - ) - - err = a.withSession(ctx, func(session *session) error { + var publisher api.LogBroker_PublishLogsClient + err := a.withSession(ctx, func(session *session) error { + var err error publisher, err = api.NewLogBrokerClient(session.conn.ClientConn).PublishLogs(ctx) return err }) @@ -602,9 +599,9 @@ func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogP SubscriptionID: subscriptionID, Close: true, }) - // close the stream forreal. ignore the return value and the error, + // close the stream for real. ignore the return value and the error, // because we don't care. - publisher.CloseAndRecv() + _, _ = publisher.CloseAndRecv() } return exec.LogPublisherFunc(func(ctx context.Context, message api.LogMessage) error { From 3ca920871d81e1499f04e6ddc9c5924a132cadf7 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 13:13:57 +0200 Subject: [PATCH 2/9] agent: worker.Subscribe: simplify subscription event loop - return early for closed event channels - use early "continue" to reduce nesting Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index 9b04c055ba..df8db08337 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -646,18 +646,23 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe defer cancel() for { select { - case v := <-ch: - task := v.(*api.Task) - if match(task) { - w.mu.RLock() - tm, ok := w.taskManagers[task.ID] - w.mu.RUnlock() - if !ok { - continue - } + case v, ok := <-ch: + if !ok { + return nil + } - go tm.Logs(ctx, *subscription.Options, publisher) + task, ok := v.(*api.Task) + if !ok || !match(task) { + continue } + w.mu.RLock() + tm, ok := w.taskManagers[task.ID] + w.mu.RUnlock() + if !ok { + continue + } + + go tm.Logs(ctx, *subscription.Options, publisher) case <-ctx.Done(): return ctx.Err() } From 53210e3376d22389ec55b3af3bd7bb2000090291 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 13:59:05 +0200 Subject: [PATCH 3/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index df8db08337..756b8fa3a2 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -613,7 +613,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe } var wg sync.WaitGroup - w.mu.Lock() + w.mu.RLock() for _, tm := range w.taskManagers { if match(tm.task) { wg.Go(func() { @@ -621,7 +621,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe }) } } - w.mu.Unlock() + w.mu.RUnlock() // If follow mode is disabled, wait for the current set of matched tasks // to finish publishing logs, then close the subscription by returning. From ad84ac1059ab019fa57ce0d88c7c086f7f87ac14 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 14:08:27 +0200 Subject: [PATCH 4/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index 756b8fa3a2..2bf5bdcd37 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -595,7 +595,16 @@ func (w *worker) updateTaskStatus(ctx context.Context, tx *bolt.Tx, taskID strin // Subscribe to log messages matching the subscription. func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMessage) error { - log.G(ctx).Debugf("Received subscription %s (selector: %v)", subscription.ID, subscription.Selector) + var options api.LogSubscriptionOptions + if subscription.Options != nil { + options = *subscription.Options + } + + log.G(ctx).WithFields(log.Fields{ + "id": subscription.ID, + "selector": subscription.Selector, + "follow": options.Follow, + }).Debug("Received subscription") publisher, cancel, err := w.publisherProvider.Publisher(ctx, subscription.ID) if err != nil { @@ -617,7 +626,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe for _, tm := range w.taskManagers { if match(tm.task) { wg.Go(func() { - tm.Logs(ctx, *subscription.Options, publisher) + tm.Logs(ctx, options, publisher) }) } } @@ -625,7 +634,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe // If follow mode is disabled, wait for the current set of matched tasks // to finish publishing logs, then close the subscription by returning. - if subscription.Options == nil || !subscription.Options.Follow { + if !options.Follow { waitCh := make(chan struct{}) go func() { defer close(waitCh) @@ -662,7 +671,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe continue } - go tm.Logs(ctx, *subscription.Options, publisher) + go tm.Logs(ctx, options, publisher) case <-ctx.Done(): return ctx.Err() } From a90581f97d2c37a8406372eda468681156b8cdd7 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 14:49:20 +0200 Subject: [PATCH 5/9] agent: worker.Subscribe: wait for log streams to finish 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 --- agent/worker.go | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index 2bf5bdcd37..afcd5f9a44 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -635,18 +635,8 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe // If follow mode is disabled, wait for the current set of matched tasks // to finish publishing logs, then close the subscription by returning. if !options.Follow { - waitCh := make(chan struct{}) - go func() { - defer close(waitCh) - wg.Wait() - }() - - select { - case <-ctx.Done(): - return ctx.Err() - case <-waitCh: - return nil - } + wg.Wait() + return ctx.Err() } // In follow mode, watch for new tasks. Don't close the subscription From acbe9c2d20ef7ff7ee8998ea979838c6de93bd03 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 14:59:23 +0200 Subject: [PATCH 6/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/agent/worker.go b/agent/worker.go index afcd5f9a44..e4b16afb6b 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -661,8 +661,11 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe continue } - go tm.Logs(ctx, options, publisher) + wg.Go(func() { + tm.Logs(ctx, options, publisher) + }) case <-ctx.Done(): + wg.Wait() return ctx.Err() } } From 22d7bfaef11c6ffe54c1098b355042a5e1ff345a Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 15:07:09 +0200 Subject: [PATCH 7/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index e4b16afb6b..4104a4e4aa 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -621,16 +621,20 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe slices.Contains(sel.NodeIDs, t.NodeID) } - var wg sync.WaitGroup w.mu.RLock() + taskManagers := make([]*taskManager, 0, len(w.taskManagers)) for _, tm := range w.taskManagers { if match(tm.task) { - wg.Go(func() { - tm.Logs(ctx, options, publisher) - }) + taskManagers = append(taskManagers, tm) } } w.mu.RUnlock() + var wg sync.WaitGroup + for _, tm := range taskManagers { + wg.Go(func() { + tm.Logs(ctx, options, publisher) + }) + } // If follow mode is disabled, wait for the current set of matched tasks // to finish publishing logs, then close the subscription by returning. From e233c068385e4f0c8fbcce8c9a9a2a6efbdfa330 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 16:48:57 +0200 Subject: [PATCH 8/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 43 +++++++++++++++++++------------------------ 1 file changed, 19 insertions(+), 24 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index 4104a4e4aa..13d06c2273 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -5,6 +5,7 @@ import ( "slices" "sync" + "github.com/docker/go-events" "github.com/moby/swarmkit/v2/agent/exec" "github.com/moby/swarmkit/v2/api" "github.com/moby/swarmkit/v2/log" @@ -621,6 +622,16 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe slices.Contains(sel.NodeIDs, t.NodeID) } + 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) + })) + } + w.mu.RLock() taskManagers := make([]*taskManager, 0, len(w.taskManagers)) for _, tm := range w.taskManagers { @@ -636,28 +647,12 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe }) } - // If follow mode is disabled, wait for the current set of matched tasks - // to finish publishing logs, then close the subscription by returning. - if !options.Follow { - wg.Wait() - return ctx.Err() - } - - // In follow mode, watch for new tasks. Don't close the subscription - // until it's cancelled. - ch, cancel := w.taskevents.Watch() - defer cancel() - for { - select { - case v, ok := <-ch: - if !ok { - return nil - } + // In follow mode, watch for new matching tasks until the subscription + // context is cancelled. + if options.Follow { + for v := range ch { + task := v.(*api.Task) - task, ok := v.(*api.Task) - if !ok || !match(task) { - continue - } w.mu.RLock() tm, ok := w.taskManagers[task.ID] w.mu.RUnlock() @@ -668,11 +663,11 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe wg.Go(func() { tm.Logs(ctx, options, publisher) }) - case <-ctx.Done(): - wg.Wait() - return ctx.Err() } } + + wg.Wait() + return ctx.Err() } func (w *worker) Wait(ctx context.Context) error { From f71c0111ab7c42261d6a4c785ff4243214e91bd9 Mon Sep 17 00:00:00 2001 From: Sebastiaan van Stijn Date: Sun, 26 Jul 2026 17:01:35 +0200 Subject: [PATCH 9/9] 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. Signed-off-by: Sebastiaan van Stijn --- agent/worker.go | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/agent/worker.go b/agent/worker.go index 13d06c2273..dd2d8b773a 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -641,11 +641,24 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe } w.mu.RUnlock() var wg sync.WaitGroup - for _, tm := range taskManagers { + + // A task may be present in the initial snapshot and also be delivered by + // the watcher. Start at most one log stream per task for this subscription. + started := make(map[string]struct{}) + startLogs := func(tm *taskManager) { + taskID := tm.task.ID + if _, ok := started[taskID]; ok { + return + } + started[taskID] = struct{}{} + wg.Go(func() { tm.Logs(ctx, options, publisher) }) } + for _, tm := range taskManagers { + startLogs(tm) + } // In follow mode, watch for new matching tasks until the subscription // context is cancelled. @@ -660,9 +673,7 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe continue } - wg.Go(func() { - tm.Logs(ctx, options, publisher) - }) + startLogs(tm) } }