direct agent transfer fixes

Author: sureshanaparti

api/src/main/java/com/cloud/host/Status.java

@@ -127,6 +127,7 @@ public static String[] toStrings(Status... states) {
         s_fsm.addTransition(Status.Connecting, Event.HostDown, Status.Down);
         s_fsm.addTransition(Status.Connecting, Event.Ping, Status.Connecting);
         s_fsm.addTransition(Status.Connecting, Event.ManagementServerDown, Status.Disconnected);
+        s_fsm.addTransition(Status.Connecting, Event.StartAgentRebalance, Status.Rebalancing);
         s_fsm.addTransition(Status.Connecting, Event.AgentDisconnected, Status.Alert);
         s_fsm.addTransition(Status.Up, Event.PingTimeout, Status.Alert);
         s_fsm.addTransition(Status.Up, Event.AgentDisconnected, Status.Alert);

core/src/main/java/com/cloud/agent/api/TransferAgentCommand.java

@@ -25,6 +25,7 @@ public class TransferAgentCommand extends Command {
     protected long agentId;
     protected long futureOwner;
     protected long currentOwner;
+    protected boolean isConnectionTransfer;
     Event event;
 
     protected TransferAgentCommand() {
@@ -37,6 +38,11 @@ public TransferAgentCommand(long agentId, long currentOwner, long futureOwner, E
         this.event = event;
     }
 
+    public TransferAgentCommand(long agentId, long currentOwner, long futureOwner, Event event, boolean isConnectionTransfer) {
+        this(agentId, currentOwner, futureOwner, event);
+        this.isConnectionTransfer = isConnectionTransfer;
+    }
+
     public long getAgentId() {
         return agentId;
     }
@@ -53,6 +59,10 @@ public long getCurrentOwner() {
         return currentOwner;
     }
 
+    public boolean isConnectionTransfer() {
+        return isConnectionTransfer;
+    }
+
     @Override
     public boolean executeInSequence() {
         return false;

core/src/main/java/com/cloud/resource/ServerResource.java

@@ -50,6 +50,10 @@ public interface ServerResource extends Manager {
      */
     StartupCommand[] initialize();
 
+    default StartupCommand[] initialize(boolean isTransferredConnection) {
+        return initialize();
+    }
+
     /**
      * @param id id of the server to put in the PingCommand
      * @return PingCommand

engine/components-api/src/main/java/com/cloud/resource/ResourceManager.java

@@ -85,6 +85,8 @@ public interface ResourceManager extends ResourceService, Configurable {
 
     public Host createHostAndAgent(Long hostId, ServerResource resource, Map<String, String> details, boolean old, List<String> hostTags, boolean forRebalance);
 
+    public Host createHostAndAgent(Long hostId, ServerResource resource, Map<String, String> details, boolean old, List<String> hostTags, boolean forRebalance, boolean isTransferredConnection);
+
     public Host addHost(long zoneId, ServerResource resource, Type hostType, Map<String, String> hostDetails);
 
     public HostVO createHostVOForConnectedAgent(StartupCommand[] cmds);

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

@@ -772,6 +772,10 @@ public void rescan() {
     }
 
     protected boolean loadDirectlyConnectedHost(final HostVO host, final boolean forRebalance) {
+        return loadDirectlyConnectedHost(host, forRebalance, false);
+    }
+
+    protected boolean loadDirectlyConnectedHost(final HostVO host, final boolean forRebalance, final boolean isTransferredConnection) {
         boolean initialized = false;
         ServerResource resource = null;
         try {
@@ -800,7 +804,7 @@ protected boolean loadDirectlyConnectedHost(final HostVO host, final boolean for
 
         if (forRebalance) {
             tapLoadingAgents(host.getId(), TapAgentsAction.Add);
-            final Host h = _resourceMgr.createHostAndAgent(host.getId(), resource, host.getDetails(), false, null, true);
+            final Host h = _resourceMgr.createHostAndAgent(host.getId(), resource, host.getDetails(), false, null, true, isTransferredConnection);
             tapLoadingAgents(host.getId(), TapAgentsAction.Del);
 
             return h == null ? false : true;

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

@@ -756,12 +756,17 @@ public void removeAgent(final AgentAttache attache, final Status nextState) {
 
     @Override
     public boolean executeRebalanceRequest(final long agentId, final long currentOwnerId, final long futureOwnerId, final Event event) throws AgentUnavailableException, OperationTimedoutException {
+        return executeRebalanceRequest(agentId, currentOwnerId, futureOwnerId, event, false);
+    }
+
+    @Override
+    public boolean executeRebalanceRequest(final long agentId, final long currentOwnerId, final long futureOwnerId, final Event event, boolean isConnectionTransfer) throws AgentUnavailableException, OperationTimedoutException {
         boolean result = false;
         if (event == Event.RequestAgentRebalance) {
             return setToWaitForRebalance(agentId, currentOwnerId, futureOwnerId);
         } else if (event == Event.StartAgentRebalance) {
             try {
-                result = rebalanceHost(agentId, currentOwnerId, futureOwnerId);
+                result = rebalanceHost(agentId, currentOwnerId, futureOwnerId, isConnectionTransfer);
             } catch (final Exception e) {
                 logger.warn("Unable to rebalance host id={}",  agentId, e);
             }
@@ -883,7 +888,11 @@ public void startRebalanceAgents() {
     }
 
     private Answer[] sendRebalanceCommand(final long peer, final long agentId, final long currentOwnerId, final long futureOwnerId, final Event event) {
-        final TransferAgentCommand transfer = new TransferAgentCommand(agentId, currentOwnerId, futureOwnerId, event);
+        return sendRebalanceCommand(peer, agentId, currentOwnerId, futureOwnerId, event, false);
+    }
+
+    private Answer[] sendRebalanceCommand(final long peer, final long agentId, final long currentOwnerId, final long futureOwnerId, final Event event, final boolean isConnectionTransfer) {
+        final TransferAgentCommand transfer = new TransferAgentCommand(agentId, currentOwnerId, futureOwnerId, event, isConnectionTransfer);
         final Commands commands = new Commands(Command.OnError.Stop);
         commands.addCommand(transfer);
 
@@ -1016,7 +1025,10 @@ private boolean setToWaitForRebalance(final long hostId, final long currentOwner
     }
 
     protected boolean rebalanceHost(final long hostId, final long currentOwnerId, final long futureOwnerId) throws AgentUnavailableException {
+        return rebalanceHost(hostId, currentOwnerId, futureOwnerId, false);
+    }
 
+    protected boolean rebalanceHost(final long hostId, final long currentOwnerId, final long futureOwnerId, final boolean isConnectionTransfer) throws AgentUnavailableException {
         boolean result = true;
         if (currentOwnerId == _nodeId) {
             if (!startRebalance(hostId)) {
@@ -1025,7 +1037,7 @@ protected boolean rebalanceHost(final long hostId, final long currentOwnerId, fi
                 return false;
             }
             try {
-                final Answer[] answer = sendRebalanceCommand(futureOwnerId, hostId, currentOwnerId, futureOwnerId, Event.StartAgentRebalance);
+                final Answer[] answer = sendRebalanceCommand(futureOwnerId, hostId, currentOwnerId, futureOwnerId, Event.StartAgentRebalance, isConnectionTransfer);
                 if (answer == null || !answer[0].getResult()) {
                     result = false;
                 }
@@ -1055,7 +1067,7 @@ protected boolean rebalanceHost(final long hostId, final long currentOwnerId, fi
 
                 if (result) {
                     logger.debug("Loading directly connected host {}({}) to the management server {} as a part of rebalance process", host.getId(), host.getName(), _nodeId);
-                    result = loadDirectlyConnectedHost(host, true);
+                    result = loadDirectlyConnectedHost(host, true, isConnectionTransfer);
                 } else {
                     logger.warn("Failed to disconnect {}({}) as a part of rebalance process without notification" + host.getId(), host.getName());
                 }
@@ -1260,10 +1272,10 @@ public String dispatch(final ClusterServicePdu pdu) {
             } else if (cmds.length == 1 && cmds[0] instanceof TransferAgentCommand) {
                 final TransferAgentCommand cmd = (TransferAgentCommand)cmds[0];
 
-                logger.debug("Intercepting command for agent rebalancing: agent {} event: {}", cmd.getAgentId(), cmd.getEvent());
+                logger.debug("Intercepting command for agent rebalancing: agent: {}, event: {}, connection transfer: {}", cmd.getAgentId(), cmd.getEvent(), cmd.isConnectionTransfer());
                 boolean result = false;
                 try {
-                    result = rebalanceAgent(cmd.getAgentId(), cmd.getEvent(), cmd.getCurrentOwner(), cmd.getFutureOwner());
+                    result = rebalanceAgent(cmd.getAgentId(), cmd.getEvent(), cmd.getCurrentOwner(), cmd.getFutureOwner(), cmd.isConnectionTransfer());
                     logger.debug("Result is {}", result);
 
                 } catch (final AgentUnavailableException e) {
@@ -1419,7 +1431,7 @@ public boolean transferDirectAgentsFromMS(String fromMsUuid, long fromMsId, long
                     _mshostCounter++;
 
                     _hostTransferDao.startAgentTransfering(host.getId(), fromMsId, msHost.getMsid());
-                    if (!rebalanceAgent(host.getId(), Event.StartAgentRebalance, fromMsId, msHost.getMsid())) {
+                    if (!rebalanceAgent(host.getId(), Event.StartAgentRebalance, fromMsId, msHost.getMsid(), true)) {
                         agentTransferFailedCount++;
                     } else {
                         updateLastManagementServer(host.getId(), fromMsId);
@@ -1511,6 +1523,10 @@ public boolean rebalanceAgent(final long agentId, final Event event, final long
         return executeRebalanceRequest(agentId, currentOwnerId, futureOwnerId, event);
     }
 
+    public boolean rebalanceAgent(final long agentId, final Event event, final long currentOwnerId, final long futureOwnerId, boolean isConnectionTransfer) throws AgentUnavailableException, OperationTimedoutException {
+        return executeRebalanceRequest(agentId, currentOwnerId, futureOwnerId, event, isConnectionTransfer);
+    }
+
     public boolean isAgentRebalanceEnabled() {
         return EnableLB.value();
     }

engine/orchestration/src/main/java/com/cloud/cluster/ClusteredAgentRebalanceService.java

@@ -27,4 +27,5 @@ public interface ClusteredAgentRebalanceService {
 
     boolean executeRebalanceRequest(long agentId, long currentOwnerId, long futureOwnerId, Event event) throws AgentUnavailableException, OperationTimedoutException;
 
+    boolean executeRebalanceRequest(long agentId, long currentOwnerId, long futureOwnerId, Event event, boolean isConnectionTransfer) throws AgentUnavailableException, OperationTimedoutException;
 }

plugins/hypervisors/vmware/src/main/java/com/cloud/hypervisor/vmware/resource/VmwareResource.java

@@ -6036,6 +6036,11 @@ public Type getType() {
 
     @Override
     public StartupCommand[] initialize() {
+        return initialize(false);
+    }
+
+    @Override
+    public StartupCommand[] initialize(boolean isTransferredConnection) {
         try {
             String hostApiVersion = "4.1";
             VmwareContext context = getServiceContext();
@@ -6064,6 +6069,7 @@ public StartupCommand[] initialize() {
             cmd.setHypervisorType(HypervisorType.VMware);
             cmd.setCluster(_cluster);
             cmd.setHypervisorVersion(hostApiVersion);
+            cmd.setConnectionTransferred(isTransferredConnection);
 
             List<StartupStorageCommand> storageCmds = initializeLocalStorage();
             StartupCommand[] answerCmds = new StartupCommand[1 + storageCmds.size()];

plugins/hypervisors/xenserver/src/main/java/com/cloud/hypervisor/xenserver/resource/CitrixResourceBase.java

@@ -3723,6 +3723,11 @@ public String handleVmStartFailure(final Connection conn, final String vmName, f
 
     @Override
     public StartupCommand[] initialize() throws IllegalArgumentException {
+        return initialize(false);
+    }
+
+    @Override
+    public StartupCommand[] initialize(boolean isTransferredConnection) throws IllegalArgumentException {
         final Connection conn = getConnection();
         if (!getHostInfo(conn)) {
             logger.warn("Unable to get host information for " + _host.getIp());
@@ -3733,6 +3738,7 @@ public StartupCommand[] initialize() throws IllegalArgumentException {
         cmd.setHypervisorType(HypervisorType.XenServer);
         cmd.setCluster(_cluster);
         cmd.setPoolSync(false);
+        cmd.setConnectionTransferred(isTransferredConnection);
 
         try {
             final Pool pool = Pool.getByUuid(conn, _host.getPool());

plugins/hypervisors/xenserver/src/main/java/com/cloud/hypervisor/xenserver/resource/XenServer56Resource.java

@@ -128,7 +128,12 @@ public Boolean checkHeartbeat(final String hostuuid) {
 
     @Override
     public StartupCommand[] initialize() {
+        return initialize(false);
+    }
+
+    @Override
+    public StartupCommand[] initialize(boolean isTransferredConnection) {
         pingXAPI();
-        return super.initialize();
+        return super.initialize(isTransferredConnection);
     }
 }