Merge branch '236-promote-strategy' into 'master'

fix: check the sync instance state before running the promotion stage (#236)

Closes #236

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

Author: agneum

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

@@ -298,6 +298,14 @@ func (p *PhysicalInitial) run(ctx context.Context) (err error) {
 		}
 	}()
 
+	var syncErr error
+
+	if p.options.Promotion.Enabled {
+		if syncErr = p.checkSyncInstance(ctx); syncErr != nil {
+			log.Dbg(fmt.Sprintf("failed to check the sync instance before snapshotting: %v", syncErr), "Changing the promotion strategy")
+		}
+	}
+
 	// Prepare pre-snapshot.
 	snapshotName, err := p.cloneManager.CreateSnapshot("", preDataStateAt+pre)
 	if err != nil {
@@ -326,7 +334,7 @@ func (p *PhysicalInitial) run(ctx context.Context) (err error) {
 
 	// Promotion.
 	if p.options.Promotion.Enabled {
-		if err := p.promoteInstance(ctx, path.Join(p.fsPool.ClonesDir(), cloneName, p.fsPool.DataSubDir)); err != nil {
+		if err := p.promoteInstance(ctx, path.Join(p.fsPool.ClonesDir(), cloneName, p.fsPool.DataSubDir), syncErr); err != nil {
 			return errors.Wrap(err, "failed to promote instance")
 		}
 	}
@@ -353,6 +361,29 @@ func (p *PhysicalInitial) run(ctx context.Context) (err error) {
 	return nil
 }
 
+func (p *PhysicalInitial) checkSyncInstance(ctx context.Context) error {
+	syncContainer, err := p.dockerClient.ContainerInspect(ctx, p.syncInstanceName())
+	if err != nil {
+		return err
+	}
+
+	if err := tools.CheckContainerReadiness(ctx, p.dockerClient, syncContainer.ID); err != nil {
+		return errors.Wrap(err, "failed to readiness check")
+	}
+
+	log.Msg("Sync instance has been checked. It is running")
+
+	if err := p.checkpoint(ctx, syncContainer.ID); err != nil {
+		return errors.Wrap(err, "failed to make a checkpoint for sync instance")
+	}
+
+	return nil
+}
+
+func (p *PhysicalInitial) syncInstanceName() string {
+	return cont.SyncInstanceContainerPrefix + p.globalCfg.InstanceID
+}
+
 func (p *PhysicalInitial) startScheduler(ctx context.Context) {
 	if p.scheduler == nil || !p.hasSchedulingOptions() {
 		return
@@ -409,7 +440,7 @@ func (p *PhysicalInitial) promoteContainerName() string {
 	return promoteContainerPrefix + p.globalCfg.InstanceID
 }
 
-func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string) (err error) {
+func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string, syncErr error) (err error) {
 	p.promotionMutex.Lock()
 	defer p.promotionMutex.Unlock()
 
@@ -436,10 +467,17 @@ func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string)
 		}
 	}
 
-	recoveryConfig := buildRecoveryConfig(recoveryFileConfig, p.options.Promotion.Recovery)
+	recoveryConfig := make(map[string]string)
+
+	// Item 5. Remove a recovery file: https://gitlab.com/postgres-ai/database-lab/-/issues/236#note_513401256
+	if syncErr != nil {
+		recoveryConfig = buildRecoveryConfig(recoveryFileConfig, p.options.Promotion.Recovery)
 
-	if err := cfgManager.ApplyRecovery(recoveryFileConfig); err != nil {
-		return errors.Wrap(err, "failed to apply recovery configuration")
+		if err := cfgManager.ApplyRecovery(recoveryFileConfig); err != nil {
+			return errors.Wrap(err, "failed to apply recovery configuration")
+		}
+	} else if err := cfgManager.RemoveRecoveryConfig(); err != nil {
+		log.Err(errors.Wrap(err, "failed to remove recovery config file"))
 	}
 
 	// Apply promotion configs.
@@ -557,6 +595,13 @@ func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string)
 		return errors.Wrap(err, "failed to store prepared configuration")
 	}
 
+	const pgStopTimeout = 600
+
+	if err := tools.StopPostgres(ctx, p.dockerClient, promoteCont.ID, clonePath, pgStopTimeout); err != nil {
+		log.Msg("Failed to stop Postgres", err)
+		tools.PrintContainerLogs(ctx, p.dockerClient, promoteCont.ID)
+	}
+
 	return nil
 }
 

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

@@ -70,7 +70,7 @@ func StopControlContainers(ctx context.Context, dockerClient *client.Client, ins
 		if shouldStopInternalProcess(controlLabel) {
 			log.Msg("Stopping control container: ", containerName)
 
-			if err := tools.StopPostgres(ctx, dockerClient, controlCont.ID, dataDir); err != nil {
+			if err := tools.StopPostgres(ctx, dockerClient, controlCont.ID, dataDir, tools.DefaultStopTimeout); err != nil {
 				log.Msg("Failed to stop Postgres", err)
 				tools.PrintContainerLogs(ctx, dockerClient, controlCont.ID)
 

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

@@ -40,6 +40,9 @@ const (
 	maxValuesToReturn     = 1
 	essentialLogsInterval = "10s"
 
+	// DefaultStopTimeout defines the default timeout for Postgres stop.
+	DefaultStopTimeout = 60
+
 	// ViewLogsCmd tells the command to view docker container logs.
 	ViewLogsCmd = "docker logs --since 1m -f"
 
@@ -156,15 +159,20 @@ func GetMountsFromMountPoints(dataDir string, mountPoints []types.MountPoint) []
 }
 
 // StopPostgres stops Postgres inside container.
-func StopPostgres(ctx context.Context, dockerClient *client.Client, containerID, dataDir string) error {
+func StopPostgres(ctx context.Context, dockerClient *client.Client, containerID, dataDir string, timeout int) error {
 	pgVersion, err := DetectPGVersion(dataDir)
 	if err != nil {
 		return errors.Wrap(err, "failed to detect PostgreSQL version")
 	}
 
+	stopCommand := []string{fmt.Sprintf("/usr/lib/postgresql/%g/bin/pg_ctl", pgVersion),
+		"-D", dataDir, "-w", "--timeout", strconv.Itoa(timeout), "stop"}
+
+	log.Msg("Stopping PostgreSQL instance", stopCommand)
+
 	if err := ExecCommand(ctx, dockerClient, containerID, types.ExecConfig{
 		User: defaults.Username,
-		Cmd:  []string{fmt.Sprintf("/usr/lib/postgresql/%g/bin/pg_ctl", pgVersion), "-D", dataDir, "stop"},
+		Cmd:  stopCommand,
 	}); err != nil {
 		return errors.Wrap(err, "failed to stop Postgres")
 	}

pkg/services/provision/databases/postgres/pgconfig/configuration.go

@@ -42,6 +42,12 @@ const (
 	// recoveryConfName defines the name of recovery Postgres (<11) config.
 	recoveryConfName = "recovery.conf"
 
+	// standbySignal defines the name of the file which means that standby mode is initialized for Postgres (>=12).
+	standbySignal = "standby.signal"
+
+	// recoverySignal defines the name of the file which means that recovery is initialized for Postgres (>=12).
+	recoverySignal = "recovery.signal"
+
 	// Database Lab configuration files.
 	// configPrefix defines a file prefix for Database Lab configuration files.
 	configPrefix = "postgresql.dblab."
@@ -307,7 +313,7 @@ func (m *Manager) ApplyRecovery(cfg map[string]string) error {
 	}
 
 	if m.pgVersion >= defaults.PGVersion12 {
-		if err := tools.TouchFile(path.Join(m.dataDir, "standby.signal")); err != nil {
+		if err := tools.TouchFile(path.Join(m.dataDir, standbySignal)); err != nil {
 			return err
 		}
 	}
@@ -335,10 +341,30 @@ func (m *Manager) TruncateRecoveryConfig() error {
 
 // RemoveRecoveryConfig removes a recovery configuration file.
 func (m *Manager) RemoveRecoveryConfig() error {
-	err := os.Remove(m.recoveryPath())
-	if pathError, ok := err.(*os.PathError); ok {
-		log.Dbg("failed to remove a recovery configuration file: ", pathError.Error())
+	if err := m.removeOptionally(m.recoveryPath()); err != nil {
+		return err
+	}
 
+	if m.pgVersion < defaults.PGVersion12 {
+		return nil
+	}
+
+	if err := m.removeOptionally(m.standbySignalPath()); err != nil {
+		return err
+	}
+
+	if err := m.removeOptionally(m.recoverySignalPath()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *Manager) removeOptionally(filepath string) error {
+	err := os.Remove(filepath)
+
+	if pathError, ok := err.(*os.PathError); ok {
+		log.Dbg(pathError.Error())
 		return nil
 	}
 
@@ -420,6 +446,16 @@ func (m Manager) recoveryFilename() string {
 	return recoveryConfName
 }
 
+// recoverySignalPath returns the path of the recovery signal file.
+func (m Manager) recoverySignalPath() string {
+	return path.Join(m.dataDir, recoverySignal)
+}
+
+// standbySignalPath returns the path of the standby signal file.
+func (m Manager) standbySignalPath() string {
+	return path.Join(m.dataDir, standbySignal)
+}
+
 // rewriteConfig completely rewrite a configuration file with provided parameters.
 func (m *Manager) rewriteConfig(pgConf string, extraConfig map[string]string) error {
 	log.Dbg("Applying configuration: ", pgConf)