Merge branch '350-postgres-shutdown' into 'master'

fix(engine): implement a general approach to gracefully shutting down Postgres containers (#350)

Closes #350

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

Author: agneum

Diff for: engine/cmd/database-lab/main.go

@@ -126,7 +126,7 @@ func main() {
 	emergencyShutdown := func() {
 		cancel()
 
-		shutdownDatabaseLabEngine(context.Background(), docker, engProps, pm.First())
+		shutdownDatabaseLabEngine(context.Background(), docker, &cfg.Global.Database, engProps.InstanceID, pm.First())
 	}
 
 	cloningSvc := cloning.NewBase(&cfg.Cloning, provisioner, tm, observingChan)
@@ -191,7 +191,7 @@ func main() {
 		log.Msg(err)
 	}
 
-	shutdownDatabaseLabEngine(ctxBackground, docker, engProps, pm.First())
+	shutdownDatabaseLabEngine(ctxBackground, docker, &cfg.Global.Database, engProps.InstanceID, pm.First())
 	cloningSvc.SaveClonesState()
 	tm.SendEvent(ctxBackground, telemetry.EngineStoppedEvent, telemetry.EngineStopped{Uptime: server.Uptime()})
 }
@@ -290,14 +290,14 @@ func setShutdownListener() chan os.Signal {
 	return c
 }
 
-func shutdownDatabaseLabEngine(ctx context.Context, dockerCLI *client.Client, engProps global.EngineProps, fsm pool.FSManager) {
+func shutdownDatabaseLabEngine(ctx context.Context, docker *client.Client, dbCfg *global.Database, instanceID string, fsm pool.FSManager) {
 	log.Msg("Stopping auxiliary containers")
 
-	if err := cont.StopControlContainers(ctx, dockerCLI, engProps.InstanceID, fsm); err != nil {
+	if err := cont.StopControlContainers(ctx, docker, dbCfg, instanceID, fsm); err != nil {
 		log.Err("Failed to stop control containers", err)
 	}
 
-	if err := cont.CleanUpSatelliteContainers(ctx, dockerCLI, engProps.InstanceID); err != nil {
+	if err := cont.CleanUpSatelliteContainers(ctx, docker, instanceID); err != nil {
 		log.Err("Failed to stop satellite containers", err)
 	}
 

Diff for: engine/internal/retrieval/engine/postgres/logical/dump.go

@@ -357,6 +357,10 @@ func (d *DumpJob) Run(ctx context.Context) (err error) {
 			return errors.Wrap(err, "failed to recalculate statistics after restore")
 		}
 
+		if err := tools.RunCheckpoint(ctx, d.dockerClient, containerID, d.globalCfg.Database.User(), d.globalCfg.Database.DBName); err != nil {
+			return errors.Wrap(err, "failed to run checkpoint before stop")
+		}
+
 		if err := tools.StopPostgres(ctx, d.dockerClient, containerID, dataDir, tools.DefaultStopTimeout); err != nil {
 			return errors.Wrap(err, "failed to stop Postgres instance")
 		}

Diff for: engine/internal/retrieval/engine/postgres/logical/restore.go

@@ -259,6 +259,10 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 		return errors.Wrap(err, "failed to recalculate statistics after restore")
 	}
 
+	if err := tools.RunCheckpoint(ctx, r.dockerClient, containerID, r.globalCfg.Database.User(), r.globalCfg.Database.DBName); err != nil {
+		return errors.Wrap(err, "failed to run checkpoint before stop")
+	}
+
 	if err := tools.StopPostgres(ctx, r.dockerClient, containerID, dataDir, tools.DefaultStopTimeout); err != nil {
 		return errors.Wrap(err, "failed to stop Postgres instance")
 	}

Diff for: engine/internal/retrieval/engine/postgres/snapshot/physical.go

@@ -149,8 +149,10 @@ type syncState struct {
 }
 
 // NewPhysicalInitialJob creates a new physical initial job.
-func NewPhysicalInitialJob(cfg config.JobConfig, global *global.Config, engineProps global.EngineProps, cloneManager pool.FSManager,
-	tm *telemetry.Agent) (*PhysicalInitial, error) {
+func NewPhysicalInitialJob(
+	cfg config.JobConfig, global *global.Config, engineProps global.EngineProps, cloneManager pool.FSManager,
+	tm *telemetry.Agent,
+) (*PhysicalInitial, error) {
 	p := &PhysicalInitial{
 		name:         cfg.Spec.Name,
 		cloneManager: cloneManager,
@@ -397,7 +399,13 @@ func (p *PhysicalInitial) checkSyncInstance(ctx context.Context) (string, error)
 
 	log.Msg("Sync instance has been checked. It is running")
 
-	if err := p.checkpoint(ctx, syncContainer.ID); err != nil {
+	if err := tools.RunCheckpoint(
+		ctx,
+		p.dockerClient,
+		syncContainer.ID,
+		p.globalCfg.Database.User(),
+		p.globalCfg.Database.Name(),
+	); err != nil {
 		return "", errors.Wrap(err, "failed to make a checkpoint for sync instance")
 	}
 
@@ -616,9 +624,8 @@ func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string,
 		}
 	}
 
-	// Checkpoint.
-	if err := p.checkpoint(ctx, containerID); err != nil {
-		return err
+	if err := tools.RunCheckpoint(ctx, p.dockerClient, containerID, p.globalCfg.Database.User(), p.globalCfg.Database.Name()); err != nil {
+		return errors.Wrap(err, "failed to run checkpoint")
 	}
 
 	if err := cfgManager.RemoveRecoveryConfig(); err != nil {
@@ -646,7 +653,10 @@ func (p *PhysicalInitial) promoteInstance(ctx context.Context, clonePath string,
 	return nil
 }
 
-func (p *PhysicalInitial) getDSAFromWAL(ctx context.Context, pgVersion float64, containerID, cloneDir string) (string, error) {
+func (p *PhysicalInitial) getDSAFromWAL(ctx context.Context, pgVersion float64, containerID, cloneDir string) (
+	string,
+	error,
+) {
 	log.Dbg(cloneDir)
 
 	walDirectory := walDir(cloneDir, pgVersion)
@@ -692,7 +702,12 @@ func walDir(cloneDir string, pgVersion float64) string {
 	return path.Join(cloneDir, dir)
 }
 
-func (p *PhysicalInitial) parseWAL(ctx context.Context, containerID string, pgVersion float64, walFilePath string) string {
+func (p *PhysicalInitial) parseWAL(
+	ctx context.Context,
+	containerID string,
+	pgVersion float64,
+	walFilePath string,
+) string {
 	cmd := walCommand(pgVersion, walFilePath)
 
 	output, err := tools.ExecCommandWithOutput(ctx, p.dockerClient, containerID, types.ExecConfig{
@@ -768,7 +783,11 @@ func buildRecoveryConfig(fileConfig, userRecoveryConfig map[string]string) map[s
 	return recoveryConf
 }
 
-func (p *PhysicalInitial) markDSA(ctx context.Context, defaultDSA, containerID, dataDir string, pgVersion float64) error {
+func (p *PhysicalInitial) markDSA(
+	ctx context.Context,
+	defaultDSA, containerID, dataDir string,
+	pgVersion float64,
+) error {
 	extractedDataStateAt, err := p.extractDataStateAt(ctx, containerID, dataDir, pgVersion, defaultDSA)
 	if err != nil {
 		if defaultDSA == "" {
@@ -895,8 +914,10 @@ and the source doesn't have enough activity.
 Step 3. Use the timestamp of the latest checkpoint. This is extracted from PGDATA using the
 pg_controldata utility. Note that this is not an exact value of the latest activity in the source
 before we took a copy of PGDATA, but we suppose it is not far from it. */
-func (p *PhysicalInitial) extractDataStateAt(ctx context.Context, containerID, dataDir string, pgVersion float64,
-	defaultDSA string) (string, error) {
+func (p *PhysicalInitial) extractDataStateAt(
+	ctx context.Context, containerID, dataDir string, pgVersion float64,
+	defaultDSA string,
+) (string, error) {
 	output, err := p.getLastXActReplayTimestamp(ctx, containerID)
 	if err != nil {
 		log.Dbg("unable to get last replay timestamp from the promotion container: ", err)
@@ -1002,20 +1023,6 @@ func (p *PhysicalInitial) runPromoteCommand(ctx context.Context, containerID, cl
 	return nil
 }
 
-func (p *PhysicalInitial) checkpoint(ctx context.Context, containerID string) error {
-	commandCheckpoint := []string{"psql", "-U", p.globalCfg.Database.User(), "-d", p.globalCfg.Database.Name(), "-XAtc", "checkpoint"}
-	log.Msg("Run checkpoint command", commandCheckpoint)
-
-	output, err := tools.ExecCommandWithOutput(ctx, p.dockerClient, containerID, types.ExecConfig{Cmd: commandCheckpoint})
-	if err != nil {
-		return errors.Wrap(err, "failed to make checkpoint")
-	}
-
-	log.Msg("Checkpoint result: ", output)
-
-	return nil
-}
-
 func (p *PhysicalInitial) markDatabaseData() error {
 	if err := p.dbMarker.CreateConfig(); err != nil {
 		return errors.Wrap(err, "failed to create a DBMarker config of the database")

Diff for: engine/internal/retrieval/engine/postgres/tools/cont/container.go

@@ -14,12 +14,13 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
-	units "github.com/docker/go-units"
+	"github.com/docker/go-units"
 	"github.com/pkg/errors"
 
 	"gitlab.com/postgres-ai/database-lab/v3/internal/provision/pool"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/tools"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/options"
+	"gitlab.com/postgres-ai/database-lab/v3/pkg/config/global"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/log"
 )
 
@@ -64,7 +65,8 @@ const (
 // TODO(akartasov): Control container manager.
 
 // StopControlContainers stops control containers run by Database Lab Engine.
-func StopControlContainers(ctx context.Context, dockerClient *client.Client, instanceID string, fsm pool.FSManager) error {
+func StopControlContainers(ctx context.Context, dockerClient *client.Client, dbCfg *global.Database, instanceID string,
+	fsm pool.FSManager) error {
 	log.Msg("Stop control containers")
 
 	list, err := getContainerList(ctx, dockerClient, instanceID, getControlContainerFilters())
@@ -84,6 +86,11 @@ func StopControlContainers(ctx context.Context, dockerClient *client.Client, ins
 		if shouldStopInternalProcess(controlLabel) && fsm != nil {
 			log.Msg("Stopping control container: ", containerName)
 
+			if err := tools.RunCheckpoint(ctx, dockerClient, controlCont.ID, dbCfg.User(), dbCfg.Name()); err != nil {
+				log.Msg("Failed to make a checkpoint:", err)
+				tools.PrintContainerLogs(ctx, dockerClient, controlCont.ID)
+			}
+
 			if err := tools.StopPostgres(ctx, dockerClient, controlCont.ID, fsm.Pool().DataDir(), tools.DefaultStopTimeout); err != nil {
 				log.Msg("Failed to stop Postgres", err)
 				tools.PrintContainerLogs(ctx, dockerClient, controlCont.ID)

Diff for: engine/internal/retrieval/engine/postgres/tools/tools.go

@@ -258,6 +258,40 @@ func StartPostgres(ctx context.Context, dockerClient *client.Client, containerID
 	return nil
 }
 
+// RunCheckpoint runs checkpoint, usually before the postgres stop
+func RunCheckpoint(
+	ctx context.Context,
+	dockerClient *client.Client,
+	containerID string,
+	user string,
+	database string,
+) error {
+	commandCheckpoint := []string{
+		"psql",
+		"-U",
+		user,
+		"-d",
+		database,
+		"-XAtc",
+		"checkpoint",
+	}
+	log.Msg("Run checkpoint command", commandCheckpoint)
+
+	output, err := ExecCommandWithOutput(
+		ctx,
+		dockerClient,
+		containerID,
+		types.ExecConfig{Cmd: commandCheckpoint},
+	)
+	if err != nil {
+		return errors.Wrap(err, "failed to make checkpoint")
+	}
+
+	log.Msg("Checkpoint result: ", output)
+
+	return nil
+}
+
 // StopPostgres stops Postgres inside container.
 func StopPostgres(ctx context.Context, dockerClient *client.Client, containerID, dataDir string, timeout int) error {
 	pgVersion, err := DetectPGVersion(dataDir)