feat: postgres database resource

Adds a new `PostgresDatabaseResource` that shifts the Postgres database
creation logic to happen later in the database creation process. Now,
the `InstanceResource` is only responsible for creating the database
users, and the `PostgresDatabase` resource is responsible for creating
the Postgres database, granting role privileges to the new database, and
initializing the Spock node.

This change is necessary because we need to use `pg_service.conf` for
SystemD support, and we need that file to exist when we create the Spock
node. This change shifts the Spock node initialization until after all
of the instances are created and all nodes are ready, so we're able to
construct the DSNs for all nodes before we initialize Spock.

This change has an added benefit that it opens up the possibility of
support for multiple Postgres databases per Control Plane database since
we're able to make multiple databases/spock nodes per
`NodeResource`/Patroni cluster. I've added the `DatabaseName` field to
most, if not all, of the resources that would need to change to
accommodate this. However, I did stop short of incorporating the
database name into these resource's identities, which will be necessary
to complete that multi-database support.

PLAT-417

Author: jason-lynch

docker/control-plane-dev/config.json

@@ -1,4 +1,9 @@
 {
   "profiling_enabled": true,
-  "client_addresses": ["127.0.0.1"]
+  "client_addresses": ["127.0.0.1"],
+  "logging": {
+    "component_levels": {
+      "api_server": "error"
+    }
+  }
 }

server/internal/api/apiv1/validate.go

@@ -67,6 +67,7 @@ func validateDatabaseSpec(spec *api.DatabaseSpec) error {
 	errs = append(errs, validateCPUs(spec.Cpus, []string{"cpus"})...)
 	errs = append(errs, validateMemory(spec.Memory, []string{"memory"})...)
 	errs = append(errs, validatePorts(spec.Port, spec.PatroniPort, []string{"port"}))
+	errs = append(errs, validateUsers(spec.DatabaseUsers, []string{"database_users"})...)
 
 	// Track node-name uniqueness and prepare set for cross-node checks.
 	seenNodeNames := make(ds.Set[string], len(spec.Nodes))
@@ -344,6 +345,33 @@ func validatePorts(postgresPort, patroniPort *int, path []string) error {
 	return nil
 }
 
+func validateUsers(users []*api.DatabaseUserSpec, path []string) []error {
+	var errs []error
+
+	seenNames := ds.NewSet[string]()
+	var hasOwner bool
+	for i, user := range users {
+		userPath := appendPath(path, arrayIndexPath(i))
+
+		if seenNames.Has(user.Username) {
+			err := errors.New("usernames must be unique within a database")
+			errs = append(errs, newValidationError(err, userPath))
+		}
+		if hasOwner {
+			err := errors.New("cannot have multiple users with db_owner = true")
+			errs = append(errs, newValidationError(err, userPath))
+		}
+
+		seenNames.Add(user.Username)
+
+		if user.DbOwner != nil && *user.DbOwner {
+			hasOwner = true
+		}
+	}
+
+	return errs
+}
+
 func validateBackupConfig(cfg *api.BackupConfigSpec, path []string) []error {
 	var errs []error
 

server/internal/api/apiv1/validate_test.go

@@ -741,6 +741,39 @@ func TestValidateDatabaseSpec(t *testing.T) {
 				"nodes[1].port: postgres and patroni ports must not conflict",
 			},
 		},
+		{
+			name: "invalid users",
+			spec: &api.DatabaseSpec{
+				DatabaseUsers: []*api.DatabaseUserSpec{
+					{
+						Username: "duplicate_user",
+					},
+					{
+						Username: "duplicate_user",
+					},
+					{
+						Username: "duplicate_owner_1",
+						DbOwner:  utils.PointerTo(true),
+					},
+					{
+						Username: "duplicate_owner_2",
+						DbOwner:  utils.PointerTo(true),
+					},
+				},
+				Nodes: []*api.DatabaseNodeSpec{
+					{
+						Name: "n1",
+						HostIds: []api.Identifier{
+							api.Identifier("host-1"),
+						},
+					},
+				},
+			},
+			expected: []string{
+				"users[1]: usernames must be unique within a database",
+				"users[3]: cannot have multiple users with db_owner = true",
+			},
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			err := validateDatabaseSpec(tc.spec)

server/internal/database/instance_resource.go

@@ -2,7 +2,6 @@ package database
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"slices"
@@ -16,7 +15,6 @@ import (
 	"github.com/pgEdge/control-plane/server/internal/patroni"
 	"github.com/pgEdge/control-plane/server/internal/postgres"
 	"github.com/pgEdge/control-plane/server/internal/resource"
-	"github.com/pgEdge/control-plane/server/internal/utils"
 )
 
 var _ resource.Resource = (*InstanceResource)(nil)
@@ -122,6 +120,10 @@ func (r *InstanceResource) Delete(ctx context.Context, rc *resource.Context) err
 }
 
 func (r *InstanceResource) Connection(ctx context.Context, rc *resource.Context, dbName string) (*pgx.Conn, error) {
+	if rc.HostID != r.Spec.HostID {
+		return nil, fmt.Errorf("cannot connect to an instance running on a different host. executing host = '%s', instance host = '%s'", rc.HostID, r.Spec.HostID)
+	}
+
 	certs, err := do.Invoke[*certificates.Service](rc.Injector)
 	if err != nil {
 		return nil, err
@@ -143,17 +145,17 @@ func (r *InstanceResource) Connection(ctx context.Context, rc *resource.Context,
 }
 
 func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.Context) error {
-	certs, err := do.Invoke[*certificates.Service](rc.Injector)
-	if err != nil {
-		return err
-	}
+	// certs, err := do.Invoke[*certificates.Service](rc.Injector)
+	// if err != nil {
+	// 	return err
+	// }
 
 	if err := r.updateConnectionInfo(ctx, rc); err != nil {
 		return err
 	}
 
 	patroniClient := r.patroniClient()
-	err = WaitForPatroniRunning(ctx, patroniClient, 0)
+	err := WaitForPatroniRunning(ctx, patroniClient, 0)
 	if err != nil {
 		return fmt.Errorf("failed to wait for patroni to enter running state: %w", err)
 	}
@@ -173,98 +175,41 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.
 		return nil
 	}
 
-	tlsCfg, err := certs.PostgresUserTLS(ctx, r.Spec.InstanceID, r.InstanceHostname, "pgedge")
-	if err != nil {
-		return fmt.Errorf("failed to get TLS config: %w", err)
-	}
-
-	var spockSets []postgres.ReplicationSet
-	var spockTables []postgres.ReplicationSetTable
-	if r.Spec.RestoreConfig != nil && r.isFirstTimeSetup(rc) {
-		err = r.renameDB(ctx, tlsCfg)
-		if err != nil {
-			return fmt.Errorf("failed to rename database %q: %w", r.Spec.DatabaseName, err)
-		}
-
-		spockSets, spockTables, err = r.backupReplicationSets(ctx, tlsCfg)
-		if err != nil {
-			return err
-		}
-
-		err = r.dropSpock(ctx, tlsCfg)
-		if err != nil {
-			return fmt.Errorf("failed to drop spock: %w", err)
-		}
-	}
-
-	err = r.createDB(ctx, tlsCfg)
+	conn, err := r.Connection(ctx, rc, "postgres")
 	if err != nil {
-		return fmt.Errorf("failed to create database %q: %w", r.Spec.DatabaseName, err)
-	}
-
-	conn, err := ConnectToInstance(ctx, &ConnectionOptions{
-		DSN: r.ConnectionInfo.AdminDSN(r.Spec.DatabaseName),
-		TLS: tlsCfg,
-	})
-	if err != nil {
-		return fmt.Errorf("failed to connect to database %q: %w", r.Spec.DatabaseName, err)
+		return err
 	}
 	defer conn.Close(ctx)
 
-	tx, err := conn.Begin(ctx)
+	// Spock shouldn't exist in the 'postgres' database, but we want to err on
+	// the side of caution.
+	tx, err := postgres.StartRepairModeTxn(ctx, conn)
 	if err != nil {
-		return fmt.Errorf("failed to begin transaction: %w", err)
+		return fmt.Errorf("failed to start repair mode transaction: %w", err)
 	}
 	defer tx.Rollback(ctx)
 
-	enabled, err := postgres.IsSpockEnabled().Scalar(ctx, tx)
-	if err != nil {
-		return fmt.Errorf("failed to check if spock is enabled: %w", err)
-	}
-
-	if enabled {
-		err = postgres.EnableRepairMode().Exec(ctx, tx)
-		if err != nil {
-			return fmt.Errorf("failed to enable repair mode: %w", err)
-		}
-	}
-
-	err = postgres.InitializePgEdgeExtensions(
-		r.Spec.NodeName,
-		r.ConnectionInfo.PeerDSN(r.Spec.DatabaseName),
-	).Exec(ctx, conn)
-	if err != nil {
-		return fmt.Errorf("failed to initialize pgedge extensions: %w", err)
-	}
-	if len(spockSets) > 0 || len(spockTables) > 0 {
-		if err := postgres.RestoreReplicationSets(spockSets, spockTables).Exec(ctx, conn); err != nil {
-			return fmt.Errorf("failed to restore spock metadata: %w", err)
-		}
-	}
 	roleStatements, err := postgres.CreateBuiltInRoles(postgres.BuiltinRoleOptions{
 		PGVersion: r.Spec.PgEdgeVersion.PostgresVersion.String(),
-		DBName:    r.Spec.DatabaseName,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to generate built-in role statements: %w", err)
 	}
-	if err := roleStatements.Exec(ctx, conn); err != nil {
+	if err := roleStatements.Exec(ctx, tx); err != nil {
 		return fmt.Errorf("failed to create built-in roles: %w", err)
 	}
 
 	for _, user := range r.Spec.DatabaseUsers {
 		statement, err := postgres.CreateUserRole(postgres.UserRoleOptions{
 			Name:       user.Username,
 			Password:   user.Password,
-			DBName:     r.Spec.DatabaseName,
-			DBOwner:    user.DBOwner,
 			Attributes: user.Attributes,
 			Roles:      user.Roles,
 		})
 		if err != nil {
 			return fmt.Errorf("failed to produce create user role statement %q: %w", user.Username, err)
 		}
-		if err := statement.Exec(ctx, conn); err != nil {
+		if err := statement.Exec(ctx, tx); err != nil {
 			return fmt.Errorf("failed to create user role %q: %w", user.Username, err)
 		}
 	}
@@ -332,103 +277,3 @@ func (r *InstanceResource) updateConnectionInfo(ctx context.Context, rc *resourc
 func (r *InstanceResource) patroniClient() *patroni.Client {
 	return patroni.NewClient(r.ConnectionInfo.PatroniURL(), nil)
 }
-
-func (r *InstanceResource) createDB(ctx context.Context, tlsCfg *tls.Config) error {
-	createDBConn, err := ConnectToInstance(ctx, &ConnectionOptions{
-		DSN: r.ConnectionInfo.AdminDSN("postgres"),
-		TLS: tlsCfg,
-	})
-	if err != nil {
-		return fmt.Errorf("failed to connect to 'postgres' database on instance: %w", err)
-	}
-	defer createDBConn.Close(ctx)
-
-	err = postgres.CreateDatabase(r.Spec.DatabaseName).Exec(ctx, createDBConn)
-	if err != nil {
-		return fmt.Errorf("failed to create database %q: %w", r.Spec.DatabaseName, err)
-	}
-
-	return nil
-}
-
-func (r *InstanceResource) renameDB(ctx context.Context, tlsCfg *tls.Config) error {
-	// Short circuit if the restore config doesn't include a dbname or if the
-	// database name is the same.
-	if r.Spec.RestoreConfig.SourceDatabaseName == "" || r.Spec.RestoreConfig.SourceDatabaseName == r.Spec.DatabaseName {
-		return nil
-	}
-
-	// This operation can be flaky because of other processes connected to the
-	// database. We retry it a few times to avoid failing the entire create
-	// operation.
-	err := utils.Retry(3, 500*time.Millisecond, func() error {
-		createDBConn, err := ConnectToInstance(ctx, &ConnectionOptions{
-			DSN: r.ConnectionInfo.AdminDSN("postgres"),
-			TLS: tlsCfg,
-		})
-		if err != nil {
-			return fmt.Errorf("failed to connect to 'postgres' database on instance: %w", err)
-		}
-		defer createDBConn.Close(ctx)
-
-		return postgres.
-			RenameDB(r.Spec.RestoreConfig.SourceDatabaseName, r.Spec.DatabaseName).
-			Exec(ctx, createDBConn)
-	})
-	if err != nil {
-		return fmt.Errorf("failed to rename database %q: %w", r.Spec.DatabaseName, err)
-	}
-
-	return nil
-}
-
-func (r *InstanceResource) dropSpock(ctx context.Context, tlsCfg *tls.Config) error {
-	conn, err := ConnectToInstance(ctx, &ConnectionOptions{
-		DSN: r.ConnectionInfo.AdminDSN(r.Spec.DatabaseName),
-		TLS: tlsCfg,
-	})
-	if err != nil {
-		return fmt.Errorf("failed to connect to database %q: %w", r.Spec.DatabaseName, err)
-	}
-	defer conn.Close(ctx)
-
-	err = postgres.DropSpockAndCleanupSlots(r.Spec.DatabaseName).Exec(ctx, conn)
-	if err != nil {
-		return fmt.Errorf("failed to drop spock: %w", err)
-	}
-
-	return nil
-}
-
-func (r *InstanceResource) isFirstTimeSetup(rc *resource.Context) bool {
-	// This instance will already exist in the state if it's been successfully
-	// created before.
-	_, ok := rc.State.Get(r.Identifier())
-	return !ok
-}
-
-func (r *InstanceResource) backupReplicationSets(
-	ctx context.Context,
-	tlsCfg *tls.Config,
-) ([]postgres.ReplicationSet, []postgres.ReplicationSetTable, error) {
-	conn, err := ConnectToInstance(ctx, &ConnectionOptions{
-		DSN: r.ConnectionInfo.AdminDSN(r.Spec.DatabaseName),
-		TLS: tlsCfg,
-	})
-	if err != nil {
-		return nil, nil, fmt.Errorf("failed to connect to database %q: %w", r.Spec.DatabaseName, err)
-	}
-	defer conn.Close(ctx)
-
-	sets, err := postgres.GetReplicationSets().Structs(ctx, conn)
-	if err != nil {
-		return nil, nil, fmt.Errorf("spock backup failed to get replication sets: %w", err)
-	}
-
-	tabs, err := postgres.GetReplicationSetTables().Structs(ctx, conn)
-	if err != nil {
-		return nil, nil, fmt.Errorf("spock backup failed to get replication set tables: %w", err)
-	}
-
-	return sets, tabs, nil
-}

server/internal/database/lag_tracker_commit_ts_resource.go

@@ -26,6 +26,7 @@ type LagTrackerCommitTimestampResource struct {
 	// Planner fields
 	OriginNode   string `json:"origin_node"`
 	ReceiverNode string `json:"receiver_node"`
+	DatabaseName string `json:"database_name"`
 
 	// Dependency wiring
 	ExtraDependencies []resource.Identifier `json:"dependent_resources,omitempty"`
@@ -56,7 +57,8 @@ func (r *LagTrackerCommitTimestampResource) Identifier() resource.Identifier {
 
 func (r *LagTrackerCommitTimestampResource) Dependencies() []resource.Identifier {
 	deps := []resource.Identifier{
-		NodeResourceIdentifier(r.ReceiverNode),
+		PostgresDatabaseResourceIdentifier(r.ReceiverNode, r.DatabaseName),
+		PostgresDatabaseResourceIdentifier(r.OriginNode, r.DatabaseName),
 	}
 	deps = append(deps, r.ExtraDependencies...)
 	return deps