refactor(engine): redesign the Retrieval service (#375)

* redefine public interface of the data retrieval service
* parse jobs into target groups
* add new retrieval states (`renewed` and `snapshotting`) to describe a new refresh model
* configure jobs depending on filesystem manager and detach them to run independently
* prepare a basis for jobs validation system
* combine calls of all available jobs using `RefreshData()` and `SnapshotData()`

Author: agneum

engine/cmd/database-lab/main.go

@@ -113,7 +113,11 @@ func main() {
 	}
 
 	// Create a new retrieval service to prepare a data directory and start snapshotting.
-	retrievalSvc := retrieval.New(cfg, engProps, docker, pm, tm, runner)
+	retrievalSvc, err := retrieval.New(cfg, engProps, docker, pm, tm, runner)
+	if err != nil {
+		log.Errf(errors.WithMessage(err, `error in the "retrieval" section of the config`).Error())
+		return
+	}
 
 	// Create a cloning service to provision new clones.
 	provisioner, err := provision.New(ctx, &cfg.Provision, dbCfg, docker, pm, engProps.InstanceID, internalNetworkID)
@@ -146,7 +150,7 @@ func main() {
 		EngineVersion: version.GetVersion(),
 		DBVersion:     provisioner.DetectDBVersion(),
 		Pools:         pm.CollectPoolStat(),
-		Restore:       retrievalSvc.CollectRestoreTelemetry(),
+		Restore:       retrievalSvc.ReportState(),
 	})
 
 	embeddedUI := embeddedui.New(cfg.EmbeddedUI, engProps, runner, docker)
@@ -233,7 +237,8 @@ func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *
 		return err
 	}
 
-	if err := retrieval.IsValidConfig(cfg); err != nil {
+	newRetrievalConfig, err := retrieval.ValidateConfig(&cfg.Retrieval)
+	if err != nil {
 		return err
 	}
 
@@ -257,7 +262,7 @@ func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *
 
 	provisionSvc.Reload(cfg.Provision, dbCfg)
 	tm.Reload(cfg.Global)
-	retrievalSvc.Reload(ctx, cfg)
+	retrievalSvc.Reload(ctx, newRetrievalConfig)
 	cloningSvc.Reload(cfg.Cloning)
 	platformSvc.Reload(newPlatformSvc)
 	est.Reload(cfg.Estimator)

engine/internal/provision/pool/pool_manager.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -346,6 +347,11 @@ func (pm *Manager) examineEntries(entries []os.DirEntry) (map[string]FSManager,
 			continue
 		}
 
+		if err := preparePoolDirectories(fsm); err != nil {
+			log.Msg("failed to prepare pool directories:", err.Error())
+			continue
+		}
+
 		fsm.RefreshSnapshotList()
 
 		fsManagers[pool.Name] = fsm
@@ -398,6 +404,22 @@ func extractDataStateAt(dataPath string) (*time.Time, error) {
 	return &dsa, nil
 }
 
+func preparePoolDirectories(fsm FSManager) error {
+	dataDir := fsm.Pool().DataDir()
+	if err := os.MkdirAll(dataDir, 0700); err != nil {
+		return err
+	}
+
+	return filepath.Walk(dataDir, func(name string, info os.FileInfo, err error) error {
+		if err == nil {
+			// PGDATA dir permissions must be 0700 to avoid errors.
+			err = os.Chmod(name, 0700)
+		}
+
+		return err
+	})
+}
+
 func (pm *Manager) describeAvailablePools() []string {
 	availablePools := []string{}