Merge pull request #63 from fly-apps/restore-support

Add support for Fly Volume Restores

Author: davissp14

Dockerfile

@@ -11,6 +11,8 @@ RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/event_handler ./cmd/event_h
 RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/failover_validation ./cmd/failover_validation
 RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/standby_cleaner ./cmd/standby_cleaner
 RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/pg_unregister ./cmd/pg_unregister
+RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/start_admin_server ./cmd/admin_server
+
 RUN CGO_ENABLED=0 GOOS=linux go build -v -o /fly/bin/start ./cmd/start
 
 COPY ./bin/* /fly/bin/

cmd/admin_server/main.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"github.com/fly-apps/postgres-flex/internal/api"
+	"github.com/fly-apps/postgres-flex/internal/flypg"
+)
+
+func main() {
+	node, err := flypg.NewNode()
+	if err != nil {
+		panic(err)
+	}
+
+	api.StartHttpServer(node)
+}

cmd/start/main.go

@@ -54,7 +54,12 @@ func main() {
 	svisor.AddProcess("repmgrd", fmt.Sprintf("gosu postgres repmgrd -f %s --daemonize=false", node.RepMgr.ConfigPath),
 		supervisor.WithRestart(0, 5*time.Second),
 	)
-	svisor.AddProcess("standby_cleaner", "/usr/local/bin/standby_cleaner", supervisor.WithRestart(0, 5*time.Second))
+	svisor.AddProcess("standby_cleaner", "/usr/local/bin/standby_cleaner",
+		supervisor.WithRestart(0, 5*time.Second),
+	)
+	svisor.AddProcess("admin", "/usr/local/bin/start_admin_server",
+		supervisor.WithRestart(0, 5*time.Second),
+	)
 
 	exporterEnv := map[string]string{
 		"DATA_SOURCE_URI":                     fmt.Sprintf("[%s]:%d/postgres?sslmode=disable", node.PrivateIP, node.Port),
@@ -63,11 +68,12 @@ func main() {
 		"PG_EXPORTER_EXCLUDE_DATABASE":        "template0,template1",
 		"PG_EXPORTER_AUTO_DISCOVER_DATABASES": "true",
 	}
-
-	svisor.AddProcess("exporter", "postgres_exporter", supervisor.WithEnv(exporterEnv), supervisor.WithRestart(0, 1*time.Second))
+	svisor.AddProcess("exporter", "postgres_exporter",
+		supervisor.WithEnv(exporterEnv),
+		supervisor.WithRestart(0, 1*time.Second),
+	)
 
 	svisor.StopOnSignal(syscall.SIGINT, syscall.SIGTERM)
-	svisor.StartHttpListener(node)
 
 	if err := svisor.Run(); err != nil {
 		fmt.Println(err)

internal/flypg/node.go

@@ -123,10 +123,26 @@ func NewNode() (*Node, error) {
 }
 
 func (n *Node) Init(ctx context.Context) error {
+	// Ensure directory and files have proper permissions
 	if err := setDirOwnership(); err != nil {
 		return err
 	}
 
+	// Initiate a restore
+	if os.Getenv("FLY_RESTORED_FROM") != "" {
+		// Check to see if there's an active restore.
+		active, err := isRestoreActive()
+		if err != nil {
+			return err
+		}
+
+		if active {
+			if err := Restore(ctx, n); err != nil {
+				return fmt.Errorf("failed to issue restore: %s", err)
+			}
+		}
+	}
+
 	if ZombieLockExists() {
 		fmt.Println("Zombie lock detected!")
 		primaryStr, err := readZombieLock()
@@ -188,7 +204,7 @@ func (n *Node) Init(ctx context.Context) error {
 		}
 	}
 
-	err := WriteSSHKey()
+	err := writeSSHKey()
 	if err != nil {
 		return fmt.Errorf("failed initialize ssh. %v", err)
 	}
@@ -211,12 +227,10 @@ func (n *Node) Init(ctx context.Context) error {
 
 		if !clusterInitialized {
 			// Initialize ourselves as the primary.
-			fmt.Println("Initializing postgres")
 			if err := n.initializePG(); err != nil {
 				return fmt.Errorf("failed to initialize postgres %s", err)
 			}
 
-			fmt.Println("Setting default HBA")
 			if err := n.setDefaultHBA(); err != nil {
 				return fmt.Errorf("failed updating pg_hba.conf: %s", err)
 			}
@@ -233,7 +247,6 @@ func (n *Node) Init(ctx context.Context) error {
 		}
 	}
 
-	fmt.Println("Initializing Postgres configuration")
 	if err := n.configurePostgres(store); err != nil {
 		return fmt.Errorf("failed to configure postgres: %s", err)
 	}
@@ -245,44 +258,6 @@ func (n *Node) Init(ctx context.Context) error {
 	return nil
 }
 
-func WriteSSHKey() error {
-	err := os.Mkdir("/data/.ssh", 0700)
-	if err != nil && !os.IsExist(err) {
-		return err
-	}
-
-	key := os.Getenv("SSH_KEY")
-
-	keyFile, err := os.Create("/data/.ssh/id_rsa")
-	if err != nil {
-		return err
-	}
-	defer keyFile.Close()
-	_, err = keyFile.Write([]byte(key))
-	if err != nil {
-		return err
-	}
-
-	cert := os.Getenv("SSH_CERT")
-
-	certFile, err := os.Create("/data/.ssh/id_rsa-cert.pub")
-	if err != nil {
-		return err
-	}
-	defer certFile.Close()
-	_, err = certFile.Write([]byte(cert))
-	if err != nil {
-		return err
-	}
-
-	err = setSSHOwnership()
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
 // PostInit are operations that should be executed against a running Postgres on boot.
 func (n *Node) PostInit(ctx context.Context) error {
 	if ZombieLockExists() {
@@ -529,17 +504,14 @@ func (n *Node) isPGInitialized() bool {
 }
 
 func (n *Node) configure(ctx context.Context, store *state.Store) error {
-	fmt.Println("Initializing internal config")
 	if err := n.configureInternal(store); err != nil {
 		return fmt.Errorf("failed to set internal config: %s", err)
 	}
 
-	fmt.Println("Initializing replication manager")
 	if err := n.configureRepmgr(store); err != nil {
 		return fmt.Errorf("failed to configure repmgr config: %s", err)
 	}
 
-	fmt.Println("Initializing pgbouncer")
 	if err := n.configurePGBouncer(store); err != nil {
 		return fmt.Errorf("failed to configure pgbouncer: %s", err)
 	}
@@ -553,6 +525,44 @@ func (n *Node) configure(ctx context.Context, store *state.Store) error {
 	return nil
 }
 
+func writeSSHKey() error {
+	err := os.Mkdir("/data/.ssh", 0700)
+	if err != nil && !os.IsExist(err) {
+		return err
+	}
+
+	key := os.Getenv("SSH_KEY")
+
+	keyFile, err := os.Create("/data/.ssh/id_rsa")
+	if err != nil {
+		return err
+	}
+	defer keyFile.Close()
+	_, err = keyFile.Write([]byte(key))
+	if err != nil {
+		return err
+	}
+
+	cert := os.Getenv("SSH_CERT")
+
+	certFile, err := os.Create("/data/.ssh/id_rsa-cert.pub")
+	if err != nil {
+		return err
+	}
+	defer certFile.Close()
+	_, err = certFile.Write([]byte(cert))
+	if err != nil {
+		return err
+	}
+
+	err = setSSHOwnership()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (n *Node) configureInternal(store *state.Store) error {
 	if err := n.InternalConfig.initialize(); err != nil {
 		return fmt.Errorf("failed to initialize internal config: %s", err)