Merge pull request #1263 from jcechace/PBM-1665-pitr-backup-profile-storage

PBM-1665 keep PITR running and prioritize different node for profile backups

Author: jcechace

cmd/pbm-agent/backup.go

@@ -93,8 +93,10 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		}
 	}
 
-	if cmd.Type == defs.LogicalBackup {
-		// wakeup the slicer to not wait for the tick
+	if cmd.Type == defs.LogicalBackup && cmd.Profile == "" {
+		// For backups to the main storage,
+		// wake up the slicer to not wait for the tick.
+		// This will slice and pause the main PITR
 		go a.sliceNow(opid)
 	}
 
@@ -156,20 +158,25 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		// not replset. So an `incremental && not_base` backup should land on
 		// the agent that made a previous (src) backup.
 		const srcHostMultiplier = 3.0
-		var c map[string]float64
+		c := make(map[string]float64)
 		if cmd.Type == defs.IncrementalBackup && !cmd.IncrBase {
 			src, err := backup.LastIncrementalBackup(ctx, a.leadConn)
 			if err != nil {
 				// try backup anyway
 				l.Warning("define source backup: %v", err)
 			} else {
-				c = make(map[string]float64)
 				for _, rs := range src.Replsets {
 					c[rs.Node] = srcHostMultiplier
 				}
 			}
 		}
 
+		// When a logical backup targets an external profile (different storage),
+		// PITR keeps running. Deprioritize nodes currently running PITR slicer
+		if cmd.Type == defs.LogicalBackup && cmd.Profile != "" {
+			c = a.deprioritizePITRNodes(ctx, c, l)
+		}
+
 		agents, err := topo.ListSteadyAgents(ctx, a.leadConn)
 		if err != nil {
 			l.Error("get agents list: %v", err)
@@ -267,6 +274,41 @@ func (a *Agent) getValidCandidates(agents []topo.AgentStat, backupType defs.Back
 	return validCandidates
 }
 
+// deprioritizePITRNodes adds low-priority coefficients for nodes currently running PITR slicing.
+// It only modifies the coefficient map for nodes not already present (e.g., incremental src host).
+// Returns the (possibly modified) coefficient map.
+func (a *Agent) deprioritizePITRNodes(
+	ctx context.Context,
+	coefficients map[string]float64,
+	l log.LogEvent,
+) map[string]float64 {
+	pitrLocks, err := lock.GetOpLocks(ctx, a.leadConn, &lock.LockHeader{Type: ctrl.CmdPITR})
+	if err != nil {
+		l.Warning("get pitr locks for deprioritization: %v", err)
+		return coefficients
+	}
+
+	ts, err := topo.GetClusterTime(ctx, a.leadConn)
+	if err != nil {
+		l.Warning("get cluster time for pitr deprioritization: %v", err)
+		return coefficients
+	}
+
+	for i := range pitrLocks {
+		pl := &pitrLocks[i]
+		if pl.Heartbeat.T+defs.StaleFrameSec < ts.T {
+			continue // stale lock, ignore
+		}
+
+		// Only set if not already present (preserve previous priorities)
+		if _, exists := coefficients[pl.Node]; !exists {
+			coefficients[pl.Node] = prio.DefaultScore - 0.1
+		}
+	}
+
+	return coefficients
+}
+
 const renominationFrame = 5 * time.Second
 
 func (a *Agent) nominateRS(ctx context.Context, bcp, rs string, nodes [][]string) error {

pbm/prio/priority.go

@@ -9,9 +9,9 @@ import (
 )
 
 const (
-	defaultScore     = 1.0
-	scoreForPrimary  = defaultScore / 2
-	scoreForHidden   = defaultScore * 2
+	DefaultScore     = 1.0
+	scoreForPrimary  = DefaultScore / 2
+	scoreForHidden   = DefaultScore * 2
 	scoreForExcluded = 0
 )
 
@@ -114,23 +114,23 @@ func CalcPriorityForNode(node *topo.NodeInfo) float64 {
 		return scoreForHidden
 	}
 
-	return defaultScore
+	return DefaultScore
 }
 
 // implicitPrioCalc provides priority calculation based on topology rules.
 // Instead of using explicitly specified priority numbers, topology rules are
 // applied for primary, secondary and hidden member.
 func implicitPrioCalc(a *topo.AgentStat, rule map[string]float64) float64 {
 	if coeff, ok := rule[a.Node]; ok && rule != nil {
-		return defaultScore * coeff
+		return DefaultScore * coeff
 	} else if a.State == defs.NodeStatePrimary {
 		return scoreForPrimary
 	} else if a.DelaySecs > 0 {
 		return scoreForExcluded
 	} else if a.Hidden {
 		return scoreForHidden
 	}
-	return defaultScore
+	return DefaultScore
 }
 
 // explicitPrioCalc uses priority numbers from configuration to calculate
@@ -139,7 +139,7 @@ func implicitPrioCalc(a *topo.AgentStat, rule map[string]float64) float64 {
 func explicitPrioCalc(a *topo.AgentStat, rule map[string]float64) float64 {
 	sc, ok := rule[a.Node]
 	if !ok || sc < 0 {
-		return defaultScore
+		return DefaultScore
 	}
 
 	return sc

pbm/prio/priority_test.go

@@ -396,8 +396,8 @@ func TestCalcPriorityForNode(t *testing.T) {
 		}
 
 		p := CalcPriorityForNode(nodeInfo)
-		if p != defaultScore {
-			t.Errorf("wrong priority for secondary: want=%v, got=%v", defaultScore, p)
+		if p != DefaultScore {
+			t.Errorf("wrong priority for secondary: want=%v, got=%v", DefaultScore, p)
 		}
 	})
 
@@ -470,8 +470,8 @@ func TestImplicitPrioCalc(t *testing.T) {
 
 		p := implicitPrioCalc(agentStat, nil)
 
-		if p != defaultScore {
-			t.Errorf("wrong priority for secondary: want=%v, got=%v", defaultScore, p)
+		if p != DefaultScore {
+			t.Errorf("wrong priority for secondary: want=%v, got=%v", DefaultScore, p)
 		}
 	})