CLOUDSTACK-10132: Extend support for management servers LB for agents (#2469)

The new CA framework introduced basic support for comma-separated
list of management servers for agent, which makes an external LB
unnecessary.

This extends that feature to implement LB sorting algorithms that
sorts the management server list before they are sent to the agents.
This adds a central intelligence in the management server and adds
additional enhancements to Agent class to be algorithm aware and
have a background mechanism to check/fallback to preferred management
server (assumed as the first in the list). This is support for any
indirect agent such as the KVM, CPVM and SSVM agent, and would
provide support for management server host migration during upgrade
(when instead of in-place, new hosts are used to setup new mgmt server).

This FR introduces two new global settings:

- `indirect.agent.lb.algorithm`: The algorithm for the indirect agent LB.
- `indirect.agent.lb.check.interval`: The preferred host check interval
  for the agent's background task that checks and switches to agent's
  preferred host.

The indirect.agent.lb.algorithm supports following algorithm options:

- static: use the list as provided.
- roundrobin: evenly spreads hosts across management servers based on
  host's id.
- shuffle: (pseudo) randomly sorts the list (not recommended for production).

Any changes to the global settings - `indirect.agent.lb.algorithm` and
`host` does not require restarting of the mangement server(s) and the
agents. A message bus based system dynamically reacts to change in these
global settings and propagates them to all connected agents.

Comma-separated management server list is propagated to agents on
following cases:
- Addition of a host (including ssvm, cpvm systevms).
- Connection or reconnection by the agents to a management server.
- After admin changes the 'host' and/or the
  'indirect.agent.lb.algorithm' global settings.

On the agent side, the 'host' setting is saved in its properties file as:
`host=<comma separated addresses>@<algorithm name>`.

First the agent connects to the management server and sends its current
management server list, which is compared by the management server and
in case of failure a new/update list is sent for the agent to persist.

From the agent's perspective, the first address in the propagated list
will be considered the preferred host. A new background task can be
activated by configuring the `indirect.agent.lb.check.interval` which is
a cluster level global setting from CloudStack and admins can also
override this by configuring the 'host.lb.check.interval' in the
`agent.properties` file.

Every time agent gets a ms-host list and the algorithm, the host specific
background check interval is also sent and it dynamically reconfigures
the background task without need to restart agents.

Note: The 'static' and 'roundrobin' algorithms, strictly checks for the
order as expected by them, however, the 'shuffle' algorithm just checks
for content and not the order of the comma separate ms host addresses.

Signed-off-by: Rohit Yadav <rohit.yadav@shapeblue.com>

Author: yadvr

agent/conf/agent.properties

@@ -30,6 +30,17 @@ workers=5
 #host= The IP address of management server
 host=localhost
 
+# The time interval in seconds after which agent will check if connected host
+# is the preferred host (the first host in the comma-separated list is preferred
+# one) and will attempt to reconnect to the preferred host when it's connected
+# to one of the secondary/backup hosts. The timer task is scheduled after agent
+# connects to a management server. On connection, it receives admin configured
+# cluster-level 'indirect.agent.lb.check.interval' setting that will be used by
+# the agent as the preferred host check interval however the following setting
+# if defined overrides the received value. The value 0 and lb algorithm 'shuffle'
+# disables this background task.
+#host.lb.check.interval=0
+
 #port = The port management server listening on, default is 8250
 port=8250
 

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

@@ -21,6 +21,8 @@
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.net.UnknownHostException;
 import java.nio.channels.ClosedChannelException;
 import java.nio.charset.Charset;
@@ -38,12 +40,15 @@
 import javax.naming.ConfigurationException;
 
 import org.apache.cloudstack.agent.directdownload.SetupDirectDownloadCertificate;
+import org.apache.cloudstack.agent.lb.SetupMSListAnswer;
+import org.apache.cloudstack.agent.lb.SetupMSListCommand;
 import org.apache.cloudstack.ca.SetupCertificateAnswer;
 import org.apache.cloudstack.ca.SetupCertificateCommand;
 import org.apache.cloudstack.ca.SetupKeyStoreCommand;
 import org.apache.cloudstack.ca.SetupKeystoreAnswer;
 import org.apache.cloudstack.managed.context.ManagedContextTimerTask;
 import org.apache.cloudstack.utils.security.KeyStoreUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.io.FileUtils;
 import org.apache.log4j.Logger;
 import org.slf4j.MDC;
@@ -65,6 +70,7 @@
 import com.cloud.exception.AgentControlChannelException;
 import com.cloud.resource.ServerResource;
 import com.cloud.utils.PropertiesUtil;
+import com.cloud.utils.StringUtils;
 import com.cloud.utils.backoff.BackoffAlgorithm;
 import com.cloud.utils.concurrency.NamedThreadFactory;
 import com.cloud.utils.exception.CloudRuntimeException;
@@ -121,6 +127,7 @@ public int value() {
     Long _id;
 
     Timer _timer = new Timer("Agent Timer");
+    Timer hostLBTimer;
 
     List<WatchTask> _watchList = new ArrayList<WatchTask>();
     long _sequence = 0;
@@ -144,7 +151,7 @@ public Agent(final IAgentShell shell) {
         _shell = shell;
         _link = null;
 
-        _connection = new NioClient("Agent", _shell.getHost(), _shell.getPort(), _shell.getWorkers(), this);
+        _connection = new NioClient("Agent", _shell.getNextHost(), _shell.getPort(), _shell.getWorkers(), this);
 
         Runtime.getRuntime().addShutdownHook(new ShutdownThread(this));
 
@@ -179,7 +186,7 @@ public Agent(final IAgentShell shell, final int localAgentId, final ServerResour
             throw new ConfigurationException("Unable to configure " + _resource.getName());
         }
 
-        final String host = _shell.getHost();
+        final String host = _shell.getNextHost();
         _connection = new NioClient("Agent", host, _shell.getPort(), _shell.getWorkers(), this);
 
         // ((NioClient)_connection).setBindAddress(_shell.getPrivateIp());
@@ -255,7 +262,7 @@ public void start() {
             s_logger.info("Attempted to connect to the server, but received an unexpected exception, trying again...");
         }
         while (!_connection.isStartup()) {
-            final String host = _shell.getHost();
+            final String host = _shell.getNextHost();
             _shell.getBackoffAlgorithm().waitBeforeRetry();
             _connection = new NioClient("Agent", host, _shell.getPort(), _shell.getWorkers(), this);
             s_logger.info("Connecting to host:" + host);
@@ -266,6 +273,7 @@ public void start() {
                 s_logger.info("Attempted to connect to the server, but received an unexpected exception, trying again...");
             }
         }
+        _shell.updateConnectedHost();
     }
 
     public void stop(final String reason, final String detail) {
@@ -310,6 +318,17 @@ public void setId(final Long id) {
         _shell.setPersistentProperty(getResourceName(), "id", Long.toString(id));
     }
 
+    private synchronized void scheduleHostLBCheckerTask(final long checkInterval) {
+        if (hostLBTimer != null) {
+            hostLBTimer.cancel();
+        }
+        if (checkInterval > 0L) {
+            s_logger.info("Scheduling preferred host timer task with host.lb.interval=" + checkInterval + "ms");
+            hostLBTimer = new Timer("Host LB Timer");
+            hostLBTimer.scheduleAtFixedRate(new PreferredHostCheckerTask(), checkInterval, checkInterval);
+        }
+    }
+
     public void scheduleWatch(final Link link, final Request request, final long delay, final long period) {
         synchronized (_watchList) {
             if (s_logger.isDebugEnabled()) {
@@ -332,18 +351,20 @@ protected void cancelTasks() {
             _watchList.clear();
         }
     }
-    public synchronized void lockStartupTask(final Link link)
-    {
+
+    public synchronized void lockStartupTask(final Link link) {
         _startup = new StartupTask(link);
         _timer.schedule(_startup, _startupWait);
     }
 
     public void sendStartup(final Link link) {
         final StartupCommand[] startup = _resource.initialize();
         if (startup != null) {
+            final String msHostList = _shell.getPersistentProperty(null, "host");
             final Command[] commands = new Command[startup.length];
             for (int i = 0; i < startup.length; i++) {
                 setupStartupCommand(startup[i]);
+                startup[i].setMSHostList(msHostList);
                 commands[i] = startup[i];
             }
             final Request request = new Request(_id != null ? _id : -1, -1, commands, false, false);
@@ -402,19 +423,23 @@ protected void reconnect(final Link link) {
             }
         }
 
-        link.close();
-        link.terminated();
+        if (link != null) {
+            link.close();
+            link.terminated();
+        }
 
         setLink(null);
         cancelTasks();
 
         _resource.disconnected();
 
+        final String lastConnectedHost = _shell.getConnectedHost();
+
         int inProgress = 0;
         do {
             _shell.getBackoffAlgorithm().waitBeforeRetry();
 
-            s_logger.info("Lost connection to the server. Dealing with the remaining commands...");
+            s_logger.info("Lost connection to host: " + lastConnectedHost + ". Dealing with the remaining commands...");
 
             inProgress = _inProgress.get();
             if (inProgress > 0) {
@@ -434,7 +459,7 @@ protected void reconnect(final Link link) {
             _shell.getBackoffAlgorithm().waitBeforeRetry();
         }
 
-        final String host = _shell.getHost();
+        final String host = _shell.getNextHost();
         do {
             _connection = new NioClient("Agent", host, _shell.getPort(), _shell.getWorkers(), this);
             s_logger.info("Reconnecting to host:" + host);
@@ -452,7 +477,8 @@ protected void reconnect(final Link link) {
             }
             _shell.getBackoffAlgorithm().waitBeforeRetry();
         } while (!_connection.isStartup());
-        s_logger.info("Connected to the server");
+        _shell.updateConnectedHost();
+        s_logger.info("Connected to the host: " + _shell.getConnectedHost());
     }
 
     public void processStartupAnswer(final Answer answer, final Response response, final Link link) {
@@ -554,6 +580,8 @@ protected void processRequest(final Request request, final Link link) {
                         answer = setupAgentCertificate((SetupCertificateCommand) cmd);
                     } else if (cmd instanceof SetupDirectDownloadCertificate) {
                         answer = setupDirectDownloadCertificate((SetupDirectDownloadCertificate) cmd);
+                    } else if (cmd instanceof SetupMSListCommand) {
+                        answer = setupManagementServerList((SetupMSListCommand) cmd);
                     } else {
                         if (cmd instanceof ReadyCommand) {
                             processReadyCommand(cmd);
@@ -708,6 +736,30 @@ private Answer setupAgentCertificate(final SetupCertificateCommand cmd) {
         return new SetupCertificateAnswer(true);
     }
 
+    private void processManagementServerList(final List<String> msList, final String lbAlgorithm, final Long lbCheckInterval) {
+        if (CollectionUtils.isNotEmpty(msList) && !Strings.isNullOrEmpty(lbAlgorithm)) {
+            try {
+                final String newMSHosts = String.format("%s%s%s", StringUtils.toCSVList(msList), IAgentShell.hostLbAlgorithmSeparator, lbAlgorithm);
+                _shell.setPersistentProperty(null, "host", newMSHosts);
+                _shell.setHosts(newMSHosts);
+                _shell.resetHostCounter();
+                s_logger.info("Processed new management server list: " + newMSHosts);
+            } catch (final Exception e) {
+                throw new CloudRuntimeException("Could not persist received management servers list", e);
+            }
+        }
+        if ("shuffle".equals(lbAlgorithm)) {
+            scheduleHostLBCheckerTask(0);
+        } else {
+            scheduleHostLBCheckerTask(_shell.getLbCheckerInterval(lbCheckInterval));
+        }
+    }
+
+    private Answer setupManagementServerList(final SetupMSListCommand cmd) {
+        processManagementServerList(cmd.getMsList(), cmd.getLbAlgorithm(), cmd.getLbCheckInterval());
+        return new SetupMSListAnswer(true);
+    }
+
     public void processResponse(final Response response, final Link link) {
         final Answer answer = response.getAnswer();
         if (s_logger.isDebugEnabled()) {
@@ -728,15 +780,16 @@ public void processResponse(final Response response, final Link link) {
     }
 
     public void processReadyCommand(final Command cmd) {
-
         final ReadyCommand ready = (ReadyCommand)cmd;
 
-        s_logger.info("Proccess agent ready command, agent id = " + ready.getHostId());
+        s_logger.info("Processing agent ready command, agent id = " + ready.getHostId());
         if (ready.getHostId() != null) {
             setId(ready.getHostId());
         }
-        s_logger.info("Ready command is processed: agent id = " + getId());
 
+        processManagementServerList(ready.getMsHostList(), ready.getLbAlgorithm(), ready.getLbCheckInterval());
+
+        s_logger.info("Ready command is processed for agent id = " + getId());
     }
 
     public void processOtherTask(final Task task) {
@@ -1018,4 +1071,44 @@ public void doTask(final Task task) throws TaskExecutionException {
             }
         }
     }
+
+    public class PreferredHostCheckerTask extends ManagedContextTimerTask {
+
+        @Override
+        protected void runInContext() {
+            try {
+                final String[] msList = _shell.getHosts();
+                if (msList == null || msList.length < 1) {
+                    return;
+                }
+                final String preferredHost  = msList[0];
+                final String connectedHost = _shell.getConnectedHost();
+                if (s_logger.isTraceEnabled()) {
+                    s_logger.trace("Running preferred host checker task, connected host=" + connectedHost + ", preferred host=" + preferredHost);
+                }
+                if (preferredHost != null && !preferredHost.equals(connectedHost) && _link != null) {
+                    boolean isHostUp = true;
+                    try (final Socket socket = new Socket()) {
+                        socket.connect(new InetSocketAddress(preferredHost, _shell.getPort()), 5000);
+                    } catch (final IOException e) {
+                        isHostUp = false;
+                        if (s_logger.isTraceEnabled()) {
+                            s_logger.trace("Host: " + preferredHost + " is not reachable");
+                        }
+                    }
+                    if (isHostUp && _link != null && _inProgress.get() == 0) {
+                        if (s_logger.isDebugEnabled()) {
+                            s_logger.debug("Preferred host " + preferredHost + " is found to be reachable, trying to reconnect");
+                        }
+                        _shell.resetHostCounter();
+                        reconnect(_link);
+                    }
+                }
+            } catch (Throwable t) {
+                s_logger.error("Error caught while attempting to connect to preferred host", t);
+            }
+        }
+
+    }
+
 }

agent/src/com/cloud/agent/AgentShell.java

@@ -50,6 +50,7 @@
 import com.cloud.utils.backoff.BackoffAlgorithm;
 import com.cloud.utils.backoff.impl.ConstantTimeBackoff;
 import com.cloud.utils.exception.CloudRuntimeException;
+import com.google.common.base.Strings;
 
 public class AgentShell implements IAgentShell, Daemon {
     private static final Logger s_logger = Logger.getLogger(AgentShell.class.getName());
@@ -72,6 +73,9 @@ public class AgentShell implements IAgentShell, Daemon {
     private volatile boolean _exit = false;
     private int _pingRetries;
     private final List<Agent> _agents = new ArrayList<Agent>();
+    private String hostToConnect;
+    private String connectedHost;
+    private Long preferredHostCheckInterval;
 
     public AgentShell() {
     }
@@ -107,18 +111,54 @@ public String getPod() {
     }
 
     @Override
-    public String getHost() {
-        final String[] hosts = _host.split(",");
+    public String getNextHost() {
+        final String[] hosts = getHosts();
         if (_hostCounter >= hosts.length) {
             _hostCounter = 0;
         }
-        final String host = hosts[_hostCounter % hosts.length];
+        hostToConnect = hosts[_hostCounter % hosts.length];
         _hostCounter++;
-        return host;
+        return hostToConnect;
     }
 
-    public void setHost(final String host) {
-        _host = host;
+    @Override
+    public String getConnectedHost() {
+        return connectedHost;
+    }
+
+    @Override
+    public long getLbCheckerInterval(final Long receivedLbInterval) {
+        if (preferredHostCheckInterval != null) {
+            return preferredHostCheckInterval * 1000L;
+        }
+        if (receivedLbInterval != null) {
+            return receivedLbInterval * 1000L;
+        }
+        return 0L;
+    }
+
+    @Override
+    public void updateConnectedHost() {
+        connectedHost = hostToConnect;
+    }
+
+
+    @Override
+    public void resetHostCounter() {
+        _hostCounter = 0;
+    }
+
+    @Override
+    public String[] getHosts() {
+        return _host.split(",");
+    }
+
+    @Override
+    public void setHosts(final String host) {
+        if (!Strings.isNullOrEmpty(host)) {
+            _host = host.split(hostLbAlgorithmSeparator)[0];
+            resetHostCounter();
+        }
     }
 
     @Override
@@ -251,7 +291,8 @@ protected boolean parseCommand(final String[] args) throws ConfigurationExceptio
         if (host == null) {
             host = "localhost";
         }
-        _host = host;
+
+        setHosts(host);
 
         if (zone != null)
             _zone = zone;
@@ -291,6 +332,9 @@ protected boolean parseCommand(final String[] args) throws ConfigurationExceptio
             _properties.setProperty("guid", _guid);
         }
 
+        String val = getProperty(null, preferredHostIntervalKey);
+        preferredHostCheckInterval = (Strings.isNullOrEmpty(val) ? null : Long.valueOf(val));
+
         return true;
     }