PBM-1057: Forbidden running Backup & PITR on non-suitable agents (#1001)

* Add ListStadyAgents for fetching agents which are ready for Backup and PITR
* Use only steady agents for backup nomination
* Extract and expand backup candidates rules
* Add PITR candidates rules logic
* Extract MaxReplicationLagTimeSec to defs package
* Expand agent's stat with replication lag
* Add replication lag guard for Backup and PITR candidates

Author: boris-ilijic

cmd/pbm-agent/agent.go

@@ -330,6 +330,13 @@ func (a *Agent) HbStatus(ctx context.Context) {
 			} else {
 				hb.State = n.State
 				hb.StateStr = n.StateStr
+
+				rLag, err := topo.ReplicationLag(ctx, a.nodeConn, a.brief.Me)
+				if err != nil {
+					l.Error("get replication lag: %v", err)
+					hb.Err += fmt.Sprintf("get replication lag: %v", err)
+				}
+				hb.ReplicationLag = rLag
 			}
 		}
 

cmd/pbm-agent/backup.go

@@ -165,22 +165,15 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 			}
 		}
 
-		agents, err := topo.ListAgentStatuses(ctx, a.leadConn)
+		agents, err := topo.ListSteadyAgents(ctx, a.leadConn)
 		if err != nil {
 			l.Error("get agents list: %v", err)
 			return
 		}
 
-		validCandidates := make([]topo.AgentStat, 0, len(agents))
-		for _, s := range agents {
-			if version.FeatureSupport(s.MongoVersion()).BackupType(cmd.Type) != nil {
-				continue
-			}
-
-			validCandidates = append(validCandidates, s)
-		}
+		candidates := a.getValidCandidates(agents, cmd.Type)
 
-		nodes := prio.CalcNodesPriority(c, cfg.Backup.Priority, validCandidates)
+		nodes := prio.CalcNodesPriority(c, cfg.Backup.Priority, candidates)
 
 		shards, err := topo.ClusterMembers(ctx, a.leadConn.MongoClient())
 		if err != nil {
@@ -256,6 +249,19 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 	}
 }
 
+// getValidCandidates filters out all agents that are not suitable for the backup.
+func (a *Agent) getValidCandidates(agents []topo.AgentStat, backupType defs.BackupType) []topo.AgentStat {
+	validCandidates := []topo.AgentStat{}
+	for _, agent := range agents {
+		if version.FeatureSupport(agent.MongoVersion()).BackupType(backupType) != nil {
+			continue
+		}
+		validCandidates = append(validCandidates, agent)
+	}
+
+	return validCandidates
+}
+
 const renominationFrame = 5 * time.Second
 
 func (a *Agent) nominateRS(ctx context.Context, bcp, rs string, nodes [][]string) error {

cmd/pbm-agent/pitr.go

@@ -412,13 +412,13 @@ func (a *Agent) leadNomination(
 		return
 	}
 
-	agents, err := topo.ListAgentStatuses(ctx, a.leadConn)
+	candidates, err := topo.ListSteadyAgents(ctx, a.leadConn)
 	if err != nil {
 		l.Error("get agents list: %v", err)
 		return
 	}
 
-	nodes := prio.CalcNodesPriority(nil, cfgPrio, agents)
+	nodes := prio.CalcNodesPriority(nil, cfgPrio, candidates)
 
 	shards, err := topo.ClusterMembers(ctx, a.leadConn.MongoClient())
 	if err != nil {
@@ -433,7 +433,7 @@ func (a *Agent) leadNomination(
 		return
 	}
 
-	err = a.reconcileReadyStatus(ctx, agents)
+	err = a.reconcileReadyStatus(ctx, candidates)
 	if err != nil {
 		l.Error("reconciling ready status: %v", err)
 		return
@@ -645,7 +645,7 @@ func (a *Agent) reconcileReadyStatus(ctx context.Context, agents []topo.AgentSta
 			if err := oplog.SetClusterStatus(ctx, a.leadConn, oplog.StatusUnset); err != nil {
 				l.Error("error while cleaning cluster status: %v", err)
 			}
-			return errors.New("timeout while roconciling ready status")
+			return errors.New("timeout while reconciling ready status")
 		}
 	}
 }

pbm/defs/defs.go

@@ -136,6 +136,8 @@ const (
 
 const StaleFrameSec uint32 = 30
 
+const MaxReplicationLagTimeSec = 21
+
 const (
 	// MetadataFileSuffix is a suffix for the metadata file on a storage
 	MetadataFileSuffix = ".pbm.json"

pbm/topo/agent.go

@@ -45,6 +45,9 @@ type AgentStat struct {
 	// DelaySecs is the node configured replication delay (lag).
 	DelaySecs int32 `bson:"delay"`
 
+	// Replication lag for mongod.
+	ReplicationLag int `bson:"repl_lag"`
+
 	// AgentVer has the PBM Agent version (looks like `v2.3.4`)
 	AgentVer string `bson:"v"`
 
@@ -192,6 +195,31 @@ func ListAgentStatuses(ctx context.Context, m connect.Client) ([]AgentStat, erro
 	return ListAgents(ctx, m)
 }
 
+// ListSteadyAgents returns agents which are in steady state for backup or PITR.
+func ListSteadyAgents(ctx context.Context, m connect.Client) ([]AgentStat, error) {
+	agents, err := ListAgentStatuses(ctx, m)
+	if err != nil {
+		return nil, errors.Wrap(err, "listing agents")
+	}
+	steadyAgents := []AgentStat{}
+	for _, a := range agents {
+		if a.State != defs.NodeStatePrimary &&
+			a.State != defs.NodeStateSecondary {
+			continue
+		}
+		if a.Arbiter || a.DelaySecs > 0 {
+			continue
+		}
+		if a.ReplicationLag >= defs.MaxReplicationLagTimeSec {
+			continue
+		}
+
+		steadyAgents = append(steadyAgents, a)
+	}
+
+	return steadyAgents, nil
+}
+
 func ListAgents(ctx context.Context, m connect.Client) ([]AgentStat, error) {
 	cur, err := m.AgentsStatusCollection().Find(ctx, bson.D{})
 	if err != nil {

pbm/topo/topo.go

@@ -148,8 +148,6 @@ func collectTopoCheckErrors(
 	return nil
 }
 
-const maxReplicationLagTimeSec = 21
-
 // NodeSuits checks if node can perform backup
 func NodeSuits(ctx context.Context, m *mongo.Client, inf *NodeInfo) (bool, error) {
 	status, err := GetNodeStatus(ctx, m, inf.Me)
@@ -165,7 +163,7 @@ func NodeSuits(ctx context.Context, m *mongo.Client, inf *NodeInfo) (bool, error
 		return false, errors.Wrap(err, "get node replication lag")
 	}
 
-	return replLag < maxReplicationLagTimeSec && status.Health == defs.NodeHealthUp &&
+	return replLag < defs.MaxReplicationLagTimeSec && status.Health == defs.NodeHealthUp &&
 			(status.State == defs.NodeStatePrimary || status.State == defs.NodeStateSecondary),
 		nil
 }