feat: add a storage pool rotation (#217)

Author: agneum

configs/config.example.logical_generic.yml

@@ -97,7 +97,7 @@ provision:
   # Custom parameters for clone containers, see
   # https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources
   containerConfig:
-    "shm-size": 256MB
+    "shm-size": 1gb
 
 # Data retrieval flow. This section defines both initial retrieval, and rules
 # to keep the data directory in a synchronized state with the source. Both are optional:

configs/config.example.logical_rds_iam.yml

@@ -96,7 +96,7 @@ provision:
 
   # Custom parameters for clone containers (https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources).
   containerConfig:
-    "shm-size": 256MB
+    "shm-size": 1gb
 
 # Data retrieval flow. This section defines both initial retrieval, and rules
 # to keep the data directory in a synchronized state with the source. Both are optional:

configs/config.example.physical_generic.yml

@@ -96,7 +96,7 @@ provision:
 
   # Custom parameters for clone containers (https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources).
   containerConfig:
-    "shm-size": 256MB
+    "shm-size": 1gb
 
 # Data retrieval flow. This section defines both initial retrieval, and rules
 # to keep the data directory in a synchronized state with the source. Both are optional:

configs/config.example.physical_walg.yml

@@ -96,7 +96,7 @@ provision:
 
   # Custom parameters for clone containers (https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources).
   containerConfig:
-    "shm-size": 256MB
+    "shm-size": 1gb
 
 # Data retrieval flow. This section defines both initial retrieval, and rules
 # to keep the data directory in a synchronized state with the source. Both are optional:

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

@@ -285,7 +285,7 @@ func (r *RestoreJob) retrieveDataStateAt(ctx context.Context, contID string) (st
 	return dataStateAt, nil
 }
 
-// updateDataStateAt updates dataStateAt for in-memory representation of a filesystem pool.
+// updateDataStateAt updates dataStateAt for in-memory representation of a storage pool.
 func (r *RestoreJob) updateDataStateAt() {
 	dsaTime, err := time.Parse(util.DataStateAtFormat, r.dbMark.DataStateAt)
 	if err != nil {

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

@@ -671,7 +671,7 @@ func (p *PhysicalInitial) markDatabaseData() error {
 	return p.dbMarker.SaveConfig(p.dbMark)
 }
 
-// updateDataStateAt updates dataStateAt for in-memory representation of a filesystem pool.
+// updateDataStateAt updates dataStateAt for in-memory representation of a storage pool.
 func (p *PhysicalInitial) updateDataStateAt() {
 	dsaTime, err := time.Parse(util.DataStateAtFormat, p.dbMark.DataStateAt)
 	if err != nil {

pkg/retrieval/retrieval.go

@@ -240,7 +240,7 @@ func (r *Retrieval) refreshFunc(ctx context.Context) func() {
 	}
 }
 
-// fullRefresh makes a full refresh for an old filesystem pool.
+// fullRefresh performs full refresh for an unused storage pool and makes it active.
 func (r *Retrieval) fullRefresh(ctx context.Context) error {
 	// Stop previous runs and snapshot schedulers.
 	if r.ctxCancel != nil {
@@ -249,13 +249,18 @@ func (r *Retrieval) fullRefresh(ctx context.Context) error {
 
 	runCtx, cancel := context.WithCancel(ctx)
 	r.ctxCancel = cancel
-	poolToUpdate := r.poolManager.Oldest()
+	elementToUpdate := r.poolManager.GetPoolToUpdate()
 
-	if poolToUpdate == nil {
-		log.Msg("Pool to a full refresh not found. Skip refreshing.")
+	if elementToUpdate == nil || elementToUpdate.Value == nil {
+		log.Msg("Pool to perform full refresh not found. Skip refreshing")
 		return nil
 	}
 
+	poolToUpdate, err := r.poolManager.GetFSManager(elementToUpdate.Value.(string))
+	if err != nil {
+		return errors.Wrap(err, "failed to get FSManager")
+	}
+
 	log.Msg("Pool to a full refresh: ", poolToUpdate.Pool())
 
 	if err := preparePoolToRefresh(poolToUpdate); err != nil {
@@ -269,14 +274,11 @@ func (r *Retrieval) fullRefresh(ctx context.Context) error {
 		return cleanUpErr
 	}
 
-	current := r.poolManager.Active()
-
 	if err := r.run(runCtx, poolToUpdate); err != nil {
 		return err
 	}
 
-	r.poolManager.SetOldest(current)
-	r.poolManager.SetActive(poolToUpdate)
+	r.poolManager.SetActive(elementToUpdate)
 
 	return nil
 }

pkg/services/provision/pool/pool_manager.go

@@ -5,6 +5,7 @@
 package pool
 
 import (
+	"container/list"
 	"io/ioutil"
 	"os"
 	"path"
@@ -34,9 +35,8 @@ const (
 type Manager struct {
 	cfg              *Config
 	mu               *sync.Mutex
+	fsManagerList    *list.List
 	fsManagerPool    map[string]FSManager
-	fsManager        FSManager
-	oldFsManager     FSManager
 	runner           runners.Runner
 	blockDeviceTypes map[string]string
 }
@@ -58,6 +58,7 @@ func NewPoolManager(cfg *Config, runner runners.Runner) *Manager {
 		fsManagerPool:    make(map[string]FSManager),
 		runner:           runner,
 		blockDeviceTypes: make(map[string]string),
+		fsManagerList:    list.New(),
 	}
 }
 
@@ -68,24 +69,56 @@ func (pm *Manager) Reload(cfg Config) error {
 	return pm.ReloadPools()
 }
 
-// Active returns the active filesystem pool manager.
-func (pm *Manager) Active() FSManager {
-	return pm.fsManager
+// SetActive sets a new active pool manager element.
+func (pm *Manager) SetActive(element *list.Element) {
+	pm.fsManagerList.MoveToFront(element)
 }
 
-// SetActive sets a new active pool manager.
-func (pm *Manager) SetActive(active FSManager) {
-	pm.fsManager = active
+// Active returns the active storage pool manager.
+func (pm *Manager) Active() FSManager {
+	active := pm.fsManagerList.Front()
+
+	if active == nil || active.Value == nil {
+		return nil
+	}
+
+	return pm.getFSManager(active.Value.(string))
 }
 
-// Oldest returns the oldest filesystem pool manager.
-func (pm *Manager) Oldest() FSManager {
-	return pm.oldFsManager
+func (pm *Manager) getFSManager(pool string) FSManager {
+	pm.mu.Lock()
+	fsm := pm.fsManagerPool[pool]
+	pm.mu.Unlock()
+
+	return fsm
 }
 
-// SetOldest sets a pool manager to update.
-func (pm *Manager) SetOldest(pool FSManager) {
-	pm.oldFsManager = pool
+// GetPoolToUpdate returns the element to update.
+func (pm *Manager) GetPoolToUpdate() *list.Element {
+	for element := pm.fsManagerList.Back(); element != nil; element = element.Prev() {
+		if element.Value == nil {
+			return nil
+		}
+
+		// The active pool cannot be updated as it leads to downtime.
+		if element == pm.fsManagerList.Front() {
+			return nil
+		}
+
+		fsm := pm.getFSManager(element.Value.(string))
+
+		clones, err := fsm.ListClonesNames()
+		if err != nil {
+			log.Err("failed to list clones", err)
+			return nil
+		}
+
+		if len(clones) == 0 {
+			return element
+		}
+	}
+
+	return nil
 }
 
 // GetFSManager returns a filesystem manager by name if exists.
@@ -130,24 +163,14 @@ func (pm *Manager) ReloadPools() error {
 	fsPools := pm.examineEntries(entries)
 
 	if len(fsPools) == 0 {
-		return errors.New("no available filesystem pools")
-	}
-
-	active, old := pm.detectWorkingPools(fsPools)
-
-	if active == nil {
-		return errors.New("active pool not found: make sure it exists")
+		return errors.New("no available pools")
 	}
 
 	pm.mu.Lock()
-
 	pm.fsManagerPool = fsPools
-	pm.SetActive(active)
-	pm.SetOldest(old)
-
 	pm.mu.Unlock()
 
-	log.Msg("Available FS pools: ", pm.describeAvailablePools())
+	log.Msg("Available storage pools: ", pm.describeAvailablePools())
 	log.Msg("Active pool: ", pm.Active().Pool().Name)
 
 	return nil
@@ -208,6 +231,13 @@ func (pm *Manager) examineEntries(entries []os.FileInfo) map[string]FSManager {
 
 		// TODO(akartasov): extract pool name.
 		fsManagers[entry.Name()] = fsm
+
+		if pm.Active() == nil || pm.Active().Pool().DSA.Before(pool.DSA) {
+			pm.fsManagerList.PushFront(fsm.Pool().Name)
+			continue
+		}
+
+		pm.fsManagerList.PushBack(fsm.Pool().Name)
 	}
 
 	return fsManagers
@@ -226,33 +256,6 @@ func (pm *Manager) reloadBlockDevices() error {
 	return nil
 }
 
-func (pm *Manager) detectWorkingPools(fsm map[string]FSManager) (FSManager, FSManager) {
-	var fsManager, old FSManager
-
-	for _, manager := range fsm {
-		if fsManager == nil {
-			fsManager = manager
-			continue
-		}
-
-		if fsManager.Pool().DSA.Before(manager.Pool().DSA) {
-			if old == nil {
-				old = fsManager
-			}
-
-			fsManager = manager
-
-			continue
-		}
-
-		if old == nil || manager.Pool().DSA.Before(old.Pool().DSA) {
-			old = manager
-		}
-	}
-
-	return fsManager, old
-}
-
 func extractDataStateAt(dataPath string) (*time.Time, error) {
 	marker := dbmarker.NewMarker(dataPath)
 
@@ -294,13 +297,14 @@ func (pm *Manager) getFSInfo(path string) (string, error) {
 func (pm *Manager) describeAvailablePools() []string {
 	availablePools := []string{}
 
-	pm.mu.Lock()
+	for el := pm.fsManagerList.Front(); el != nil; el = el.Next() {
+		if el.Value == nil {
+			log.Err("empty element: skip listing")
+			continue
+		}
 
-	for _, fsm := range pm.fsManagerPool {
-		availablePools = append(availablePools, fsm.Pool().DataDir())
+		availablePools = append(availablePools, el.Value.(string))
 	}
 
-	pm.mu.Unlock()
-
 	return availablePools
 }