Merge branch '303-disable-physical-refresh' into 'master'

fix: disable full-refresh by default in physical mode (#303)

Closes #303

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

Author: agneum

cmd/database-lab/main.go

@@ -48,6 +48,7 @@ func main() {
 	}
 
 	log.Msg("Database Lab Instance ID:", cfg.Global.InstanceID)
+	log.Msg("Database Lab Engine version:", version.GetVersion())
 
 	if cfg.Server.VerificationToken == "" {
 		log.Warn("Verification Token is empty. Database Lab Engine is insecure")

pkg/models/retrieval.go

@@ -16,6 +16,8 @@ const (
 	Physical RetrievalMode = "physical"
 	// Logical defines logical retrieval mode.
 	Logical RetrievalMode = "logical"
+	// Unknown defines the case when retrieval mode is unknown or is not set.
+	Unknown RetrievalMode = "unknown"
 )
 
 // RefreshStatus defines status of refreshing data.

pkg/retrieval/retrieval.go

@@ -62,7 +62,7 @@ type Scheduler struct {
 
 // New creates a new data retrieval.
 func New(cfg *dblabCfg.Config, docker *client.Client, pm *pool.Manager, tm *telemetry.Agent, runner runners.Runner) *Retrieval {
-	return &Retrieval{
+	r := &Retrieval{
 		cfg:         &cfg.Retrieval,
 		global:      &cfg.Global,
 		docker:      docker,
@@ -74,6 +74,20 @@ func New(cfg *dblabCfg.Config, docker *client.Client, pm *pool.Manager, tm *tele
 			Status: models.Inactive,
 		},
 	}
+
+	for _, jobName := range r.cfg.Jobs {
+		jobSpec, ok := r.cfg.JobsSpec[jobName]
+		if !ok {
+			continue
+		}
+
+		jobSpec.Name = jobName
+		r.jobSpecs[jobName] = jobSpec
+	}
+
+	r.defineRetrievalMode()
+
+	return r
 }
 
 // Reload reloads retrieval configuration.
@@ -100,7 +114,9 @@ func (r *Retrieval) Run(ctx context.Context) error {
 	runCtx, cancel := context.WithCancel(ctx)
 	r.ctxCancel = cancel
 
-	fsManager, err := r.getPoolToDataRefresh()
+	log.Msg("Retrieval mode:", r.State.Mode)
+
+	fsManager, err := r.getPoolToDataRetrieving()
 	if err != nil {
 		var skipError *SkipRefreshingError
 		if errors.As(err, &skipError) {
@@ -118,7 +134,7 @@ func (r *Retrieval) Run(ctx context.Context) error {
 		return fmt.Errorf("failed to choose pool to refresh: %w", err)
 	}
 
-	log.Msg("Pool to perform a full refresh: ", fsManager.Pool().Name)
+	log.Msg("Pool to perform data retrieving: ", fsManager.Pool().Name)
 
 	if err := r.run(runCtx, fsManager); err != nil {
 		r.tm.SendEvent(ctx, telemetry.AlertEvent, telemetry.Alert{Level: models.RefreshFailed, Message: err.Error()})
@@ -130,7 +146,7 @@ func (r *Retrieval) Run(ctx context.Context) error {
 	return nil
 }
 
-func (r *Retrieval) getPoolToDataRefresh() (pool.FSManager, error) {
+func (r *Retrieval) getPoolToDataRetrieving() (pool.FSManager, error) {
 	firstPool := r.poolManager.First()
 	if firstPool == nil {
 		return nil, errors.New("no available pools")
@@ -140,6 +156,12 @@ func (r *Retrieval) getPoolToDataRefresh() (pool.FSManager, error) {
 		return firstPool, nil
 	}
 
+	// For physical or unknown modes, changing the pool is possible only by the refresh timetable.
+	if r.State.Mode != models.Logical {
+		return firstPool, nil
+	}
+
+	// For logical mode try to find another pool to avoid rewriting prepared data.
 	elementToRefresh := r.poolManager.GetPoolToUpdate()
 
 	if elementToRefresh == nil || elementToRefresh.Value == nil {
@@ -220,8 +242,6 @@ func (r *Retrieval) configure(fsm pool.FSManager) error {
 		return errors.Wrap(err, "invalid data retrieval configuration")
 	}
 
-	r.defineRetrievalMode()
-
 	return nil
 }
 
@@ -237,14 +257,11 @@ func (r *Retrieval) parseJobs(fsm pool.FSManager) error {
 	r.jobs = make([]components.JobRunner, 0, len(r.cfg.Jobs))
 
 	for _, jobName := range r.cfg.Jobs {
-		jobSpec, ok := r.cfg.JobsSpec[jobName]
+		jobSpec, ok := r.jobSpecs[jobName]
 		if !ok {
 			return errors.Errorf("Job %q not found", jobName)
 		}
 
-		jobSpec.Name = jobName
-		r.jobSpecs[jobName] = jobSpec
-
 		jobCfg := config.JobConfig{
 			Spec:   jobSpec,
 			Docker: r.docker,
@@ -315,11 +332,15 @@ func (r *Retrieval) hasPhysicalJob() bool {
 func (r *Retrieval) defineRetrievalMode() {
 	if r.hasPhysicalJob() {
 		r.State.Mode = models.Physical
+		return
 	}
 
 	if r.hasLogicalJob() {
 		r.State.Mode = models.Logical
+		return
 	}
+
+	r.State.Mode = models.Unknown
 }
 
 func (r *Retrieval) prepareEnvironment(fsm pool.FSManager) error {