Merge branch '201-asynchronous-sync-instance' into 'master'

feat: start sync instance asynchronously (#201)

Closes #201

See merge request postgres-ai/database-lab!207

Author: agneum

configs/config.example.physical_generic.yml

@@ -238,6 +238,13 @@ retrieval:
             # Limit defines how many snapshots should be hold.
             limit: 4
 
+        # Set environment variables here. See https://www.postgresql.org/docs/current/libpq-envars.html
+        envs:
+          PGUSER: "postgres"
+          PGPASSWORD: "postgres"
+          PGHOST: "source.hostname"
+          PGPORT: 5432
+
 cloning:
   # Host that will be specified in database connection info for all clones
   # Use public IP address if database connections are allowed from outside

configs/config.example.physical_walg.yml

@@ -228,6 +228,11 @@ retrieval:
             # Limit defines how many snapshots should be hold.
             limit: 4
 
+        # Passes custom environment variables to the promotion Docker container.
+        envs:
+          WALG_GS_PREFIX: "gs://{BUCKET}/{SCOPE}"
+          GOOGLE_APPLICATION_CREDENTIALS: "/tmp/sa.json"
+
 cloning:
   # Host that will be specified in database connection info for all clones
   # Use public IP address if database connections are allowed from outside

pkg/retrieval/engine/postgres/physical/physical.go

@@ -153,10 +153,12 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 
 	defer func() {
 		if err == nil && r.CopyOptions.Sync.Enabled {
-			if syncErr := r.runSyncInstance(ctx); syncErr != nil {
-				log.Err("Failed to run sync instance: ", syncErr)
-				tools.StopContainer(ctx, r.dockerClient, r.syncInstanceName(), time.Second)
-			}
+			go func() {
+				if syncErr := r.runSyncInstance(ctx); syncErr != nil {
+					log.Err("Failed to run sync instance: ", syncErr)
+					tools.StopContainer(ctx, r.dockerClient, r.syncInstanceName(), time.Second)
+				}
+			}()
 		}
 	}()
 
@@ -438,7 +440,7 @@ func (r *RestoreJob) adjustRecoveryConfiguration(pgVersion, pgDataDir string) er
 		recoveryFilename = "recovery.conf"
 	}
 
-	return appendConfigFile(path.Join(pgDataDir, recoveryFilename), r.restorer.GetRecoveryConfig())
+	return fs.AppendFile(path.Join(pgDataDir, recoveryFilename), r.restorer.GetRecoveryConfig())
 }
 
 func (r *RestoreJob) applyInitParams(ctx context.Context, contID, pgVersion, dataDir string) error {
@@ -521,20 +523,5 @@ func appendInitConfigs(initConfiguration map[string]string, pgDataDir string) er
 		buffer.WriteString(fmt.Sprintf("%s = '%s'\n", key, value))
 	}
 
-	return appendConfigFile(path.Join(pgDataDir, "postgresql.conf"), buffer.Bytes())
-}
-
-func appendConfigFile(file string, data []byte) error {
-	configFile, err := os.OpenFile(file, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
-	if err != nil {
-		return err
-	}
-
-	defer func() { _ = configFile.Close() }()
-
-	if _, err := configFile.Write(data); err != nil {
-		return err
-	}
-
-	return nil
+	return fs.AppendFile(path.Join(pgDataDir, "postgresql.conf"), buffer.Bytes())
 }

pkg/retrieval/engine/postgres/snapshot/physical.go

@@ -9,7 +9,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path"
 	"strconv"
@@ -31,6 +30,7 @@ import (
 	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/engine/postgres/tools"
 	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/engine/postgres/tools/cont"
 	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/engine/postgres/tools/defaults"
+	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/engine/postgres/tools/fs"
 	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/engine/postgres/tools/health"
 	"gitlab.com/postgres-ai/database-lab/pkg/retrieval/options"
 	"gitlab.com/postgres-ai/database-lab/pkg/services/provision/databases/postgres/configuration"
@@ -80,6 +80,7 @@ type PhysicalOptions struct {
 	PreprocessingScript string            `yaml:"preprocessingScript"`
 	Configs             map[string]string `yaml:"configs"`
 	Sysctls             map[string]string `yaml:"sysctls"`
+	Envs                map[string]string `yaml:"envs"`
 	Scheduler           *Scheduler        `yaml:"scheduler"`
 }
 
@@ -529,11 +530,9 @@ func (p *PhysicalInitial) adjustRecoveryConfiguration(pgVersion, clonePGDataDir
 		replicationFilename = "recovery.conf"
 
 		buffer.WriteString("standby_mode = 'on'\n")
-		buffer.WriteString("primary_conninfo = ''\n")
-		buffer.WriteString("restore_command = ''\n")
 	}
 
-	if err := ioutil.WriteFile(path.Join(clonePGDataDir, replicationFilename), buffer.Bytes(), 0666); err != nil {
+	if err := fs.AppendFile(path.Join(clonePGDataDir, replicationFilename), buffer.Bytes()); err != nil {
 		return err
 	}
 
@@ -557,10 +556,7 @@ func (p *PhysicalInitial) buildContainerConfig(clonePath, promoteImage, password
 			cont.DBLabControlLabel:    cont.DBLabPromoteLabel,
 			cont.DBLabInstanceIDLabel: p.globalCfg.InstanceID,
 		},
-		Env: []string{
-			"PGDATA=" + clonePath,
-			"POSTGRES_PASSWORD=" + password,
-		},
+		Env:   p.getEnvironmentVariables(clonePath, password),
 		Image: promoteImage,
 		Healthcheck: health.GetConfig(
 			p.globalCfg.Database.User(),
@@ -571,6 +567,20 @@ func (p *PhysicalInitial) buildContainerConfig(clonePath, promoteImage, password
 	}
 }
 
+func (p *PhysicalInitial) getEnvironmentVariables(clonePath, password string) []string {
+	envVariables := []string{
+		"PGDATA=" + clonePath,
+		"POSTGRES_PASSWORD=" + password,
+	}
+
+	// Add user-defined environment variables.
+	for env, value := range p.options.Envs {
+		envVariables = append(envVariables, fmt.Sprintf("%s=%s", env, value))
+	}
+
+	return envVariables
+}
+
 func (p *PhysicalInitial) buildHostConfig(ctx context.Context, clonePath string) (*container.HostConfig, error) {
 	hostConfig := &container.HostConfig{
 		Sysctls: p.options.Sysctls,
@@ -621,6 +631,9 @@ func (p *PhysicalInitial) runPromoteCommand(ctx context.Context, containerID, cl
 	output, err := tools.ExecCommandWithOutput(ctx, p.dockerClient, containerID, types.ExecConfig{
 		User: defaults.Username,
 		Cmd:  promoteCommand,
+		Env: []string{
+			fmt.Sprintf("PGCTLTIMEOUT=%d", p.options.Promotion.HealthCheck.MaxRetries*int(p.options.Promotion.HealthCheck.Interval)),
+		},
 	})
 	if err != nil {
 		return errors.Wrap(err, "failed to promote instance")

pkg/retrieval/engine/postgres/tools/fs/tools.go

@@ -57,3 +57,19 @@ func copyFile(sourceFilename, destinationFilename string) error {
 
 	return nil
 }
+
+// AppendFile appends data to a file.
+func AppendFile(file string, data []byte) error {
+	configFile, err := os.OpenFile(file, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		return err
+	}
+
+	defer func() { _ = configFile.Close() }()
+
+	if _, err := configFile.Write(data); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/retrieval/engine/postgres/tools/tools.go

@@ -189,7 +189,7 @@ func CheckContainerReadiness(ctx context.Context, dockerClient *client.Client, c
 	for {
 		select {
 		case <-ctx.Done():
-			return nil
+			return ctx.Err()
 		default:
 		}