PMM-14442 Fix the data race

.github/workflows/clean.yml

@@ -12,14 +12,7 @@ jobs:
     name: Clean caches
     timeout-minutes: 5
 
-    strategy:
-      fail-fast: false
-      matrix:
-        go:
-          - version: 1.24.x
-            may-fail: false
-
-    continue-on-error: ${{ matrix.go.may-fail }}
+    continue-on-error: false
     runs-on: ubuntu-22.04
 
     env:
@@ -31,15 +24,15 @@ jobs:
           ROBOT_TOKEN: ${{ secrets.ROBOT_TOKEN }}
         run: echo "machine github.com login percona-robot password ${{ secrets.ROBOT_TOKEN }}" > $HOME/.netrc
 
-      - name: Set up Go release
+      - name: Set up Go
         env:
           # to avoid error due to `go version` accepting -v flag with an argument since 1.15
           GOFLAGS: ""
         uses: actions/setup-go@4dc6199c7b1a012772edbd06daecab0f50c9053c # v6.1.0
         with:
-          go-version: ${{ matrix.go.version }}
+          go-version-file: ${{ github.workspace }}/go.mod
 
-      - name: Check out code into the Go module directory
+      - name: Check out code
         uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # v6.0.0
         with:
           lfs: true
@@ -48,18 +41,17 @@ jobs:
         uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
         with:
           path: ~/go/pkg/mod
-          key: ${{ matrix.go.version }}-modules-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ matrix.go.version }}-modules-
+          key: ${{ runner.os }}-go-modules-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ runner.os }}-go-modules-
 
       - name: Enable Go build cache
         uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
         with:
           path: ~/.cache/go-build
-          key: ${{ matrix.go.version }}-build-${{ github.ref }}-${{ hashFiles('**') }}
+          key: ${{ runner.os }}-go-build-${{ github.ref }}-${{ hashFiles('**') }}
           restore-keys: |
-            ${{ matrix.go.version }}-build-${{ github.ref }}-
-            ${{ matrix.go.version }}-build-
+            ${{ runner.os }}-go-build-${{ github.ref }}-
+            ${{ runner.os }}-go-build-
 
       - name: Clean Go modules cache
         run: go clean -modcache

agent/docker-compose.yml

@@ -1,6 +1,4 @@
 ---
-version: '3.7'
-
 services:
   pmm-server:
     image: ${PMM_SERVER_IMAGE:-perconalab/pmm-server:3-dev-latest}

api-tests/docker-compose.yml

@@ -82,15 +82,15 @@ services:
       mysql:
         condition: service_healthy
     environment:
-      PMM_AGENT_SERVER_ADDRESS: pmm-server
+      PMM_AGENT_SERVER_ADDRESS: pmm-server:8443
       PMM_AGENT_SERVER_USERNAME: admin
       PMM_AGENT_SERVER_PASSWORD: admin
       PMM_AGENT_SERVER_INSECURE_TLS: 1
       PMM_AGENT_CONFIG_FILE: config/pmm-agent.yaml
       PMM_AGENT_SETUP: 1
       PMM_AGENT_SETUP_FORCE: 1
       PMM_AGENT_SIDECAR: 1
-#      PMM_AGENT_PRERUN_SCRIPT: "pmm-admin status --wait=10s; pmm-admin add mysql --username=pmm-agent --password=pmm-agent-password --host=pmm-agent_mysql --port=3306"
+      # PMM_AGENT_PRERUN_SCRIPT: "pmm-admin status --wait=10s; pmm-admin add mysql --username=pmm-agent --password=pmm-agent-password --host=pmm-agent_mysql --port=3306"
 
   sysbench:
     image: perconalab/sysbench

managed/cmd/pmm-managed/main.go

@@ -1082,40 +1082,30 @@ func main() { //nolint:maintidx,cyclop
 	l.Info("Starting services...")
 	var wg sync.WaitGroup
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		authServer.Run(ctx)
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		vmalert.Run(ctx)
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		vmdb.Run(ctx)
-	}()
+	})
 
 	haService.AddLeaderService(ha.NewContextService("checks", func(ctx context.Context) error {
 		checksService.Run(ctx)
 		return nil
 	}))
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		supervisord.Run(ctx)
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		updater.Run(ctx)
-	}()
+	})
 
 	wg.Add(1)
 	haService.AddLeaderService(ha.NewContextService("telemetry", func(ctx context.Context) error {
@@ -1134,9 +1124,7 @@ func main() { //nolint:maintidx,cyclop
 		return nil
 	}))
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		runGRPCServer(ctx,
 			&gRPCServerDeps{
 				actions:              actionsService,
@@ -1170,22 +1158,18 @@ func main() { //nolint:maintidx,cyclop
 				vmClient:             &vmClient,
 				vmdb:                 vmdb,
 			})
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		runHTTP1Server(ctx, &http1ServerDeps{
 			logs:       logs,
 			authServer: authServer,
 		})
-	}()
+	})
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		runDebugServer(ctx)
-	}()
+	})
 
 	haService.AddLeaderService(ha.NewContextService("cleaner", func(ctx context.Context) error {
 		cleaner.Run(ctx, cleanInterval, cleanOlderThan)

managed/services/ha/highavailability.go

@@ -93,9 +93,7 @@ func New(params *models.HAParams) *Service {
 
 // Run runs the high availability service.
 func (s *Service) Run(ctx context.Context) error {
-	s.wg.Add(1)
-	go func() {
-		defer s.wg.Done()
+	s.wg.Go(func() {
 		for {
 			select {
 			case <-s.services.Refresh():
@@ -107,12 +105,12 @@ func (s *Service) Run(ctx context.Context) error {
 				return
 			}
 		}
-	}()
+	})
 
 	if !s.params.Enabled {
 		s.l.Infoln("High availability is disabled")
-		s.services.Wait()
 		s.wg.Wait()
+		s.services.Wait()
 		return nil
 	}
 
@@ -197,22 +195,18 @@ func (s *Service) Run(ctx context.Context) error {
 			return fmt.Errorf("failed to join memberlist cluster: %w", err)
 		}
 	}
-	s.wg.Add(1)
-	go func() {
-		defer s.wg.Done()
+	s.wg.Go(func() {
 		s.runLeaderObserver(ctx)
-	}()
+	})
 
-	s.wg.Add(1)
-	go func() {
-		defer s.wg.Done()
+	s.wg.Go(func() {
 		s.runRaftNodesSynchronizer(ctx)
-	}()
+	})
 
 	<-ctx.Done()
 
-	s.services.Wait()
 	s.wg.Wait()
+	s.services.Wait()
 
 	return nil
 }

managed/services/ha/services.go

@@ -50,7 +50,7 @@ func (s *services) Add(service LeaderService) error {
 
 	id := service.ID()
 	if _, ok := s.all[id]; ok {
-		return fmt.Errorf("service with id %s is already exist", id)
+		return fmt.Errorf("service with id %s already exists", id)
 	}
 	s.all[id] = service
 	select {
@@ -61,33 +61,46 @@ func (s *services) Add(service LeaderService) error {
 }
 
 func (s *services) StartAllServices(ctx context.Context) {
-	s.rw.Lock()
-	defer s.rw.Unlock()
+	type startItem struct {
+		svc LeaderService
+		id  string
+	}
 
+	s.rw.Lock()
+	toStart := make([]startItem, 0, len(s.all))
 	for id, service := range s.all {
 		if _, ok := s.running[id]; !ok {
-			s.wg.Add(1)
 			s.running[id] = service
-			go func() {
-				s.l.Infoln("Starting", service.ID())
-				err := service.Start(ctx)
-				if err != nil {
-					s.l.Errorln(err)
-					s.removeService(service.ID())
-				}
-			}()
+			toStart = append(toStart, startItem{svc: service, id: id})
 		}
 	}
+	s.rw.Unlock()
+
+	for _, service := range toStart {
+		s.wg.Add(1)
+		go func(svc LeaderService, svcID string) {
+			s.l.Infoln("Starting", svcID)
+			err := svc.Start(ctx)
+			if err != nil {
+				s.l.Errorln(err)
+				s.removeService(svcID)
+			}
+		}(service.svc, service.id)
+	}
 }
 
 func (s *services) StopRunningServices() {
 	s.rw.Lock()
-	defer s.rw.Unlock()
-
+	toStop := make([]LeaderService, 0, len(s.running))
 	for id, service := range s.running {
+		toStop = append(toStop, service)
+		delete(s.running, id)
+	}
+	s.rw.Unlock()
+
+	for _, service := range toStop {
 		s.l.Infoln("Stopping", service.ID())
 		service.Stop()
-		delete(s.running, id)
 		s.wg.Done()
 	}
 }
@@ -102,7 +115,7 @@ func (s *services) Wait() {
 
 func (s *services) removeService(id string) {
 	s.rw.Lock()
-	defer s.rw.Unlock()
 	delete(s.running, id)
+	s.rw.Unlock()
 	s.wg.Done()
 }

managed/services/victoriametrics/victoriametrics.go

@@ -43,8 +43,8 @@ import (
 )
 
 const (
-	updateBatchDelay           = time.Second
-	configurationUpdateTimeout = 3 * time.Second
+	updateBatchDelay           = 3 * time.Second
+	configurationUpdateTimeout = 5 * time.Second
 
 	victoriametricsDir     = "/srv/victoriametrics"
 	victoriametricsDataDir = "/srv/victoriametrics/data"
@@ -242,7 +242,7 @@ func (svc *Service) validateConfig(ctx context.Context, cfg []byte) error {
 		_ = os.Remove(f.Name())
 	}()
 
-	args := []string{"-dryRun", "-promscrape.config", f.Name()}
+	args := []string{"-promscrape.config.dryRun=true", "-promscrape.config", f.Name()}
 	cmd := exec.CommandContext(ctx, "victoriametrics", args...) //nolint:gosec
 	pdeathsig.Set(cmd, unix.SIGKILL)