Skip to content

Commit 96ce8a7

Browse files
agneumagneum
committed
fix: logical dump and restore issues (#152)
1 parent f211af4 commit 96ce8a7

File tree

7 files changed

+66
-65
lines changed

7 files changed

+66
-65
lines changed

‎pkg/retrieval/engine/postgres/initialize/logical/dump.go

Lines changed: 33 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ type DumpJob struct {
6565

6666
// DumpOptions defines a logical dump options.
6767
type DumpOptions struct {
68-
DumpFile string `yaml:"dumpLocation"`
68+
DumpLocation string `yaml:"dumpLocation"`
6969
DockerImage string `yaml:"dockerImage"`
7070
Connection Connection `yaml:"connection"`
7171
Source Source `yaml:"source"`
@@ -273,36 +273,19 @@ func (d *DumpJob) Run(ctx context.Context) (err error) {
273273
log.Msg("Partial dump will be run. Tables for dumping: ", strings.Join(d.Partial.Tables, ", "))
274274
}
275275

276-
var output io.Writer = os.Stdout
277-
278-
if d.DumpOptions.DumpFile != "" {
279-
dumpFile, err := os.Create(d.getDumpContainerPath())
280-
if err != nil {
281-
return errors.Wrap(err, "failed to create file")
282-
}
283-
284-
defer func() {
285-
if err := dumpFile.Close(); err != nil {
286-
log.Err("failed to close dump file", err)
287-
}
288-
}()
289-
290-
output = dumpFile
291-
}
292-
293-
if err := d.performDumpCommand(ctx, output, cont.ID, execCommand.ID); err != nil {
276+
if err := d.performDumpCommand(ctx, os.Stdout, cont.ID, execCommand.ID); err != nil {
294277
return errors.Wrap(err, "failed to dump a database")
295278
}
296279

297-
if err := d.markDatabaseData(); err != nil {
298-
return errors.Wrap(err, "failed to mark the created dump")
299-
}
300-
301280
if d.DumpOptions.Restore != nil {
281+
if err := d.markDatabaseData(); err != nil {
282+
return errors.Wrap(err, "failed to mark the created dump")
283+
}
284+
302285
if err := recalculateStats(ctx, d.dockerClient, cont.ID, buildAnalyzeCommand(Connection{
303286
DBName: d.config.db.DBName,
304287
Username: defaults.Username,
305-
})); err != nil {
288+
}, d.DumpOptions.ParallelJobs)); err != nil {
306289
return errors.Wrap(err, "failed to recalculate statistics after restore")
307290
}
308291
}
@@ -345,16 +328,16 @@ func (d *DumpJob) performDumpCommand(ctx context.Context, cmdOutput io.Writer, c
345328
return nil
346329
}
347330

348-
func (d *DumpJob) getDumpContainerPath() string {
349-
return d.DumpFile
350-
}
351-
352331
func (d *DumpJob) getEnvironmentVariables() []string {
353332
envs := []string{
354-
"PGDATA=" + d.globalCfg.DataDir,
355333
"POSTGRES_HOST_AUTH_METHOD=trust",
356334
}
357335

336+
// Avoid initialization of PostgreSQL directory in case of preparing of a dump.
337+
if d.DumpOptions.Restore != nil {
338+
envs = append(envs, "PGDATA="+d.globalCfg.DataDir)
339+
}
340+
358341
if d.DumpOptions.Source.Type == sourceTypeLocal && d.DumpOptions.Source.Connection.Port == defaults.Port {
359342
log.Msg(fmt.Sprintf("The default PostgreSQL port is busy, trying to use an alternative one: %d", reservePort))
360343
envs = append(envs, "PGPORT="+strconv.Itoa(reservePort))
@@ -365,6 +348,7 @@ func (d *DumpJob) getEnvironmentVariables() []string {
365348

366349
func (d *DumpJob) buildContainerConfig() *container.Config {
367350
return &container.Config{
351+
Labels: map[string]string{"label": tools.DBLabControlLabel},
368352
Env: d.getEnvironmentVariables(),
369353
Image: d.DockerImage,
370354
Healthcheck: health.GetConfig(),
@@ -388,11 +372,11 @@ func (d *DumpJob) buildHostConfig() (*container.HostConfig, error) {
388372
func (d *DumpJob) getMountVolumes() []mount.Mount {
389373
mounts := d.dumper.GetMounts()
390374

391-
if d.DumpOptions.DumpFile != "" {
375+
if d.DumpOptions.DumpLocation != "" {
392376
mounts = append(mounts, mount.Mount{
393377
Type: mount.TypeBind,
394-
Source: filepath.Dir(d.DumpOptions.DumpFile),
395-
Target: filepath.Dir(d.DumpOptions.DumpFile),
378+
Source: filepath.Dir(d.DumpOptions.DumpLocation),
379+
Target: filepath.Dir(d.DumpOptions.DumpLocation),
396380
})
397381
}
398382

@@ -426,21 +410,27 @@ func (d *DumpJob) getExecEnvironmentVariables() []string {
426410
}
427411

428412
func (d *DumpJob) buildLogicalDumpCommand() []string {
429-
dumpCmd := []string{"pg_dump", "-C", "-Fc"}
413+
format := "custom"
414+
415+
if d.DumpOptions.ParallelJobs > defaultParallelJobs {
416+
format = "directory"
417+
}
430418

431419
optionalArgs := map[string]string{
432-
"-h": d.config.db.Host,
433-
"-p": strconv.Itoa(d.config.db.Port),
434-
"-U": d.config.db.Username,
435-
"-d": d.config.db.DBName,
436-
"-j": strconv.Itoa(d.DumpOptions.ParallelJobs),
420+
"--host": d.config.db.Host,
421+
"--port": strconv.Itoa(d.config.db.Port),
422+
"--username": d.config.db.Username,
423+
"--dbname": d.config.db.DBName,
424+
"--jobs": strconv.Itoa(d.DumpOptions.ParallelJobs),
437425
}
438-
dumpCmd = append(dumpCmd, prepareCmdOptions(optionalArgs)...)
426+
427+
dumpCmd := append([]string{"pg_dump", "--create", "--format", format}, prepareCmdOptions(optionalArgs)...)
439428

440429
for _, table := range d.Partial.Tables {
441-
dumpCmd = append(dumpCmd, "-t", table)
430+
dumpCmd = append(dumpCmd, "--table", table)
442431
}
443432

433+
// Define if restore directly or export to dump location.
444434
if d.DumpOptions.Restore != nil {
445435
dumpCmd = append(dumpCmd, d.buildLogicalRestoreCommand()...)
446436
cmd := strings.Join(dumpCmd, " ")
@@ -450,11 +440,13 @@ func (d *DumpJob) buildLogicalDumpCommand() []string {
450440
return []string{"sh", "-c", cmd}
451441
}
452442

443+
dumpCmd = append(dumpCmd, "--file", d.DumpOptions.DumpLocation)
444+
453445
return dumpCmd
454446
}
455447

456448
func (d *DumpJob) buildLogicalRestoreCommand() []string {
457-
restoreCmd := []string{"|", "pg_restore", "-U", defaults.Username, "-C", "-d", defaults.DBName, "--no-privileges"}
449+
restoreCmd := []string{"|", "pg_restore", "--username", defaults.Username, "--create", "--dbname", defaults.DBName, "--no-privileges"}
458450

459451
if d.Restore.ForceInit {
460452
restoreCmd = append(restoreCmd, "--clean", "--if-exists")

‎pkg/retrieval/engine/postgres/initialize/logical/logical.go

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ package logical
77

88
import (
99
"context"
10+
"strconv"
1011

1112
"github.com/docker/docker/api/types"
1213
"github.com/docker/docker/client"
@@ -40,12 +41,13 @@ func recalculateStats(ctx context.Context, dockerClient *client.Client, contID s
4041
return nil
4142
}
4243

43-
func buildAnalyzeCommand(conn Connection) []string {
44+
func buildAnalyzeCommand(conn Connection, parallelJobs int) []string {
4445
analyzeCmd := []string{
45-
"psql",
46-
"-U", conn.Username,
47-
"-d", conn.DBName,
48-
"-c", "vacuum freeze analyze;",
46+
"vacuumdb",
47+
"--analyze",
48+
"--jobs", strconv.Itoa(parallelJobs),
49+
"--username", conn.Username,
50+
"--dbname", conn.DBName,
4951
}
5052

5153
return analyzeCmd

‎pkg/retrieval/engine/postgres/initialize/logical/restore.go

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -148,10 +148,6 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
148148

149149
log.Msg(fmt.Sprintf("Running container: %s. ID: %v", r.restoreContainerName(), cont.ID))
150150

151-
if err := r.markDatabase(ctx, cont.ID); err != nil {
152-
return errors.Wrap(err, "failed to mark the database")
153-
}
154-
155151
if err := tools.CheckContainerReadiness(ctx, r.dockerClient, cont.ID); err != nil {
156152
return errors.Wrap(err, "failed to readiness check")
157153
}
@@ -182,10 +178,14 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
182178
return errors.Wrap(err, "failed to exec restore command")
183179
}
184180

181+
if err := r.markDatabase(ctx, cont.ID); err != nil {
182+
return errors.Wrap(err, "failed to mark the database")
183+
}
184+
185185
if err := recalculateStats(ctx, r.dockerClient, cont.ID, buildAnalyzeCommand(Connection{
186-
DBName: r.RestoreOptions.DBName,
187186
Username: defaults.Username,
188-
})); err != nil {
187+
DBName: r.RestoreOptions.DBName,
188+
}, r.RestoreOptions.ParallelJobs)); err != nil {
189189
return errors.Wrap(err, "failed to recalculate statistics after restore")
190190
}
191191

@@ -196,8 +196,10 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
196196

197197
func (r *RestoreJob) buildContainerConfig() *container.Config {
198198
return &container.Config{
199+
Labels: map[string]string{"label": tools.DBLabControlLabel},
199200
Env: []string{
200201
"PGDATA=" + r.globalCfg.DataDir,
202+
"POSTGRES_HOST_AUTH_METHOD=trust",
201203
},
202204
Image: r.RestoreOptions.DockerImage,
203205
Healthcheck: health.GetConfig(),
@@ -244,7 +246,7 @@ func (r *RestoreJob) markDatabase(ctx context.Context, contID string) error {
244246
}
245247

246248
func (r *RestoreJob) retrieveDataStateAt(ctx context.Context, contID string) (string, error) {
247-
restoreMetaCmd := []string{"sh", "-c", "pg_restore -l " + r.RestoreOptions.DumpFile + " | head -n 10"}
249+
restoreMetaCmd := []string{"sh", "-c", "pg_restore --list " + r.RestoreOptions.DumpFile + " | head -n 10"}
248250

249251
log.Dbg("Running a restore metadata command: ", restoreMetaCmd)
250252

@@ -273,18 +275,19 @@ func (r *RestoreJob) retrieveDataStateAt(ctx context.Context, contID string) (st
273275
}
274276

275277
func (r *RestoreJob) buildLogicalRestoreCommand() []string {
276-
restoreCmd := []string{"pg_restore", "-U", defaults.Username, "-C"}
278+
restoreCmd := []string{"pg_restore", "--username", defaults.Username, "--dbname", defaults.DBName, "--create", "--no-privileges"}
277279

278280
if r.ForceInit {
279-
restoreCmd = append(restoreCmd, "-d", defaults.DBName, "--clean", "--if-exists")
280-
} else {
281-
restoreCmd = append(restoreCmd, "-d", r.RestoreOptions.DBName)
281+
restoreCmd = append(restoreCmd, "--clean", "--if-exists")
282282
}
283+
//else {
284+
// restoreCmd = append(restoreCmd)
285+
//}
283286

284-
restoreCmd = append(restoreCmd, "-j", strconv.Itoa(r.ParallelJobs))
287+
restoreCmd = append(restoreCmd, "--jobs", strconv.Itoa(r.ParallelJobs))
285288

286289
for _, table := range r.Partial.Tables {
287-
restoreCmd = append(restoreCmd, "-t", table)
290+
restoreCmd = append(restoreCmd, "--table", table)
288291
}
289292

290293
restoreCmd = append(restoreCmd, r.RestoreOptions.DumpFile)

‎pkg/retrieval/engine/postgres/initialize/logical/restore_test.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,14 @@ func TestRestoreCommandBuilding(t *testing.T) {
2424
ForceInit: false,
2525
DumpFile: "/tmp/db.dump",
2626
},
27-
Command: []string{"pg_restore", "-U", "postgres", "-C", "-d", "testDB", "-j", "1", "/tmp/db.dump"},
27+
Command: []string{"pg_restore", "--username", "postgres", "--dbname", "postgres", "--create", "--no-privileges", "--jobs", "1", "/tmp/db.dump"},
2828
},
2929
{
3030
CopyOptions: RestoreOptions{
3131
ParallelJobs: 4,
3232
ForceInit: true,
3333
},
34-
Command: []string{"pg_restore", "-U", "postgres", "-C", "-d", "postgres", "--clean", "--if-exists", "-j", "4", ""},
34+
Command: []string{"pg_restore", "--username", "postgres", "--dbname", "postgres", "--create", "--no-privileges", "--clean", "--if-exists", "--jobs", "4", ""},
3535
},
3636
{
3737
CopyOptions: RestoreOptions{
@@ -40,7 +40,7 @@ func TestRestoreCommandBuilding(t *testing.T) {
4040
Partial: Partial{Tables: []string{"test", "users"}},
4141
DumpFile: "/tmp/db.dump",
4242
},
43-
Command: []string{"pg_restore", "-U", "postgres", "-C", "-d", "testDB", "-j", "1", "-t", "test", "-t", "users", "/tmp/db.dump"},
43+
Command: []string{"pg_restore", "--username", "postgres", "--dbname", "postgres", "--create", "--no-privileges", "--jobs", "1", "--table", "test", "--table", "users", "/tmp/db.dump"},
4444
},
4545
}
4646

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -360,8 +360,9 @@ func (r *RestoreJob) getEnvironmentVariables() []string {
360360

361361
func (r *RestoreJob) buildContainerConfig() *container.Config {
362362
return &container.Config{
363-
Env: r.getEnvironmentVariables(),
364-
Image: r.DockerImage,
363+
Labels: map[string]string{"label": tools.DBLabControlLabel},
364+
Env: r.getEnvironmentVariables(),
365+
Image: r.DockerImage,
365366
}
366367
}
367368

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -465,7 +465,7 @@ func (p *PhysicalInitial) adjustRecoveryConfiguration(pgVersion, clonePGDataDir
465465

466466
func (p *PhysicalInitial) buildContainerConfig(clonePath, promoteImage string) *container.Config {
467467
return &container.Config{
468-
Labels: map[string]string{"label": "dblab_control"},
468+
Labels: map[string]string{"label": tools.DBLabControlLabel},
469469
Env: []string{
470470
"PGDATA=" + clonePath,
471471
"POSTGRES_HOST_AUTH_METHOD=trust",

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ const (
3838

3939
// SyncInstanceContainerPrefix defines a sync container name.
4040
SyncInstanceContainerPrefix = "dblab_sync_"
41+
42+
// DBLabControlLabel defines a label to mark service containers.
43+
DBLabControlLabel = "dblab_control"
4144
)
4245

4346
// IsEmptyDirectory checks whether a directory is empty.

0 commit comments

Comments
 (0)