diff --git a/go.mod b/go.mod index 08972723c1..9a3fe8a45a 100644 --- a/go.mod +++ b/go.mod @@ -74,7 +74,7 @@ require ( github.com/moby/profiles/apparmor v0.2.1 github.com/moby/profiles/seccomp v0.2.3 github.com/moby/pubsub v1.0.0 - github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35 + github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9 github.com/moby/sys/atomicwriter v0.1.0 github.com/moby/sys/mount v0.3.5 github.com/moby/sys/mountinfo v0.7.2 diff --git a/go.sum b/go.sum index ff1f7beffa..ee5f6fb80e 100644 --- a/go.sum +++ b/go.sum @@ -563,8 +563,8 @@ github.com/moby/profiles/seccomp v0.2.3 h1:nrHNSiECQQvq4WjgceCUgJXIXUJBIswVQ133k github.com/moby/profiles/seccomp v0.2.3/go.mod h1:8m3qkkWZXrRsMqlUUN2zyccnYmBf3EAdQYPMVJ3NBhk= github.com/moby/pubsub v1.0.0 h1:jkp/imWsmJz2f6LyFsk7EkVeN2HxR/HTTOY8kHrsxfA= github.com/moby/pubsub v1.0.0/go.mod h1:bXSO+3h5MNXXCaEG+6/NlAIk7MMZbySZlnB+cUQhKKc= -github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35 h1:8D44vdVynl7eWS5XblHfyctMHcrzbLS4YSVcZBvjwKY= -github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35/go.mod h1:9rg8/RJeagBT8uRrsNFcetujUVZT/ynfya2hnabQY/A= +github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9 h1:uv2uVSLLXQtAZOK9s4vXyBy7vOZHGKXEUf+qCp0KSCs= +github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9/go.mod h1:9rg8/RJeagBT8uRrsNFcetujUVZT/ynfya2hnabQY/A= github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw= github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs= github.com/moby/sys/capability v0.4.0 h1:4D4mI6KlNtWMCM1Z/K0i7RV1FkX+DBDHKVJpCndZoHk= diff --git a/vendor/github.com/moby/swarmkit/v2/manager/logbroker/broker.go b/vendor/github.com/moby/swarmkit/v2/manager/logbroker/broker.go index f53e297d1b..9546e02720 100644 --- a/vendor/github.com/moby/swarmkit/v2/manager/logbroker/broker.go +++ b/vendor/github.com/moby/swarmkit/v2/manager/logbroker/broker.go @@ -105,40 +105,38 @@ func (lb *LogBroker) newSubscription(selector *api.LogSelector, options *api.Log lb.mu.RLock() defer lb.mu.RUnlock() - subscription := newSubscription(lb.store, &api.SubscriptionMessage{ + return newSubscription(lb.store, &api.SubscriptionMessage{ ID: identity.NewID(), Selector: selector, Options: options, }, lb.subscriptionQueue) - - return subscription } func (lb *LogBroker) getSubscription(id string) *subscription { lb.mu.RLock() defer lb.mu.RUnlock() - subscription, ok := lb.registeredSubscriptions[id] + sub, ok := lb.registeredSubscriptions[id] if !ok { return nil } - return subscription + return sub } -func (lb *LogBroker) registerSubscription(subscription *subscription) { +func (lb *LogBroker) registerSubscription(sub *subscription) { lb.mu.Lock() defer lb.mu.Unlock() - lb.registeredSubscriptions[subscription.message.ID] = subscription - lb.subscriptionQueue.Publish(subscription) + lb.registeredSubscriptions[sub.ID()] = sub + lb.subscriptionQueue.Publish(sub) - for _, node := range subscription.Nodes() { + for _, node := range sub.Nodes() { if _, ok := lb.subscriptionsByNode[node]; !ok { // Mark nodes that won't receive the message as done. - subscription.Done(node, fmt.Errorf("node %s is not available", node)) + sub.Done(node, fmt.Errorf("node %s is not available", node)) } else { // otherwise, add the subscription to the node's subscriptions list - lb.subscriptionsByNode[node][subscription] = struct{}{} + lb.subscriptionsByNode[node][sub] = struct{}{} } } } @@ -147,7 +145,7 @@ func (lb *LogBroker) unregisterSubscription(subscription *subscription) { lb.mu.Lock() defer lb.mu.Unlock() - delete(lb.registeredSubscriptions, subscription.message.ID) + delete(lb.registeredSubscriptions, subscription.ID()) // remove the subscription from all of the nodes for _, node := range subscription.Nodes() { @@ -234,25 +232,25 @@ func (lb *LogBroker) SubscribeLogs(request *api.SubscribeLogsRequest, stream api return errNotRunning } - subscription := lb.newSubscription(request.Selector, request.Options) - subscription.Run(pctx) - defer subscription.Stop() + sub := lb.newSubscription(request.Selector, request.Options) + sub.Run(pctx) + defer sub.Stop() logger := log.G(ctx).WithFields( log.Fields{ "method": "(*LogBroker).SubscribeLogs", - "subscription.id": subscription.message.ID, + "subscription.id": sub.ID(), }, ) logger.Debug("subscribed") - publishCh, publishCancel := lb.subscribe(subscription.message.ID) + publishCh, publishCancel := lb.subscribe(sub.ID()) defer publishCancel() - lb.registerSubscription(subscription) - defer lb.unregisterSubscription(subscription) + lb.registerSubscription(sub) + defer lb.unregisterSubscription(sub) - completed := subscription.Wait(ctx) + completed := sub.Wait(ctx) for { select { case <-ctx.Done(): @@ -273,10 +271,10 @@ func (lb *LogBroker) SubscribeLogs(request *api.SubscribeLogsRequest, stream api completed = nil lb.logQueue.Publish(&logMessage{ PublishLogsMessage: &api.PublishLogsMessage{ - SubscriptionID: subscription.message.ID, + SubscriptionID: sub.ID(), }, completed: true, - err: subscription.Err(), + err: sub.Err(), }) } } @@ -332,7 +330,7 @@ func (lb *LogBroker) ListenSubscriptions(_ *api.ListenSubscriptionsRequest, stre activeSubscriptions := make(map[string]*subscription) // Start by sending down all active subscriptions. - for _, subscription := range subscriptions { + for _, sub := range subscriptions { select { case <-stream.Context().Done(): return stream.Context().Err() @@ -341,30 +339,30 @@ func (lb *LogBroker) ListenSubscriptions(_ *api.ListenSubscriptionsRequest, stre default: } - if err := stream.Send(subscription.message); err != nil { - logger.Error(err) + if err := stream.Send(sub.Message()); err != nil { + logger.WithError(err).Error("failed to send initial subscription") return err } - activeSubscriptions[subscription.message.ID] = subscription + activeSubscriptions[sub.ID()] = sub } // Send down new subscriptions. for { select { case v := <-subscriptionCh: - subscription := v.(*subscription) + sub := v.(*subscription) - if subscription.Closed() { - delete(activeSubscriptions, subscription.message.ID) + if sub.Closed() { + delete(activeSubscriptions, sub.ID()) } else { // Avoid sending down the same subscription multiple times - if _, ok := activeSubscriptions[subscription.message.ID]; ok { + if _, ok := activeSubscriptions[sub.ID()]; ok { continue } - activeSubscriptions[subscription.message.ID] = subscription + activeSubscriptions[sub.ID()] = sub } - if err := stream.Send(subscription.message); err != nil { - logger.Error(err) + if err := stream.Send(sub.Message()); err != nil { + logger.WithError(err).Error("failed to send subscription update") return err } case <-stream.Context().Done(): @@ -408,7 +406,7 @@ func (lb *LogBroker) PublishLogs(stream api.LogBroker_PublishLogsServer) (err er return status.Errorf(codes.NotFound, "unknown subscription ID") } } else { - if logMsg.SubscriptionID != currentSubscription.message.ID { + if logMsg.SubscriptionID != currentSubscription.ID() { return status.Errorf(codes.InvalidArgument, "different subscription IDs in the same session") } } diff --git a/vendor/github.com/moby/swarmkit/v2/manager/logbroker/subscription.go b/vendor/github.com/moby/swarmkit/v2/manager/logbroker/subscription.go index e7aa08b8df..bffbd3f286 100644 --- a/vendor/github.com/moby/swarmkit/v2/manager/logbroker/subscription.go +++ b/vendor/github.com/moby/swarmkit/v2/manager/logbroker/subscription.go @@ -15,6 +15,10 @@ import ( ) type subscription struct { + id string + + follow bool + mu sync.RWMutex wg sync.WaitGroup @@ -32,6 +36,8 @@ type subscription struct { func newSubscription(store *store.MemoryStore, message *api.SubscriptionMessage, changed *watch.Queue) *subscription { return &subscription{ + id: message.ID, + follow: message.Options != nil && message.Options.Follow, store: store, message: message, changed: changed, @@ -40,8 +46,8 @@ func newSubscription(store *store.MemoryStore, message *api.SubscriptionMessage, } } -func (s *subscription) follow() bool { - return s.message.Options != nil && s.message.Options.Follow +func (s *subscription) ID() string { + return s.id } func (s *subscription) Contains(nodeID string) bool { @@ -66,12 +72,12 @@ func (s *subscription) Nodes() []string { func (s *subscription) Run(ctx context.Context) { s.ctx, s.cancel = context.WithCancel(ctx) - if s.follow() { + if s.follow { wq := s.store.WatchQueue() ch, cancel := state.Watch(wq, api.EventCreateTask{}, api.EventUpdateTask{}) go func() { defer cancel() - s.watch(ch) + _ = s.watch(ch) }() } @@ -86,7 +92,7 @@ func (s *subscription) Stop() { func (s *subscription) Wait(_ context.Context) <-chan struct{} { // Follow subscriptions never end - if s.follow() { + if s.follow { return nil } @@ -106,7 +112,7 @@ func (s *subscription) Done(nodeID string, err error) { s.errors = append(s.errors, err) } - if s.follow() { + if s.follow { return } @@ -152,6 +158,7 @@ func (s *subscription) Closed() bool { func (s *subscription) match() { s.mu.Lock() defer s.mu.Unlock() + selector := s.message.Selector add := func(t *api.Task) { if t.NodeID == "" { @@ -165,17 +172,17 @@ func (s *subscription) match() { } s.store.View(func(tx store.ReadTx) { - for _, nid := range s.message.Selector.NodeIDs { + for _, nid := range selector.NodeIDs { s.nodes[nid] = struct{}{} } - for _, tid := range s.message.Selector.TaskIDs { + for _, tid := range selector.TaskIDs { if task := store.GetTask(tx, tid); task != nil { add(task) } } - for _, sid := range s.message.Selector.ServiceIDs { + for _, sid := range selector.ServiceIDs { tasks, err := store.FindTasks(tx, store.ByServiceID(sid)) if err != nil { log.L.Warning(err) @@ -183,7 +190,7 @@ func (s *subscription) match() { } for _, task := range tasks { // if we're not following, don't add tasks that aren't running yet - if !s.follow() && task.Status.State < api.TaskStateRunning { + if !s.follow && task.Status.State < api.TaskStateRunning { continue } add(task) @@ -193,13 +200,15 @@ func (s *subscription) match() { } func (s *subscription) watch(ch <-chan events.Event) error { + selector := s.message.Selector + matchTasks := map[string]struct{}{} - for _, tid := range s.message.Selector.TaskIDs { + for _, tid := range selector.TaskIDs { matchTasks[tid] = struct{}{} } matchServices := map[string]struct{}{} - for _, sid := range s.message.Selector.ServiceIDs { + for _, sid := range selector.ServiceIDs { matchServices[sid] = struct{}{} } @@ -257,3 +266,18 @@ func (s *subscription) watch(ch <-chan events.Event) error { } } } + +// Message returns a snapshot of the current subscription message. +// +// The underlying message is owned by the subscription and may be updated +// as part of its lifecycle. Callers should use this accessor rather than +// accessing the underlying message directly. +func (s *subscription) Message() *api.SubscriptionMessage { + s.mu.RLock() + defer s.mu.RUnlock() + + // Return a snapshot of the message to avoid races while it is being + // marshaled. See https://github.com/moby/moby/issues/47322. + msg := *s.message + return &msg +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 47b6434ea3..a7a450c5bb 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1331,7 +1331,7 @@ github.com/moby/profiles/seccomp # github.com/moby/pubsub v1.0.0 ## explicit; go 1.19 github.com/moby/pubsub -# github.com/moby/swarmkit/v2 v2.1.3-0.20260725133749-28ad14b9eb35 +# github.com/moby/swarmkit/v2 v2.1.3-0.20260727142002-e53c2e1ca8a9 ## explicit; go 1.24.0 github.com/moby/swarmkit/v2/agent github.com/moby/swarmkit/v2/agent/configs