diff --git a/.golangci.yml b/.golangci.yml index 2d41e08e04..e4138482bb 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -9,6 +9,7 @@ linters: enable: - misspell - ineffassign + - modernize - revive - unconvert - unused diff --git a/agent/worker.go b/agent/worker.go index bd62f81bfb..9b04c055ba 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -612,15 +612,13 @@ func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMe slices.Contains(sel.NodeIDs, t.NodeID) } - wg := sync.WaitGroup{} + var wg sync.WaitGroup w.mu.Lock() for _, tm := range w.taskManagers { if match(tm.task) { - wg.Add(1) - go func(tm *taskManager) { - defer wg.Done() + wg.Go(func() { tm.Logs(ctx, *subscription.Options, publisher) - }(tm) + }) } } w.mu.Unlock() diff --git a/go.mod b/go.mod index 525094ad0b..5a7e5fc901 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/moby/swarmkit/v2 -go 1.24.0 +go 1.25.0 require ( code.cloudfoundry.org/clock v1.1.0 diff --git a/integration/cluster_test.go b/integration/cluster_test.go index 0d8fcdf74e..618a1d9cd3 100644 --- a/integration/cluster_test.go +++ b/integration/cluster_test.go @@ -14,7 +14,6 @@ import ( "github.com/moby/swarmkit/v2/identity" "github.com/moby/swarmkit/v2/log" "github.com/moby/swarmkit/v2/manager/encryption" - "github.com/moby/swarmkit/v2/node" "github.com/moby/swarmkit/v2/testutils" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -164,20 +163,17 @@ func (c *testCluster) runNode(n *testNode, nodeOrder int) error { defer cancel() defer close(done) - c.wg.Add(2) - go func() { + c.wg.Go(func() { c.errs <- n.node.Start(ctx) - c.wg.Done() - }() - go func(n *node.Node) { - err := n.Err(errCtx) + }) + c.wg.Go(func() { + err := n.node.Err(errCtx) select { case <-errCtx.Done(): default: done <- err } - c.wg.Done() - }(n.node) + }) select { case <-n.node.Ready(): diff --git a/manager/allocator/allocator.go b/manager/allocator/allocator.go index f1dee41ab4..b8db097a57 100644 --- a/manager/allocator/allocator.go +++ b/manager/allocator/allocator.go @@ -106,28 +106,19 @@ func (a *Allocator) Run(ctx context.Context) error { a.registerToVote(aa.taskVoter) } - // Assign a pointer for variable capture - aaPtr := &aa actor := func() error { - wg.Add(1) - defer wg.Done() - // init might return an allocator specific context // which is a child of the passed in context to hold // allocator specific state - watch, watchCancel, err := a.init(ctx, aaPtr) + watch, watchCancel, err := a.init(ctx, &aa) if err != nil { return err } - wg.Add(1) - go func(watch <-chan events.Event, watchCancel func()) { - defer func() { - watchCancel() - wg.Done() - }() - a.run(ctx, *aaPtr, watch) - }(watch, watchCancel) + wg.Go(func() { + defer watchCancel() + a.run(ctx, aa, watch) + }) return nil } diff --git a/manager/controlapi/service.go b/manager/controlapi/service.go index 013be7dc57..ef79d776cf 100644 --- a/manager/controlapi/service.go +++ b/manager/controlapi/service.go @@ -4,6 +4,7 @@ import ( "context" "errors" "reflect" + "slices" "strings" "time" @@ -247,10 +248,8 @@ func validateGenericRuntimeSpec(taskSpec api.TaskSpec) error { } reservedNames := []string{"container", "attachment"} - for _, n := range reservedNames { - if strings.ToLower(generic.Kind) == n { - return status.Errorf(codes.InvalidArgument, "Generic runtime: %q is a reserved name", generic.Kind) - } + if slices.Contains(reservedNames, strings.ToLower(generic.Kind)) { + return status.Errorf(codes.InvalidArgument, "Generic runtime: %q is a reserved name", generic.Kind) } payload := generic.Payload diff --git a/manager/logbroker/broker_test.go b/manager/logbroker/broker_test.go index f13dd54b22..cea026b7a5 100644 --- a/manager/logbroker/broker_test.go +++ b/manager/logbroker/broker_test.go @@ -72,15 +72,12 @@ func TestLogBrokerLogs(t *testing.T) { taskID := fmt.Sprintf("%v.task-%v", serviceID, task) for node := range nNodes { - nodeID := fmt.Sprintf("node-%v", node) - if (task+1)%(node+1) != 0 { continue } messagesExpected += nLogMessagesPerTask - wg.Add(1) - go func(nodeID, serviceID, taskID string) { + wg.Go(func() { <-hold // Each goroutine gets its own publisher @@ -90,7 +87,6 @@ func TestLogBrokerLogs(t *testing.T) { defer func() { _, err := publisher.CloseAndRecv() require.NoError(t, err) - wg.Done() }() msgctx := api.LogContext{ @@ -104,7 +100,7 @@ func TestLogBrokerLogs(t *testing.T) { Messages: []api.LogMessage{newLogMessage(msgctx, "log message number %d", i)}, })) } - }(nodeID, serviceID, taskID) + }) } } } diff --git a/manager/orchestrator/update/updater.go b/manager/orchestrator/update/updater.go index 55bbac08f4..fbad48e0d5 100644 --- a/manager/orchestrator/update/updater.go +++ b/manager/orchestrator/update/updater.go @@ -189,13 +189,11 @@ func (u *Updater) Run(ctx context.Context, slots []orchestrator.Slot) { // Start the workers. slotQueue := make(chan orchestrator.Slot) - wg := sync.WaitGroup{} - wg.Add(parallelism) + var wg sync.WaitGroup for range parallelism { - go func() { + wg.Go(func() { u.worker(ctx, slotQueue, updateConfig) - wg.Done() - }() + }) } var failedTaskWatch chan events.Event diff --git a/manager/state/raft/raft.go b/manager/state/raft/raft.go index 6dfac5f2c2..c41223f172 100644 --- a/manager/state/raft/raft.go +++ b/manager/state/raft/raft.go @@ -114,8 +114,8 @@ type Node struct { reqIDGen *idutil.Generator wait *wait campaignWhenAble bool - signalledLeadership uint32 - isMember uint32 + signalledLeadership atomic.Uint32 + isMember atomic.Uint32 bootstrapMembers []*api.RaftMember // waitProp waits for all the proposals to be terminated before @@ -161,7 +161,7 @@ type Node struct { // an raft DEK during a raft DEK rotation, so that we won't finish a rotation until // a snapshot covering that index has been written encrypted with the new raft DEK waitForAppliedIndex uint64 - ticksWithNoLeader uint32 + ticksWithNoLeader atomic.Uint32 } // NodeOptions provides node-level options. @@ -275,7 +275,7 @@ func (n *Node) IsIDRemoved(id uint64) bool { // Part of transport.Raft interface. func (n *Node) NodeRemoved() { n.removeRaftOnce.Do(func() { - atomic.StoreUint32(&n.isMember, 0) + n.isMember.Store(0) close(n.RemovedFromRaft) }) } @@ -331,7 +331,7 @@ func (n *Node) SetAddr(ctx context.Context, addr string) error { ctx, cancelCtx := n.WithContext(ctx) defer cancelCtx() - isLeader := atomic.LoadUint32(&n.signalledLeadership) == 1 + isLeader := n.signalledLeadership.Load() == 1 for !isLeader { select { case leadershipChange := <-leadershipCh: @@ -383,7 +383,7 @@ func (n *Node) JoinAndStart(ctx context.Context) (err error) { n.stopMu.Unlock() n.done() } else { - atomic.StoreUint32(&n.isMember, 1) + n.isMember.Store(1) } }() @@ -574,9 +574,9 @@ func (n *Node) Run(ctx context.Context) error { n.raftNode.Tick() if n.leader() == raft.None { - atomic.AddUint32(&n.ticksWithNoLeader, 1) + n.ticksWithNoLeader.Add(1) } else { - atomic.StoreUint32(&n.ticksWithNoLeader, 0) + n.ticksWithNoLeader.Store(0) } case rd := <-n.raftNode.Ready(): raftConfig := n.getCurrentRaftConfig() @@ -646,8 +646,8 @@ func (n *Node) Run(ctx context.Context) error { wasLeader = false log.G(ctx).Error("soft state changed, node no longer a leader, resetting and cancelling all waits") - if atomic.LoadUint32(&n.signalledLeadership) == 1 { - atomic.StoreUint32(&n.signalledLeadership, 0) + if n.signalledLeadership.Load() == 1 { + n.signalledLeadership.Store(0) n.leadershipBroadcast.Publish(IsFollower) } @@ -686,11 +686,11 @@ func (n *Node) Run(ctx context.Context) error { n.triggerSnapshot(ctx, raftConfig) } - if wasLeader && atomic.LoadUint32(&n.signalledLeadership) != 1 { + if wasLeader && n.signalledLeadership.Load() != 1 { // If all the entries in the log have become // committed, broadcast our leadership status. if n.caughtUp() { - atomic.StoreUint32(&n.signalledLeadership, 1) + n.signalledLeadership.Store(1) n.leadershipBroadcast.Publish(IsLeader) } } @@ -860,7 +860,7 @@ func (n *Node) stop(ctx context.Context) { n.raftNode.Stop() n.ticker.Stop() n.raftLogger.Close(ctx) - atomic.StoreUint32(&n.isMember, 0) + n.isMember.Store(0) // TODO(stevvooe): Handle ctx.Done() } @@ -910,7 +910,7 @@ func (n *Node) Leader() (uint64, error) { // saying that it has become the leader. This means it is ready to accept // proposals. func (n *Node) ReadyForProposals() bool { - return atomic.LoadUint32(&n.signalledLeadership) == 1 + return n.signalledLeadership.Load() == 1 } func (n *Node) caughtUp() bool { @@ -1517,7 +1517,7 @@ func (n *Node) LeaderConn(ctx context.Context) (*grpc.ClientConn, error) { if err == raftselector.ErrIsLeader { return nil, err } - if atomic.LoadUint32(&n.ticksWithNoLeader) > lostQuorumTimeout { + if n.ticksWithNoLeader.Load() > lostQuorumTimeout { return nil, errLostQuorum } @@ -1732,7 +1732,7 @@ func (n *Node) GetNodeIDByRaftID(raftID uint64) (string, error) { // IsMember checks if the raft node has effectively joined // a cluster of existing members. func (n *Node) IsMember() bool { - return atomic.LoadUint32(&n.isMember) == 1 + return n.isMember.Load() == 1 } // Saves a log entry to our Store @@ -1801,7 +1801,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa ch := n.wait.register(r.ID, cb, cancel) // Do this check after calling register to avoid a race. - if atomic.LoadUint32(&n.signalledLeadership) != 1 { + if n.signalledLeadership.Load() != 1 { log.G(ctx).Error("node is no longer leader, aborting propose") n.wait.cancel(r.ID) return nil, ErrLostLeadership @@ -1829,7 +1829,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa if !ok { // Wait notification channel was closed. This should only happen if the wait was cancelled. log.G(ctx).Error("wait cancelled") - if atomic.LoadUint32(&n.signalledLeadership) == 1 { + if n.signalledLeadership.Load() == 1 { log.G(ctx).Error("wait cancelled but node is still a leader") } return nil, ErrLostLeadership @@ -1841,7 +1841,7 @@ func (n *Node) processInternalRaftRequest(ctx context.Context, r *api.InternalRa x, ok := <-ch if !ok { log.G(ctx).WithError(waitCtx.Err()).Error("wait context cancelled") - if atomic.LoadUint32(&n.signalledLeadership) == 1 { + if n.signalledLeadership.Load() == 1 { log.G(ctx).Error("wait context cancelled but node is still a leader") } return nil, ErrLostLeadership diff --git a/manager/state/store/memory_test.go b/manager/state/store/memory_test.go index 786bc7b1a5..549f9636f1 100644 --- a/manager/state/store/memory_test.go +++ b/manager/state/store/memory_test.go @@ -2115,9 +2115,7 @@ func BenchmarkNodeConcurrency(b *testing.B) { // Run 5 writer goroutines and 5 reader goroutines var wg sync.WaitGroup for c := range 5 { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for i := range b.N { _ = s.Update(func(tx1 Tx) error { _ = UpdateNode(tx1, &api.Node{ @@ -2131,19 +2129,17 @@ func BenchmarkNodeConcurrency(b *testing.B) { return nil }) } - }() + }) } for range 5 { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { s.View(func(tx1 ReadTx) { for i := range b.N { _ = GetNode(tx1, nodeIDs[i%benchmarkNumNodes]) } }) - }() + }) } wg.Wait() diff --git a/node/node.go b/node/node.go index eda5266a73..e82fc8d755 100644 --- a/node/node.go +++ b/node/node.go @@ -425,7 +425,6 @@ func (n *Node) run(ctx context.Context) (err error) { // the node object until all 3 of these components have terminated, so we // create a waitgroup to block termination of the node until then var wg sync.WaitGroup - wg.Add(3) // These two blocks update some of the metrics settings. nodeInfo.WithValues( @@ -444,7 +443,7 @@ func (n *Node) run(ctx context.Context) (err error) { // CertificateUpdates, and launch a goroutine to handle this. Updates is a // channel we iterate containing the results of certificate renewals. updates := renewer.Start(ctx) - go func() { + wg.Go(func() { for certUpdate := range updates { if certUpdate.Err != nil { log.G(ctx).Warnf("error renewing TLS certificate: %v", certUpdate.Err) @@ -464,9 +463,7 @@ func (n *Node) run(ctx context.Context) (err error) { nodeManager.Set(0) } } - - wg.Done() - }() + }) // and, finally, start the two main components: the manager and the agent role := n.role @@ -479,18 +476,17 @@ func (n *Node) run(ctx context.Context) (err error) { // respective goroutines below. var managerErr error var agentErr error - go func() { + wg.Go(func() { // superviseManager is a routine that watches our manager role managerErr = n.superviseManager(ctx, securityConfig, paths.RootCA, managerReady, renewer) // store err and loop - wg.Done() cancel() - }() - go func() { + }) + wg.Go(func() { + defer close(agentDone) + agentErr = n.runAgent(ctx, db, securityConfig, agentReady) - wg.Done() cancel() - close(agentDone) - }() + }) // This goroutine is what signals that the node has fully started by // closing the n.ready channel. First, it waits for the agent to start. diff --git a/swarmd/go.mod b/swarmd/go.mod index 65caacb83d..12b3ea7be9 100644 --- a/swarmd/go.mod +++ b/swarmd/go.mod @@ -1,6 +1,6 @@ module github.com/moby/swarmkit/swarmd -go 1.24.0 +go 1.25.0 require ( github.com/cloudflare/cfssl v1.6.4 diff --git a/swarmd/go.work b/swarmd/go.work index ac9df34508..6f887b01da 100644 --- a/swarmd/go.work +++ b/swarmd/go.work @@ -1,4 +1,4 @@ -go 1.24.0 +go 1.25.0 use ( . diff --git a/watch/watch_test.go b/watch/watch_test.go index b937a6ae3e..b7431d53d0 100644 --- a/watch/watch_test.go +++ b/watch/watch_test.go @@ -223,21 +223,19 @@ func benchmarkWatchForQueue(q *Queue, b *testing.B, nlisteners, npublishers int, publishersRunning sync.WaitGroup ) + eventsPerPublisher := b.N / npublishers + eventsPerWatcher := eventsPerPublisher * npublishers for range nlisteners { watchersAttached.Add(1) - watchersRunning.Add(1) - go func(n int) { + watchersRunning.Go(func() { w, cancel := q.Watch() defer cancel() watchersAttached.Done() - for range n { + for range eventsPerWatcher { <-w } - if waitForWatchers { - watchersRunning.Done() - } - }(b.N / npublishers * npublishers) + }) } // Wait for watchers to be in place before we start publishing events. @@ -246,13 +244,11 @@ func benchmarkWatchForQueue(q *Queue, b *testing.B, nlisteners, npublishers int, b.ResetTimer() for range npublishers { - publishersRunning.Add(1) - go func(n int) { - for range n { + publishersRunning.Go(func() { + for range eventsPerPublisher { q.Publish("myevent") } - publishersRunning.Done() - }(b.N / npublishers) + }) } publishersRunning.Wait()