PMM-15360: Gate OpenManager's service on the switch

managed/cmd/pmm-managed/main.go

@@ -581,7 +581,7 @@
 		deps.l.Warnf("Failed to get settings: %s.", err)
 		return false
 	}
 	err = deps.supervisord.UpdateConfiguration(settings)
 	if err != nil {
 		deps.l.Warnf("Failed to update supervisord configuration: %s.", err)
 		return false
@@ -1065,6 +1065,15 @@
 		VMURL:         *victoriaMetricsURLF,
 	})
 
+	// Where SEP is, optional. Empty means OM builds its document from PMM's own inventory
+	// and metrics alone and records the probe source as disabled.
+	//
+	// Constructed here, ahead of serverParams below, so server.Server can hold it for the
+	// OpenManager enable/disable switch (Enabled gate, IsAvailable check).
+	omService := om.New(db, v1.NewAPI(vmClient), haService, logrus.WithField("component", "om"))
+	omService.WithProbeSource(*sepURLF, *sepTokenF)
+	prom.MustRegister(om.NewMetricsCollector(omService))
+
 	serverParams := &server.Params{
 		DB:                   db,
 		VMDB:                 vmdb,
@@ -1081,6 +1090,7 @@
 		HAService:            haService,
 		Nomad:                nomad,
 		QANClient:            qanClient,
+		OmService:            omService,
 	}
 
 	server, err := server.NewServer(serverParams)
@@ -1193,12 +1203,6 @@
 		return nil
 	}))
 
-	// Where SEP is, optional. Empty means OM builds its document from PMM's own inventory
-	// and metrics alone and records the probe source as disabled.
-	omService := om.New(db, v1.NewAPI(vmClient), haService, logrus.WithField("component", "om"))
-	omService.WithProbeSource(*sepURLF, *sepTokenF)
-	prom.MustRegister(om.NewMetricsCollector(omService))
-
 	// Leader-only, like every other periodic writer here. A collection persists a run and
 	// its snapshot and then prunes the shared history, so running it on every node of an
 	// HA cluster would have each node writing runs and pruning the others' -- and the

managed/services/om/sep_client.go

@@ -102,3 +102,50 @@ func (a sepApp) request(ctx context.Context, method, path string, query url.Valu
 	}
 	return req, nil
 }
+
+// patchConfig sends fields to the app's PATCH /config endpoint and discards the
+// response body -- callers that only want to write settings, not read the
+// SettingResponse rows PATCH returns, use this instead of building the request
+// themselves.
+func (a sepApp) patchConfig(ctx context.Context, fields map[string]any) error {
+	req, err := a.request(ctx, http.MethodPatch, "config", nil, fields, false)
+	if err != nil {
+		return fmt.Errorf("failed to build the request: %w", err)
+	}
+
+	resp, err := a.client.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("PATCH %s: %w", a.endpoint("config"), err)
+	}
+	defer resp.Body.Close() //nolint:errcheck
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("PATCH %s: unexpected status %s", a.endpoint("config"), resp.Status)
+	}
+	return nil
+}
+
+// triggerRun queues a full-estate probe sweep via the app's POST /runs endpoint and
+// discards the accepted run's body. Body is sent empty rather than omitted -- the
+// endpoint takes an optional scope object, and FastAPI rejects a bodyless POST
+// against a typed body.
+//
+// A 409 (a sweep already in flight) is not treated as a failure to log: the estate
+// is about to be swept either way, which is exactly the outcome a caller here wants.
+func (a sepApp) triggerRun(ctx context.Context) error {
+	req, err := a.request(ctx, http.MethodPost, "runs", nil, nil, true)
+	if err != nil {
+		return fmt.Errorf("failed to build the request: %w", err)
+	}
+
+	resp, err := a.client.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("POST %s: %w", a.endpoint("runs"), err)
+	}
+	defer resp.Body.Close() //nolint:errcheck
+
+	if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusConflict {
+		return fmt.Errorf("POST %s: unexpected status %s", a.endpoint("runs"), resp.Status)
+	}
+	return nil
+}

managed/services/om/service.go

@@ -157,6 +157,71 @@ func (s *Service) WithProbeSource(sepURL, token string) *Service {
 	return s
 }
 
+// Enabled returns true if OpenManager is enabled, so every /v1/om/* RPC and the
+// scheduled collection in Run refuse while it is off, via the same generic
+// gRPC-service-enabled interceptor BackupService and the other preview features use.
+func (s *Service) Enabled() bool {
+	settings, err := models.GetSettings(s.db)
+	if err != nil {
+		s.l.WithError(err).Error("can't get settings")
+		return false
+	}
+	return settings.IsOMEnabled()
+}
+
+// IsAvailable reports whether SEP's om_inventory app is configured and reachable.
+//
+// Used to gate turning OpenManager on: an admin flipping the switch with no inventory
+// app to talk to would enable a UI backed by a source that can never answer, with no
+// way to tell "off" from "broken" apart from reading logs. It does not drive anything
+// on SEP's side -- this is the same read every scheduled collection already performs
+// via probeSource.collect, just run once up front rather than waited out.
+func (s *Service) IsAvailable(ctx context.Context) bool {
+	if s.probe == nil || s.probe.app.client == nil {
+		return false
+	}
+	_, err := s.probe.fetch(ctx)
+	return err == nil
+}
+
+// SyncInventoryEnabled tells SEP's om_inventory app whether OpenManager is on, and
+// on enabling, kicks an immediate sweep instead of leaving the estate to wait out
+// SCHEDULE's own interval.
+//
+// PATCHes ENABLED rather than SCHEDULE: the app keeps its own configured cadence
+// (an operator's SCHEDULE override) independent of whether OpenManager is turned
+// on, so toggling this switch off and back on does not reset a customized interval
+// back to the app's default. See OmInventorySettings in SEP for the other half.
+//
+// The immediate sweep exists because a freshly (re-)enabled periodic task in SEP's
+// beat store is not due until one full SCHEDULE interval has elapsed -- there is no
+// "run once now, then repeat" concept in an interval schedule, so a 60-minute
+// cadence would otherwise leave the estate empty for up to an hour after being
+// turned on. Mirrors triggerOMCollectionIfJustEnabled, PMM's own equivalent kick for
+// its topology page.
+//
+// Both calls are best-effort: a stale write, or a sweep that does not fire, means
+// SEP is briefly out of step with PMM's switch, not a broken settings change, so
+// failure is logged rather than returned to the caller -- matching
+// triggerOMCollectionIfJustEnabled, the other side effect ChangeSettings fires on
+// this same transition.
+func (s *Service) SyncInventoryEnabled(ctx context.Context, enabled bool) {
+	if s.probe == nil || s.probe.app.client == nil {
+		return
+	}
+	if err := s.probe.app.patchConfig(ctx, map[string]any{"ENABLED": enabled}); err != nil {
+		s.l.WithError(err).WithField("enabled", enabled).
+			Warn("failed to sync OpenManager's on/off state to SEP's om_inventory app")
+		return
+	}
+	if !enabled {
+		return
+	}
+	if err := s.probe.app.triggerRun(ctx); err != nil {
+		s.l.WithError(err).Warn("failed to trigger an immediate SEP inventory sweep after enabling OpenManager")
+	}
+}
+
 // GetTopology returns the whole MongoDB estate as one document.
 //
 // A pure read path: memory, then the stored snapshot, never a collection. Collection is
@@ -261,6 +326,9 @@ func (s *Service) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
+			if !s.Enabled() {
+				continue
+			}
 			_, err := s.discover(ctx)
 			if err != nil && ctx.Err() == nil {
 				s.l.Warnf("scheduled collection failed: %s", err)

managed/services/server/deps.go

@@ -20,6 +20,7 @@ import (
 	"net/url"
 	"time"
 
+	omv1 "github.com/percona/pmm/api/om/v1"
 	serverv1 "github.com/percona/pmm/api/server/v1"
 	"github.com/percona/pmm/managed/models"
 )
@@ -118,3 +119,14 @@ type victoriaMetricsParams interface {
 type nomadService interface {
 	UpdateConfiguration(settings *models.Settings) error
 }
+
+// omService is a subset of methods of om.Service used by this package.
+// We use it instead of the real type to avoid a dependency cycle.
+type omService interface {
+	// IsAvailable reports whether SEP's OpenManager Inventory app is configured and
+	// reachable, gating whether OpenManager may be enabled.
+	IsAvailable(ctx context.Context) bool
+	TriggerTopologyCollection(ctx context.Context, req *omv1.TriggerTopologyCollectionRequest) (*omv1.TriggerTopologyCollectionResponse, error)
+	// SyncInventoryEnabled tells SEP's om_inventory app whether OpenManager is on.
+	SyncInventoryEnabled(ctx context.Context, enabled bool)
+}

managed/services/server/server.go

@@ -38,6 +38,7 @@
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"gopkg.in/reform.v1"
 
+	omv1 "github.com/percona/pmm/api/om/v1"
 	serverv1 "github.com/percona/pmm/api/server/v1"
 	"github.com/percona/pmm/managed/models"
 	"github.com/percona/pmm/managed/utils/distribution"
@@ -66,6 +67,7 @@
 	haService            haService
 	updater              *Updater
 	nomad                nomadService
+	omService            omService
 
 	l *logrus.Entry
 
@@ -92,6 +94,7 @@
 	Dus                  *distribution.Service
 	HAService            haService
 	Nomad                nomadService
+	OmService            omService
 }
 
 // NewServer returns new server for Server service.
@@ -117,6 +120,7 @@
 		updater:              params.Updater,
 		haService:            params.HAService,
 		nomad:                params.Nomad,
+		omService:            params.OmService,
 		l:                    logrus.WithField("component", "server"),
 		envSettings:          &models.ChangeSettingsParams{},
 	}
@@ -452,7 +456,7 @@
 	}, nil
 }
 
 func (s *Server) validateChangeSettingsRequest(ctx context.Context, req *serverv1.ChangeSettingsRequest) error {
 	metricsRes := req.MetricsResolutions
 
 	if req.SshKey != nil {
@@ -484,6 +488,21 @@
 		return status.Error(codes.FailedPrecondition, "Azure Discover is configured via PMM_ENABLE_AZURE_DISCOVER environment variable.")
 	}
 
+	if req.EnableOm != nil && s.envSettings.EnableOM != nil && *req.EnableOm != *s.envSettings.EnableOM {
+		return status.Error(codes.FailedPrecondition, "OpenManager is configured via PMM_ENABLE_OM environment variable.")
+	}
+
+	if req.EnableOm != nil && *req.EnableOm {
+		currentSettings, err := models.GetSettings(s.db.WithContext(ctx))
+		if err != nil {
+			return status.Errorf(codes.Internal, "failed to get server settings: %s", err)
+		}
+
+		if !currentSettings.IsOMEnabled() && (s.omService == nil || !s.omService.IsAvailable(ctx)) {
+			return status.Error(codes.FailedPrecondition, "OpenManager cannot be enabled: the OpenManager Inventory app is not available in SEP.")
+		}
+	}
+
 	if !canUpdateDurationSetting(metricsRes.GetHr().AsDuration(), s.envSettings.MetricsResolutions.HR) {
 		return status.Error(
 			codes.FailedPrecondition,
@@ -536,6 +555,7 @@
 			EnableBackupManagement: req.EnableBackupManagement,
 			EnableAccessControl:    req.EnableAccessControl,
 			EnableInternalPgQAN:    req.EnableInternalPgQan,
+			EnableOM:               req.EnableOm,
 			AdvisorsRunInterval: models.AdvisorsRunIntervals{
 				RareInterval:     advisorsRunInterval.GetRareInterval().AsDuration(),
 				StandardInterval: advisorsRunInterval.GetStandardInterval().AsDuration(),
@@ -627,11 +647,36 @@
 		}
 	}
 
+	s.triggerOMCollectionIfJustEnabled(ctx, oldSettings, newSettings)
+	s.syncOMInventoryEnabledIfChanged(ctx, oldSettings, newSettings)
+
 	return &serverv1.ChangeSettingsResponse{
 		Settings: s.convertSettings(newSettings, disableInternalPgQan),
 	}, nil
 }
 
+// triggerOMCollectionIfJustEnabled kicks a topology collection so OpenManager's page is
+// not empty on first view, instead of waiting out the next scheduled tick.
+func (s *Server) triggerOMCollectionIfJustEnabled(ctx context.Context, oldSettings, newSettings *models.Settings) {
+	if oldSettings.IsOMEnabled() || !newSettings.IsOMEnabled() || s.omService == nil {
+		return
+	}
+	_, err := s.omService.TriggerTopologyCollection(ctx, &omv1.TriggerTopologyCollectionRequest{})
+	if err != nil {
+		s.l.WithError(err).Warn("failed to trigger OpenManager topology collection after enabling")
+	}
+}
+
+// syncOMInventoryEnabledIfChanged tells SEP's om_inventory app to start or stop its
+// own estate sweep alongside this switch, on either transition -- unlike
+// triggerOMCollectionIfJustEnabled, which only reacts to enabling.
+func (s *Server) syncOMInventoryEnabledIfChanged(ctx context.Context, oldSettings, newSettings *models.Settings) {
+	if oldSettings.IsOMEnabled() == newSettings.IsOMEnabled() || s.omService == nil {
+		return
+	}
+	s.omService.SyncInventoryEnabled(ctx, newSettings.IsOMEnabled())
+}
+
 func (s *Server) getInternalPgQANAgent(q *reform.Querier) (*models.Agent, error) {
 	agents, err := models.FindAgents(q, models.AgentFilters{
 		PMMAgentID: models.PMMServerAgentID,