PMM-14324 PMM HA monitoring

managed/cmd/pmm-managed/main.go

@@ -796,6 +796,7 @@ func main() { //nolint:maintidx,cyclop
 		SSLCAPath:   *postgresSSLCAPathF,
 		SSLKeyPath:  *postgresSSLKeyPathF,
 		SSLCertPath: *postgresSSLCertPathF,
+		HANodeID:    *haNodeID,
 	}
 
 	sqlDB, err := models.OpenDB(setupParams)
@@ -823,10 +824,16 @@ func main() { //nolint:maintidx,cyclop
 
 	cleaner := clean.New(db)
 	externalRules := vmalert.NewExternalRules()
-	vmdb, err := victoriametrics.NewVictoriaMetrics(*victoriaMetricsConfigF, db, vmParams)
+	vmdb, err := victoriametrics.NewVictoriaMetrics(*victoriaMetricsConfigF, db, vmParams, haService)
 	if err != nil {
 		l.Panicf("VictoriaMetrics service problem: %+v", err)
 	}
+
+	// This ensures scrape config regeneration happens when leadership changes
+	if haParams.Enabled {
+		haService.AddLeaderService(vmdb)
+	}
+
 	vmalert, err := vmalert.NewVMAlert(externalRules, *victoriaMetricsVMAlertURLF)
 	if err != nil {
 		l.Panicf("VictoriaMetrics VMAlert service problem: %+v", err)
@@ -839,11 +846,6 @@ func main() { //nolint:maintidx,cyclop
 
 	agentsRegistry := agents.NewRegistry(db, vmParams)
 
-	// TODO remove once PMM cluster will be Active-Active
-	haService.AddLeaderService(ha.NewStandardService("agentsRegistry", func(_ context.Context) error { return nil }, func() {
-		agentsRegistry.KickAll(ctx)
-	}))
-
 	pbmPITRService := backup.NewPBMPITRService()
 	backupRemovalService := backup.NewRemovalService(db, pbmPITRService)
 	backupRetentionService := backup.NewRetentionService(db, backupRemovalService)
@@ -877,19 +879,6 @@ func main() { //nolint:maintidx,cyclop
 			HAParams: haParams,
 		})
 
-	haService.AddLeaderService(ha.NewStandardService("pmm-agent-runner", func(_ context.Context) error {
-		err := supervisord.StartSupervisedService("pmm-agent")
-		if err != nil {
-			l.Warnf("couldn't start pmm-agent: %q", err)
-		}
-		return err
-	}, func() {
-		err := supervisord.StopSupervisedService("pmm-agent")
-		if err != nil {
-			l.Warnf("couldn't stop pmm-agent: %q", err)
-		}
-	}))
-
 	platformAddress, err := envvars.GetPlatformAddress()
 	if err != nil {
 		l.Fatal(err)

managed/models/database.go

@@ -1213,6 +1213,7 @@ type SetupDBParams struct {
 	SSLCAPath        string
 	SSLKeyPath       string
 	SSLCertPath      string
+	HANodeID         string
 	SetupFixtures    SetupFixturesMode
 	MigrationVersion *int
 }
@@ -1462,14 +1463,19 @@ func setupPMMServerAgents(q *reform.Querier, params SetupDBParams) error {
 		return err
 	}
 	if params.Address != DefaultPostgreSQLAddr {
+		nodeName := PMMServerPostgreSQLNodeName
+		if params.HANodeID != "" {
+			nodeName = params.HANodeID
+		}
 		if node, err = CreateNode(q, RemoteNodeType, &CreateNodeParams{
-			NodeName: PMMServerPostgreSQLNodeName,
+			NodeName: nodeName,
 			Address:  address,
 		}); err != nil {
 			return err
 		}
 	} else {
-		params.Name = "" // using postgres database in order to get metrics from entrypoint extension setup for QAN.
+		// using postgres database in order to get metrics from entrypoint extension setup for QAN.
+		params.Name = ""
 	}
 
 	// create PostgreSQL Service and associated Agents

managed/models/node_helpers.go

@@ -198,7 +198,7 @@ func createNodeWithID(q *reform.Querier, id string, nodeType NodeType, params *C
 		return nil, err
 	}
 
-	// do not check that machine-id is unique: https://jira.percona.com/browse/PMM-4196
+	// do not check that machine-id is unique: https://perconadev.atlassian.net/browse/PMM-4196
 
 	if nodeType == RemoteRDSNodeType {
 		if strings.Contains(params.InstanceID, ".") {
@@ -211,7 +211,7 @@ func createNodeWithID(q *reform.Querier, id string, nodeType NodeType, params *C
 	}
 
 	// Trim trailing \n received from broken 2.0.0 clients.
-	// See https://jira.percona.com/browse/PMM-4720
+	// See https://perconadev.atlassian.net/browse/PMM-4720
 	machineID := pointer.ToStringOrNil(strings.TrimSpace(pointer.GetString(params.MachineID)))
 
 	node := &Node{

managed/services/agents/deps.go

@@ -31,7 +31,7 @@ import (
 // FIXME Rename to victoriaMetrics.Service, update tests.
 type prometheusService interface {
 	RequestConfigurationUpdate()
-	BuildScrapeConfigForVMAgent(pmmAgentID string) ([]byte, error)
+	BuildScrapeConfigForVMAgent(ctx context.Context, pmmAgentID string) ([]byte, error)
 }
 
 // qanClient is a subset of methods of qan.Client used by this package.

managed/services/agents/registry.go

@@ -417,13 +417,6 @@ func (r *Registry) Collect(ch chan<- prom.Metric) {
 	r.mClockDrift.Collect(ch)
 }
 
-// KickAll sends a signal to all registered agents in the registry to perform a kick action.
-func (r *Registry) KickAll(ctx context.Context) {
-	for _, agentInfo := range r.agents {
-		r.Kick(ctx, agentInfo.id)
-	}
-}
-
 // check interfaces.
 var (
 	_ prom.Collector = (*Registry)(nil)

managed/services/agents/state.go

@@ -188,7 +188,7 @@ func (u *StateUpdater) sendSetStateRequest(ctx context.Context, agent *pmmAgentI
 		case models.PMMAgentType:
 			continue
 		case models.VMAgentType:
-			scrapeCfg, err := u.vmdb.BuildScrapeConfigForVMAgent(agent.id)
+			scrapeCfg, err := u.vmdb.BuildScrapeConfigForVMAgent(ctx, agent.id)
 			if err != nil {
 				return errors.Wrapf(err, "cannot get agent scrape config for agent: %s", agent.id)
 			}

managed/services/alerting/service.go

@@ -409,7 +409,7 @@ func (s *Service) CreateTemplate(ctx context.Context, req *alerting.CreateTempla
 
 	templates, err := alert.Parse(strings.NewReader(req.Yaml), pParams)
 	if err != nil {
-		s.l.Errorf("failed to parse rule template form request: +%v", err)
+		s.l.Errorf("failed to parse rule template form request: %+v", err)
 		return nil, status.Errorf(codes.InvalidArgument, "Failed to parse rule template: %v.", err)
 	}
 
@@ -462,7 +462,7 @@ func (s *Service) UpdateTemplate(ctx context.Context, req *alerting.UpdateTempla
 
 	templates, err := alert.Parse(strings.NewReader(req.Yaml), parseParams)
 	if err != nil {
-		s.l.Errorf("failed to parse rule template form request: +%v", err)
+		s.l.Errorf("failed to parse rule template form request: %+v", err)
 		return nil, status.Error(codes.InvalidArgument, "Failed to parse rule template.")
 	}
 

managed/services/ha/highavailability.go

@@ -179,7 +179,7 @@ func (s *Service) Run(ctx context.Context) error {
 					s.services.StartAllServices(ctx)
 				}
 			case <-ctx.Done():
-				s.services.StopRunningServices()
+				s.services.StopAllServices()
 				return
 			}
 		}
@@ -413,8 +413,7 @@ func (s *Service) runLeaderObserver(ctx context.Context) {
 		case isLeader := <-node.LeaderCh():
 			if isLeader {
 				s.services.StartAllServices(ctx)
-				// This node is the leader
-				s.l.Printf("I am the leader!")
+				s.l.Info("I am the leader!")
 				peers := s.memberlist.Members()
 				for _, peer := range peers {
 					if peer.Name == s.params.NodeID {
@@ -423,8 +422,8 @@ func (s *Service) runLeaderObserver(ctx context.Context) {
 					s.addMemberlistNodeToRaft(peer)
 				}
 			} else {
-				s.l.Printf("I am not a leader!")
-				s.services.StopRunningServices()
+				s.l.Info("I am not a leader!")
+				s.services.StopAllServices()
 			}
 		case <-t.C:
 			address, serverID := s.raftNode.LeaderWithID()
@@ -439,7 +438,7 @@ func (s *Service) runLeaderObserver(ctx context.Context) {
 func (s *Service) AddLeaderService(leaderService LeaderService) {
 	err := s.services.Add(leaderService)
 	if err != nil {
-		s.l.Errorf("couldn't add HA service: +%v", err)
+		s.l.Errorf("couldn't add HA service: %+v", err)
 	}
 }
 
@@ -460,14 +459,14 @@ func (s *Service) BroadcastMessage(message []byte) error {
 	return nil
 }
 
-// IsLeader checks if the current instance of the high availability service is the leader.
+// IsLeader checks if the current instance of HA service is the leader.
 func (s *Service) IsLeader() bool {
 	s.rw.RLock()
 	defer s.rw.RUnlock()
 	return (s.raftNode != nil && s.raftNode.State() == raft.Leader) || !s.params.Enabled
 }
 
-// Bootstrap performs the necessary steps to initialize the high availability service.
+// Bootstrap returns true if HA service should bootstrap (true in non-HA setups).
 func (s *Service) Bootstrap() bool {
 	return s.params.Bootstrap || !s.params.Enabled
 }

managed/services/ha/leaderservice.go

@@ -27,44 +27,6 @@ type LeaderService interface {
 	ID() string
 }
 
-// StandardService represents a standard service in the high-availability setup.
-type StandardService struct {
-	id string
-
-	mu sync.Mutex
-
-	startFunc func(context.Context) error
-	stopFunc  func()
-}
-
-// NewStandardService creates a new standard service.
-func NewStandardService(id string, startFunc func(context.Context) error, stopFunc func()) *StandardService {
-	return &StandardService{
-		id:        id,
-		startFunc: startFunc,
-		stopFunc:  stopFunc,
-	}
-}
-
-// ID returns the ID of the standard service.
-func (s *StandardService) ID() string {
-	return s.id
-}
-
-// Start starts the standard service.
-func (s *StandardService) Start(ctx context.Context) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.startFunc(ctx)
-}
-
-// Stop stops the standard service.
-func (s *StandardService) Stop() {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.stopFunc()
-}
-
 // ContextService represents a context service.
 type ContextService struct {
 	id string

managed/services/ha/services.go

@@ -35,6 +35,7 @@ type services struct {
 	l *logrus.Entry
 }
 
+// newServices creates a new services manager.
 func newServices() *services {
 	return &services{
 		all:     make(map[string]LeaderService),
@@ -44,13 +45,14 @@ func newServices() *services {
 	}
 }
 
+// Add registers a new leader service.
 func (s *services) Add(service LeaderService) error {
 	s.rw.Lock()
 	defer s.rw.Unlock()
 
 	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 {
@@ -60,6 +62,7 @@ func (s *services) Add(service LeaderService) error {
 	return nil
 }
 
+// StartAllServices starts all registered services that are not currently running.
 func (s *services) StartAllServices(ctx context.Context) {
 	s.rw.Lock()
 	defer s.rw.Unlock()
@@ -80,7 +83,8 @@ func (s *services) StartAllServices(ctx context.Context) {
 	}
 }
 
-func (s *services) StopRunningServices() {
+// StopAllServices stops all running services.
+func (s *services) StopAllServices() {
 	s.rw.Lock()
 	defer s.rw.Unlock()
 
@@ -92,14 +96,17 @@ func (s *services) StopRunningServices() {
 	}
 }
 
+// Refresh returns a channel that signals when services should be refreshed.
 func (s *services) Refresh() chan struct{} {
 	return s.refresh
 }
 
+// Wait waits for all services to stop.
 func (s *services) Wait() {
 	s.wg.Wait()
 }
 
+// removeService removes a service from the registry of running services.
 func (s *services) removeService(id string) {
 	s.rw.Lock()
 	defer s.rw.Unlock()

managed/services/supervisord/pmm_config.go

@@ -181,7 +181,7 @@ redirect_stderr = true
 priority = 15
 command = /usr/sbin/pmm-agent --config-file=/usr/local/percona/pmm/config/pmm-agent.yaml --paths-tempdir=/srv/pmm-agent/tmp --paths-nomad-data-dir=/srv/nomad/data
 autorestart = true
-autostart = false
+autostart = true
 startretries = 1000
 startsecs = 1
 stopsignal = TERM

managed/services/supervisord/supervisord.go

@@ -413,24 +413,6 @@ func (s *Service) UpdateConfiguration(settings *models.Settings, ssoDetails *mod
 	return err
 }
 
-// RestartSupervisedService restarts given service.
-func (s *Service) RestartSupervisedService(serviceName string) error {
-	_, err := s.supervisorctl("restart", serviceName)
-	return err
-}
-
-// StartSupervisedService starts given service.
-func (s *Service) StartSupervisedService(serviceName string) error {
-	_, err := s.supervisorctl("start", serviceName)
-	return err
-}
-
-// StopSupervisedService stops given service.
-func (s *Service) StopSupervisedService(serviceName string) error {
-	_, err := s.supervisorctl("stop", serviceName)
-	return err
-}
-
 //nolint:lll
 var templates = template.Must(template.New("").Option("missingkey=error").Parse(`
 

managed/services/victoriametrics/prometheus.go

@@ -29,7 +29,7 @@ import (
 // AddScrapeConfigs - adds agents scrape configuration to given scrape config,
 // pmm_agent_id and push_metrics used for filtering.
 func AddScrapeConfigs(l *logrus.Entry, cfg *config.Config, q *reform.Querier, //nolint:cyclop,maintidx
-	globalResolutions *models.MetricsResolutions, pmmAgentID *string, pushMetrics bool,
+	globalResolutions *models.MetricsResolutions, pmmAgentID *string, pushMetrics bool, skipExternalAgents bool,
 ) error {
 	agents, err := models.FindAgentsForScrapeConfig(q, pmmAgentID, pushMetrics)
 	if err != nil {
@@ -187,6 +187,10 @@ func AddScrapeConfigs(l *logrus.Entry, cfg *config.Config, q *reform.Querier, //
 			continue
 
 		case models.RDSExporterType:
+			if skipExternalAgents && pointer.GetString(agent.RunsOnNodeID) == models.PMMServerNodeID {
+				l.Debugf("Skip the scrape config for RDSExporter %s running on PMM Server in HA non-leader mode", agent.AgentID)
+				continue
+			}
 			rdsParams = append(rdsParams, &scrapeConfigParams{
 				host:              paramsHost,
 				node:              paramsNode,
@@ -197,6 +201,10 @@ func AddScrapeConfigs(l *logrus.Entry, cfg *config.Config, q *reform.Querier, //
 			continue
 
 		case models.ExternalExporterType:
+			if skipExternalAgents && pointer.GetString(agent.RunsOnNodeID) == models.PMMServerNodeID {
+				l.Debugf("Skip the scrape config for ExternalExporter %s running on PMM Server in HA non-leader mode", agent.AgentID)
+				continue
+			}
 			scfgs, err = scrapeConfigsForExternalExporter(&mr, &scrapeConfigParams{
 				host:              paramsHost,
 				node:              paramsNode,