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/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)
@@ -119,6 +117,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
@@ -175,77 +177,17 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.
 		return fmt.Errorf("failed to get TLS config: %w", err)
 	}
 
-	firstTimeSetup, err := r.isFirstTimeSetup(rc)
-	if err != nil {
-		return err
-	}
-
-	var spockSets []postgres.ReplicationSet
-	var spockTables []postgres.ReplicationSetTable
-	if r.Spec.RestoreConfig != nil && firstTimeSetup {
-		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)
-	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),
+		DSN: r.ConnectionInfo.AdminDSN("postgres"),
 		TLS: tlsCfg,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to connect to database %q: %w", r.Spec.DatabaseName, err)
 	}
 	defer conn.Close(ctx)
 
-	tx, err := conn.Begin(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to begin 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)
@@ -258,8 +200,6 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.
 		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,
 		})
@@ -271,10 +211,6 @@ func (r *InstanceResource) initializeInstance(ctx context.Context, rc *resource.
 		}
 	}
 
-	if err := tx.Commit(ctx); err != nil {
-		return fmt.Errorf("failed to commit transaction: %w", err)
-	}
-
 	err = r.updateInstanceState(ctx, rc, &InstanceUpdateOptions{State: InstanceStateAvailable})
 	if err != nil {
 		return r.recordError(ctx, rc, err)
@@ -334,109 +270,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, error) {
-	// This instance will already exist in the state if it's been successfully
-	// created before.
-	_, err := resource.FromContext[*InstanceResource](rc, r.Identifier())
-	if errors.Is(err, resource.ErrNotFound) {
-		return true, nil
-	} else if err != nil {
-		return false, fmt.Errorf("failed to check state for previous version of this instance: %w", err)
-	}
-
-	return false, nil
-}
-
-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

server/internal/database/node_resource.go

@@ -3,7 +3,10 @@ package database
 import (
 	"context"
 	"fmt"
+	"slices"
+	"strings"
 
+	"github.com/pgEdge/control-plane/server/internal/postgres"
 	"github.com/pgEdge/control-plane/server/internal/resource"
 )
 
@@ -83,3 +86,62 @@ func (n *NodeResource) Update(ctx context.Context, rc *resource.Context) error {
 func (n *NodeResource) Delete(ctx context.Context, rc *resource.Context) error {
 	return nil
 }
+
+func (n *NodeResource) DSN(
+	ctx context.Context,
+	rc *resource.Context,
+	fromInstance *InstanceResource,
+	dbName string,
+) (*postgres.DSN, error) {
+	// Sort the instances so that our final DSN is deterministic
+	instanceIDs := slices.Clone(n.InstanceIDs)
+	slices.SortFunc(instanceIDs, func(a, b string) int {
+		// Always sort the primary instance to the beginning of the list since
+		// these DSNs are used for subscriptions, so we'll want to try
+		// connecting to the known primary instance first.
+		switch {
+		case a == n.PrimaryInstanceID:
+			return -1
+		case b == n.PrimaryInstanceID:
+			return 1
+		default:
+			return strings.Compare(a, b)
+		}
+	})
+
+	hosts := make([]string, len(instanceIDs))
+	ports := make([]int, len(instanceIDs))
+	for i, instanceID := range instanceIDs {
+		instance, err := resource.FromContext[*InstanceResource](rc, InstanceResourceIdentifier(instanceID))
+		if err != nil {
+			return nil, fmt.Errorf("failed to get instance '%s': %w", instanceID, err)
+		}
+		hosts[i] = instance.ConnectionInfo.PeerHost
+		ports[i] = instance.ConnectionInfo.PeerPort
+	}
+
+	return &postgres.DSN{
+		Hosts:       hosts,
+		Ports:       ports,
+		DBName:      dbName,
+		User:        "pgedge",
+		SSLCert:     fromInstance.ConnectionInfo.PeerSSLCert,
+		SSLKey:      fromInstance.ConnectionInfo.PeerSSLKey,
+		SSLRootCert: fromInstance.ConnectionInfo.PeerSSLRootCert,
+		Extra: map[string]string{
+			"target_session_attrs": "primary",
+		},
+	}, nil
+}
+
+func (n *NodeResource) PrimaryInstance(ctx context.Context, rc *resource.Context) (*InstanceResource, error) {
+	if n.PrimaryInstanceID == "" {
+		return nil, fmt.Errorf("%w: primary instance id not set", resource.ErrNotFound)
+	}
+	instance, err := resource.FromContext[*InstanceResource](rc, InstanceResourceIdentifier(n.PrimaryInstanceID))
+	if err != nil {
+		return nil, fmt.Errorf("failed to get primary instance '%s': %w", n.PrimaryInstanceID, err)
+	}
+
+	return instance, nil
+}

server/internal/database/operations/add_nodes.go

@@ -3,7 +3,6 @@ package operations
 import (
 	"fmt"
 
-	"github.com/pgEdge/control-plane/server/internal/database"
 	"github.com/pgEdge/control-plane/server/internal/resource"
 )
 
@@ -17,14 +16,12 @@ func AddNode(node *NodeResources) ([]*resource.State, error) {
 		return nil, fmt.Errorf("got empty instances for node %s", node.NodeName)
 	}
 
-	instanceIDs := make([]string, 0, len(node.InstanceResources))
 	states := make([]*resource.State, 0, 2)
 
 	primary, err := instanceState(node.InstanceResources[0])
 	if err != nil {
 		return nil, err
 	}
-	instanceIDs = append(instanceIDs, node.InstanceResources[0].InstanceID())
 	states = append(states, primary)
 
 	var replicas *resource.State
@@ -38,20 +35,17 @@ func AddNode(node *NodeResources) ([]*resource.State, error) {
 		} else {
 			replicas.Merge(replica)
 		}
-		instanceIDs = append(instanceIDs, inst.InstanceID())
 	}
 
 	if replicas != nil {
 		states = append(states, replicas)
 	}
 
-	err = addNodeResource(states, &database.NodeResource{
-		Name:        node.NodeName,
-		InstanceIDs: instanceIDs,
-	})
+	nodeState, err := node.nodeResourceState()
 	if err != nil {
 		return nil, err
 	}
+	states[len(states)-1].Merge(nodeState)
 
 	return states, nil
 }