kvm: Handle storage issue on NFS/KVM in multiple ways

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

@@ -29,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -77,6 +78,8 @@
 import com.cloud.agent.transport.Request;
 import com.cloud.agent.transport.Response;
 import com.cloud.alert.AlertManager;
+import com.cloud.cluster.ManagementServerHostVO;
+import com.cloud.cluster.dao.ManagementServerHostDao;
 import com.cloud.configuration.ManagementServiceConfiguration;
 import com.cloud.dc.ClusterVO;
 import com.cloud.dc.DataCenterVO;
@@ -105,6 +108,7 @@
 import com.cloud.utils.concurrency.NamedThreadFactory;
 import com.cloud.utils.db.DB;
 import com.cloud.utils.db.EntityManager;
+import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchCriteria.Op;
 import com.cloud.utils.db.TransactionLegacy;
@@ -121,6 +125,11 @@
 import com.cloud.utils.time.InaccurateClock;
 import com.google.common.base.Strings;
 
+import static com.cloud.configuration.ConfigurationManagerImpl.KVM_HEARTBEAT_FAILURE_ACTION;
+import static com.cloud.configuration.ConfigurationManagerImpl.KVM_HEARTBEAT_UPDATE_MAX_RETRIES;
+import static com.cloud.configuration.ConfigurationManagerImpl.KVM_HEARTBEAT_UPDATE_RETRY_SLEEP;
+import static com.cloud.configuration.ConfigurationManagerImpl.KVM_HEARTBEAT_UPDATE_TIMEOUT;
+
 /**
  * Implementation of the Agent Manager. This class controls the connection to the agents.
  **/
@@ -157,6 +166,8 @@ public class AgentManagerImpl extends ManagerBase implements AgentManager, Handl
  protected ConfigurationDao _configDao = null;
  @Inject
  protected ClusterDao _clusterDao = null;
+ @Inject
+ protected ManagementServerHostDao _msHostDao;
 
  @Inject
  protected HighAvailabilityManager _haMgr = null;
@@ -178,12 +189,16 @@ public class AgentManagerImpl extends ManagerBase implements AgentManager, Handl
  protected ScheduledExecutorService _directAgentExecutor;
  protected ScheduledExecutorService _cronJobExecutor;
  protected ScheduledExecutorService _monitorExecutor;
+ protected ScheduledExecutorService _scanHostsExecutor;
+ protected ScheduledExecutorService _investigatorExecutor;
 
  private int _directAgentThreadCap;
 
  protected StateMachine2<Status, Status.Event, Host> _statusStateMachine = Status.getStateMachine();
  private final ConcurrentHashMap<Long, Long> _pingMap = new ConcurrentHashMap<Long, Long>(10007);
 
+ private final ConcurrentHashMap<Long, ScheduledFuture<?>> _investigateTasksMap = new ConcurrentHashMap<>();
+
  @Inject
  ResourceManager _resourceMgr;
  @Inject
@@ -203,6 +218,14 @@ public class AgentManagerImpl extends ManagerBase implements AgentManager, Handl
  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);
 
+ protected final ConfigKey<Boolean> InvestigateDisconnectedHosts = new ConfigKey<>("Advanced", Boolean.class, "investigate.disconnected.hosts",
+ "false", "Determines whether to investigate VMs on disconnected hosts", false);
+ protected final ConfigKey<Integer> InvestigateDisconnectedHostsInterval = new ConfigKey<>("Advanced", Integer.class, "investigate.disconnected.hosts.interval",
+ "300", "The time (in seconds) between VM investigation on disconnected hosts.", false);
+ protected final ConfigKey<Integer> InvestigateDisconnectedHostsPoolSize = new ConfigKey<Integer>("Advanced", Integer.class, "investigate.disconnected.hosts.pool.size", "10",
+ "Default pool size to investigate disconnected hosts", false);
+
+
  @Override
  public boolean configure(final String name, final Map<String, Object> params) throws ConfigurationException {
 
@@ -238,6 +261,9 @@ public boolean configure(final String name, final Map<String, Object> params) th
 
  _monitorExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AgentMonitor"));
 
+ _scanHostsExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("HostsScanner"));
+ _investigatorExecutor = new ScheduledThreadPoolExecutor(InvestigateDisconnectedHostsPoolSize.value(), new NamedThreadFactory("DisconnectHostsInvestigator"));
+
  return true;
  }
 
@@ -614,6 +640,10 @@ public boolean start() {
 
  _monitorExecutor.scheduleWithFixedDelay(new MonitorTask(), mgmtServiceConf.getPingInterval(), mgmtServiceConf.getPingInterval(), TimeUnit.SECONDS);
 
+ if (InvestigateDisconnectedHosts.value()) {
+ _scanHostsExecutor.scheduleAtFixedRate(new ScanDisconnectedHostsTask(), 60, 60, TimeUnit.SECONDS);
+ }
+
  return true;
  }
 
@@ -773,6 +803,8 @@ public boolean stop() {
 
  _connectExecutor.shutdownNow();
  _monitorExecutor.shutdownNow();
+ _investigatorExecutor.shutdownNow();
+
  return true;
  }
 
@@ -1725,7 +1757,8 @@ public String getConfigComponentName() {
  @Override
  public ConfigKey<?>[] getConfigKeys() {
  return new ConfigKey<?>[] { CheckTxnBeforeSending, Workers, Port, Wait, AlertWait, DirectAgentLoadSize, DirectAgentPoolSize,
- DirectAgentThreadCap };
+ DirectAgentThreadCap,
+ InvestigateDisconnectedHosts, InvestigateDisconnectedHostsInterval, InvestigateDisconnectedHostsPoolSize };
  }
 
  protected class SetHostParamsListener implements Listener {
@@ -1758,8 +1791,11 @@ public void processConnect(final Host host, final StartupCommand cmd, final bool
  if (cmd instanceof StartupRoutingCommand) {
  if (((StartupRoutingCommand)cmd).getHypervisorType() == HypervisorType.KVM || ((StartupRoutingCommand)cmd).getHypervisorType() == HypervisorType.LXC) {
  Map<String, String> params = new HashMap<String, String>();
- params.put("router.aggregation.command.each.timeout", _configDao.getValue("router.aggregation.command.each.timeout"));
-
+ Arrays.asList(KVM_HEARTBEAT_UPDATE_MAX_RETRIES,
+ KVM_HEARTBEAT_UPDATE_RETRY_SLEEP,
+ KVM_HEARTBEAT_UPDATE_TIMEOUT,
+ KVM_HEARTBEAT_FAILURE_ACTION)
+ .forEach(c -> params.put(c, _configDao.getValue(c)));
  try {
  SetHostParamsCommand cmds = new SetHostParamsCommand(params);
  Commands c = new Commands(cmds);
@@ -1838,4 +1874,91 @@ public void propagateChangeToAgents(Map<String, String> params) {
  sendCommandToAgents(hostsPerZone, params);
  }
  }
+
+ protected class ScanDisconnectedHostsTask extends ManagedContextRunnable {
+
+ @Override
+ protected void runInContext() {
+ try {
+ ManagementServerHostVO msHost = _msHostDao.findOneInUpState(new Filter(ManagementServerHostVO.class, "id", true, 0L, 1L));
+ if (msHost == null || (msHost.getMsid() != _nodeId)) {
+ s_logger.debug("Skipping disconnected hosts scan task");
+ for (Long hostId : _investigateTasksMap.keySet()) {
+ cancelInvestigationTask(hostId);
+ }
+ return;
+ }
+ for (HostVO host : _hostDao.listByType(Host.Type.Routing)) {
+ if (host.getStatus() == Status.Disconnected) {
+ scheduleInvestigationTask(host.getId());
+ }
+ }
+ } catch (final Exception e) {
+ s_logger.error("Exception caught while scanning disconnected hosts : ", e);
+ }
+ }
+ }
+
+ protected class InvestigationTask extends ManagedContextRunnable {
+ Long _hostId;
+
+ InvestigationTask(final Long hostId) {
+ _hostId = hostId;
+ }
+
+ @Override
+ protected void runInContext() {
+ try {
+ final long hostId = _hostId;
+ s_logger.info("Investigating host " + hostId + " to determine its actual state");
+ HostVO host = _hostDao.findById(hostId);
+ if (host == null) {
+ s_logger.info("Cancelling investigation on host " + hostId + " which might has been removed");
+ cancelInvestigationTask(hostId);
+ return;
+ }
+ if (host.getStatus() != Status.Disconnected) {
+ s_logger.info("Cancelling investigation on host " + hostId + " in status " + host.getStatus());
+ cancelInvestigationTask(hostId);
+ return;
+ }
+ Status determinedState = _haMgr.investigate(hostId);
+ s_logger.info("Investigators determine the status of host " + hostId + " is " + determinedState);
+ if (determinedState == Status.Down) {
+ agentStatusTransitTo(host, Status.Event.HostDown, _nodeId);
+ s_logger.info("Scheduling VMs restart on host " + hostId + " which is Down");
+ _haMgr.scheduleRestartForVmsOnHost(host, true);
+ s_logger.info("Cancelling investigation on host " + hostId + " which is Down");
+ cancelInvestigationTask(hostId);
+ }
+ } catch (final Exception e) {
+ s_logger.error("Exception caught while handling investigation task: ", e);
+ }
+ }
+ }
+
+ private void scheduleInvestigationTask(final Long hostId) {
+ ScheduledFuture future = _investigateTasksMap.get(hostId);
+ if (future != null) {
+ s_logger.info("There is already a task to investigate host " + hostId);
+ } else {
+ ScheduledFuture scheduledFuture = _investigatorExecutor.scheduleWithFixedDelay(new InvestigationTask(hostId), InvestigateDisconnectedHostsInterval.value(),
+ InvestigateDisconnectedHostsInterval.value(), TimeUnit.SECONDS);
+ _investigateTasksMap.put(hostId, scheduledFuture);
+ s_logger.info("Scheduled a task to investigate host " + hostId);
+ }
+ }
+
+ private void cancelInvestigationTask(final Long hostId) {
+ ScheduledFuture future = _investigateTasksMap.get(hostId);
+ if (future != null) {
+ try {
+ future.cancel(false);
+ s_logger.info("Cancelled a task to investigate host " + hostId);
+ _investigateTasksMap.remove(hostId);
+ } catch (Exception e) {
+ s_logger.error("Exception caught while cancelling investigation task: ", e);
+ }
+ }
+ }
 }

plugins/hypervisors/kvm/src/main/java/com/cloud/ha/KVMInvestigator.java

@@ -85,6 +85,15 @@ public Status isAgentAlive(Host agent) {
  break;
  }
  }
+ if (!hasNfs) {
+ List<StoragePoolVO> zonePools = _storagePoolDao.findZoneWideStoragePoolsByHypervisor(agent.getDataCenterId(), agent.getHypervisorType());
+ for (StoragePoolVO pool : zonePools) {
+ if (pool.getPoolType() == StoragePoolType.NetworkFilesystem) {
+ hasNfs = true;
+ break;
+ }
+ }
+ }
  if (!hasNfs) {
  s_logger.warn(
  "Agent investigation was requested on host " + agent + ", but host does not support investigation because it has no NFS storage. Skipping investigation.");

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/KVMHABase.java

@@ -32,15 +32,37 @@ public class KVMHABase {
  private static final Logger s_logger = Logger.getLogger(KVMHABase.class);
  private long _timeout = 60000; /* 1 minutes */
  protected static String s_heartBeatPath;
- protected long _heartBeatUpdateTimeout = 60000;
- protected long _heartBeatUpdateFreq = 60000;
- protected long _heartBeatUpdateMaxTries = 5;
- protected long _heartBeatUpdateRetrySleep = 10000;
+ protected static long s_heartBeatUpdateTimeout = 60000;
+ protected static long s_heartBeatUpdateFreq = 60000;
+ protected static long s_heartBeatUpdateMaxRetries = 5;
+ protected static long s_heartBeatUpdateRetrySleep = 10000;
+ protected static HeartBeatAction s_heartBeatFailureAction = HeartBeatAction.HARDRESET;
 
  public static enum PoolType {
  PrimaryStorage, SecondaryStorage
  }
 
+ public static enum HeartBeatAction {
+ HARDRESET("hardreset", "-c"),
+ DESTROYVMS("destroyvms", "-d"),
+ NOACTION("noaction", "-n"),
+ STOPAGENT("stopagent", "-s");
+
+ String _action;
+ String _flag;
+ HeartBeatAction(String action, String flag) {
+ _action = action;
+ _flag = flag;
+ }
+ @Override
+ public String toString() {
+ return _action;
+ }
+ public String getFlag() {
+ return _flag;
+ }
+ }
+
  public static class NfsStoragePool {
  String _poolUUID;
  String _poolIp;

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/KVMHAChecker.java

@@ -50,7 +50,7 @@ public Boolean checkingHeartBeat() {
  cmd.add("-m", pool._mountDestPath);
  cmd.add("-h", _hostIP);
  cmd.add("-r");
- cmd.add("-t", String.valueOf(_heartBeatUpdateFreq / 1000));
+ cmd.add("-t", String.valueOf(s_heartBeatUpdateFreq / 1000));
  OutputInterpreter.OneLineParser parser = new OutputInterpreter.OneLineParser();
  String result = cmd.execute(parser);
  s_logger.debug("KVMHAChecker pool: " + pool._poolIp);