Skip to content

Commit 57efcff

Browse files
agneumagneum
committed
feat: provide the DLE container name to control container labels in order to determine which DLE a sync instance belongs to:
* find a DLE container name * create a struct to simplify internal Engine properties management * provide the DLE container name to control container labels
1 parent 50da1e0 commit 57efcff

File tree

14 files changed

+101
-48
lines changed

14 files changed

+101
-48
lines changed

‎cmd/database-lab/main.go‎

Lines changed: 34 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,10 @@ package main
1010

1111
import (
1212
"context"
13+
"fmt"
1314
"os"
1415
"os/signal"
16+
"strings"
1517
"syscall"
1618
"time"
1719

@@ -69,18 +71,19 @@ func main() {
6971
ctx, cancel := context.WithCancel(context.Background())
7072
defer cancel()
7173

72-
hostname := os.Getenv("HOSTNAME")
73-
if hostname == "" {
74-
log.Err("hostname is empty")
74+
engProps, err := getEngineProperties(ctx, dockerCLI, cfg.Global.InstanceID)
75+
if err != nil {
76+
log.Err("failed to get Database Lab Engine properties:", err.Error())
77+
return
7578
}
7679

77-
internalNetworkID, err := networks.Setup(ctx, dockerCLI, cfg.Global.InstanceID, hostname)
80+
internalNetworkID, err := networks.Setup(ctx, dockerCLI, cfg.Global.InstanceID, engProps.ContainerName)
7881
if err != nil {
7982
log.Errf(err.Error())
8083
return
8184
}
8285

83-
defer networks.Stop(dockerCLI, internalNetworkID, hostname)
86+
defer networks.Stop(dockerCLI, internalNetworkID, engProps.ContainerName)
8487

8588
// Create a platform service to make requests to Platform.
8689
platformSvc, err := platform.New(ctx, cfg.Platform)
@@ -100,17 +103,17 @@ func main() {
100103
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout)
101104
defer shutdownCancel()
102105

103-
shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, cfg.Global, pm.First().Pool())
106+
shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, engProps, pm.First().Pool())
104107
}
105108

106-
tm, err := telemetry.New(cfg.Global)
109+
tm, err := telemetry.New(cfg.Global, engProps)
107110
if err != nil {
108111
log.Errf(errors.WithMessage(err, "failed to initialize a telemetry service").Error())
109112
return
110113
}
111114

112115
// Create a new retrieval service to prepare a data directory and start snapshotting.
113-
retrievalSvc := retrieval.New(cfg, dockerCLI, pm, tm, runner)
116+
retrievalSvc := retrieval.New(cfg, engProps, dockerCLI, pm, tm, runner)
114117

115118
if err := retrievalSvc.Run(ctx); err != nil {
116119
log.Err("Failed to run the data retrieval service:", err)
@@ -149,7 +152,7 @@ func main() {
149152
Restore: retrievalSvc.CollectRestoreTelemetry(),
150153
})
151154

152-
server := srv.NewServer(&cfg.Server, &cfg.Global, cloningSvc, retrievalSvc, platformSvc, dockerCLI, obs, est, pm, tm)
155+
server := srv.NewServer(&cfg.Server, &cfg.Global, engProps, cloningSvc, retrievalSvc, platformSvc, dockerCLI, obs, est, pm, tm)
153156
shutdownCh := setShutdownListener()
154157

155158
go setReloadListener(ctx, provisionSvc, tm, retrievalSvc, pm, cloningSvc, platformSvc, est, server)
@@ -173,11 +176,30 @@ func main() {
173176
log.Msg(err)
174177
}
175178

176-
shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, cfg.Global, pm.First().Pool())
179+
shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, engProps, pm.First().Pool())
177180
cloningSvc.SaveClonesState()
178181
tm.SendEvent(ctx, telemetry.EngineStoppedEvent, telemetry.EngineStopped{Uptime: server.Uptime()})
179182
}
180183

184+
func getEngineProperties(ctx context.Context, dockerCLI *client.Client, instanceID string) (global.EngineProps, error) {
185+
hostname := os.Getenv("HOSTNAME")
186+
if hostname == "" {
187+
return global.EngineProps{}, errors.New("hostname is empty")
188+
}
189+
190+
dleContainer, err := dockerCLI.ContainerInspect(ctx, hostname)
191+
if err != nil {
192+
return global.EngineProps{}, fmt.Errorf("failed to inspect DLE container: %w", err)
193+
}
194+
195+
engProps := global.EngineProps{
196+
InstanceID: instanceID,
197+
ContainerName: strings.Trim(dleContainer.Name, "/"),
198+
}
199+
200+
return engProps, nil
201+
}
202+
181203
func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *telemetry.Agent, retrievalSvc *retrieval.Retrieval,
182204
pm *pool.Manager, cloningSvc *cloning.Base, platformSvc *platform.Service, est *estimator.Estimator, server *srv.Server) error {
183205
cfg, err := config.LoadConfiguration()
@@ -241,10 +263,10 @@ func setShutdownListener() chan os.Signal {
241263
return c
242264
}
243265

244-
func shutdownDatabaseLabEngine(ctx context.Context, dockerCLI *client.Client, global global.Config, fsp *resources.Pool) {
266+
func shutdownDatabaseLabEngine(ctx context.Context, dockerCLI *client.Client, engProps global.EngineProps, fsp *resources.Pool) {
245267
log.Msg("Stopping control containers")
246268

247-
if err := cont.StopControlContainers(ctx, dockerCLI, global.InstanceID, fsp.DataDir()); err != nil {
269+
if err := cont.StopControlContainers(ctx, dockerCLI, engProps.InstanceID, fsp.DataDir()); err != nil {
248270
log.Err("Failed to stop control containers", err)
249271
}
250272

‎pkg/config/global/config.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,3 +47,9 @@ type Telemetry struct {
4747
Enabled bool `yaml:"enabled"`
4848
URL string `yaml:"url"`
4949
}
50+
51+
// EngineProps contains internal Database Lab Engine properties.
52+
type EngineProps struct {
53+
InstanceID string
54+
ContainerName string
55+
}

‎pkg/retrieval/engine/engine.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,11 @@ import (
1616
)
1717

1818
// JobBuilder provides a new job builder.
19-
func JobBuilder(globalCfg *global.Config, cloneManager pool.FSManager, tm *telemetry.Agent) (components.JobBuilder, error) {
19+
func JobBuilder(globalCfg *global.Config, engineProps global.EngineProps, cloneManager pool.FSManager,
20+
tm *telemetry.Agent) (components.JobBuilder, error) {
2021
switch globalCfg.Engine {
2122
case postgres.EngineType:
22-
return postgres.NewJobBuilder(globalCfg, cloneManager, tm), nil
23+
return postgres.NewJobBuilder(globalCfg, engineProps, cloneManager, tm), nil
2324

2425
default:
2526
return nil, errors.New("failed to get engine")

‎pkg/retrieval/engine/postgres/job_builder.go‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,15 @@ const (
2727
type JobBuilder struct {
2828
cloneManager pool.FSManager
2929
globalCfg *global.Config
30+
engineProps global.EngineProps
3031
tm *telemetry.Agent
3132
}
3233

3334
// NewJobBuilder create a new job builder.
34-
func NewJobBuilder(global *global.Config, cm pool.FSManager, tm *telemetry.Agent) *JobBuilder {
35+
func NewJobBuilder(global *global.Config, engineProps global.EngineProps, cm pool.FSManager, tm *telemetry.Agent) *JobBuilder {
3536
return &JobBuilder{
3637
globalCfg: global,
38+
engineProps: engineProps,
3739
cloneManager: cm,
3840
tm: tm,
3941
}
@@ -43,19 +45,19 @@ func NewJobBuilder(global *global.Config, cm pool.FSManager, tm *telemetry.Agent
4345
func (s *JobBuilder) BuildJob(jobCfg config.JobConfig) (components.JobRunner, error) {
4446
switch jobCfg.Spec.Name {
4547
case logical.DumpJobType:
46-
return logical.NewDumpJob(jobCfg, s.globalCfg)
48+
return logical.NewDumpJob(jobCfg, s.globalCfg, s.engineProps)
4749

4850
case logical.RestoreJobType:
49-
return logical.NewJob(jobCfg, s.globalCfg)
51+
return logical.NewJob(jobCfg, s.globalCfg, s.engineProps)
5052

5153
case physical.RestoreJobType:
52-
return physical.NewJob(jobCfg, s.globalCfg)
54+
return physical.NewJob(jobCfg, s.globalCfg, s.engineProps)
5355

5456
case snapshot.LogicalSnapshotType:
55-
return snapshot.NewLogicalInitialJob(jobCfg, s.globalCfg, s.cloneManager, s.tm)
57+
return snapshot.NewLogicalInitialJob(jobCfg, s.globalCfg, s.engineProps, s.cloneManager, s.tm)
5658

5759
case snapshot.PhysicalSnapshotType:
58-
return snapshot.NewPhysicalInitialJob(jobCfg, s.globalCfg, s.cloneManager, s.tm)
60+
return snapshot.NewPhysicalInitialJob(jobCfg, s.globalCfg, s.engineProps, s.cloneManager, s.tm)
5961
}
6062

6163
return nil, errors.Errorf("unknown job type: %q", jobCfg.Spec.Name)

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

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ type DumpJob struct {
6868
dockerClient *client.Client
6969
fsPool *resources.Pool
7070
globalCfg *global.Config
71+
engineProps global.EngineProps
7172
config dumpJobConfig
7273
dumper dumper
7374
dbMarker *dbmarker.Marker
@@ -135,12 +136,13 @@ type ImmediateRestore struct {
135136
}
136137

137138
// NewDumpJob creates a new DumpJob.
138-
func NewDumpJob(jobCfg config.JobConfig, global *global.Config) (*DumpJob, error) {
139+
func NewDumpJob(jobCfg config.JobConfig, global *global.Config, engineProps global.EngineProps) (*DumpJob, error) {
139140
dumpJob := &DumpJob{
140141
name: jobCfg.Spec.Name,
141142
dockerClient: jobCfg.Docker,
142143
fsPool: jobCfg.FSPool,
143144
globalCfg: global,
145+
engineProps: engineProps,
144146
dbMarker: jobCfg.Marker,
145147
dbMark: &dbmarker.Config{
146148
DataType: dbmarker.LogicalDataType,
@@ -208,7 +210,7 @@ func (d *DumpJob) setupDumper() error {
208210
}
209211

210212
func (d *DumpJob) dumpContainerName() string {
211-
return dumpContainerPrefix + d.globalCfg.InstanceID
213+
return dumpContainerPrefix + d.engineProps.InstanceID
212214
}
213215

214216
// Name returns a name of the job.
@@ -563,7 +565,8 @@ func (d *DumpJob) buildContainerConfig(password string) *container.Config {
563565
return &container.Config{
564566
Labels: map[string]string{
565567
cont.DBLabControlLabel: cont.DBLabDumpLabel,
566-
cont.DBLabInstanceIDLabel: d.globalCfg.InstanceID,
568+
cont.DBLabInstanceIDLabel: d.engineProps.InstanceID,
569+
cont.DBLabEngineNameLabel: d.engineProps.ContainerName,
567570
},
568571
Env: d.getEnvironmentVariables(password),
569572
Image: d.DockerImage,

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

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ type RestoreJob struct {
8484
dockerClient *client.Client
8585
fsPool *resources.Pool
8686
globalCfg *global.Config
87+
engineProps global.EngineProps
8788
dbMarker *dbmarker.Marker
8889
dbMark *dbmarker.Config
8990
isDumpLocationDir bool
@@ -107,12 +108,13 @@ type Partial struct {
107108
}
108109

109110
// NewJob create a new logical restore job.
110-
func NewJob(cfg config.JobConfig, global *global.Config) (*RestoreJob, error) {
111+
func NewJob(cfg config.JobConfig, global *global.Config, engineProps global.EngineProps) (*RestoreJob, error) {
111112
restoreJob := &RestoreJob{
112113
name: cfg.Spec.Name,
113114
dockerClient: cfg.Docker,
114115
fsPool: cfg.FSPool,
115116
globalCfg: global,
117+
engineProps: engineProps,
116118
dbMarker: cfg.Marker,
117119
dbMark: &dbmarker.Config{DataType: dbmarker.LogicalDataType},
118120
}
@@ -134,7 +136,7 @@ func (r *RestoreJob) setDefaults() {
134136
}
135137

136138
func (r *RestoreJob) restoreContainerName() string {
137-
return restoreContainerPrefix + r.globalCfg.InstanceID
139+
return restoreContainerPrefix + r.engineProps.InstanceID
138140
}
139141

140142
// Name returns a name of the job.
@@ -585,7 +587,8 @@ func (r *RestoreJob) buildContainerConfig(password string) *container.Config {
585587
return &container.Config{
586588
Labels: map[string]string{
587589
cont.DBLabControlLabel: cont.DBLabRestoreLabel,
588-
cont.DBLabInstanceIDLabel: r.globalCfg.InstanceID,
590+
cont.DBLabInstanceIDLabel: r.engineProps.InstanceID,
591+
cont.DBLabEngineNameLabel: r.engineProps.ContainerName,
589592
},
590593
Env: append(os.Environ(), []string{
591594
"PGDATA=" + r.fsPool.DataDir(),

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

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ type RestoreJob struct {
5858
dockerClient *client.Client
5959
fsPool *resources.Pool
6060
globalCfg *global.Config
61+
engineProps global.EngineProps
6162
dbMarker *dbmarker.Marker
6263
restorer restorer
6364
CopyOptions
@@ -98,11 +99,12 @@ type restorer interface {
9899
}
99100

100101
// NewJob creates a new physical restore job.
101-
func NewJob(cfg config.JobConfig, global *global.Config) (*RestoreJob, error) {
102+
func NewJob(cfg config.JobConfig, global *global.Config, engineProps global.EngineProps) (*RestoreJob, error) {
102103
physicalJob := &RestoreJob{
103104
name: cfg.Spec.Name,
104105
dockerClient: cfg.Docker,
105106
globalCfg: global,
107+
engineProps: engineProps,
106108
dbMarker: cfg.Marker,
107109
fsPool: cfg.FSPool,
108110
}
@@ -135,7 +137,7 @@ func (r *RestoreJob) getRestorer(tool string) (restorer, error) {
135137
}
136138

137139
func (r *RestoreJob) restoreContainerName() string {
138-
return restoreContainerPrefix + r.globalCfg.InstanceID
140+
return restoreContainerPrefix + r.engineProps.InstanceID
139141
}
140142

141143
// Name returns a name of the job.
@@ -290,7 +292,7 @@ func (r *RestoreJob) startContainer(ctx context.Context, containerName string, c
290292
}
291293

292294
func (r *RestoreJob) syncInstanceName() string {
293-
return cont.SyncInstanceContainerPrefix + r.globalCfg.InstanceID
295+
return cont.SyncInstanceContainerPrefix + r.engineProps.InstanceID
294296
}
295297

296298
func (r *RestoreJob) runSyncInstance(ctx context.Context) (err error) {
@@ -373,7 +375,8 @@ func (r *RestoreJob) buildContainerConfig(label, password string) *container.Con
373375
return &container.Config{
374376
Labels: map[string]string{
375377
cont.DBLabControlLabel: label,
376-
cont.DBLabInstanceIDLabel: r.globalCfg.InstanceID,
378+
cont.DBLabInstanceIDLabel: r.engineProps.InstanceID,
379+
cont.DBLabEngineNameLabel: r.engineProps.ContainerName,
377380
},
378381
Env: r.getEnvironmentVariables(password),
379382
Image: r.CopyOptions.DockerImage,

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

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ type LogicalInitial struct {
4747
dockerClient *client.Client
4848
options LogicalOptions
4949
globalCfg *global.Config
50+
engineProps global.EngineProps
5051
dbMarker *dbmarker.Marker
5152
queryProcessor *queryProcessor
5253
}
@@ -67,14 +68,15 @@ type DataPatching struct {
6768
}
6869

6970
// NewLogicalInitialJob creates a new logical initial job.
70-
func NewLogicalInitialJob(cfg config.JobConfig, global *global.Config, cloneManager pool.FSManager,
71+
func NewLogicalInitialJob(cfg config.JobConfig, global *global.Config, engineProps global.EngineProps, cloneManager pool.FSManager,
7172
tm *telemetry.Agent) (*LogicalInitial, error) {
7273
li := &LogicalInitial{
7374
name: cfg.Spec.Name,
7475
cloneManager: cloneManager,
7576
fsPool: cfg.FSPool,
7677
dockerClient: cfg.Docker,
7778
globalCfg: global,
79+
engineProps: engineProps,
7880
dbMarker: cfg.Marker,
7981
tm: tm,
8082
}
@@ -99,7 +101,7 @@ func (s *LogicalInitial) Name() string {
99101

100102
// patchContainerName returns container name.
101103
func (s *LogicalInitial) patchContainerName() string {
102-
return patchContainerPrefix + s.globalCfg.InstanceID
104+
return patchContainerPrefix + s.engineProps.InstanceID
103105
}
104106

105107
// Reload reloads job configuration.
@@ -237,7 +239,8 @@ func (s *LogicalInitial) buildContainerConfig(clonePath, patchImage, password st
237239
return &container.Config{
238240
Labels: map[string]string{
239241
cont.DBLabControlLabel: cont.DBLabPatchLabel,
240-
cont.DBLabInstanceIDLabel: s.globalCfg.InstanceID,
242+
cont.DBLabInstanceIDLabel: s.engineProps.InstanceID,
243+
cont.DBLabEngineNameLabel: s.engineProps.ContainerName,
241244
},
242245
Env: []string{
243246
"PGDATA=" + clonePath,

0 commit comments

Comments
 (0)