PMM-15326 Add a Prometheus collector for OM collection health

OM registered no collector where agents, inventory, ha, advisor and
backups all do, per 4nte's review: the per-run counters it already
computes ended up only in om_topology_runs (pruned at runHistory) and a
log line at Info, so a degraded pass was invisible on PMM Health.

Adds MetricsCollector, following the ha/inventory/backup shape (manual
prom.Desc + prom.MustNewConstMetric, Describe delegating to
DescribeByCollect), registered next to omService's own construction in
main.go:

  - pmm_managed_om_runs_total{status} -- completed runs by outcome,
    tallied by three atomic counters on Service and incremented once
    per run by the new recordRunOutcome.
  - pmm_managed_om_document_age_seconds -- age of the newest
    observation in the document currently served, deliberately the
    same age Snapshot.Stale is already computed from, so the existing
    staleness signal becomes alertable rather than only renderable.
    Unset until the estate has been observed at least once.
  - pmm_managed_om_services{state} -- estate size versus how much
    answered, from the last pass's Summary.

Verified live against the running dev stack: runs_total advanced on
the next collection tick, document_age_seconds tracked real elapsed
time, and services{state} matched GetTopology's own summary (14/14).

Signed-off-by: Pawel Lebioda <pawel.lebioda@percona.com>

Author: plebioda

managed/cmd/pmm-managed/main.go

@@ -1201,6 +1201,7 @@ func main() { //nolint:gocognit,maintidx,cyclop
 	// 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

managed/services/om/om_metrics.go

@@ -0,0 +1,105 @@
+// Copyright (C) 2023 Percona LLC
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+package om
+
+import (
+	"time"
+
+	prom "github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	prometheusNamespace = "pmm_managed"
+	prometheusSubsystem = "om"
+)
+
+// MetricsCollector exposes OM's own collection health to Prometheus -- otherwise a
+// degraded pass is visible only in the run history, which prunes at runHistory, and in a
+// log line at Info.
+//
+// The following metrics are exposed:
+//
+//   - pmm_managed_om_runs_total{status="success|partial|failed"} -- completed collection
+//     runs so far, by outcome. A rising "partial" or "failed" rate says collection itself
+//     is unhealthy, before any caller notices the document is stale.
+//
+//   - pmm_managed_om_document_age_seconds -- how old the newest observation in the
+//     document PMM currently serves is. This is the same age Snapshot.Stale is computed
+//     from, exposed as a number so it can be alerted on rather than only rendered. Unset
+//     until the estate has been observed at least once.
+//
+//   - pmm_managed_om_services{state="total|up|down"} -- how much of the estate the last
+//     collection pass saw, and how much of it answered.
+type MetricsCollector struct {
+	svc *Service
+
+	mRunsTotal   *prom.Desc
+	mDocumentAge *prom.Desc
+	mServices    *prom.Desc
+}
+
+// NewMetricsCollector creates a new MetricsCollector backed by the given OM service.
+func NewMetricsCollector(svc *Service) *MetricsCollector {
+	return &MetricsCollector{
+		svc: svc,
+		mRunsTotal: prom.NewDesc(
+			prom.BuildFQName(prometheusNamespace, prometheusSubsystem, "runs_total"),
+			"Total number of OM topology collection runs, by outcome status.",
+			[]string{"status"},
+			nil,
+		),
+		mDocumentAge: prom.NewDesc(
+			prom.BuildFQName(prometheusNamespace, prometheusSubsystem, "document_age_seconds"),
+			"Age, in seconds, of the newest observation in the topology document PMM currently serves.",
+			nil,
+			nil,
+		),
+		mServices: prom.NewDesc(
+			prom.BuildFQName(prometheusNamespace, prometheusSubsystem, "services"),
+			"Services in the topology document PMM currently serves, by state.",
+			[]string{"state"},
+			nil,
+		),
+	}
+}
+
+// Describe implements prom.Collector.
+func (c *MetricsCollector) Describe(ch chan<- *prom.Desc) {
+	prom.DescribeByCollect(c, ch)
+}
+
+// Collect implements prom.Collector.
+func (c *MetricsCollector) Collect(ch chan<- prom.Metric) {
+	ch <- prom.MustNewConstMetric(c.mRunsTotal, prom.CounterValue, float64(c.svc.runsSuccess.Load()), runStatusSuccess)
+	ch <- prom.MustNewConstMetric(c.mRunsTotal, prom.CounterValue, float64(c.svc.runsPartial.Load()), runStatusPartial)
+	ch <- prom.MustNewConstMetric(c.mRunsTotal, prom.CounterValue, float64(c.svc.runsFailed.Load()), runStatusFailed)
+
+	snap := c.svc.snapshot()
+	if snap.GetSnapshot().GetObservedAt() == nil {
+		// Nothing has been observed yet -- a fresh estate whose leader has not completed
+		// its first pass. Reporting age 0 would read as "current" rather than "no data".
+		return
+	}
+	age := time.Since(snap.Snapshot.ObservedAt.AsTime()).Seconds()
+	ch <- prom.MustNewConstMetric(c.mDocumentAge, prom.GaugeValue, age)
+
+	summary := snap.Summary
+	ch <- prom.MustNewConstMetric(c.mServices, prom.GaugeValue, float64(summary.GetTotalServices()), "total")
+	ch <- prom.MustNewConstMetric(c.mServices, prom.GaugeValue, float64(summary.GetUpServices()), "up")
+	ch <- prom.MustNewConstMetric(c.mServices, prom.GaugeValue, float64(summary.GetDownServices()), "down")
+}
+
+var _ prom.Collector = (*MetricsCollector)(nil)

managed/services/om/service.go

@@ -37,6 +37,7 @@ import (
 	"fmt"
 	"net/http"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -110,6 +111,13 @@ type Service struct {
 
 	mu     sync.Mutex
 	latest *omv1.GetTopologyResponse
+
+	// Completed-run tally by status, read by MetricsCollector on every Prometheus scrape
+	// and written once per run by recordRunOutcome. Atomic because a scrape can land while
+	// a collection is in flight.
+	runsSuccess atomic.Int64
+	runsPartial atomic.Int64
+	runsFailed  atomic.Int64
 }
 
 // New returns a new OM service.
@@ -314,6 +322,7 @@ func (s *Service) collect(ctx context.Context) (*omv1.GetTopologyResponse, *omv1
 		Environments:  doc.environments,
 	}
 	run := buildRun(runID, startedAt, generatedAt, services, merged, doc, results)
+	s.recordRunOutcome(run.Status)
 
 	s.mu.Lock()
 	s.latest = response
@@ -539,6 +548,21 @@ func buildRun(
 	return run
 }
 
+// recordRunOutcome tallies one completed run by status, for MetricsCollector's
+// pmm_managed_om_runs_total counter.
+func (s *Service) recordRunOutcome(status omv1.RunStatus) {
+	switch status {
+	case omv1.RunStatus_RUN_STATUS_SUCCESS:
+		s.runsSuccess.Add(1)
+	case omv1.RunStatus_RUN_STATUS_PARTIAL:
+		s.runsPartial.Add(1)
+	case omv1.RunStatus_RUN_STATUS_FAILED:
+		s.runsFailed.Add(1)
+	default:
+		// buildRun only ever sets one of the three above.
+	}
+}
+
 // detailStrings renders a source's counters for the wire, which carries them as strings
 // so a source can report whatever it has without the contract naming every counter.
 func detailStrings(detail map[string]any) map[string]string {