New feature: Reconcile commands (CopyCommand, MigrateCommand, MigrateVolumeCommand)

Author: weizhouapache

agent/src/main/java/com/cloud/agent/Agent.java

@@ -797,6 +797,9 @@ protected void processRequest(final Request request, final Link link) {
                         }
                         commandsInProgress.incrementAndGet();
                         try {
+                            if (cmd.isReconcile()) {
+                                cmd.setRequestSequence(request.getSequence());
+                            }
                             answer = serverResource.executeRequest(cmd);
                         } finally {
                             commandsInProgress.decrementAndGet();
@@ -1003,9 +1006,13 @@ public void processResponse(final Response response, final Link link) {
             for (final IAgentControlListener listener : controlListeners) {
                 listener.processControlResponse(response, (AgentControlAnswer)answer);
             }
-        } else if (answer instanceof PingAnswer && (((PingAnswer) answer).isSendStartup()) && reconnectAllowed) {
-            logger.info("Management server requested startup command to reinitialize the agent");
-            sendStartup(link);
+        } else if (answer instanceof PingAnswer) {
+            if (((PingAnswer) answer).isSendStartup() && reconnectAllowed) {
+                logger.info("Management server requested startup command to reinitialize the agent");
+                sendStartup(link);
+            } else {
+                serverResource.processPingAnswer((PingAnswer) answer);
+            }
         } else {
             updateLastPingResponseTime();
         }
@@ -1073,6 +1080,9 @@ public void processOtherTask(final Task task) {
             Answer answer = null;
             commandsInProgress.incrementAndGet();
             try {
+                if (command.isReconcile()) {
+                    command.setRequestSequence(req.getSequence());
+                }
                 answer = serverResource.executeRequest(command);
             } finally {
                 commandsInProgress.decrementAndGet();

api/src/main/java/com/cloud/agent/api/Command.java

@@ -35,13 +35,31 @@ public static enum OnError {
         Continue, Stop
     }
 
+    public enum State {
+        CREATED,        // Command is created by management server
+        STARTED,        // Command is started by agent
+        PROCESSING,     // Processing by agent
+        PROCESSING_IN_BACKEND,  // Processing in backend by agent
+        COMPLETED,      // Operation succeeds by agent or management server
+        FAILED,         // Operation fails by agent
+        RECONCILE_RETRY,        // Ready for retry of reconciliation
+        RECONCILING,    // Being reconciled by management server
+        RECONCILED,     // Reconciled by management server
+        RECONCILE_SKIPPED,  // Skip the reconciliation as the resource state is inconsistent with the command
+        RECONCILE_FAILED,       // Fail to reconcile by management server
+        TIMED_OUT,      // Timed out on management server or agent
+        INTERRUPTED,    // Interrupted by management server or agent (for example agent is restarted),
+        DANGLED_IN_BACKEND     // Backend process which cannot be processed normally (for example agent is restarted)
+    }
+
     public static final String HYPERVISOR_TYPE = "hypervisorType";
 
     // allow command to carry over hypervisor or other environment related context info
     @LogLevel(Log4jLevel.Trace)
     protected Map<String, String> contextMap = new HashMap<String, String>();
     private int wait;  //in second
     private boolean bypassHostMaintenance = false;
+    private transient long requestSequence = 0L;
 
     protected Command() {
         this.wait = 0;
@@ -82,6 +100,10 @@ public String getContextParam(String name) {
         return contextMap.get(name);
     }
 
+    public Map<String, String> getContextMap() {
+        return contextMap;
+    }
+
     public boolean allowCaching() {
         return true;
     }
@@ -94,6 +116,18 @@ public void setBypassHostMaintenance(boolean bypassHostMaintenance) {
         this.bypassHostMaintenance = bypassHostMaintenance;
     }
 
+    public boolean isReconcile() {
+        return false;
+    }
+
+    public long getRequestSequence() {
+        return requestSequence;
+    }
+
+    public void setRequestSequence(long requestSequence) {
+        this.requestSequence = requestSequence;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

api/src/main/java/com/cloud/agent/api/to/DiskTO.java

@@ -46,7 +46,7 @@ public class DiskTO {
     private Long diskSeq;
     private String path;
     private Volume.Type type;
-    private Map<String, String> _details;
+    private Map<String, String> details;
 
     public DiskTO() {
 
@@ -92,10 +92,10 @@ public void setType(Volume.Type type) {
     }
 
     public void setDetails(Map<String, String> details) {
-        _details = details;
+        this.details = details;
     }
 
     public Map<String, String> getDetails() {
-        return _details;
+        return details;
     }
 }

api/src/main/java/com/cloud/agent/api/to/NetworkTO.java

@@ -36,7 +36,7 @@ public class NetworkTO {
     protected TrafficType type;
     protected URI broadcastUri;
     protected URI isolationUri;
-    protected boolean isSecurityGroupEnabled;
+    protected boolean securityGroupEnabled;
     protected String name;
     protected String ip6address;
     protected String ip6gateway;
@@ -112,7 +112,7 @@ public String getName() {
     }
 
     public void setSecurityGroupEnabled(boolean enabled) {
-        this.isSecurityGroupEnabled = enabled;
+        this.securityGroupEnabled = enabled;
     }
 
     /**
@@ -221,7 +221,7 @@ public void setIsolationuri(URI isolationUri) {
     }
 
     public boolean isSecurityGroupEnabled() {
-        return this.isSecurityGroupEnabled;
+        return this.securityGroupEnabled;
     }
 
     public void setIp6Dns1(String ip6Dns1) {

api/src/main/java/com/cloud/agent/api/to/NicTO.java

@@ -86,6 +86,14 @@ public void setUuid(String uuid) {
         this.nicUuid = uuid;
     }
 
+    public String getNicUuid() {
+        return nicUuid;
+    }
+
+    public void setNicUuid(String nicUuid) {
+        this.nicUuid = nicUuid;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder("[Nic:").append(type).append("-").append(ip).append("-").append(broadcastUri).append("]").toString();

api/src/main/java/com/cloud/agent/api/to/VirtualMachineTO.java

@@ -61,7 +61,7 @@ public class VirtualMachineTO {
     @LogLevel(LogLevel.Log4jLevel.Off)
     String vncPassword;
     String vncAddr;
-    Map<String, String> params;
+    Map<String, String> details;
     String uuid;
     String bootType;
     String bootMode;
@@ -191,7 +191,11 @@ public Integer getMaxSpeed() {
         return maxSpeed;
     }
 
-    public boolean getLimitCpuUse() {
+    public boolean isEnableHA() {
+        return enableHA;
+    }
+
+    public boolean isLimitCpuUse() {
         return limitCpuUse;
     }
 
@@ -289,11 +293,11 @@ public void setVncAddr(String vncAddr) {
     }
 
     public Map<String, String> getDetails() {
-        return params;
+        return details;
     }
 
     public void setDetails(Map<String, String> params) {
-        this.params = params;
+        this.details = params;
     }
 
     public String getUuid() {

api/src/main/java/org/apache/cloudstack/command/ReconcileCommandService.java

@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.command;
+
+
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.Command;
+import com.cloud.hypervisor.Hypervisor;
+import org.apache.cloudstack.framework.config.ConfigKey;
+
+import java.util.Arrays;
+import java.util.List;
+
+public interface ReconcileCommandService {
+
+    ConfigKey<Boolean> ReconcileCommandsEnabled = new ConfigKey<>("Advanced", Boolean.class,
+            "reconcile.commands.enabled", "true",
+            "Indicates whether the background task to reconcile the commands is enabled or not",
+            false);
+
+    ConfigKey<Integer> ReconcileCommandsInterval = new ConfigKey<>("Advanced", Integer.class,
+            "reconcile.commands.interval", "60",
+            "Interval (in seconds) for the background task to reconcile the commands",
+            false);
+    ConfigKey<Integer> ReconcileCommandsMaxAttempts = new ConfigKey<>("Advanced", Integer.class,
+            "reconcile.commands.max.attempts", "30",
+            "The maximum number of attempts to reconcile the commands",
+            true);
+
+    ConfigKey<Integer> ReconcileCommandsWorkers = new ConfigKey<>("Advanced", Integer.class,
+            "reconcile.commands.workers", "100",
+            "The Number of worker threads to reconcile the commands",
+            false);
+
+    List<Hypervisor.HypervisorType> SupportedHypervisorTypes = Arrays.asList(Hypervisor.HypervisorType.KVM);
+
+    void persistReconcileCommands(Long hostId, Long requestSequence, Command[] cmd);
+
+    boolean updateReconcileCommand(long requestSeq, Command command, Answer answer, Command.State newStateByManagement, Command.State newStateByAgent);
+
+    void processCommand(Command pingCommand, Answer pingAnswer);
+
+    void processAnswers(long requestSeq, Command[] commands, Answer[] answers);
+
+    void updateReconcileCommandToInterruptedByManagementServerId(long managementServerId);
+
+    void updateReconcileCommandToInterruptedByHostId(long hostId);
+}

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

@@ -29,14 +29,14 @@
 
 public class MigrateCommand extends Command {
     private String vmName;
-    private String destIp;
+    private String destinationIp;
     private Map<String, MigrateDiskInfo> migrateStorage;
     private boolean migrateStorageManaged;
     private boolean migrateNonSharedInc;
     private boolean autoConvergence;
     private String hostGuid;
-    private boolean isWindows;
-    private VirtualMachineTO vmTO;
+    private boolean windows;
+    private VirtualMachineTO virtualMachine;
     private boolean executeInSequence = false;
     private List<MigrateDiskInfo> migrateDiskInfoList = new ArrayList<>();
     private Map<String, DpdkTO> dpdkInterfaceMapping = new HashMap<>();
@@ -64,11 +64,11 @@ public void setVlanToPersistenceMap(Map<String, Boolean> vlanToPersistenceMap) {
     protected MigrateCommand() {
     }
 
-    public MigrateCommand(String vmName, String destIp, boolean isWindows, VirtualMachineTO vmTO, boolean executeInSequence) {
+    public MigrateCommand(String vmName, String destinationIp, boolean windows, VirtualMachineTO virtualMachine, boolean executeInSequence) {
         this.vmName = vmName;
-        this.destIp = destIp;
-        this.isWindows = isWindows;
-        this.vmTO = vmTO;
+        this.destinationIp = destinationIp;
+        this.windows = windows;
+        this.virtualMachine = virtualMachine;
         this.executeInSequence = executeInSequence;
     }
 
@@ -105,15 +105,15 @@ public boolean isAutoConvergence() {
     }
 
     public boolean isWindows() {
-        return isWindows;
+        return windows;
     }
 
     public VirtualMachineTO getVirtualMachine() {
-        return vmTO;
+        return virtualMachine;
     }
 
     public String getDestinationIp() {
-        return destIp;
+        return destinationIp;
     }
 
     public String getVmName() {
@@ -233,4 +233,9 @@ public void setSourceDiskOnStorageFileSystem(boolean isDiskOnFileSystemStorage)
             this.isSourceDiskOnStorageFileSystem = isDiskOnFileSystemStorage;
         }
     }
+
+    @Override
+    public boolean isReconcile() {
+        return true;
+    }
 }

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

@@ -19,11 +19,16 @@
 
 package com.cloud.agent.api;
 
+import java.util.ArrayList;
+import java.util.List;
+
 public class PingAnswer extends Answer {
     private PingCommand _command = null;
 
     private boolean sendStartup = false;
 
+    private List<String> reconcileCommands = new ArrayList<>();
+
     protected PingAnswer() {
     }
 
@@ -44,4 +49,16 @@ public boolean isSendStartup() {
     public void setSendStartup(boolean sendStartup) {
         this.sendStartup = sendStartup;
     }
+
+    public List<String> getReconcileCommands() {
+        return reconcileCommands;
+    }
+
+    public void setReconcileCommands(List<String> reconcileCommands) {
+        this.reconcileCommands = reconcileCommands;
+    }
+
+    public void addReconcileCommand(String reconcileCommand) {
+        this.reconcileCommands.add(reconcileCommand);
+    }
 }

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

@@ -20,11 +20,14 @@
 package com.cloud.agent.api;
 
 import com.cloud.host.Host;
+import org.apache.cloudstack.command.CommandInfo;
 
 public class PingCommand extends Command {
     Host.Type hostType;
     long hostId;
     boolean outOfBand;
+    @LogLevel(LogLevel.Log4jLevel.Trace)
+    private CommandInfo[] commandInfos = new CommandInfo[] {};
 
     protected PingCommand() {
     }
@@ -78,4 +81,12 @@ public int hashCode() {
         result = 31 * result + (int) (hostId ^ (hostId >>> 32));
         return result;
     }
+
+    public CommandInfo[] getCommandInfos() {
+        return commandInfos;
+    }
+
+    public void setCommandInfos(CommandInfo[] commandInfos) {
+        this.commandInfos = commandInfos;
+    }
 }