make agent send retriable for agent peer relocation; avoid stop command on liveMigrate of VM when agent connect fails before starting migrate

Author: Glover, Rene (rg9975)

engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java

@@ -204,6 +204,8 @@ public class AgentManagerImpl extends ManagerBase implements AgentManager, Handl
             "Default size for DirectAgentPool", false);
     protected final ConfigKey<Float> DirectAgentThreadCap = new ConfigKey<Float>("Advanced", Float.class, "direct.agent.thread.cap", "1",
             "Percentage (as a value between 0 and 1) of direct.agent.pool.size to be used as upper thread cap for a single direct agent to process requests", false);
+        protected final ConfigKey<Integer> PeerNotFoundRetryAttempts = new ConfigKey<Integer>("Advanced", Integer.class, "agent.peer.not.found.retry.attempts", "1",
+            "Number of retry attempts for agent sends when 'Unable to find peer' is encountered", true);
     protected final ConfigKey<Boolean> CheckTxnBeforeSending = new ConfigKey<Boolean>("Developer", Boolean.class, "check.txn.before.sending.agent.commands", "false",
             "This parameter allows developers to enable a check to see if a transaction wraps commands that are sent to the resource.  This is not to be enabled on production systems.", true);
 
@@ -421,6 +423,14 @@ private Command[] checkForCommandsAndTag(final Commands commands) {
         return cmds;
     }
 
+    private boolean isPeerNotFoundException(final AgentUnavailableException exception) {
+        return exception != null && "Unable to find peer".equals(exception.getMessage());
+    }
+
+    private int getPeerNotFoundRetryAttempts() {
+        return PeerNotFoundRetryAttempts.value();
+    }
+
     /**
      * @param commands
      * @param cmds
@@ -452,19 +462,34 @@ public Answer[] send(final Long hostId, final Commands commands, int timeout) th
         }
 
         final Command[] cmds = checkForCommandsAndTag(commands);
+        int retry = 0;
+        final int maxRetries = getPeerNotFoundRetryAttempts();
+        while (true) {
+            //check what agent is returned.
+            final AgentAttache agent = getAttache(hostId);
+            if (agent == null || agent.isClosed()) {
+                throw new AgentUnavailableException("agent not logged into this management server", hostId);
+            }
 
-        //check what agent is returned.
-        final AgentAttache agent = getAttache(hostId);
-        if (agent == null || agent.isClosed()) {
-            throw new AgentUnavailableException("agent not logged into this management server", hostId);
+            final Request req = new Request(hostId, agent.getName(), _nodeId, cmds, commands.stopOnError(), true);
+            req.setSequence(agent.getNextSequence());
+            try {
+                final Answer[] answers = agent.send(req, timeout);
+                notifyAnswersToMonitors(hostId, req.getSequence(), answers);
+                commands.setAnswers(answers);
+                return answers;
+            } catch (final AgentUnavailableException e) {
+                if (isPeerNotFoundException(e) && retry < maxRetries) {
+                    retry++;
+                    if (s_logger.isDebugEnabled()) {
+                        s_logger.debug(String.format("Retrying send to host %s due to transient peer mismatch (%d/%d)", hostId, retry,
+                                maxRetries));
+                    }
+                    continue;
+                }
+                throw e;
+            }
         }
-
-        final Request req = new Request(hostId, agent.getName(), _nodeId, cmds, commands.stopOnError(), true);
-        req.setSequence(agent.getNextSequence());
-        final Answer[] answers = agent.send(req, timeout);
-        notifyAnswersToMonitors(hostId, req.getSequence(), answers);
-        commands.setAnswers(answers);
-        return answers;
     }
 
     protected Status investigate(final AgentAttache agent) {
@@ -502,18 +527,32 @@ protected AgentAttache getAttache(final Long hostId) throws AgentUnavailableExce
 
     @Override
     public long send(final Long hostId, final Commands commands, final Listener listener) throws AgentUnavailableException {
-        final AgentAttache agent = getAttache(hostId);
-        if (agent.isClosed()) {
-            throw new AgentUnavailableException("Agent " + agent.getId() + " is closed", agent.getId());
-        }
-
         final Command[] cmds = checkForCommandsAndTag(commands);
+        int retry = 0;
+        final int maxRetries = getPeerNotFoundRetryAttempts();
+        while (true) {
+            final AgentAttache agent = getAttache(hostId);
+            if (agent.isClosed()) {
+                throw new AgentUnavailableException("Agent " + agent.getId() + " is closed", agent.getId());
+            }
 
-        final Request req = new Request(hostId, agent.getName(), _nodeId, cmds, commands.stopOnError(), true);
-        req.setSequence(agent.getNextSequence());
-
-        agent.send(req, listener);
-        return req.getSequence();
+            final Request req = new Request(hostId, agent.getName(), _nodeId, cmds, commands.stopOnError(), true);
+            req.setSequence(agent.getNextSequence());
+            try {
+                agent.send(req, listener);
+                return req.getSequence();
+            } catch (final AgentUnavailableException e) {
+                if (isPeerNotFoundException(e) && retry < maxRetries) {
+                    retry++;
+                    if (s_logger.isDebugEnabled()) {
+                        s_logger.debug(String.format("Retrying async send to host %s due to transient peer mismatch (%d/%d)", hostId, retry,
+                                maxRetries));
+                    }
+                    continue;
+                }
+                throw e;
+            }
+        }
     }
 
     public void removeAgent(final AgentAttache attache, final Status nextState) {

engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java

@@ -2792,6 +2792,7 @@ protected void migrate(final VMInstanceVO vm, final long srcHostId, final Deploy
 
         boolean migrated = false;
         Map<String, DpdkTO> dpdkInterfaceMapping = new HashMap<>();
+        boolean agentUnavailable = false;
         try {
             final MigrateCommand mc = buildMigrateCommand(vm, to, dest, pfma, dpdkInterfaceMapping);
 
@@ -2807,6 +2808,8 @@ protected void migrate(final VMInstanceVO vm, final long srcHostId, final Deploy
                     _haMgr.scheduleRestart(vm, true);
                 }
                 throw new AgentUnavailableException("Operation timed out on migrating " + vm, dstHostId);
+            } catch (AgentUnavailableException e) {
+                agentUnavailable = true;
             }
 
             try {
@@ -2842,7 +2845,11 @@ protected void migrate(final VMInstanceVO vm, final long srcHostId, final Deploy
                         "Unable to migrate vm " + vm.getInstanceName() + " from host " + fromHost.getName() + " in zone " + dest.getDataCenter().getName() + " and pod " +
                                 dest.getPod().getName(), "Migrate Command failed.  Please check logs.");
                 try {
-                    _agentMgr.send(dstHostId, new Commands(cleanup(vm, dpdkInterfaceMapping)), null);
+                    // if migrate didn't start (i.e. we couldn't send the command to the agent at all), don't do the
+                    // cleanup/stop action as that causes an outage
+                    if (!agentUnavailable) {
+                        _agentMgr.send(dstHostId, new Commands(cleanup(vm, dpdkInterfaceMapping)), null);
+                    }
                 } catch (final AgentUnavailableException ae) {
                     s_logger.warn("Looks like the destination Host is unavailable for cleanup", ae);
                 }