diff --git a/go/core/pkg/app/app.go b/go/core/pkg/app/app.go index 6cadbdd6d..1cf46122c 100644 --- a/go/core/pkg/app/app.go +++ b/go/core/pkg/app/app.go @@ -137,9 +137,10 @@ type Config struct { // that originates TLS upstream. Off by default; MCPEgressPlaintext bool Database struct { - Url string - UrlFile string - VectorEnabled bool + Url string + UrlFile string + VectorEnabled bool + SkipMigrations bool } Substrate struct { AteAPIEndpoint string @@ -181,6 +182,7 @@ func (cfg *Config) SetFlags(commandLine *flag.FlagSet) { commandLine.StringVar(&cfg.Database.Url, "postgres-database-url", "postgres://postgres:kagent@kagent-postgresql.kagent.svc.cluster.local:5432/postgres", "The URL of the PostgreSQL database.") commandLine.StringVar(&cfg.Database.UrlFile, "postgres-database-url-file", "", "Path to a file containing the PostgreSQL database URL. Takes precedence over --postgres-database-url.") commandLine.BoolVar(&cfg.Database.VectorEnabled, "database-vector-enabled", true, "Enable pgvector extension and memory table. Requires pgvector to be installed on the PostgreSQL server.") + commandLine.BoolVar(&cfg.Database.SkipMigrations, "skip-migrations", false, "Do not run database migrations at startup; instead verify the database is already migrated and fail if it is not. Migrations must be applied out-of-band (e.g. from a pipeline or pre-upgrade hook). Settable via the SKIP_MIGRATIONS env var.") commandLine.StringVar(&cfg.WatchNamespaces, "watch-namespaces", "", "The namespaces to watch for .") @@ -470,13 +472,25 @@ func Start(getExtensionConfig GetExtensionConfig, extraSources []migrations.Sour // Run migrations before connecting; schema must exist before queries. // Built-in sources run first, then any downstream-registered extras. - setupLog.Info("running database migrations") + // With --skip-migrations (SKIP_MIGRATIONS) the server applies nothing and + // instead verifies the database is already migrated, so migrations can run + // out-of-band and this connection needs no DDL privileges. sources := append(migrations.BuiltinSources(cfg.Database.VectorEnabled), extraSources...) - if err := migrations.RunUp(ctx, dbURL, sources); err != nil { - setupLog.Error(err, "database migration failed") - os.Exit(1) + if cfg.Database.SkipMigrations { + setupLog.Info("skipping database migrations; verifying schema is migrated") + if err := migrations.VerifyMigrated(ctx, dbURL, sources); err != nil { + setupLog.Error(err, "database migration verification failed") + os.Exit(1) + } + setupLog.Info("database schema verified") + } else { + setupLog.Info("running database migrations") + if err := migrations.RunUp(ctx, dbURL, sources); err != nil { + setupLog.Error(err, "database migration failed") + os.Exit(1) + } + setupLog.Info("database migrations complete") } - setupLog.Info("database migrations complete") // Connect to database db, err := database.Connect(ctx, &database.PostgresConfig{ diff --git a/go/core/pkg/migrations/runner.go b/go/core/pkg/migrations/runner.go index a90dd946a..06762e75c 100644 --- a/go/core/pkg/migrations/runner.go +++ b/go/core/pkg/migrations/runner.go @@ -179,6 +179,77 @@ func RunUp(ctx context.Context, url string, sources []Source) error { return nil } +// VerifyMigrated checks, without applying or reverting anything, that every +// source's migrations have been applied to the database. It is the boot-time +// guard for the SKIP_MIGRATIONS deployment mode, where migrations run +// out-of-band (a pipeline or pre-upgrade hook) and the server must refuse to +// serve a wrong-shaped schema. It issues only SELECTs — never golang-migrate, +// which creates the tracking table on open — so it is safe on a connection +// whose role has no DDL privileges. +// +// Per source: a missing tracking table or a version behind this binary's +// embedded max is an error; a dirty tracking table is an error; a database +// ahead of the binary is tolerated (compatibility mode), matching RunUp. +func VerifyMigrated(ctx context.Context, url string, sources []Source) error { + if len(sources) == 0 { + return nil + } + if err := validateSources(sources); err != nil { + return err + } + + db, err := sql.Open("pgx", url) + if err != nil { + return fmt.Errorf("open database to verify migrations: %w", err) + } + defer db.Close() + + for _, src := range sources { + if err := ctx.Err(); err != nil { + return fmt.Errorf("migration verification cancelled before %s: %w", src.Name, err) + } + maxVer, err := maxEmbeddedVersion(src.FS, src.Dir) + if err != nil { + return fmt.Errorf("determine max embedded version for %s: %w", src.Name, err) + } + + // For Schema == "" the unqualified name resolves via the connection's + // search_path — the same place RunUp put the table. + table := quoteIdentifier(src.TrackingTable) + if src.Schema != "" { + table = quoteIdentifier(src.Schema) + "." + table + } + + var exists bool + if err := db.QueryRowContext(ctx, "SELECT to_regclass($1) IS NOT NULL", table).Scan(&exists); err != nil { + return fmt.Errorf("check tracking table for %s: %w", src.Name, err) + } + if !exists { + return fmt.Errorf("source %s: tracking table %s does not exist - the database has not been migrated; apply migrations out-of-band or unset SKIP_MIGRATIONS", src.Name, table) + } + + var version int64 + var dirty bool + err = db.QueryRowContext(ctx, "SELECT version, dirty FROM "+table+" LIMIT 1").Scan(&version, &dirty) + if errors.Is(err, sql.ErrNoRows) { + version, dirty = 0, false // table exists but nothing applied yet + } else if err != nil { + return fmt.Errorf("read tracking table for %s: %w", src.Name, err) + } + + switch { + case dirty: + return fmt.Errorf("source %s is dirty at version %d: a previous migration attempt failed and must be resolved before starting with SKIP_MIGRATIONS", src.Name, version) + case version < int64(maxVer): + return fmt.Errorf("source %s is at version %d but this binary requires version %d: apply migrations out-of-band or unset SKIP_MIGRATIONS", src.Name, version, maxVer) + case version > int64(maxVer): + log.Info("database schema is ahead of this binary; running in compatibility mode", + "track", src.Name, "dbVersion", version, "binaryMax", maxVer) + } + } + return nil +} + // validateSources rejects a source set that cannot be run safely. It checks two // things. // diff --git a/go/core/pkg/migrations/runner_test.go b/go/core/pkg/migrations/runner_test.go index 81b36e48e..9019aef82 100644 --- a/go/core/pkg/migrations/runner_test.go +++ b/go/core/pkg/migrations/runner_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "maps" + "net/url" "strings" "testing" "testing/fstest" @@ -972,3 +973,96 @@ func TestApplySource_DirtyStateRecoveryOnRestart(t *testing.T) { t.Errorf("after restart: version = %d, want 1 (dirty cleared, rolled back)", got) } } + +// TestVerifyMigrated covers the SKIP_MIGRATIONS boot guard: refuse an +// un-migrated or behind or dirty database, tolerate exact-match and ahead +// (compatibility mode), and work on a connection whose role has no DDL +// privileges — the deployment mode the guard exists for. +func TestVerifyMigrated(t *testing.T) { + ctx := context.Background() + connStr := startTestDB(t) + full := []Source{coreSource(goodCoreFS)} // max embedded version 2 + + t.Run("unmigrated database is refused", func(t *testing.T) { + err := VerifyMigrated(ctx, connStr, full) + if err == nil || !strings.Contains(err.Error(), "has not been migrated") { + t.Fatalf("error = %v, want missing-tracking-table refusal", err) + } + }) + + t.Run("pending migrations are refused", func(t *testing.T) { + if _, err := applySource(ctx, connStr, coreSource(oneCoreFS)); err != nil { + t.Fatalf("apply v1: %v", err) + } + err := VerifyMigrated(ctx, connStr, full) + if err == nil || !strings.Contains(err.Error(), "requires version 2") { + t.Fatalf("error = %v, want behind-binary refusal", err) + } + }) + + t.Run("fully migrated database passes", func(t *testing.T) { + if _, err := applySource(ctx, connStr, coreSource(goodCoreFS)); err != nil { + t.Fatalf("apply v2: %v", err) + } + if err := VerifyMigrated(ctx, connStr, full); err != nil { + t.Fatalf("VerifyMigrated() = %v, want nil", err) + } + }) + + t.Run("works without DDL privileges", func(t *testing.T) { + db, err := sql.Open("pgx", connStr) + if err != nil { + t.Fatal(err) + } + defer db.Close() + for _, q := range []string{ + `CREATE ROLE readonly LOGIN PASSWORD 'ro'`, + `GRANT USAGE ON SCHEMA public TO readonly`, + `GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly`, + } { + if _, err := db.ExecContext(ctx, q); err != nil { + t.Fatalf("%s: %v", q, err) + } + } + u, err := url.Parse(connStr) + if err != nil { + t.Fatal(err) + } + u.User = url.UserPassword("readonly", "ro") + if err := VerifyMigrated(ctx, u.String(), full); err != nil { + t.Fatalf("VerifyMigrated() as readonly = %v, want nil", err) + } + }) + + t.Run("database ahead of binary passes", func(t *testing.T) { + if err := VerifyMigrated(ctx, connStr, []Source{coreSource(oneCoreFS)}); err != nil { + t.Fatalf("VerifyMigrated() with older binary = %v, want nil (compatibility mode)", err) + } + }) + + t.Run("dirty tracking table is refused", func(t *testing.T) { + db, err := sql.Open("pgx", connStr) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.ExecContext(ctx, `UPDATE schema_migrations SET dirty = true`); err != nil { + t.Fatal(err) + } + defer func() { + if _, err := db.ExecContext(ctx, `UPDATE schema_migrations SET dirty = false`); err != nil { + t.Fatal(err) + } + }() + err = VerifyMigrated(ctx, connStr, full) + if err == nil || !strings.Contains(err.Error(), "dirty") { + t.Fatalf("error = %v, want dirty refusal", err) + } + }) + + t.Run("no sources is a no-op", func(t *testing.T) { + if err := VerifyMigrated(ctx, "postgres://unused", nil); err != nil { + t.Fatalf("VerifyMigrated() with no sources = %v, want nil", err) + } + }) +} diff --git a/helm/kagent/templates/NOTES.txt b/helm/kagent/templates/NOTES.txt index 1bffe87b1..96f4c75ab 100644 --- a/helm/kagent/templates/NOTES.txt +++ b/helm/kagent/templates/NOTES.txt @@ -92,6 +92,16 @@ DOCUMENTATION: database.postgres.bundled.enabled=false {{- end }} {{- end }} +{{- if .Values.database.postgres.skipMigrations }} +################################################################################ +# NOTE: STARTUP MIGRATIONS ARE DISABLED # +################################################################################ + database.postgres.skipMigrations is set: the controller will not run database + migrations at startup. It verifies the schema is already migrated and fails + to start if it is not. + + Ensure migrations are applied out-of-band before installing or upgrading. +{{- end }} {{- if .Values.substrate.enabled }} ################################################################################ # WARNING: SUBSTRATE IS EXPERIMENTAL, USE AT OWN RISK # diff --git a/helm/kagent/templates/controller-configmap.yaml b/helm/kagent/templates/controller-configmap.yaml index 0436ac15f..cadb2df43 100644 --- a/helm/kagent/templates/controller-configmap.yaml +++ b/helm/kagent/templates/controller-configmap.yaml @@ -55,6 +55,7 @@ data: PROXY_URL: {{ .Values.proxy.url | quote }} {{- end }} DATABASE_VECTOR_ENABLED: {{ .Values.database.postgres.vectorEnabled | quote }} + SKIP_MIGRATIONS: {{ .Values.database.postgres.skipMigrations | default false | quote }} WATCH_NAMESPACES: {{ include "kagent.watchNamespaces" . | quote }} MCP_EGRESS_PLAINTEXT: {{ .Values.controller.mcpEgressPlaintext | default false | quote }} {{- if .Values.controller.a2aClientTimeout }} diff --git a/helm/kagent/values.yaml b/helm/kagent/values.yaml index 5fbe1bd0e..6117696eb 100644 --- a/helm/kagent/values.yaml +++ b/helm/kagent/values.yaml @@ -76,6 +76,10 @@ database: # Required to use features that depend on database vector capability. (e.g. long-term memory) # Set to true when using an external PostgreSQL that has the pgvector extension installed. vectorEnabled: false + # -- Skip running database migrations at controller startup. + # The controller instead verifies the database is already migrated and fails if it is not. + # Migrations must be applied out-of-band (e.g. from a CI/CD pipeline) before install/upgrade. + skipMigrations: false # -- Bundled PostgreSQL instance — for development and evaluation only. # Not suitable for production. Deployed when enabled is true and url/urlFile are not set. bundled: