PMM-15326: Add the OM topology tables and models

Migration 119 creates om_topology_runs, one row per collection pass, and
om_topology_snapshots, the topology document each pass produced. The
document is JSONB rather than a relational tree because the topology model
is still moving; schema_version is what a reader checks. A snapshot is
deleted with its run, which is what bounds retention.

Nothing reads these yet - the service that does comes in a later PR.

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

Author: plebioda

managed/models/database.go

@@ -1185,6 +1185,40 @@ var databaseSchema = [][]string{
 		`ALTER TABLE dumps ADD COLUMN encrypted boolean NOT NULL DEFAULT false`,
 		`UPDATE dumps SET encrypted = false`,
 	},
+	119: {
+		// OM (OpenManager) topology collection: one row per pass, and the
+		// topology document it produced.
+		`CREATE TABLE om_topology_runs (
+			run_id            VARCHAR PRIMARY KEY,
+			started_at        TIMESTAMP NOT NULL,
+			finished_at       TIMESTAMP,
+			status            VARCHAR NOT NULL,
+			services_total    INTEGER NOT NULL DEFAULT 0,
+			services_resolved INTEGER NOT NULL DEFAULT 0,
+			services_orphaned INTEGER NOT NULL DEFAULT 0,
+			probes_ok         INTEGER NOT NULL DEFAULT 0,
+			services_stale    INTEGER NOT NULL DEFAULT 0,
+			origin_node       VARCHAR NOT NULL DEFAULT '',
+			sources           JSONB,
+			errors            JSONB,
+			created_at        TIMESTAMP NOT NULL
+		)`,
+		`CREATE INDEX om_topology_runs_started_at_idx ON om_topology_runs (started_at DESC)`,
+
+		// The document is JSONB rather than a relational tree because the topology model
+		// is still moving; schema_version is what a reader checks. It is deleted with its
+		// run, which is what bounds retention.
+		`CREATE TABLE om_topology_snapshots (
+			run_id         VARCHAR PRIMARY KEY REFERENCES om_topology_runs (run_id) ON DELETE CASCADE,
+			generated_at   TIMESTAMP NOT NULL,
+			observed_at    TIMESTAMP,
+			stale          BOOLEAN NOT NULL DEFAULT false,
+			schema_version INTEGER NOT NULL,
+			document       JSONB NOT NULL,
+			created_at     TIMESTAMP NOT NULL
+		)`,
+		`CREATE INDEX om_topology_snapshots_generated_at_idx ON om_topology_snapshots (generated_at DESC)`,
+	},
 }
 
 // ^^^ Avoid default values in schema definition. ^^^

managed/models/om_helpers.go

@@ -0,0 +1,111 @@
+// 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 models
+
+import (
+	"errors"
+	"fmt"
+
+	"gopkg.in/reform.v1"
+)
+
+// CreateOmTopologyRun stores one run and the topology document it produced.
+//
+// Both in one call because they are one fact: a run with no document is not a state any
+// reader should have to handle, and the caller writes them inside a transaction so no
+// reader ever sees one without the other.
+func CreateOmTopologyRun(q *reform.Querier, run *OmTopologyRun, snapshot *OmTopologySnapshot) error {
+	err := q.Insert(run)
+	if err != nil {
+		return fmt.Errorf("failed to insert OM run: %w", err)
+	}
+	if snapshot != nil {
+		snapshot.RunID = run.RunID
+		err = q.Insert(snapshot)
+		if err != nil {
+			return fmt.Errorf("failed to insert OM snapshot: %w", err)
+		}
+	}
+	return nil
+}
+
+// FindOmTopologyRuns returns the most recent runs, newest first.
+func FindOmTopologyRuns(q *reform.Querier, limit int) ([]*OmTopologyRun, error) {
+	if limit <= 0 {
+		return []*OmTopologyRun{}, nil
+	}
+	structs, err := q.SelectAllFrom(OmTopologyRunTable, "ORDER BY started_at DESC, run_id DESC LIMIT "+q.Placeholder(1), limit)
+	if err != nil {
+		return nil, fmt.Errorf("failed to select OM runs: %w", err)
+	}
+	runs := make([]*OmTopologyRun, len(structs))
+	for i, s := range structs {
+		runs[i] = s.(*OmTopologyRun) //nolint:forcetypeassert
+	}
+	return runs, nil
+}
+
+// FindOmTopologyRunByID returns one run, or ErrNotFound when there is no such run.
+func FindOmTopologyRunByID(q *reform.Querier, runID string) (*OmTopologyRun, error) {
+	if runID == "" {
+		return nil, NewInvalidArgumentError("run_id shouldn't be empty")
+	}
+	run := &OmTopologyRun{RunID: runID}
+	err := q.Reload(run)
+	if err != nil {
+		if errors.Is(err, reform.ErrNoRows) {
+			return nil, ErrNotFound
+		}
+		return nil, fmt.Errorf("failed to select OM run: %w", err)
+	}
+	return run, nil
+}
+
+// FindLatestOmTopologySnapshot returns the newest stored topology document, or ErrNotFound when
+// no collection has ever run.
+//
+// This is what lets a restarted pmm-managed serve the estate before it has collected
+// anything of its own.
+func FindLatestOmTopologySnapshot(q *reform.Querier) (*OmTopologySnapshot, error) {
+	structs, err := q.SelectAllFrom(OmTopologySnapshotTable, "ORDER BY generated_at DESC LIMIT 1")
+	if err != nil {
+		return nil, fmt.Errorf("failed to select the latest OM snapshot: %w", err)
+	}
+	if len(structs) == 0 {
+		return nil, ErrNotFound
+	}
+	return structs[0].(*OmTopologySnapshot), nil //nolint:forcetypeassert
+}
+
+// PruneOmTopologyRuns deletes all but the newest keep runs, cascading to their snapshots.
+//
+// Retention has to be bounded here rather than by an operator: collection runs on a timer
+// and on every read past the cache, so the table grows on its own. Pruning on write keeps
+// it at a fixed size without a second scheduled job to forget about.
+func PruneOmTopologyRuns(q *reform.Querier, keep int) error {
+	if keep <= 0 {
+		return NewInvalidArgumentError("keep should be positive")
+	}
+	_, err := q.Exec(`
+		DELETE FROM om_topology_runs
+		WHERE run_id NOT IN (
+			SELECT run_id FROM om_topology_runs ORDER BY started_at DESC, run_id DESC LIMIT `+q.Placeholder(1)+`
+		)`, keep)
+	if err != nil {
+		return fmt.Errorf("failed to prune OM runs: %w", err)
+	}
+	return nil
+}

managed/models/om_helpers_test.go

@@ -0,0 +1,155 @@
+// 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 models_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/reform.v1"
+	"gopkg.in/reform.v1/dialects/postgresql"
+
+	"github.com/percona/pmm/managed/models"
+	"github.com/percona/pmm/managed/utils/testdb"
+)
+
+func TestOmTopologyRuns(t *testing.T) {
+	sqlDB := testdb.Open(t, models.SkipFixtures, nil)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+
+	db := reform.NewDB(sqlDB, postgresql.Dialect, reform.NewPrintfLogger(t.Logf))
+	base := time.Date(2026, 8, 11, 12, 0, 0, 0, time.UTC)
+
+	// insert records one run and its document, minutes apart so ordering is unambiguous.
+	insert := func(t *testing.T, q *reform.Querier, id string, offset time.Duration, document string) {
+		t.Helper()
+		started := base.Add(offset)
+		finished := started.Add(time.Second)
+		observed := started.Add(-5 * time.Second)
+
+		run := &models.OmTopologyRun{
+			RunID: id, StartedAt: started, FinishedAt: &finished,
+			Status: models.OmTopologyRunSuccess, ServicesTotal: 14, ServicesResolved: 14,
+			ProbesOK: 13, ServicesStale: 1, OriginNode: "pmm-server",
+			Sources: models.OmTopologySourceReports{
+				{Source: "inventory", Status: "ok", Facts: 81},
+				{Source: "metrics", Status: "ok", Facts: 219, Detail: map[string]string{"queries": "13"}},
+			},
+			Errors: models.OmTopologyRunErrors{
+				{Scope: "query", Code: "vm_query_failed", Message: "boom"},
+			},
+		}
+		snapshot := &models.OmTopologySnapshot{
+			GeneratedAt: finished, ObservedAt: &observed,
+			SchemaVersion: 3, Document: []byte(document),
+		}
+		require.NoError(t, models.CreateOmTopologyRun(q, run, snapshot))
+	}
+
+	t.Run("a run round-trips with its receipt and its document", func(t *testing.T) {
+		q := db.Querier
+		insert(t, q, "run-1", 0, `{"origin_node":"pmm-server"}`)
+
+		run, err := models.FindOmTopologyRunByID(q, "run-1")
+		require.NoError(t, err)
+		assert.Equal(t, models.OmTopologyRunSuccess, run.Status)
+		assert.Equal(t, int32(14), run.ServicesTotal)
+		assert.Equal(t, int32(1), run.ServicesStale)
+		assert.Equal(t, "pmm-server", run.OriginNode)
+
+		// The per-source receipt is the part that makes a thin document legible.
+		require.Len(t, run.Sources, 2)
+		assert.Equal(t, "metrics", run.Sources[1].Source)
+		assert.Equal(t, "13", run.Sources[1].Detail["queries"])
+		require.Len(t, run.Errors, 1)
+		assert.Equal(t, "vm_query_failed", run.Errors[0].Code)
+
+		snapshot, err := models.FindLatestOmTopologySnapshot(q)
+		require.NoError(t, err)
+		assert.Equal(t, "run-1", snapshot.RunID)
+		assert.JSONEq(t, `{"origin_node":"pmm-server"}`, string(snapshot.Document))
+		assert.Equal(t, int32(3), snapshot.SchemaVersion)
+	})
+
+	t.Run("runs come back newest first", func(t *testing.T) {
+		q := db.Querier
+		insert(t, q, "run-2", time.Minute, `{"n":2}`)
+		insert(t, q, "run-3", 2*time.Minute, `{"n":3}`)
+
+		runs, err := models.FindOmTopologyRuns(q, 2)
+		require.NoError(t, err)
+		require.Len(t, runs, 2)
+		assert.Equal(t, "run-3", runs[0].RunID)
+		assert.Equal(t, "run-2", runs[1].RunID)
+
+		// And the newest document is the one a cold start restores.
+		snapshot, err := models.FindLatestOmTopologySnapshot(q)
+		require.NoError(t, err)
+		assert.Equal(t, "run-3", snapshot.RunID)
+	})
+
+	t.Run("pruning bounds the history and takes the documents with it", func(t *testing.T) {
+		q := db.Querier
+		for i := range 5 {
+			insert(t, q, fmt.Sprintf("prune-%d", i), time.Duration(10+i)*time.Minute, `{}`)
+		}
+
+		require.NoError(t, models.PruneOmTopologyRuns(q, 3))
+
+		runs, err := models.FindOmTopologyRuns(q, 100)
+		require.NoError(t, err)
+		require.Len(t, runs, 3, "only the newest are kept")
+		assert.Equal(t, "prune-4", runs[0].RunID)
+
+		// The snapshot cascades, so retention needs no second sweep to stay bounded.
+		_, err = models.FindOmTopologyRunByID(q, "prune-0")
+		require.ErrorIs(t, err, models.ErrNotFound)
+
+		var count int
+		require.NoError(t, db.QueryRow("SELECT count(*) FROM om_topology_snapshots").Scan(&count))
+		assert.Equal(t, 3, count)
+	})
+
+	t.Run("absent things report absence, not an empty value", func(t *testing.T) {
+		q := db.Querier
+
+		_, err := models.FindOmTopologyRunByID(q, "nope")
+		require.ErrorIs(t, err, models.ErrNotFound)
+
+		_, err = models.FindOmTopologyRunByID(q, "")
+		require.Error(t, err, "an empty id is a caller mistake, not a miss")
+
+		runs, err := models.FindOmTopologyRuns(q, 0)
+		require.NoError(t, err)
+		assert.Empty(t, runs)
+
+		require.Error(t, models.PruneOmTopologyRuns(q, 0), "keeping nothing is never what a caller meant")
+	})
+}
+
+func TestOmTopologySnapshotAbsentOnEmptyDatabase(t *testing.T) {
+	sqlDB := testdb.Open(t, models.SkipFixtures, nil)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+
+	db := reform.NewDB(sqlDB, postgresql.Dialect, reform.NewPrintfLogger(t.Logf))
+
+	// The cold-start path depends on this being a clean miss rather than an error.
+	_, err := models.FindLatestOmTopologySnapshot(db.Querier)
+	require.ErrorIs(t, err, models.ErrNotFound)
+}