diff --git a/.env.example b/.env.example index 53eedf1..ca3651c 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,9 @@ OM_PUBLIC_URL=http://localhost:8080 # --- API --- OM_API_PORT=8080 +# om serve applies pending migrations before it listens. Set false to keep +# schema changes on an explicit `om migrate` (the Helm init container does that). +# OM_AUTO_MIGRATE=true # Optional. When set, the setup form can only create this first admin. # OM_ADMIN_EMAIL= diff --git a/README.md b/README.md index 355d5b1..43599ca 100644 --- a/README.md +++ b/README.md @@ -80,15 +80,13 @@ git clone https://github.com/OpenSourceOM/core.git cd core cp .env.example .env -# Start Postgres + API (http://localhost:8080) +# Start Postgres + API (http://localhost:8080). +# om serve applies pending migrations before it listens. docker compose up -d # Build the CLI go build -o om ./cmd/om -# Apply graph schema migrations -./om migrate - # Load the sample graph (no cloud credentials) ./om scan demo @@ -141,7 +139,7 @@ An API token is created only when an admin asks for one, from the console or wit ./om scan plugin -- ./my-collector ``` -The CLI does not read `.env`. It defaults `POSTGRES_HOST` to `localhost`, which matches the Postgres port Compose publishes, so the host commands above reach the database. Set `POSTGRES_HOST=localhost` only if your shell already exports `POSTGRES_HOST=postgres`. For CLI-only use without the API container, run `docker compose up -d postgres`. +The CLI does not read `.env`. It defaults `POSTGRES_HOST` to `localhost`, which matches the Postgres port Compose publishes, so the host commands above reach the database. Set `POSTGRES_HOST=localhost` only if your shell already exports `POSTGRES_HOST=postgres`. For CLI-only use without the API container, run `docker compose up -d postgres` and then `./om migrate`. `om serve` applies the same files on startup. Set `OM_AUTO_MIGRATE=false` when a separate `om migrate` should be the only writer. The Helm chart still runs `om migrate` in an init container before `om serve`. **Kubernetes:** diff --git a/deploy/helm/opensourceom/render_test.sh b/deploy/helm/opensourceom/render_test.sh index 5599dee..c3d8509 100755 --- a/deploy/helm/opensourceom/render_test.sh +++ b/deploy/helm/opensourceom/render_test.sh @@ -53,6 +53,10 @@ helm lint "$chart" "${base[@]}" render default assert_absent "$workdir/default.yaml" "OM_API_SECRET" "default install still set a shared API secret" +if ! grep -A1 'name: OM_AUTO_MIGRATE' "$workdir/default.yaml" | grep -q 'value: "true"'; then + echo "API container does not apply migrations on startup" >&2 + exit 1 +fi assert_absent "$workdir/default.yaml" "kind: CronJob" "default install created a CronJob" assert_absent "$workdir/default.yaml" "kind: ClusterRole$" "default install created scan RBAC" assert_absent "$workdir/default.yaml" "- demo" "default install scheduled the demo scan" diff --git a/deploy/helm/opensourceom/templates/api.yaml b/deploy/helm/opensourceom/templates/api.yaml index c27de8d..16b8863 100644 --- a/deploy/helm/opensourceom/templates/api.yaml +++ b/deploy/helm/opensourceom/templates/api.yaml @@ -48,6 +48,8 @@ spec: value: {{ .Values.api.env | quote }} - name: OM_API_PORT value: "8080" + - name: OM_AUTO_MIGRATE + value: {{ .Values.api.autoMigrate | quote }} {{- with .Values.api.publicURL }} - name: OM_PUBLIC_URL value: {{ . | quote }} diff --git a/deploy/helm/opensourceom/values.yaml b/deploy/helm/opensourceom/values.yaml index 2bfa2df..f481f96 100644 --- a/deploy/helm/opensourceom/values.yaml +++ b/deploy/helm/opensourceom/values.yaml @@ -14,6 +14,8 @@ service: api: env: production + # om serve applies pending migrations. The migrate init container still runs first. + autoMigrate: true # Browser origin for the console and optional OAuth callbacks. No trailing path. publicURL: "" # When set, only this address can become the first admin. diff --git a/docker-compose.yml b/docker-compose.yml index 96e99a7..95e6dd2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 # Local dev stack: Postgres + API # Usage: cp .env.example .env && docker compose up -d +# The API applies pending migrations when it starts. # The API image compiles aws-sdk-go-v2/service/ec2. Docker needs at least 6 GiB of RAM. services: @@ -30,6 +31,7 @@ services: environment: OM_ENV: ${OM_ENV:-development} OM_API_PORT: 8080 + OM_AUTO_MIGRATE: ${OM_AUTO_MIGRATE:-true} OM_PUBLIC_URL: ${OM_PUBLIC_URL:-http://localhost:8080} OM_ADMIN_EMAIL: ${OM_ADMIN_EMAIL:-} OM_GOOGLE_CLIENT_ID: ${OM_GOOGLE_CLIENT_ID:-} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7f56794..066b0e5 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -67,8 +67,8 @@ See [ADR 001](./adr/001-graph-schema-v0.md), [ADR 002](./adr/002-phase1-findings ## Deployment -- **Dev:** Docker Compose — Postgres + API (`docker compose up -d`); CLI can also target Postgres directly -- **Prod:** Helm chart in `deploy/helm/opensourceom/` (API, optional Postgres, optional scheduled collectors) +- **Dev:** Docker Compose — Postgres + API (`docker compose up -d`). `om serve` applies pending migrations before it listens. The CLI can also target Postgres directly; `om migrate` is the schema step when the API is not running. +- **Prod:** Helm chart in `deploy/helm/opensourceom/` (API, optional Postgres, optional scheduled collectors). An init container runs `om migrate`, then the main container runs `om serve`, which applies any still-pending files unless `OM_AUTO_MIGRATE=false`. ## Phase 3 diff --git a/docs/adr/001-graph-schema-v0.md b/docs/adr/001-graph-schema-v0.md index 467e386..4dbbb8d 100644 --- a/docs/adr/001-graph-schema-v0.md +++ b/docs/adr/001-graph-schema-v0.md @@ -18,7 +18,7 @@ OpenSourceOM needs a minimal graph model to connect cloud inventory (workloads, ### Storage - **Backend:** PostgreSQL with two tables — `nodes` and `edges`. -- **Migrations:** Versioned SQL files in `migrations/`, applied by `om migrate`. +- **Migrations:** Versioned SQL files in `migrations/`, applied by `om migrate` and by `om serve` unless `OM_AUTO_MIGRATE=false`. - **Properties:** Flexible `JSONB` column on nodes and edges for provider-specific attributes. ### Node types (v0) diff --git a/internal/api/blastradius_test.go b/internal/api/blastradius_test.go index 7b3be32..92b2136 100644 --- a/internal/api/blastradius_test.go +++ b/internal/api/blastradius_test.go @@ -173,7 +173,7 @@ func openBlastRadiusStore(t *testing.T) *graph.Store { } databaseURL.Path = "/" + dbName - if err := migrate.Run(ctx, databaseURL.String(), blastRadiusMigrationsDir(t)); err != nil { + if _, err := migrate.Run(ctx, databaseURL.String(), blastRadiusMigrationsDir(t)); err != nil { t.Fatalf("migrate: %v", err) } store, err := graph.NewStore(ctx, databaseURL.String()) diff --git a/internal/auth/store_test.go b/internal/auth/store_test.go index 6ea5837..d750958 100644 --- a/internal/auth/store_test.go +++ b/internal/auth/store_test.go @@ -266,7 +266,7 @@ func openAuthStore(t *testing.T) *Store { t.Fatal("locate test file") } migrations := filepath.Join(filepath.Dir(file), "..", "..", "migrations") - if err := migrate.Run(ctx, databaseURL.String(), migrations); err != nil { + if _, err := migrate.Run(ctx, databaseURL.String(), migrations); err != nil { t.Fatalf("migrate: %v", err) } store, err := NewStore(ctx, databaseURL.String()) diff --git a/internal/cmd/helpers.go b/internal/cmd/helpers.go index 4e4c3db..43cdc55 100644 --- a/internal/cmd/helpers.go +++ b/internal/cmd/helpers.go @@ -50,10 +50,10 @@ func loadConfig() config.Config { return config.Load() } -func runMigrations(ctx context.Context, cfg config.Config) error { +func runMigrations(ctx context.Context, cfg config.Config) ([]string, error) { migrationsDir, err := repoPath("migrations") if err != nil { - return err + return nil, err } return migrate.Run(ctx, cfg.DatabaseURL, migrationsDir) } diff --git a/internal/cmd/migrate.go b/internal/cmd/migrate.go index 8d9a639..a9d4395 100644 --- a/internal/cmd/migrate.go +++ b/internal/cmd/migrate.go @@ -14,9 +14,17 @@ var migrateCmd = &cobra.Command{ Short: "Apply database migrations", RunE: func(cmd *cobra.Command, args []string) error { cfg := loadConfig() - if err := runMigrations(cmd.Context(), cfg); err != nil { + applied, err := runMigrations(cmd.Context(), cfg) + if err != nil { return err } + if len(applied) == 0 { + fmt.Println("Database schema is current.") + return nil + } + for _, version := range applied { + fmt.Printf("applied %s\n", version) + } fmt.Println("Migrations applied.") return nil }, diff --git a/internal/cmd/serve.go b/internal/cmd/serve.go index 1f30ae4..109409e 100644 --- a/internal/cmd/serve.go +++ b/internal/cmd/serve.go @@ -20,6 +20,9 @@ var serveCmd = &cobra.Command{ Short: "Start the HTTP API server", RunE: func(cmd *cobra.Command, args []string) error { cfg := loadConfig() + if err := applyMigrations(cmd.Context(), cfg); err != nil { + return err + } store, err := openGraphStore(cmd.Context(), cfg) if err != nil { return err @@ -49,6 +52,28 @@ var serveCmd = &cobra.Command{ }, } +func applyMigrations(ctx context.Context, cfg config.Config) error { + auto, err := config.ServeAutoMigrate() + if err != nil { + return err + } + if !auto { + log.Printf("skipping database migrations because OM_AUTO_MIGRATE is false") + return nil + } + + applied, err := runMigrations(ctx, cfg) + if err != nil { + return fmt.Errorf("apply database migrations: %w", err) + } + if len(applied) == 0 { + log.Printf("database schema is current") + return nil + } + log.Printf("applied database migrations: %s", strings.Join(applied, ", ")) + return nil +} + func warnOAuthConfig(cfg config.Config) { if (cfg.GoogleClientID == "") != (cfg.GoogleClientSecret == "") { log.Printf("Google console login needs both OM_GOOGLE_CLIENT_ID and OM_GOOGLE_CLIENT_SECRET") diff --git a/internal/config/config.go b/internal/config/config.go index a8a2fae..3e0d883 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -7,6 +7,7 @@ import ( "fmt" "os" "strconv" + "strings" ) type Config struct { @@ -91,6 +92,20 @@ func Load() Config { } } +// ServeAutoMigrate reports whether om serve applies pending migrations before it listens. +// An unset or empty OM_AUTO_MIGRATE means yes. A present value must be a boolean. +func ServeAutoMigrate() (bool, error) { + raw, ok := os.LookupEnv("OM_AUTO_MIGRATE") + if !ok || strings.TrimSpace(raw) == "" { + return true, nil + } + value, err := strconv.ParseBool(strings.TrimSpace(raw)) + if err != nil { + return false, fmt.Errorf("OM_AUTO_MIGRATE=%q is not a boolean", raw) + } + return value, nil +} + func getEnv(key, fallback string) string { if v := os.Getenv(key); v != "" { return v diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..00e0114 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,58 @@ +// Copyright 2026 OpenSourceOM +// SPDX-License-Identifier: Apache-2.0 + +package config + +import ( + "os" + "testing" +) + +func TestServeAutoMigrateDefaultsWhenUnset(t *testing.T) { + t.Setenv("OM_AUTO_MIGRATE", "false") + if err := os.Unsetenv("OM_AUTO_MIGRATE"); err != nil { + t.Fatal(err) + } + + got, err := ServeAutoMigrate() + if err != nil { + t.Fatal(err) + } + if !got { + t.Fatal("unset OM_AUTO_MIGRATE should apply migrations") + } +} + +func TestServeAutoMigrate(t *testing.T) { + tests := []struct { + name string + value string + want bool + wantErr bool + }{ + {name: "empty", value: " ", want: true}, + {name: "true", value: "true", want: true}, + {name: "false", value: "false", want: false}, + {name: "zero", value: "0", want: false}, + {name: "junk", value: "sometimes", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("OM_AUTO_MIGRATE", tt.value) + got, err := ServeAutoMigrate() + if tt.wantErr { + if err == nil { + t.Fatal("expected an error") + } + return + } + if err != nil { + t.Fatalf("ServeAutoMigrate: %v", err) + } + if got != tt.want { + t.Fatalf("ServeAutoMigrate() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/graph/query_test.go b/internal/graph/query_test.go index aa086a5..e022e65 100644 --- a/internal/graph/query_test.go +++ b/internal/graph/query_test.go @@ -418,7 +418,7 @@ func openQueryFixture(t *testing.T) *graph.Store { } databaseURL.Path = "/" + dbName - if err := migrate.Run(ctx, databaseURL.String(), migrationsDir(t)); err != nil { + if _, err := migrate.Run(ctx, databaseURL.String(), migrationsDir(t)); err != nil { t.Fatalf("migrate: %v", err) } store, err := graph.NewStore(ctx, databaseURL.String()) diff --git a/internal/migrate/migrate.go b/internal/migrate/migrate.go index 6644821..d26640a 100644 --- a/internal/migrate/migrate.go +++ b/internal/migrate/migrate.go @@ -11,33 +11,47 @@ import ( "sort" "strings" - "github.com/jackc/pgx/v5/pgxpool" + "github.com/jackc/pgx/v5" ) -func Run(ctx context.Context, databaseURL, migrationsDir string) error { - pool, err := pgxpool.New(ctx, databaseURL) +// lockSQL is a session advisory lock so two processes cannot apply the same file. +// The Helm init container and om serve can both call Run against one database. +const ( + lockSQL = `SELECT pg_advisory_lock(hashtext('opensourceom.migrate'))` + unlockSQL = `SELECT pg_advisory_unlock(hashtext('opensourceom.migrate'))` +) + +// Run applies pending SQL migrations. Applied versions are skipped. +// Each file and its schema_migrations row commit together. +// The returned names are the versions applied by this call. +func Run(ctx context.Context, databaseURL, migrationsDir string) ([]string, error) { + conn, err := pgx.Connect(ctx, databaseURL) if err != nil { - return fmt.Errorf("connect postgres: %w", err) + return nil, fmt.Errorf("connect postgres: %w", err) + } + defer conn.Close(context.Background()) + + if err := conn.Ping(ctx); err != nil { + return nil, fmt.Errorf("ping postgres: %w", err) } - defer pool.Close() - if err := pool.Ping(ctx); err != nil { - return fmt.Errorf("ping postgres: %w", err) + if _, err := conn.Exec(ctx, lockSQL); err != nil { + return nil, fmt.Errorf("lock migrations: %w", err) } + defer conn.Exec(context.Background(), unlockSQL) - _, err = pool.Exec(ctx, ` + if _, err := conn.Exec(ctx, ` CREATE TABLE IF NOT EXISTS schema_migrations ( version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now() ) - `) - if err != nil { - return err + `); err != nil { + return nil, err } entries, err := os.ReadDir(migrationsDir) if err != nil { - return fmt.Errorf("read migrations dir: %w", err) + return nil, fmt.Errorf("read migrations dir: %w", err) } var files []string @@ -49,11 +63,12 @@ func Run(ctx context.Context, databaseURL, migrationsDir string) error { } sort.Strings(files) + var applied []string for _, file := range files { version := strings.TrimSuffix(file, ".sql") var exists bool - if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version).Scan(&exists); err != nil { - return err + if err := conn.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version).Scan(&exists); err != nil { + return nil, err } if exists { continue @@ -61,17 +76,33 @@ func Run(ctx context.Context, databaseURL, migrationsDir string) error { sqlBytes, err := os.ReadFile(filepath.Join(migrationsDir, file)) if err != nil { - return err + return nil, err } - - if _, err := pool.Exec(ctx, string(sqlBytes)); err != nil { - return fmt.Errorf("apply migration %s: %w", file, err) + if err := applyFile(ctx, conn, version, string(sqlBytes)); err != nil { + return nil, err } + applied = append(applied, version) + } - if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1) ON CONFLICT DO NOTHING`, version); err != nil { - return err - } + return applied, nil +} + +// applyFile runs the file and records its version as one simple-protocol batch. +// Postgres treats that batch as a single transaction, so a failed statement +// leaves neither the schema change nor the version row. +func applyFile(ctx context.Context, conn *pgx.Conn, version, sqlText string) error { + statement := strings.TrimSpace(sqlText) + if statement != "" && !strings.HasSuffix(statement, ";") { + statement += ";" } + statement += "\nINSERT INTO schema_migrations (version) VALUES (" + quoteLiteral(version) + ") ON CONFLICT DO NOTHING;" + if _, err := conn.Exec(ctx, statement); err != nil { + return fmt.Errorf("apply migration %s: %w", version, err) + } return nil } + +func quoteLiteral(value string) string { + return "'" + strings.ReplaceAll(value, "'", "''") + "'" +} diff --git a/internal/migrate/migrate_test.go b/internal/migrate/migrate_test.go new file mode 100644 index 0000000..5bd5212 --- /dev/null +++ b/internal/migrate/migrate_test.go @@ -0,0 +1,181 @@ +// Copyright 2026 OpenSourceOM +// SPDX-License-Identifier: Apache-2.0 + +package migrate + +import ( + "context" + "crypto/rand" + "encoding/hex" + "net/url" + "os" + "path/filepath" + "runtime" + "sync" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestRunAppliesRepoMigrationsOnce(t *testing.T) { + ctx := context.Background() + databaseURL := newDatabase(t) + + applied, err := Run(ctx, databaseURL, repoMigrations(t)) + if err != nil { + t.Fatalf("first run: %v", err) + } + if len(applied) == 0 { + t.Fatal("first run applied nothing") + } + + again, err := Run(ctx, databaseURL, repoMigrations(t)) + if err != nil { + t.Fatalf("second run: %v", err) + } + if len(again) != 0 { + t.Fatalf("second run applied %v", again) + } +} + +func TestRunAppliesOnlyPendingFiles(t *testing.T) { + ctx := context.Background() + databaseURL := newDatabase(t) + dir := t.TempDir() + writeMigration(t, dir, "001_first.sql", "CREATE TABLE first_only (id int);") + + applied, err := Run(ctx, databaseURL, dir) + if err != nil { + t.Fatal(err) + } + if len(applied) != 1 || applied[0] != "001_first" { + t.Fatalf("first apply = %v", applied) + } + + writeMigration(t, dir, "002_second.sql", "CREATE TABLE second_only (id int);") + applied, err = Run(ctx, databaseURL, dir) + if err != nil { + t.Fatal(err) + } + if len(applied) != 1 || applied[0] != "002_second" { + t.Fatalf("pending apply = %v", applied) + } +} + +func TestRunRollsBackFailedFile(t *testing.T) { + ctx := context.Background() + databaseURL := newDatabase(t) + dir := t.TempDir() + writeMigration(t, dir, "001_bad.sql", "CREATE TABLE should_rollback (id int);\nSELECT 1/0;") + + if _, err := Run(ctx, databaseURL, dir); err == nil { + t.Fatal("expected migration error") + } + + conn, err := pgx.Connect(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close(ctx) }) + + var tableExists, versionExists bool + if err := conn.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'should_rollback')`).Scan(&tableExists); err != nil { + t.Fatal(err) + } + if err := conn.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = '001_bad')`).Scan(&versionExists); err != nil { + t.Fatal(err) + } + if tableExists || versionExists { + t.Fatalf("table=%v version=%v, want both absent", tableExists, versionExists) + } +} + +func TestRunSerializesConcurrentApplicators(t *testing.T) { + ctx := context.Background() + databaseURL := newDatabase(t) + dir := t.TempDir() + writeMigration(t, dir, "001_once.sql", "CREATE TABLE once_only (id int PRIMARY KEY);") + + var wg sync.WaitGroup + errs := make(chan error, 2) + applied := make(chan []string, 2) + for range 2 { + wg.Add(1) + go func() { + defer wg.Done() + versions, err := Run(ctx, databaseURL, dir) + errs <- err + applied <- versions + }() + } + wg.Wait() + close(errs) + close(applied) + + var got []string + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + for versions := range applied { + got = append(got, versions...) + } + if len(got) != 1 || got[0] != "001_once" { + t.Fatalf("applied = %v, want [001_once]", got) + } +} + +func writeMigration(t *testing.T, dir, name, body string) { + t.Helper() + if err := os.WriteFile(filepath.Join(dir, name), []byte(body+"\n"), 0o644); err != nil { + t.Fatal(err) + } +} + +func repoMigrations(t *testing.T) string { + t.Helper() + _, file, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("runtime.Caller failed") + } + return filepath.Join(filepath.Dir(file), "..", "..", "migrations") +} + +func newDatabase(t *testing.T) string { + t.Helper() + adminURL := os.Getenv("TEST_DATABASE_URL") + if adminURL == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + + ctx := context.Background() + admin, err := pgxpool.New(ctx, adminURL) + if err != nil { + t.Fatalf("connect postgres: %v", err) + } + t.Cleanup(admin.Close) + + var raw [8]byte + if _, err := rand.Read(raw[:]); err != nil { + t.Fatal(err) + } + dbName := "omm" + hex.EncodeToString(raw[:]) + ident := pgx.Identifier{dbName}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE DATABASE "+ident); err != nil { + t.Fatalf("create database: %v", err) + } + t.Cleanup(func() { + if _, err := admin.Exec(context.Background(), "DROP DATABASE "+ident+" WITH (FORCE)"); err != nil { + t.Errorf("drop database %s: %v", dbName, err) + } + }) + + databaseURL, err := url.Parse(adminURL) + if err != nil { + t.Fatal(err) + } + databaseURL.Path = "/" + dbName + return databaseURL.String() +} diff --git a/internal/rules/engine_test.go b/internal/rules/engine_test.go index bdc7b17..43922af 100644 --- a/internal/rules/engine_test.go +++ b/internal/rules/engine_test.go @@ -189,7 +189,7 @@ func openRulesFixture(t *testing.T) *graph.Store { t.Fatal("locate test file") } migrations := filepath.Join(filepath.Dir(file), "..", "..", "migrations") - if err := migrate.Run(ctx, databaseURL.String(), migrations); err != nil { + if _, err := migrate.Run(ctx, databaseURL.String(), migrations); err != nil { t.Fatalf("migrate: %v", err) } store, err := graph.NewStore(ctx, databaseURL.String())