Add active lock check before running the backup

cmd/pbm-agent/backup.go

@@ -54,13 +54,24 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 	l := logger.NewEvent(string(ctrl.CmdBackup), cmd.Name, opid.String(), ep.TS())
 	ctx = log.SetLogEventToContext(ctx, l)
 
+	moveOn, err := a.startBcpLockCheck(ctx)
+	if err != nil {
+		l.Error("start backup lock check: %v", err)
+		return
+	}
+	if !moveOn {
+		l.Error("unable to proceed with the backup, active lock is present")
+		return
+	}
+
 	nodeInfo, err := topo.GetNodeInfoExt(ctx, a.nodeConn)
 	if err != nil {
 		l.Error("get node info: %v", err)
 		return
 	}
 
 	isClusterLeader := nodeInfo.IsClusterLeader()
+
 	canRunBackup, err := topo.NodeSuitsExt(ctx, a.nodeConn, nodeInfo, cmd.Type)
 	if err != nil {
 		l.Error("node check: %v", err)
@@ -310,3 +321,32 @@ func (a *Agent) waitNomination(ctx context.Context, bcp string) (bool, error) {
 		}
 	}
 }
+
+// startBcpLockCheck checks if there is any active lock.
+// It fetches all existing pbm locks, and if any exists, it is also
+// checked for staleness.
+// false is returned in case a single active lock exists or error happens.
+// true means that there's no active locks.
+func (a *Agent) startBcpLockCheck(ctx context.Context) (bool, error) {
+	locks, err := lock.GetLocks(ctx, a.leadConn, &lock.LockHeader{})
+	if err != nil {
+		return false, errors.Wrap(err, "get all locks for backup start")
+	}
+	if len(locks) == 0 {
+		return true, nil
+	}
+
+	// stale lock check
+	ts, err := topo.GetClusterTime(ctx, a.leadConn)
+	if err != nil {
+		return false, errors.Wrap(err, "read cluster time")
+	}
+
+	for _, l := range locks {
+		if l.Heartbeat.T+defs.StaleFrameSec >= ts.T {
+			return false, nil
+		}
+	}
+
+	return true, nil
+}