@@ -37,10 +37,10 @@ const (
3737 dumpContainerName = "retriever_logical_dump"
3838 dumpContainerDir = "/tmp/dump"
3939
40- // Defines dump connection types.
41- connectionTypeLocal = "local"
42- //connectionTypeRemote = "remote"
43- //connectionTypeRDS = "rds"
40+ // Defines dump source types.
41+ sourceTypeLocal = "local"
42+ sourceTypeRemote = "remote"
43+ sourceTypeRDS = "rds"
4444
4545 // reservePort defines reserve port in case of a local dump.
4646 reservePort = 9999
@@ -55,6 +55,8 @@ type DumpJob struct {
5555 name string
5656 dockerClient * client.Client
5757 globalCfg * dblabCfg.Global
58+ config dumpJobConfig
59+ dumper dumper
5860 DumpOptions
5961}
6062
@@ -63,14 +65,37 @@ type DumpOptions struct {
6365 DumpFile string `yaml:"dumpLocation"`
6466 DockerImage string `yaml:"dockerImage"`
6567 Connection Connection `yaml:"connection"`
68+ Source Source `yaml:"source"`
6669 Partial Partial `yaml:"partial"`
6770 ParallelJobs int `yaml:"parallelJobs"`
6871 Restore * DirectRestore `yaml:"restore,omitempty"`
6972}
7073
74+ // Source describes source of data to dump.
75+ type Source struct {
76+ Type string `yaml:"type"`
77+ Connection Connection `yaml:"connection"`
78+ RDS * RDSConfig `yaml:"rds"`
79+ }
80+
81+ type dumpJobConfig struct {
82+ db Connection
83+ }
84+
85+ // dumper describes the interface to prepare environment for a logical dump.
86+ type dumper interface {
87+ // GetEnvVariables returns dumper environment variables.
88+ GetCmdEnvVariables () []string
89+
90+ // GetMounts returns dumper volume configurations for mounting.
91+ GetMounts () []mount.Mount
92+
93+ // SetConnectionOptions sets connection options for dumping.
94+ SetConnectionOptions (context.Context , * Connection ) error
95+ }
96+
7197// Connection provides connection options.
7298type Connection struct {
73- Type string `yaml:"type"`
7499 Host string `yaml:"host"`
75100 Port int `yaml:"port"`
76101 DBName string `yaml:"dbname"`
@@ -95,12 +120,16 @@ func NewDumpJob(cfg config.JobConfig, docker *client.Client, global *dblabCfg.Gl
95120 return nil , errors .Wrap (err , "failed to unmarshal configuration options" )
96121 }
97122
98- dumpJob .setDefaults ()
99-
100123 if err := dumpJob .validate (); err != nil {
101124 return nil , errors .Wrap (err , "invalid logical dump job" )
102125 }
103126
127+ dumpJob .setDefaults ()
128+
129+ if err := dumpJob .setupDumper (); err != nil {
130+ return nil , errors .Wrap (err , "failed to set up a dump helper" )
131+ }
132+
104133 return dumpJob , nil
105134}
106135
@@ -113,22 +142,51 @@ Either set 'numberOfJobs' equals to 1 or disable the restore section`)
113142 return nil
114143}
115144
116- // Name returns a name of the job.
117- func (d * DumpJob ) Name () string {
118- return d .name
119- }
120-
121145func (d * DumpJob ) setDefaults () {
122146 // TODO: Default yaml values in tags.
123- if d .DumpOptions .Connection .Port == 0 {
124- d .DumpOptions .Connection .Port = defaultPort
147+ if d .DumpOptions .Source . Connection .Port == 0 {
148+ d .DumpOptions .Source . Connection .Port = defaultPort
125149 }
126150
127- if d .DumpOptions .Connection .Username == "" {
128- d .DumpOptions .Connection .Username = defaultUsername
151+ if d .DumpOptions .Source .Connection .Username == "" {
152+ d .DumpOptions .Source .Connection .Username = defaultUsername
153+ }
154+
155+ if d .DumpOptions .ParallelJobs == 0 {
156+ d .DumpOptions .ParallelJobs = defaultParallelJobs
129157 }
130158}
131159
160+ // setupDumper sets up a tool to perform physical restoring.
161+ func (d * DumpJob ) setupDumper () error {
162+ switch d .Source .Type {
163+ case sourceTypeLocal , sourceTypeRemote , "" :
164+ d .dumper = newDefaultDumper ()
165+ return nil
166+
167+ case sourceTypeRDS :
168+ if d .Source .RDS == nil {
169+ return errors .New ("the RDS configuration section must not be empty when using the RDS source type" )
170+ }
171+
172+ dumper , err := newRDSDumper (d .Source .RDS )
173+ if err != nil {
174+ return errors .Wrap (err , "failed to create an RDS dumper" )
175+ }
176+
177+ d .dumper = dumper
178+
179+ return nil
180+ }
181+
182+ return errors .Errorf ("unknown source type given: %v" , d .Source .Type )
183+ }
184+
185+ // Name returns a name of the job.
186+ func (d * DumpJob ) Name () string {
187+ return d .name
188+ }
189+
132190// Run starts the job.
133191func (d * DumpJob ) Run (ctx context.Context ) error {
134192 log .Msg (fmt .Sprintf ("Run job: %s. Options: %v" , d .Name (), d .DumpOptions ))
@@ -160,6 +218,8 @@ func (d *DumpJob) Run(ctx context.Context) error {
160218 dumpContainerName ,
161219 )
162220 if err != nil {
221+ log .Err (err )
222+
163223 return errors .Wrap (err , "failed to create container" )
164224 }
165225
@@ -185,6 +245,10 @@ func (d *DumpJob) Run(ctx context.Context) error {
185245 return errors .Wrap (err , "failed to readiness check" )
186246 }
187247
248+ if err := d .setupConnectionOptions (ctx ); err != nil {
249+ return errors .Wrap (err , "failed to setup connection options" )
250+ }
251+
188252 dumpCommand := d .buildLogicalDumpCommand ()
189253 log .Msg ("Running dump command" , dumpCommand )
190254
@@ -233,6 +297,17 @@ func (d *DumpJob) Run(ctx context.Context) error {
233297 return nil
234298}
235299
300+ // setupConnectionOptions prepares connection options to perform a logical dump.
301+ func (d * DumpJob ) setupConnectionOptions (ctx context.Context ) error {
302+ d .config .db = d .DumpOptions .Source .Connection
303+
304+ if err := d .dumper .SetConnectionOptions (ctx , & d .config .db ); err != nil {
305+ return errors .Wrap (err , "failed to set connection options" )
306+ }
307+
308+ return nil
309+ }
310+
236311func (d * DumpJob ) performDumpCommand (ctx context.Context , cmdOutput io.Writer , commandID string ) error {
237312 execAttach , err := d .dockerClient .ContainerExecAttach (ctx , commandID , types.ExecStartCheck {})
238313 if err != nil {
@@ -275,9 +350,12 @@ func (d *DumpJob) getDumpContainerPath() string {
275350}
276351
277352func (d * DumpJob ) getEnvironmentVariables () []string {
278- envs := []string {"PGDATA=" + pgDataContainerDir }
353+ envs := []string {
354+ "PGDATA=" + pgDataContainerDir ,
355+ "POSTGRES_HOST_AUTH_METHOD=trust" ,
356+ }
279357
280- if d .DumpOptions .Connection .Type == connectionTypeLocal && d .DumpOptions .Connection .Port == defaultPort {
358+ if d .DumpOptions .Source .Type == sourceTypeLocal && d .DumpOptions . Source .Connection .Port == defaultPort {
281359 envs = append (envs , "PGPORT=" + strconv .Itoa (reservePort ))
282360 }
283361
@@ -294,33 +372,36 @@ func (d *DumpJob) getMountVolumes() []mount.Mount {
294372 },
295373 }
296374
297- if d .Connection . Type != connectionTypeLocal && d . DumpOptions .DumpFile != "" {
375+ if d .DumpOptions .DumpFile != "" {
298376 mounts = append (mounts , mount.Mount {
299377 Type : mount .TypeBind ,
300378 Source : filepath .Dir (d .DumpOptions .DumpFile ),
301379 Target : dumpContainerDir ,
302380 })
303381 }
304382
383+ // Add dump specific mounts.
384+ mounts = append (mounts , d .dumper .GetMounts ()... )
385+
305386 return mounts
306387}
307388
308389func (d * DumpJob ) getContainerNetworkMode () container.NetworkMode {
309390 networkMode := networkModeDefault
310391
311- if d .Connection .Type == connectionTypeLocal {
392+ if d .Source .Type == sourceTypeLocal {
312393 networkMode = networkModeHost
313394 }
314395
315396 return networkMode
316397}
317398
318399func (d * DumpJob ) getExecEnvironmentVariables () []string {
319- execEnvs := [] string {}
400+ execEnvs := d . dumper . GetCmdEnvVariables ()
320401
321- pgPassword := d .DumpOptions . Connection .Password
402+ pgPassword := d .config . db .Password
322403
323- if os .Getenv ("PGPASSWORD" ) != "" {
404+ if pgPassword == "" && os .Getenv ("PGPASSWORD" ) != "" {
324405 pgPassword = os .Getenv ("PGPASSWORD" )
325406 }
326407
@@ -335,10 +416,10 @@ func (d *DumpJob) buildLogicalDumpCommand() []string {
335416 dumpCmd := []string {"pg_dump" , "-C" }
336417
337418 optionalArgs := map [string ]string {
338- "-h" : d .DumpOptions . Connection .Host ,
339- "-p" : strconv .Itoa (d .DumpOptions . Connection .Port ),
340- "-U" : d .DumpOptions . Connection .Username ,
341- "-d" : d .DumpOptions . Connection .DBName ,
419+ "-h" : d .config . db .Host ,
420+ "-p" : strconv .Itoa (d .config . db .Port ),
421+ "-U" : d .config . db .Username ,
422+ "-d" : d .config . db .DBName ,
342423 "-j" : strconv .Itoa (d .DumpOptions .ParallelJobs ),
343424 }
344425 dumpCmd = append (dumpCmd , prepareCmdOptions (optionalArgs )... )
@@ -351,7 +432,7 @@ func (d *DumpJob) buildLogicalDumpCommand() []string {
351432 dumpCmd = append (dumpCmd , d .buildLogicalRestoreCommand ()... )
352433 cmd := strings .Join (dumpCmd , " " )
353434
354- log .Msg (cmd )
435+ log .Dbg (cmd )
355436
356437 return []string {"sh" , "-c" , cmd }
357438 }
0 commit comments