Parse the pgcontrol file on boot to ensure compatibility with our internal defaults.

davissp14 · davissp14 · commit 8b4fd9e1a003 · 2025-02-06T12:08:13.000-06:00
diff --git a/internal/flypg/node.go b/internal/flypg/node.go
@@ -218,7 +218,7 @@ func (n *Node) Init(ctx context.Context) error {
 		return fmt.Errorf("failed to initialize fly config: %s", err)
 	}
 
-	if err := n.PGConfig.initialize(store); err != nil {
+	if err := n.PGConfig.initialize(ctx, store); err != nil {
 		return fmt.Errorf("failed to initialize pg config: %s", err)
 	}
 
@@ -527,7 +527,7 @@ func (n *Node) handleRemoteRestore(ctx context.Context, store *state.Store) erro
 		}
 
 		// Set restore configuration
-		if err := n.PGConfig.initialize(store); err != nil {
+		if err := n.PGConfig.initialize(ctx, store); err != nil {
 			return fmt.Errorf("failed to initialize pg config: %s", err)
 		}
 
diff --git a/internal/flypg/pg.go b/internal/flypg/pg.go
@@ -112,7 +112,7 @@ func (c *PGConfig) Print(w io.Writer) error {
 	return e.Encode(cfg)
 }
 
-func (c *PGConfig) SetDefaults(store *state.Store) error {
+func (c *PGConfig) SetDefaults(ctx context.Context, store *state.Store) error {
 	// The default wal_segment_size in mb
 	const walSegmentSize = 16
 
@@ -184,6 +184,27 @@ func (c *PGConfig) SetDefaults(store *state.Store) error {
 		return fmt.Errorf("failed to set recovery target config: %s", err)
 	}
 
+	// Override any default settings that may conflict with pg_control.
+	pgControlMap, err := pgControlSettings(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to fetch pg_control settings: %s", err)
+	}
+
+	if pgControlMap != nil {
+		for k, v := range pgControlMap {
+			// Skip any settings that are not already specified by the internal config
+			if _, ok := c.internalConfig[k]; !ok {
+				continue
+			}
+
+			// Check for value discrepancies and log a warning if found.
+			if c.internalConfig[k] != v {
+				log.Printf("[WARN] Overriding internal config setting %s: %s -> %s", k, c.internalConfig[k], v)
+				c.internalConfig[k] = v
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -269,7 +290,7 @@ func (c *PGConfig) isInitialized() bool {
 
 // initialize will ensure the required configuration files are stubbed and the parent
 // postgresql.conf file includes them.
-func (c *PGConfig) initialize(store *state.Store) error {
+func (c *PGConfig) initialize(ctx context.Context, store *state.Store) error {
 	if err := c.setDefaultHBA(); err != nil {
 		return fmt.Errorf("failed updating pg_hba.conf: %s", err)
 	}
@@ -294,7 +315,7 @@ func (c *PGConfig) initialize(store *state.Store) error {
 		}
 	}
 
-	if err := c.SetDefaults(store); err != nil {
+	if err := c.SetDefaults(context.TODO(), store); err != nil {
 		return fmt.Errorf("failed to set pg defaults: %s", err)
 	}
 
diff --git a/internal/flypg/pg_control.go b/internal/flypg/pg_control.go
@@ -0,0 +1,62 @@
+package flypg
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"strings"
+
+	"log"
+
+	"github.com/fly-apps/postgres-flex/internal/utils"
+)
+
+const (
+	pathToPGControl = "/data/postgresql/global/pg_control"
+)
+
+func pgControlSettings(ctx context.Context) (map[string]string, error) {
+	// Short-circuit if the pg_control file doesn't exist.
+	if !utils.FileExists(pathToPGControl) {
+		log.Println("[WARN] pg_control file does not exist. Skipping evaluation.")
+		return nil, nil
+	}
+
+	result, err := utils.RunCmd(ctx, "root", "pg_controldata")
+	if err != nil {
+		return nil, fmt.Errorf("failed to run pg_controldata: %s", err)
+	}
+
+	return parsePGControlData(string(result))
+}
+
+func parsePGControlData(pgControlData string) (map[string]string, error) {
+	settings := make(map[string]string)
+
+	scanner := bufio.NewScanner(strings.NewReader(pgControlData))
+	for scanner.Scan() {
+		line := scanner.Text()
+
+		// Filter out lines that don't contain the word "setting".
+		if !strings.Contains(line, "setting:") {
+			continue
+		}
+
+		parts := strings.SplitN(line, "setting:", 2)
+		if len(parts) != 2 {
+			continue
+		}
+
+		key := strings.TrimSpace(parts[0])
+		value := strings.TrimSpace(parts[1])
+
+		settings[key] = value
+	}
+
+	// Check for any scanner errors.
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	return settings, nil
+}
diff --git a/internal/flypg/pg_control_test.go b/internal/flypg/pg_control_test.go
@@ -0,0 +1,60 @@
+package flypg
+
+import (
+	"testing"
+)
+
+func TestParseSettingsFromFile(t *testing.T) {
+	// Sample input that includes some lines with "setting:" and some without.
+	input := `pg_control version number:            1300
+Catalog version number:               202307071
+Database system identifier:           7420479024646529412
+Database cluster state:               in archive recovery
+pg_control last modified:             Tue 04 Feb 2025 10:04:52 PM UTC
+Latest checkpoint location:           2/40000060
+Latest checkpoint's REDO location:    2/40000028
+Latest checkpoint's REDO WAL file:    000000020000000200000040
+Latest checkpoint's TimeLineID:       2
+Latest checkpoint's PrevTimeLineID:   2
+Latest checkpoint's full_page_writes: on
+Latest checkpoint's NextXID:          0:34
+wal_level setting:                    replica
+wal_log_hints setting:                on
+max_connections setting:              500
+max_worker_processes setting:         8
+Some other line without the keyword
+Blocks per segment of large relation: 131072
+WAL block size:                       8192
+Bytes per WAL segment:                16777216
+Maximum length of identifiers:        64
+Maximum columns in an index:          32
+Maximum size of a TOAST chunk:        1996
+Size of a large-object chunk:         2048`
+
+	settings, err := parsePGControlData(input)
+	if err != nil {
+		t.Fatalf("parsePGControlData returned an error: %v", err)
+	}
+
+	// Define the expected key/value pairs.
+	expected := map[string]string{
+		"wal_level":            "replica",
+		"wal_log_hints":        "on",
+		"max_connections":      "500",
+		"max_worker_processes": "8",
+	}
+
+	if len(settings) != len(expected) {
+		t.Errorf("expected %d settings, got %d", len(expected), len(settings))
+	}
+
+	// Verify that the expected key/value pairs are present in the settings map.
+	for key, want := range expected {
+		got, ok := settings[key]
+		if !ok {
+			t.Errorf("expected key %q not found in settings", key)
+		} else if got != want {
+			t.Errorf("for key %q, expected value %q, got %q", key, want, got)
+		}
+	}
+}
diff --git a/internal/flypg/pg_test.go b/internal/flypg/pg_test.go
@@ -1,6 +1,7 @@
 package flypg
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"strings"
@@ -25,6 +26,8 @@ func TestPGConfigInitialization(t *testing.T) {
 	}
 	defer cleanup()
 
+	ctx := context.TODO()
+
 	pgConf := &PGConfig{
 		DataDir:                pgTestDirectory,
 		Port:                   5433,
@@ -41,7 +44,7 @@ func TestPGConfigInitialization(t *testing.T) {
 
 	t.Run("initialize", func(t *testing.T) {
 		store, _ := state.NewStore()
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 	})
@@ -98,7 +101,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("TIMESCALEDB_ENABLED", "true")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -129,7 +132,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		}
 
 		t.Run("defaults", func(t *testing.T) {
-			if err := pgConf.initialize(store); err != nil {
+			if err := pgConf.initialize(ctx, store); err != nil {
 				t.Fatal(err)
 			}
 
@@ -160,7 +163,7 @@ func TestPGConfigInitialization(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			if err := pgConf.initialize(store); err != nil {
+			if err := pgConf.initialize(ctx, store); err != nil {
 				t.Fatal(err)
 			}
 
@@ -181,7 +184,7 @@ func TestPGConfigInitialization(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			if err := pgConf.initialize(store); err != nil {
+			if err := pgConf.initialize(ctx, store); err != nil {
 				t.Fatal(err)
 			}
 
@@ -202,7 +205,7 @@ func TestPGConfigInitialization(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			if err := pgConf.initialize(store); err != nil {
+			if err := pgConf.initialize(ctx, store); err != nil {
 				t.Fatal(err)
 			}
 
@@ -223,7 +226,7 @@ func TestPGConfigInitialization(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			if err := pgConf.initialize(store); err != nil {
+			if err := pgConf.initialize(ctx, store); err != nil {
 				t.Fatal(err)
 			}
 
@@ -242,7 +245,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -257,7 +260,7 @@ func TestPGConfigInitialization(t *testing.T) {
 
 		t.Setenv("S3_ARCHIVE_CONFIG", "")
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -275,7 +278,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory?targetTime=2024-06-30T11:15:00-06:00")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -293,7 +296,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory?targetName=20240626T172443")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -311,7 +314,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory?target=immediate")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -329,7 +332,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory?targetTime=2024-06-30T11:15:00Z&targetInclusive=false")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -351,7 +354,7 @@ func TestPGConfigInitialization(t *testing.T) {
 		t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:my-secret@fly.storage.tigris.dev/my-bucket/my-directory?targetTime=2024-06-30T11:15:00-06:00&targetTimeline=2")
 		store, _ := state.NewStore()
 
-		if err := pgConf.initialize(store); err != nil {
+		if err := pgConf.initialize(ctx, store); err != nil {
 			t.Fatal(err)
 		}
 
@@ -391,7 +394,7 @@ func TestPGUserConfigOverride(t *testing.T) {
 	}
 
 	store, _ := state.NewStore()
-	if err := pgConf.initialize(store); err != nil {
+	if err := pgConf.initialize(context.TODO(), store); err != nil {
 		t.Error(err)
 	}
 
@@ -542,7 +545,7 @@ func TestValidateCompatibility(t *testing.T) {
 	}
 
 	store, _ := state.NewStore()
-	if err := pgConf.initialize(store); err != nil {
+	if err := pgConf.initialize(context.TODO(), store); err != nil {
 		t.Fatal(err)
 	}
 	t.Run("SharedPreloadLibraries", func(t *testing.T) {

Original file line number	Diff line number	Diff line change
`@@ -218,7 +218,7 @@ func (n *Node) Init(ctx context.Context) error {`
`218`	`218`	`return fmt.Errorf("failed to initialize fly config: %s", err)`
`219`	`219`	`}`
`220`	`220`
`221`		`- if err := n.PGConfig.initialize(store); err != nil {`
	`221`	`+ if err := n.PGConfig.initialize(ctx, store); err != nil {`
`222`	`222`	`return fmt.Errorf("failed to initialize pg config: %s", err)`
`223`	`223`	`}`
`224`	`224`
`@@ -527,7 +527,7 @@ func (n Node) handleRemoteRestore(ctx context.Context, store state.Store) erro`
`527`	`527`	`}`
`528`	`528`
`529`	`529`	`// Set restore configuration`
`530`		`- if err := n.PGConfig.initialize(store); err != nil {`
	`530`	`+ if err := n.PGConfig.initialize(ctx, store); err != nil {`
`531`	`531`	`return fmt.Errorf("failed to initialize pg config: %s", err)`
`532`	`532`	`}`
`533`	`533`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`package flypg`
`2`	`2`
`3`	`3`	`import (`
	`4`	`+ "context"`
`4`	`5`	`"fmt"`
`5`	`6`	`"os"`
`6`	`7`	`"strings"`
`@@ -25,6 +26,8 @@ func TestPGConfigInitialization(t *testing.T) {`
`25`	`26`	`}`
`26`	`27`	`defer cleanup()`
`27`	`28`
	`29`	`+ ctx := context.TODO()`
	`30`	`+`
`28`	`31`	`pgConf := &PGConfig{`
`29`	`32`	`DataDir: pgTestDirectory,`
`30`	`33`	`Port: 5433,`
`@@ -41,7 +44,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`41`	`44`
`42`	`45`	`t.Run("initialize", func(t *testing.T) {`
`43`	`46`	`store, _ := state.NewStore()`
`44`		`- if err := pgConf.initialize(store); err != nil {`
	`47`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`45`	`48`	`t.Fatal(err)`
`46`	`49`	`}`
`47`	`50`	`})`
`@@ -98,7 +101,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`98`	`101`	`t.Setenv("TIMESCALEDB_ENABLED", "true")`
`99`	`102`	`store, _ := state.NewStore()`
`100`	`103`
`101`		`- if err := pgConf.initialize(store); err != nil {`
	`104`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`102`	`105`	`t.Fatal(err)`
`103`	`106`	`}`
`104`	`107`
`@@ -129,7 +132,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`129`	`132`	`}`
`130`	`133`
`131`	`134`	`t.Run("defaults", func(t *testing.T) {`
`132`		`- if err := pgConf.initialize(store); err != nil {`
	`135`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`133`	`136`	`t.Fatal(err)`
`134`	`137`	`}`
`135`	`138`
`@@ -160,7 +163,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`160`	`163`	`t.Fatal(err)`
`161`	`164`	`}`
`162`	`165`
`163`		`- if err := pgConf.initialize(store); err != nil {`
	`166`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`164`	`167`	`t.Fatal(err)`
`165`	`168`	`}`
`166`	`169`
`@@ -181,7 +184,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`181`	`184`	`t.Fatal(err)`
`182`	`185`	`}`
`183`	`186`
`184`		`- if err := pgConf.initialize(store); err != nil {`
	`187`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`185`	`188`	`t.Fatal(err)`
`186`	`189`	`}`
`187`	`190`
`@@ -202,7 +205,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`202`	`205`	`t.Fatal(err)`
`203`	`206`	`}`
`204`	`207`
`205`		`- if err := pgConf.initialize(store); err != nil {`
	`208`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`206`	`209`	`t.Fatal(err)`
`207`	`210`	`}`
`208`	`211`
`@@ -223,7 +226,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`223`	`226`	`t.Fatal(err)`
`224`	`227`	`}`
`225`	`228`
`226`		`- if err := pgConf.initialize(store); err != nil {`
	`229`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`227`	`230`	`t.Fatal(err)`
`228`	`231`	`}`
`229`	`232`
`@@ -242,7 +245,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`242`	`245`	`t.Setenv("S3_ARCHIVE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory")`
`243`	`246`	`store, _ := state.NewStore()`
`244`	`247`
`245`		`- if err := pgConf.initialize(store); err != nil {`
	`248`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`246`	`249`	`t.Fatal(err)`
`247`	`250`	`}`
`248`	`251`
`@@ -257,7 +260,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`257`	`260`
`258`	`261`	`t.Setenv("S3_ARCHIVE_CONFIG", "")`
`259`	`262`
`260`		`- if err := pgConf.initialize(store); err != nil {`
	`263`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`261`	`264`	`t.Fatal(err)`
`262`	`265`	`}`
`263`	`266`
`@@ -275,7 +278,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`275`	`278`	`t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory?targetTime=2024-06-30T11:15:00-06:00")`
`276`	`279`	`store, _ := state.NewStore()`
`277`	`280`
`278`		`- if err := pgConf.initialize(store); err != nil {`
	`281`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`279`	`282`	`t.Fatal(err)`
`280`	`283`	`}`
`281`	`284`
`@@ -293,7 +296,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`293`	`296`	`t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory?targetName=20240626T172443")`
`294`	`297`	`store, _ := state.NewStore()`
`295`	`298`
`296`		`- if err := pgConf.initialize(store); err != nil {`
	`299`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`297`	`300`	`t.Fatal(err)`
`298`	`301`	`}`
`299`	`302`
`@@ -311,7 +314,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`311`	`314`	`t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory?target=immediate")`
`312`	`315`	`store, _ := state.NewStore()`
`313`	`316`
`314`		`- if err := pgConf.initialize(store); err != nil {`
	`317`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`315`	`318`	`t.Fatal(err)`
`316`	`319`	`}`
`317`	`320`
`@@ -329,7 +332,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`329`	`332`	`t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory?targetTime=2024-06-30T11:15:00Z&targetInclusive=false")`
`330`	`333`	`store, _ := state.NewStore()`
`331`	`334`
`332`		`- if err := pgConf.initialize(store); err != nil {`
	`335`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`333`	`336`	`t.Fatal(err)`
`334`	`337`	`}`
`335`	`338`
`@@ -351,7 +354,7 @@ func TestPGConfigInitialization(t *testing.T) {`
`351`	`354`	`t.Setenv("S3_ARCHIVE_REMOTE_RESTORE_CONFIG", "https://my-key:[email protected]/my-bucket/my-directory?targetTime=2024-06-30T11:15:00-06:00&targetTimeline=2")`
`352`	`355`	`store, _ := state.NewStore()`
`353`	`356`
`354`		`- if err := pgConf.initialize(store); err != nil {`
	`357`	`+ if err := pgConf.initialize(ctx, store); err != nil {`
`355`	`358`	`t.Fatal(err)`
`356`	`359`	`}`
`357`	`360`
`@@ -391,7 +394,7 @@ func TestPGUserConfigOverride(t *testing.T) {`
`391`	`394`	`}`
`392`	`395`
`393`	`396`	`store, _ := state.NewStore()`
`394`		`- if err := pgConf.initialize(store); err != nil {`
	`397`	`+ if err := pgConf.initialize(context.TODO(), store); err != nil {`
`395`	`398`	`t.Error(err)`
`396`	`399`	`}`
`397`	`400`
`@@ -542,7 +545,7 @@ func TestValidateCompatibility(t *testing.T) {`
`542`	`545`	`}`
`543`	`546`
`544`	`547`	`store, _ := state.NewStore()`
`545`		`- if err := pgConf.initialize(store); err != nil {`
	`548`	`+ if err := pgConf.initialize(context.TODO(), store); err != nil {`
`546`	`549`	`t.Fatal(err)`
`547`	`550`	`}`
`548`	`551`	`t.Run("SharedPreloadLibraries", func(t *testing.T) {`