fix: update global patroni params for systemd

e2e/patroni_global_params_test.go

@@ -0,0 +1,101 @@
+//go:build e2e_test
+
+package e2e
+
+import (
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	api "github.com/pgEdge/control-plane/api/apiv1/gen/control_plane"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpdatePatroniGlobalParams(t *testing.T) {
+	t.Parallel()
+
+	// I'm intentionally not adding a helper function for this check because
+	// this condition will be very temporary.
+	if fixture.orchestrator != "systemd" {
+		t.Skip("patroni global parameter update is currently unsupported in non-systemd clusters")
+	}
+
+	host1 := fixture.HostIDs()[0]
+
+	username := "admin"
+	password := "password"
+
+	tLog(t, "creating database")
+
+	ctx := t.Context()
+	db := fixture.NewDatabaseFixture(ctx, t, &api.CreateDatabaseRequest{
+		Spec: &api.DatabaseSpec{
+			DatabaseName: "test_global_params",
+			DatabaseUsers: []*api.DatabaseUserSpec{
+				{
+					Username:   username,
+					Password:   pointerTo(password),
+					DbOwner:    pointerTo(true),
+					Attributes: []string{"LOGIN", "SUPERUSER"},
+				},
+			},
+			Port:        pointerTo(0),
+			PatroniPort: pointerTo(0),
+			Nodes: []*api.DatabaseNodeSpec{
+				{
+					Name:    "n1",
+					HostIds: []api.Identifier{api.Identifier(host1)},
+				},
+			},
+		},
+	})
+
+	tLog(t, "querying original max connections")
+
+	opts := ConnectionOptions{
+		Username: username,
+		Password: password,
+	}
+	var originalMaxConnections int
+	db.WithConnection(ctx, opts, t, func(conn *pgx.Conn) {
+		require.NoError(t, conn.QueryRow(ctx, `SELECT setting::integer FROM pg_settings WHERE name = 'max_connections';`).Scan(&originalMaxConnections))
+	})
+
+	tLogf(t, "got max_connections=%d", originalMaxConnections)
+	tLog(t, "updating database to change max connections")
+
+	err := db.Update(ctx, UpdateOptions{
+		Spec: &api.DatabaseSpec{
+			DatabaseName: "test_global_params",
+			DatabaseUsers: []*api.DatabaseUserSpec{
+				{
+					Username:   username,
+					DbOwner:    pointerTo(true),
+					Attributes: []string{"LOGIN", "SUPERUSER"},
+				},
+			},
+			Port:        pointerTo(0),
+			PatroniPort: pointerTo(0),
+			PostgresqlConf: map[string]any{
+				"max_connections": originalMaxConnections + 1,
+			},
+			Nodes: []*api.DatabaseNodeSpec{
+				{
+					Name:    "n1",
+					HostIds: []api.Identifier{api.Identifier(host1)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	tLog(t, "querying new max connections")
+
+	var newMaxConnections int
+	db.WithConnection(ctx, opts, t, func(conn *pgx.Conn) {
+		require.NoError(t, conn.QueryRow(ctx, `SELECT setting::integer FROM pg_settings WHERE name = 'max_connections';`).Scan(&newMaxConnections))
+	})
+
+	tLogf(t, "got max_connections=%d", newMaxConnections)
+
+	require.Equal(t, originalMaxConnections+1, newMaxConnections)
+}

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/enable_fast_basebackup.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/in-place_restore.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/minimal_swarm.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: true
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/minimal_systemd.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/user_pg_hba_pg_ident_and_scram.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/with_backup_config.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/golden_test/TestPatroniConfigGenerator/with_restore_config.yaml

@@ -13,6 +13,7 @@ bootstrap:
     loop_wait: 10
     ttl: 30
     retry_timeout: 10
+    failsafe_mode: false
     postgresql:
       parameters:
         max_connections: 901

server/internal/orchestrator/common/patroni_config.go

@@ -5,8 +5,10 @@ import (
 	"errors"
 	"fmt"
 	"path/filepath"
+	"time"
 
 	"github.com/goccy/go-yaml"
+	"github.com/rs/zerolog"
 	"github.com/samber/do"
 	"github.com/spf13/afero"
 	clientv3 "go.etcd.io/etcd/client/v3"
@@ -74,11 +76,58 @@ func (c *PatroniConfig) Create(
 	systemAddresses []string,
 	extraHbaEntries []hba.Entry,
 ) error {
-	fs, err := do.Invoke[afero.Fs](rc.Injector)
+	_, err := c.create(ctx, rc, systemAddresses, extraHbaEntries)
+	return err
+}
+
+func (c *PatroniConfig) Update(
+	ctx context.Context,
+	rc *resource.Context,
+	systemAddresses []string,
+	extraHbaEntries []hba.Entry,
+	reload func(ctx context.Context, rc *resource.Context, wait time.Duration) error,
+) error {
+	logger, err := do.Invoke[zerolog.Logger](rc.Injector)
 	if err != nil {
 		return err
 	}
-	etcdClient, err := do.Invoke[*clientv3.Client](rc.Injector)
+
+	cfg, err := c.create(ctx, rc, systemAddresses, extraHbaEntries)
+	if err != nil {
+		return err
+	}
+
+	wait := patroni.DefaultLoopWaitSeconds * time.Second
+	if client := c.client(rc); client != nil {
+		w, isPrimary := c.getStatusInfo(ctx, client)
+		if w > 0 {
+			wait = w
+		}
+		if isPrimary && cfg.Bootstrap != nil && cfg.Bootstrap.DCS != nil {
+			_, err := client.PatchDynamicConfig(ctx, cfg.Bootstrap.DCS.ToDynamicConfig())
+			if err != nil {
+				logger.Warn().
+					Str("database_id", c.Generator.DatabaseID).
+					Str("instance_id", c.InstanceID).
+					Err(err).
+					Msg("failed to patch dynamic config")
+			}
+		}
+	}
+
+	// We intentionally leave the reload implementation up to the caller rather
+	// than use the Patroni API's reload method. A process signal-based reload
+	// has the advantage that it will work regardless of whether the Patroni API
+	// is healthy.
+	if err := reload(ctx, rc, wait); err != nil {
+		return fmt.Errorf("failed to trigger patroni reload: %w", err)
+	}
+
+	return nil
+}
+
+func (c *PatroniConfig) Delete(ctx context.Context, rc *resource.Context) error {
+	fs, err := do.Invoke[afero.Fs](rc.Injector)
 	if err != nil {
 		return err
 	}
@@ -88,62 +137,71 @@ func (c *PatroniConfig) Create(
 		return fmt.Errorf("failed to get parent full path: %w", err)
 	}
 
+	err = fs.Remove(filepath.Join(parentFullPath, "patroni.yaml"))
+	if errors.Is(err, afero.ErrFileNotFound) {
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("failed to remove patroni.yaml: %w", err)
+	}
+
+	return nil
+}
+
+func (c *PatroniConfig) create(
+	ctx context.Context,
+	rc *resource.Context,
+	systemAddresses []string,
+	extraHbaEntries []hba.Entry,
+) (*patroni.Config, error) {
+	fs, err := do.Invoke[afero.Fs](rc.Injector)
+	if err != nil {
+		return nil, err
+	}
+	etcdClient, err := do.Invoke[*clientv3.Client](rc.Injector)
+	if err != nil {
+		return nil, err
+	}
+
+	parentFullPath, err := filesystem.DirResourceFullPath(rc, c.ParentID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get parent full path: %w", err)
+	}
+
 	etcdCreds, err := resource.FromContext[*EtcdCreds](rc, EtcdCredsIdentifier(c.InstanceID))
 	if err != nil {
-		return fmt.Errorf("failed to get etcd creds from state: %w", err)
+		return nil, fmt.Errorf("failed to get etcd creds from state: %w", err)
 	}
 
 	etcdHosts, err := patroni.EtcdHosts(ctx, etcdClient)
 	if err != nil {
-		return fmt.Errorf("failed to get etcd hosts: %w", err)
+		return nil, fmt.Errorf("failed to get etcd hosts: %w", err)
 	}
 
 	enableFastBasebackup, err := c.isNewNode(rc)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	config := c.Generator.Generate(etcdHosts, etcdCreds, GenerateOptions{
+	cfg := c.Generator.Generate(etcdHosts, etcdCreds, GenerateOptions{
 		EnableFastBasebackup: enableFastBasebackup,
 		SystemAddresses:      systemAddresses,
 		ExtraHbaEntries:      extraHbaEntries,
 	})
 
-	content, err := yaml.Marshal(config)
+	content, err := yaml.Marshal(cfg)
 	if err != nil {
-		return fmt.Errorf("failed to marshal patroni config: %w", err)
+		return nil, fmt.Errorf("failed to marshal patroni config: %w", err)
 	}
 
 	configPath := filepath.Join(parentFullPath, "patroni.yaml")
 	if err := afero.WriteFile(fs, configPath, content, 0o600); err != nil {
-		return fmt.Errorf("failed to write %s: %w", configPath, err)
+		return nil, fmt.Errorf("failed to write %s: %w", configPath, err)
 	}
 	if err := fs.Chown(configPath, c.OwnerUID, c.OwnerGID); err != nil {
-		return fmt.Errorf("failed to change ownership for %s: %w", configPath, err)
-	}
-
-	return nil
-}
-
-func (c *PatroniConfig) Delete(ctx context.Context, rc *resource.Context) error {
-	fs, err := do.Invoke[afero.Fs](rc.Injector)
-	if err != nil {
-		return err
-	}
-
-	parentFullPath, err := filesystem.DirResourceFullPath(rc, c.ParentID)
-	if err != nil {
-		return fmt.Errorf("failed to get parent full path: %w", err)
-	}
-
-	err = fs.Remove(filepath.Join(parentFullPath, "patroni.yaml"))
-	if errors.Is(err, afero.ErrFileNotFound) {
-		return nil
-	} else if err != nil {
-		return fmt.Errorf("failed to remove patroni.yaml: %w", err)
+		return nil, fmt.Errorf("failed to change ownership for %s: %w", configPath, err)
 	}
 
-	return nil
+	return cfg, nil
 }
 
 func (c *PatroniConfig) isNewNode(rc *resource.Context) (bool, error) {
@@ -157,3 +215,50 @@ func (c *PatroniConfig) isNewNode(rc *resource.Context) (bool, error) {
 		return false, nil
 	}
 }
+
+func (c *PatroniConfig) client(rc *resource.Context) *patroni.Client {
+	// We're not using FromContext here to handle the case where the instance
+	// creation failed, but Patroni is still running.
+	data, ok := rc.State.Get(database.InstanceResourceIdentifier(c.InstanceID))
+	if !ok {
+		return nil
+	}
+	instance, err := resource.ToResource[*database.InstanceResource](data)
+	if err == nil && instance.ConnectionInfo != nil {
+		return patroni.NewClient(instance.ConnectionInfo.PatroniURL(), nil)
+	}
+
+	return nil
+}
+
+func (c *PatroniConfig) getStatusInfo(ctx context.Context, client *patroni.Client) (time.Duration, bool) {
+	cfg, err := client.GetDynamicConfig(ctx)
+	if err != nil {
+		return 0, false
+	}
+
+	var loopWait time.Duration
+	if cfg.LoopWait == nil {
+		loopWait = patroni.DefaultLoopWaitSeconds * time.Second
+	} else {
+		loopWait = time.Duration(*cfg.LoopWait) * time.Second
+	}
+
+	wait := loopWait
+	status, err := client.GetInstanceStatus(ctx)
+	if err != nil {
+		return wait, false
+	}
+	if status.DCSLastSeen != nil {
+		lastSeen := time.Unix(*status.DCSLastSeen, 0)
+		lowerBound := time.Now().Add(-2 * loopWait)
+		upperBound := time.Now()
+		// Ignore last seen if clocks are very off in either direction
+		if lastSeen.After(lowerBound) && lastSeen.Before(upperBound) {
+			// Compute the time until the next run cycle
+			wait = time.Until(lastSeen.Add(loopWait))
+		}
+	}
+
+	return wait, status.IsPrimary()
+}