Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit c666c79

Browse files
committed
fix: clean up data and manage runs during full data refresh in logical mode:
* Remove all snapshots and PGDATA during a full refresh of data in a logical mode * Remove `--clean` and `--if-exists` flags for the `pg_restore` since PGDATA is going to be empty * Remove the `forceInit` options from logical configuration files, add a warning if PGDATA still contains any files or directories. * Add a new `skipStartRefresh` option for a logical mode to skip data refreshing
1 parent 311eb32 commit c666c79

File tree

9 files changed

+50
-57
lines changed

9 files changed

+50
-57
lines changed

‎engine/configs/config.example.logical_generic.yml‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,9 @@ retrieval:
185185
# Timetable is to be defined in crontab format: https://en.wikipedia.org/wiki/Cron#Overview
186186
timetable: "0 0 * * 1"
187187

188+
# Skip data refresh while the retrieval starts.
189+
skipStartRefresh: false
190+
188191
# The jobs section must not contain physical and logical restore jobs simultaneously.
189192
jobs:
190193
- logicalDump
@@ -252,9 +255,6 @@ retrieval:
252255
# immediateRestore:
253256
# # Enable immediate restore.
254257
# enabled: true
255-
# # Restore data even if the Postgres directory (`global.dataDir`) is not empty.
256-
# # Note the existing data will be overwritten.
257-
# forceInit: false
258258
# # Option to adjust PostgreSQL configuration for a logical dump job.
259259
# # It's useful if a dumped database contains non-standard extensions.
260260
# <<: *db_configs
@@ -281,11 +281,6 @@ retrieval:
281281
# If your machine with DLE has 4 vCPUs or less, and you don't want to saturate them, use 2 or 1.
282282
parallelJobs: 4
283283

284-
285-
# Restore data even if the Postgres directory (`global.dataDir`) is not empty.
286-
# Note the existing data will be overwritten.
287-
forceInit: false
288-
289284
# Ignore errors that occurred during logical data restore. Do not ignore by default.
290285
ignoreErrors: false
291286

‎engine/configs/config.example.logical_rds_iam.yml‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,9 @@ retrieval:
185185
# Timetable is to be defined in crontab format: https://en.wikipedia.org/wiki/Cron#Overview
186186
timetable: "0 0 * * 1"
187187

188+
# Skip data refresh while the retrieval starts.
189+
skipStartRefresh: false
190+
188191
# The jobs section must not contain physical and logical restore jobs simultaneously.
189192
jobs:
190193
- logicalDump
@@ -249,9 +252,6 @@ retrieval:
249252
# immediateRestore:
250253
# # Enable immediate restore.
251254
# enabled: true
252-
# # Restore data even if the Postgres directory (`global.dataDir`) is not empty.
253-
# # Note the existing data will be overwritten.
254-
# forceInit: false
255255
# # Option to adjust PostgreSQL configuration for a logical dump job.
256256
# # It's useful if a dumped database contains non-standard extensions.
257257
# <<: *db_configs
@@ -277,10 +277,6 @@ retrieval:
277277
# If your machine with DLE has 4 vCPUs or less, and you don't want to saturate them, use 2 or 1.
278278
parallelJobs: 4
279279

280-
# Restore data even if the Postgres directory (`global.dataDir`) is not empty.
281-
# Note the existing data will be overwritten.
282-
forceInit: false
283-
284280
# Ignore errors that occurred during logical data restore. Do not ignore by default.
285281
ignoreErrors: false
286282

‎engine/internal/retrieval/config/config.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,14 +14,15 @@ import (
1414

1515
// Config describes of data retrieval jobs.
1616
type Config struct {
17-
Refresh Refresh `yaml:"refresh"`
17+
Refresh *Refresh `yaml:"refresh"`
1818
Jobs []string `yaml:"jobs,flow"`
1919
JobsSpec map[string]JobSpec `yaml:"spec"`
2020
}
2121

2222
// Refresh describes full-refresh options.
2323
type Refresh struct {
24-
Timetable string `yaml:"timetable"`
24+
Timetable string `yaml:"timetable"`
25+
SkipStartRefresh bool `yaml:"skipStartRefresh"`
2526
}
2627

2728
// JobSpec contains details about a job.

‎engine/internal/retrieval/dbmarker/dbmarker.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@ type Config struct {
3232
}
3333

3434
const (
35-
configDir = ".dblab"
35+
// ConfigDir defines the name of the dbMarker configuration directory.
36+
ConfigDir = ".dblab"
3637
configFilename = "dbmarker"
3738

3839
// LogicalDataType defines a logical data type.
@@ -44,7 +45,7 @@ const (
4445

4546
// Init inits DB marker for the data directory.
4647
func (m *Marker) initDBLabDirectory() error {
47-
dirname := path.Join(m.dataPath, configDir)
48+
dirname := path.Join(m.dataPath, ConfigDir)
4849
if err := os.MkdirAll(dirname, 0755); err != nil {
4950
return errors.Wrapf(err, "cannot create a DBMarker directory %s", dirname)
5051
}
@@ -104,5 +105,5 @@ func (m *Marker) SaveConfig(cfg *Config) error {
104105

105106
// buildFileName builds a DBMarker config filename.
106107
func (m *Marker) buildFileName() string {
107-
return path.Join(m.dataPath, configDir, configFilename)
108+
return path.Join(m.dataPath, ConfigDir, configFilename)
108109
}

‎engine/internal/retrieval/engine/postgres/logical/dump.go‎

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,6 @@ type Connection struct {
136136
// ImmediateRestore contains options for direct data restore without saving the dump file on disk.
137137
type ImmediateRestore struct {
138138
Enabled bool `yaml:"enabled"`
139-
ForceInit bool `yaml:"forceInit"`
140139
Configs map[string]string `yaml:"configs"`
141140
CustomOptions []string `yaml:"customOptions"`
142141
}
@@ -280,11 +279,7 @@ func (d *DumpJob) Run(ctx context.Context) (err error) {
280279
}
281280

282281
if d.DumpOptions.Restore.Enabled && !isEmpty {
283-
if !d.DumpOptions.Restore.ForceInit {
284-
return errors.New("the data directory is not empty. Use 'forceInit' or empty the data directory")
285-
}
286-
287-
log.Msg("The data directory is not empty. Existing data may be overwritten.")
282+
log.Warn("The data directory is not empty. Existing data will be overwritten.")
288283

289284
if err := updateConfigs(dataDir, d.DumpOptions.Restore.Configs); err != nil {
290285
return fmt.Errorf("failed to update configs: %w", err)
@@ -726,10 +721,6 @@ func (d *DumpJob) buildLogicalRestoreCommand(dbName string) []string {
726721
restoreCmd = append(restoreCmd, "--create")
727722
}
728723

729-
if d.Restore.ForceInit {
730-
restoreCmd = append(restoreCmd, "--clean", "--if-exists")
731-
}
732-
733724
restoreCmd = append(restoreCmd, d.DumpOptions.Restore.CustomOptions...)
734725

735726
return restoreCmd

‎engine/internal/retrieval/engine/postgres/logical/restore.go‎

Lines changed: 2 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,6 @@ type RestoreOptions struct {
100100
DockerImage string `yaml:"dockerImage"`
101101
ContainerConfig map[string]interface{} `yaml:"containerConfig"`
102102
Databases map[string]DumpDefinition `yaml:"databases"`
103-
ForceInit bool `yaml:"forceInit"`
104103
IgnoreErrors bool `yaml:"ignoreErrors"`
105104
ParallelJobs int `yaml:"parallelJobs"`
106105
Configs map[string]string `yaml:"configs"`
@@ -198,16 +197,7 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
198197
}
199198

200199
if !isEmpty {
201-
if !r.ForceInit {
202-
return fmt.Errorf("the data directory %q is not empty. Use 'forceInit' or empty the data directory: %w",
203-
dataDir, err)
204-
}
205-
206-
log.Msg(fmt.Sprintf("The data directory %q is not empty. Existing data may be overwritten.", dataDir))
207-
208-
if err := updateConfigs(dataDir, r.RestoreOptions.Configs); err != nil {
209-
return fmt.Errorf("failed to update configuration: %w", err)
210-
}
200+
log.Warn(fmt.Sprintf("The data directory %q is not empty. Existing data will be overwritten.", dataDir))
211201
}
212202

213203
if err := tools.PullImage(ctx, r.dockerClient, r.RestoreOptions.DockerImage); err != nil {
@@ -740,17 +730,14 @@ func (r *RestoreJob) buildPlainTextCommand(dumpName string, definition DumpDefin
740730
}
741731

742732
func (r *RestoreJob) buildPGRestoreCommand(dumpName string, definition DumpDefinition) []string {
733+
// Using the default database name because the database for connection must exist.
743734
restoreCmd := []string{"pg_restore", "--username", r.globalCfg.Database.User(), "--dbname", defaults.DBName}
744735

745736
if definition.dbName != defaults.DBName {
746737
// To avoid recreating of the default database.
747738
restoreCmd = append(restoreCmd, "--create")
748739
}
749740

750-
if r.ForceInit {
751-
restoreCmd = append(restoreCmd, "--clean", "--if-exists")
752-
}
753-
754741
restoreCmd = append(restoreCmd, "--jobs", strconv.Itoa(r.ParallelJobs))
755742

756743
if len(definition.Tables) > 0 {

‎engine/internal/retrieval/engine/postgres/logical/restore_test.go‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@ func TestRestoreCommandBuilding(t *testing.T) {
3434
{
3535
copyOptions: RestoreOptions{
3636
ParallelJobs: 1,
37-
ForceInit: false,
3837
Databases: map[string]DumpDefinition{
3938
"testDB": {
4039
Format: customFormat,
@@ -48,14 +47,12 @@ func TestRestoreCommandBuilding(t *testing.T) {
4847
{
4948
copyOptions: RestoreOptions{
5049
ParallelJobs: 4,
51-
ForceInit: true,
5250
},
53-
command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--create", "--clean", "--if-exists", "--jobs", "4"},
51+
command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--create", "--jobs", "4"},
5452
},
5553
{
5654
copyOptions: RestoreOptions{
5755
ParallelJobs: 2,
58-
ForceInit: false,
5956
Databases: map[string]DumpDefinition{"testDB": {}},
6057
DumpLocation: "/tmp/db.dump",
6158
CustomOptions: []string{"--no-privileges", "--no-owner", "--exit-on-error"},

‎engine/internal/retrieval/retrieval.go‎

Lines changed: 32 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,13 @@ func (r *Retrieval) Run(ctx context.Context) error {
190190
return fmt.Errorf("failed to collect content lists from the foundation Docker image of the logicalDump job: %w", err)
191191
}
192192

193+
if r.cfg.Refresh != nil && r.cfg.Refresh.SkipStartRefresh {
194+
log.Msg("Continue without performing initial data refresh because the `skipStartRefresh` option is enabled")
195+
r.setupScheduler(ctx)
196+
197+
return nil
198+
}
199+
193200
fsManager, err := r.getNextPoolToDataRetrieving()
194201
if err != nil {
195202
var skipError *SkipRefreshingError
@@ -393,6 +400,12 @@ func (r *Retrieval) RefreshData(ctx context.Context, poolName string) error {
393400
r.State.CurrentJob = nil
394401
}()
395402

403+
if r.State.Mode == models.Logical {
404+
if err := preparePoolToRefresh(fsm, r.runner); err != nil {
405+
return fmt.Errorf("failed to prepare pool for initial refresh: %w", err)
406+
}
407+
}
408+
396409
for _, j := range jobs {
397410
r.State.CurrentJob = j
398411

@@ -530,7 +543,7 @@ func (r *Retrieval) defineRetrievalMode() {
530543
func (r *Retrieval) setupScheduler(ctx context.Context) {
531544
r.stopScheduler()
532545

533-
if r.cfg.Refresh.Timetable == "" {
546+
if r.cfg.Refresh==nil||r.cfg.Refresh.Timetable == "" {
534547
return
535548
}
536549

@@ -608,10 +621,6 @@ func (r *Retrieval) FullRefresh(ctx context.Context) error {
608621

609622
log.Msg("Pool to a full refresh: ", poolToUpdate.Pool())
610623

611-
if err := preparePoolToRefresh(poolToUpdate); err != nil {
612-
return errors.Wrap(err, "failed to prepare the pool to a full refresh")
613-
}
614-
615624
// Stop service containers: sync-instance, etc.
616625
if cleanUpErr := cont.CleanUpControlContainers(runCtx, r.docker, r.engineProps.InstanceID); cleanUpErr != nil {
617626
log.Err("Failed to clean up service containers:", cleanUpErr)
@@ -641,7 +650,7 @@ func (r *Retrieval) stopScheduler() {
641650
}
642651
}
643652

644-
func preparePoolToRefresh(poolToUpdate pool.FSManager) error {
653+
func preparePoolToRefresh(poolToUpdate pool.FSManager, runner runners.Runner) error {
645654
cloneList, err := poolToUpdate.ListClonesNames()
646655
if err != nil {
647656
return errors.Wrap(err, "failed to check running clones")
@@ -652,6 +661,12 @@ func preparePoolToRefresh(poolToUpdate pool.FSManager) error {
652661
strings.Join(cloneList, " "))
653662
}
654663

664+
if _, err := runner.Run(fmt.Sprintf("rm -rf %s %s",
665+
filepath.Join(poolToUpdate.Pool().DataDir(), "*"),
666+
filepath.Join(poolToUpdate.Pool().DataDir(), dbmarker.ConfigDir))); err != nil {
667+
return errors.Wrap(err, "failed to clean unix socket directory")
668+
}
669+
655670
poolToUpdate.RefreshSnapshotList()
656671

657672
snapshots := poolToUpdate.SnapshotList()
@@ -660,7 +675,11 @@ func preparePoolToRefresh(poolToUpdate pool.FSManager) error {
660675
return nil
661676
}
662677

678+
log.Msg("Preparing pool for full data refresh; existing snapshots are to be destroyed")
679+
663680
for _, snapshotEntry := range snapshots {
681+
log.Msg("Destroying snapshot:", snapshotEntry.ID)
682+
664683
if err := poolToUpdate.DestroySnapshot(snapshotEntry.ID); err != nil {
665684
return errors.Wrap(err, "failed to destroy the existing snapshot")
666685
}
@@ -671,9 +690,15 @@ func preparePoolToRefresh(poolToUpdate pool.FSManager) error {
671690

672691
// ReportState collects the current restore state.
673692
func (r *Retrieval) ReportState() telemetry.Restore {
693+
var refreshingTimetable string
694+
695+
if r.cfg.Refresh != nil {
696+
refreshingTimetable = r.cfg.Refresh.Timetable
697+
}
698+
674699
return telemetry.Restore{
675700
Mode: r.State.Mode,
676-
Refreshing: r.cfg.Refresh.Timetable,
701+
Refreshing: refreshingTimetable,
677702
Jobs: r.cfg.Jobs,
678703
}
679704
}

‎engine/internal/retrieval/validator.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ func validateStructure(r *config.Config) error {
7070
}
7171

7272
func validateRefreshTimetable(r *config.Config) error {
73-
if r.Refresh.Timetable == "" {
73+
if r.Refresh==nil||r.Refresh.Timetable == "" {
7474
return nil
7575
}
7676

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /