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

// 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 com.cloud.agent;

import java.io.File;
import java.io.IOException;
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;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import javax.naming.ConfigurationException;

import org.apache.cloudstack.agent.lb.SetupMSListAnswer;
import org.apache.cloudstack.agent.lb.SetupMSListCommand;
import org.apache.cloudstack.ca.PostCertificateRenewalCommand;
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.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.ThreadContext;

import com.cloud.agent.api.AgentControlAnswer;
import com.cloud.agent.api.AgentControlCommand;
import com.cloud.agent.api.Answer;
import com.cloud.agent.api.Command;
import com.cloud.agent.api.CronCommand;
import com.cloud.agent.api.MaintainAnswer;
import com.cloud.agent.api.MaintainCommand;
import com.cloud.agent.api.PingAnswer;
import com.cloud.agent.api.PingCommand;
import com.cloud.agent.api.ReadyCommand;
import com.cloud.agent.api.ShutdownCommand;
import com.cloud.agent.api.StartupAnswer;
import com.cloud.agent.api.StartupCommand;
import com.cloud.agent.transport.Request;
import com.cloud.agent.transport.Response;
import com.cloud.exception.AgentControlChannelException;
import com.cloud.host.Host;
import com.cloud.resource.AgentStatusUpdater;
import com.cloud.resource.ResourceStatusUpdater;
import com.cloud.resource.ServerResource;
import com.cloud.utils.NumbersUtil;
import com.cloud.utils.PropertiesUtil;
import com.cloud.utils.concurrency.NamedThreadFactory;
import com.cloud.utils.exception.CloudRuntimeException;
import com.cloud.utils.exception.NioConnectionException;
import com.cloud.utils.exception.TaskExecutionException;
import com.cloud.utils.nio.HandlerFactory;
import com.cloud.utils.nio.Link;
import com.cloud.utils.nio.NioClient;
import com.cloud.utils.nio.NioConnection;
import com.cloud.utils.nio.Task;
import com.cloud.utils.script.OutputInterpreter;
import com.cloud.utils.script.Script;

/**
 * @config
 *         {@table
 *         || Param Name | Description | Values | Default ||
 *         || type | Type of server | Storage / Computing / Routing | No Default ||
 *         || workers | # of workers to process the requests | int | 1 ||
 *         || host | host to connect to | ip address | localhost ||
 *         || port | port to connect to | port number | 8250 ||
 *         || instance | Used to allow multiple agents running on the same host | String | none || * }
 *
 *         For more configuration options, see the individual types.
 *
 **/
public class Agent implements HandlerFactory, IAgentControl, AgentStatusUpdater {
    protected Logger logger = LogManager.getLogger(getClass());

    public enum ExitStatus {
        Normal(0), // Normal status = 0.
        Upgrade(65), // Exiting for upgrade.
        Configuration(66), // Exiting due to configuration problems.
        Error(67); // Exiting because of error.

        final int value;

        ExitStatus(final int value) {
            this.value = value;
        }

        public int value() {
            return value;
        }
    }

    CopyOnWriteArrayList<IAgentControlListener> controlListeners = new CopyOnWriteArrayList<>();

    IAgentShell shell;
    NioConnection connection;
    ServerResource serverResource;
    Link link;
    Long id;
    String _uuid;
    String _name;

    ScheduledExecutorService selfTaskExecutor;
    ScheduledExecutorService certExecutor;
    ScheduledExecutorService hostLbCheckExecutor;

    CopyOnWriteArrayList<ScheduledFuture<?>> watchList = new CopyOnWriteArrayList<>();
    AtomicLong sequence = new AtomicLong(0);
    AtomicLong lastPingResponseTime = new AtomicLong(0L);
    long pingInterval = 0;
    AtomicInteger commandsInProgress = new AtomicInteger(0);

    private final AtomicReference<StartupTask> startupTask = new AtomicReference<>();
    private static final long DEFAULT_STARTUP_WAIT = 180;
    long startupWait = DEFAULT_STARTUP_WAIT;
    boolean reconnectAllowed = true;

    //For time sensitive task, e.g. PingTask
    ThreadPoolExecutor outRequestHandler;
    ExecutorService requestHandler;

    Thread shutdownThread = new ShutdownThread(this);

    private String keystoreSetupSetupPath;
    private String keystoreCertImportScriptPath;

    private String hostname;

    protected String getLinkLog(final Link link) {
        if (link == null) {
            return "";
        }
        StringBuilder str = new StringBuilder();
        if (logger.isTraceEnabled()) {
            str.append(System.identityHashCode(link)).append("-");
        }
        str.append(link.getSocketAddress());
        return str.toString();
    }

    protected String getAgentName() {
        return (serverResource != null && serverResource.isAppendAgentNameToLogs() &&
                StringUtils.isNotBlank(serverResource.getName())) ?
                serverResource.getName() :
                "Agent";
    }

    protected void setupShutdownHookAndInitExecutors() {
        logger.trace("Adding shutdown hook");
        Runtime.getRuntime().addShutdownHook(shutdownThread);
        selfTaskExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Agent-SelfTask"));
        outRequestHandler = new ThreadPoolExecutor(shell.getPingRetries(), 2 * shell.getPingRetries(), 10, TimeUnit.MINUTES,
                new SynchronousQueue<>(), new NamedThreadFactory("AgentOutRequest-Handler"));
        requestHandler = new ThreadPoolExecutor(shell.getWorkers(), 5 * shell.getWorkers(), 1, TimeUnit.DAYS,
                new LinkedBlockingQueue<>(), new NamedThreadFactory("AgentRequest-Handler"));
    }

    /**
     * Constructor for the {@code Agent} class, intended for simulator use only.
     *
     * <p>This constructor initializes the agent with a provided {@link IAgentShell}.
     * It sets up the necessary NIO client connection, establishes a shutdown hook,
     * and initializes the thread executors.
     *
     * @param shell the {@link IAgentShell} instance that provides agent configuration and runtime information.
     */
    public Agent(final IAgentShell shell) {
        this.shell = shell;
        this.link = null;
        this.connection = new NioClient(
                getAgentName(),
                this.shell.getNextHost(),
                this.shell.getPort(),
                this.shell.getWorkers(),
                this.shell.getSslHandshakeTimeout(),
                this
        );
        setupShutdownHookAndInitExecutors();
    }

    public Agent(final IAgentShell shell, final int localAgentId, final ServerResource resource) throws ConfigurationException {
        this.shell = shell;
        serverResource = resource;
        link = null;
        resource.setAgentControl(this);
        final String value = shell.getPersistentProperty(getResourceName(), "id");
        _uuid = shell.getPersistentProperty(getResourceName(), "uuid");
        _name = shell.getPersistentProperty(getResourceName(), "name");
        id = value != null ? Long.parseLong(value) : null;
        logger.info("Initialising agent [id: {}, uuid: {}, name: {}]", ObjectUtils.defaultIfNull(id, ""), _uuid, _name);

        final Map<String, Object> params = new HashMap<>();
        // merge with properties from command line to let resource access command line parameters
        for (final Map.Entry<String, Object> cmdLineProp : this.shell.getCmdLineProperties().entrySet()) {
            params.put(cmdLineProp.getKey(), cmdLineProp.getValue());
        }
        if (!serverResource.configure(getResourceName(), params)) {
            throw new ConfigurationException("Unable to configure " + serverResource.getName());
        }
        ThreadContext.put("agentname", getAgentName());
        final String host = this.shell.getNextHost();
        connection = new NioClient(getAgentName(), host, this.shell.getPort(), this.shell.getWorkers(),
                this.shell.getSslHandshakeTimeout(), this);
        setupShutdownHookAndInitExecutors();
        logger.info("{} with host = {}, local id = {}", this, host, localAgentId);
    }


    @Override
    public String toString() {
        return String.format("Agent [id = %s, uuid = %s, name = %s, type = %s, zone = %s, pod = %s, workers = %d, port = %d]",
                ObjectUtils.defaultIfNull(id, "new"),
                _uuid,
                _name,
                getResourceName(),
                this.shell.getZone(),
                this.shell.getPod(),
                this.shell.getWorkers(),
                this.shell.getPort());
    }

    public String getVersion() {
        return shell.getVersion();
    }

    public String getResourceGuid() {
        final String guid = shell.getGuid();
        return guid + "-" + getResourceName();
    }

    public String getZone() {
        return shell.getZone();
    }

    public String getPod() {
        return shell.getPod();
    }

    protected void setLink(final Link link) {
        this.link = link;
    }

    public ServerResource getResource() {
        return serverResource;
    }

    public String getResourceName() {
        return serverResource.getClass().getSimpleName();
    }

    /**
     * In case of a software based agent restart, this method
     * can help to perform explicit garbage collection of any old
     * agent instances and its inner objects.
     */
    private void scavengeOldAgentObjects() {
        requestHandler.submit(() -> {
            try {
                Thread.sleep(2000L);
            } catch (final InterruptedException ignored) {
            } finally {
                System.gc();
            }
        });
    }

    public void start() {
        if (!serverResource.start()) {
            String msg = String.format("Unable to start the resource: %s", serverResource.getName());
            logger.error(msg);
            throw new CloudRuntimeException(msg);
        }

        keystoreSetupSetupPath = Script.findScript("scripts/util/", KeyStoreUtils.KS_SETUP_SCRIPT);
        if (keystoreSetupSetupPath == null) {
            throw new CloudRuntimeException(String.format("Unable to find the '%s' script", KeyStoreUtils.KS_SETUP_SCRIPT));
        }

        keystoreCertImportScriptPath = Script.findScript("scripts/util/", KeyStoreUtils.KS_IMPORT_SCRIPT);
        if (keystoreCertImportScriptPath == null) {
            throw new CloudRuntimeException(String.format("Unable to find the '%s' script", KeyStoreUtils.KS_IMPORT_SCRIPT));
        }

        try {
            connection.start();
        } catch (final NioConnectionException e) {
            logger.warn("Attempt to connect to server generated NIO Connection Exception {}, trying again", e.getLocalizedMessage());
        }
        while (!connection.isStartup()) {
            final String host = shell.getNextHost();
            shell.getBackoffAlgorithm().waitBeforeRetry();
            connection = new NioClient(getAgentName(), host, shell.getPort(), shell.getWorkers(),
                    shell.getSslHandshakeTimeout(), this);
            logger.info("Connecting to host: {}", host);
            try {
                connection.start();
            } catch (final NioConnectionException e) {
                stopAndCleanupConnection(false);
                logger.info("Attempted to connect to the server, but received an unexpected exception, trying again...", e);
            }
        }
        shell.updateConnectedHost();
        scavengeOldAgentObjects();

    }

    public void stop(final String reason, final String detail) {
        logger.info("Stopping the agent: Reason = {}{}", reason, (detail != null ? ": Detail = " + detail : ""));
        reconnectAllowed = false;
        if (connection != null) {
            final ShutdownCommand cmd = new ShutdownCommand(reason, detail);
            try {
                if (link != null) {
                    final Request req = new Request(id != null ? id : -1, -1, cmd, false);
                    link.send(req.toBytes());
                }
            } catch (final ClosedChannelException e) {
                logger.warn("Unable to send: {}", cmd.toString());
            } catch (final Exception e) {
                logger.warn("Unable to send: {} due to exception: {}", cmd.toString(), e);
            }
            logger.debug("Sending shutdown to management server");
            try {
                Thread.sleep(1000);
            } catch (final InterruptedException e) {
                logger.debug("Who the heck interrupted me here?");
            }
            connection.stop();
            connection = null;
            link = null;
        }

        if (serverResource != null) {
            serverResource.stop();
            serverResource = null;
        }

        if (startupTask.get() != null) {
            startupTask.set(null);
        }

        if (outRequestHandler != null) {
            outRequestHandler.shutdownNow();
            outRequestHandler = null;
        }

        if (requestHandler != null) {
            requestHandler.shutdown();
            requestHandler = null;
        }

        if (selfTaskExecutor != null) {
            selfTaskExecutor.shutdown();
            selfTaskExecutor = null;
        }

        if (hostLbCheckExecutor != null) {
            hostLbCheckExecutor.shutdown();
            hostLbCheckExecutor = null;
        }

        if (certExecutor != null) {
            certExecutor.shutdown();
            certExecutor = null;
        }
    }

    public Long getId() {
        return id;
    }

    public void setId(final Long id) {
        logger.debug("Set agent id {}", id);
        this.id = id;
        shell.setPersistentProperty(getResourceName(), "id", Long.toString(id));
    }

    public String getUuid() {
        return _uuid;
    }

    public void setUuid(String uuid) {
        this._uuid = uuid;
        shell.setPersistentProperty(getResourceName(), "uuid", uuid);
    }

    public String getName() {
        return _name;
    }

    public void setName(String name) {
        this._name = name;
        shell.setPersistentProperty(getResourceName(), "name", name);
    }

    private void scheduleCertificateRenewalTask() {
        String name = "CertificateRenewalTask";
        if (certExecutor != null && !certExecutor.isShutdown()) {
            certExecutor.shutdown();
            try {
                if (!certExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                    certExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                logger.debug("Forcing {} shutdown as it did not shutdown in the desired time due to: {}",
                                name, e.getMessage());
                certExecutor.shutdownNow();
            }
        }
        certExecutor = Executors.newSingleThreadScheduledExecutor((new NamedThreadFactory(name)));
        certExecutor.schedule(new PostCertificateRenewalTask(this), 5, TimeUnit.SECONDS);
    }

    private void scheduleHostLBCheckerTask(final long checkInterval) {
        String name = "HostLBCheckerTask";
        if (hostLbCheckExecutor != null && !hostLbCheckExecutor.isShutdown()) {
            hostLbCheckExecutor.shutdown();
            try {
                if (!hostLbCheckExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                    hostLbCheckExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                logger.debug("Forcing {} shutdown as it did not shutdown in the desired time due to: {}",
                        name, e.getMessage());
                hostLbCheckExecutor.shutdownNow();
            }
        }
        if (checkInterval > 0L) {
            logger.info("Scheduling preferred host task with host.lb.interval={}ms", checkInterval);
            hostLbCheckExecutor = Executors.newSingleThreadScheduledExecutor((new NamedThreadFactory(name)));
            hostLbCheckExecutor.scheduleAtFixedRate(new PreferredHostCheckerTask(), checkInterval, checkInterval,
                    TimeUnit.MILLISECONDS);
        }
    }

    public void scheduleWatch(final Link link, final Request request, final long delay, final long period) {
        logger.debug("Adding a watch list");
        final WatchTask task = new WatchTask(link, request, this);
        final ScheduledFuture<?> future = selfTaskExecutor.scheduleAtFixedRate(task, delay, period, TimeUnit.MILLISECONDS);
        watchList.add(future);
    }

    public void triggerUpdate() {
        PingCommand command = serverResource.getCurrentStatus(getId());
        command.setOutOfBand(true);
        logger.debug("Sending out of band ping");
        final Request request = new Request(id, -1, command, false);
        request.setSequence(getNextSequence());
        try {
            link.send(request.toBytes());
        } catch (final ClosedChannelException e) {
            logger.warn("Unable to send ping update: {}", request.toString());
        }
    }

    protected void cancelTasks() {
        for (final ScheduledFuture<?> task : watchList) {
            task.cancel(true);
        }
        logger.debug("Clearing watch list: {}", () -> watchList.size());
        watchList.clear();
    }

    /**
     * Cleanup agent zone properties.
     *
     * Unset zone, cluster and pod values so that host is not added back
     * when service is restarted. This will be set to proper values
     * when host is added back
     */
    protected void cleanupAgentZoneProperties() {
        shell.setPersistentProperty(null, "zone", "");
        shell.setPersistentProperty(null, "cluster", "");
        shell.setPersistentProperty(null, "pod", "");
    }

    public void lockStartupTask(final Link link) {
        logger.debug("Creating startup task for link: {}", () -> getLinkLog(link));
        StartupTask currentTask = startupTask.get();
        if (currentTask != null) {
            logger.warn("A Startup task is already locked or in progress, cannot create for link {}",
                    getLinkLog(link));
            return;
        }
        currentTask = new StartupTask(link);
        if (startupTask.compareAndSet(null, currentTask)) {
            selfTaskExecutor.schedule(currentTask, startupWait, TimeUnit.SECONDS);
            return;
        }
        logger.warn("Failed to lock a StartupTask for link: {}", getLinkLog(link));
    }

    protected boolean cancelStartupTask() {
        StartupTask task = startupTask.getAndSet(null);
        if (task != null) {
            task.cancel();
            return true;
        }
        return false;
    }

    public void sendStartup(final Link link) {
        final StartupCommand[] startup = serverResource.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);
            request.setSequence(getNextSequence());

            logger.debug("Sending Startup: {}", request.toString());
            lockStartupTask(link);
            try {
                link.send(request.toBytes());
            } catch (final ClosedChannelException e) {
                logger.warn("Unable to send request to {} due to '{}', request: {}",
                        getLinkLog(link), e.getMessage(), request);
            }

            if (serverResource instanceof ResourceStatusUpdater) {
                ((ResourceStatusUpdater) serverResource).registerStatusUpdater(this);
            }
        }
    }

    protected String retrieveHostname() {
        logger.trace("Retrieving hostname with resource={}", () -> serverResource.getClass().getSimpleName());
        final String result = Script.runSimpleBashScript(Script.getExecutableAbsolutePath("hostname"), 500);
        if (StringUtils.isNotBlank(result)) {
            return result;
        }
        try {
            InetAddress address = InetAddress.getLocalHost();
            return address.toString();
        } catch (final UnknownHostException e) {
            logger.warn("unknown host? ", e);
            throw new CloudRuntimeException("Cannot get local IP address");
        }
    }

    protected void setupStartupCommand(final StartupCommand startup) {
        startup.setId(getId());
        if (StringUtils.isBlank(startup.getName())) {
            if (StringUtils.isBlank(hostname)) {
                hostname = retrieveHostname();
            }
            startup.setName(hostname);
        }
        startup.setDataCenter(getZone());
        startup.setPod(getPod());
        startup.setGuid(getResourceGuid());
        startup.setResourceName(getResourceName());
        startup.setVersion(getVersion());
        startup.setArch(getAgentArch());
    }

    protected String getAgentArch() {
        final Script command = new Script("/usr/bin/arch", 500, logger);
        final OutputInterpreter.OneLineParser parser = new OutputInterpreter.OneLineParser();
        return command.execute(parser);
    }

    @Override
    public Task create(final Task.Type type, final Link link, final byte[] data) {
        return new ServerHandler(type, link, data);
    }

    protected void closeAndTerminateLink(final Link link) {
        if (link == null) {
            return;
        }
        link.close();
        link.terminated();
    }

    protected void stopAndCleanupConnection(boolean waitForStop) {
        if (connection == null) {
            return;
        }
        connection.stop();
        try {
            connection.cleanUp();
        } catch (final IOException e) {
            logger.warn("Fail to clean up old connection. {}", e);
        }
        if (!waitForStop) {
            return;
        }
        do {
            shell.getBackoffAlgorithm().waitBeforeRetry();
        } while (connection.isStartup());
    }

    protected void reconnect(final Link link) {
        if (!reconnectAllowed) {
            logger.debug("Reconnect requested but it is not allowed {}", () -> getLinkLog(link));
            return;
        }
        cancelStartupTask();
        closeAndTerminateLink(link);
        closeAndTerminateLink(this.link);
        setLink(null);
        cancelTasks();
        serverResource.disconnected();
        logger.info("Lost connection to host: {}. Attempting reconnection while we still have {} commands in progress.", shell.getConnectedHost(), commandsInProgress.get());
        stopAndCleanupConnection(true);
        do {
            final String host = shell.getNextHost();
            connection = new NioClient(getAgentName(), host, shell.getPort(), shell.getWorkers(), shell.getSslHandshakeTimeout(), this);
            logger.info("Reconnecting to host: {}", host);
            try {
                connection.start();
            } catch (final NioConnectionException e) {
                logger.info("Attempted to re-connect to the server, but received an unexpected exception, trying again...", e);
                stopAndCleanupConnection(false);
            }
            shell.getBackoffAlgorithm().waitBeforeRetry();
        } while (!connection.isStartup());
        shell.updateConnectedHost();
        logger.info("Connected to the host: {}", shell.getConnectedHost());
    }

    public void processStartupAnswer(final Answer answer, final Response response, final Link link) {
        boolean answerValid = cancelStartupTask();
        final StartupAnswer startup = (StartupAnswer)answer;
        if (!startup.getResult()) {
            logger.error("Not allowed to connect to the server: {}", answer.getDetails());
            if (serverResource != null && !serverResource.isExitOnFailures()) {
                logger.trace("{} does not allow exit on failure, reconnecting",
                        serverResource.getClass().getSimpleName());
                reconnect(link);
                return;
            }
            System.exit(1);
        }
        if (!answerValid) {
            logger.warn("Threw away a startup answer because we're reconnecting.");
            return;
        }

        logger.info("Process agent startup answer, agent [id: {}, uuid: {}, name: {}] connected to the server",
                startup.getHostId(), startup.getHostUuid(), startup.getHostName());

        setId(startup.getHostId());
        setUuid(startup.getHostUuid());
        setName(startup.getHostName());
        pingInterval = startup.getPingInterval() * 1000L; // change to ms.

        updateLastPingResponseTime();
        scheduleWatch(link, response, pingInterval, pingInterval);

        outRequestHandler.setKeepAliveTime(2 * pingInterval, TimeUnit.MILLISECONDS);

        logger.info("Startup Response Received: agent [id: {}, uuid: {}, name: {}]",
                startup.getHostId(), startup.getHostUuid(), startup.getHostName());
    }

    protected void processRequest(final Request request, final Link link) {
        boolean requestLogged = false;
        Response response = null;
        try {
            final Command[] cmds = request.getCommands();
            final Answer[] answers = new Answer[cmds.length];

            for (int i = 0; i < cmds.length; i++) {
                final Command cmd = cmds[i];
                Answer answer;
                try {
                    if (logger.isDebugEnabled()) {
                        if (!requestLogged) // ensures request is logged only once per method call
                        {
                            final String requestMsg = request.toString();
                            if (requestMsg != null) {
                                logger.debug("Request:{}",requestMsg);
                            }
                            requestLogged = true;
                        }
                        logger.debug("Processing command: {}", cmd.toString());
                    }

                    if (cmd instanceof CronCommand) {
                        final CronCommand watch = (CronCommand)cmd;
                        scheduleWatch(link, request, watch.getInterval() * 1000L, watch.getInterval() * 1000L);
                        answer = new Answer(cmd, true, null);
                    } else if (cmd instanceof ShutdownCommand) {
                        final ShutdownCommand shutdown = (ShutdownCommand)cmd;
                        logger.debug("Received shutdownCommand, due to: {}", shutdown.getReason());
                        cancelTasks();
                        if (shutdown.isRemoveHost()) {
                            cleanupAgentZoneProperties();
                        }
                        reconnectAllowed = false;
                        answer = new Answer(cmd, true, null);
                    } else if (cmd instanceof ReadyCommand && ((ReadyCommand)cmd).getDetails() != null) {

                        logger.debug("Not ready to connect to mgt server: {}", ((ReadyCommand)cmd).getDetails());
                        if (serverResource != null && !serverResource.isExitOnFailures()) {
                            logger.trace("{} does not allow exit on failure, reconnecting",
                                    serverResource.getClass().getSimpleName());
                            reconnect(link);
                            return;
                        }
                        System.exit(1);
                        return;
                    } else if (cmd instanceof MaintainCommand) {
                        logger.debug("Received maintainCommand, do not cancel current tasks");
                        answer = new MaintainAnswer((MaintainCommand)cmd);
                    } else if (cmd instanceof AgentControlCommand) {
                        answer = null;
                        for (final IAgentControlListener listener : controlListeners) {
                            answer = listener.processControlRequest(request, (AgentControlCommand)cmd);
                            if (answer != null) {
                                break;
                            }
                        }

                        if (answer == null) {
                            logger.warn("No handler found to process cmd: {}", cmd.toString());
                            answer = new AgentControlAnswer(cmd);
                        }
                    } else if (cmd instanceof SetupKeyStoreCommand && ((SetupKeyStoreCommand) cmd).isHandleByAgent()) {
                        answer = setupAgentKeystore((SetupKeyStoreCommand) cmd);
                    } else if (cmd instanceof SetupCertificateCommand && ((SetupCertificateCommand) cmd).isHandleByAgent()) {
                        answer = setupAgentCertificate((SetupCertificateCommand) cmd);
                        if (Host.Type.Routing.equals(serverResource.getType())) {
                            scheduleCertificateRenewalTask();
                        }
                    } else if (cmd instanceof SetupMSListCommand) {
                        answer = setupManagementServerList((SetupMSListCommand) cmd);
                    } else {
                        if (cmd instanceof ReadyCommand) {
                            processReadyCommand(cmd);
                        }
                        commandsInProgress.incrementAndGet();
                        try {
                            answer = serverResource.executeRequest(cmd);
                        } finally {
                            commandsInProgress.decrementAndGet();
                        }
                        if (answer == null) {
                            logger.debug("Response: unsupported command {}", cmd.toString());
                            answer = Answer.createUnsupportedCommandAnswer(cmd);
                        }
                    }
                } catch (final Throwable th) {
                    logger.warn("Caught: ", th);
                    final StringWriter writer = new StringWriter();
                    th.printStackTrace(new PrintWriter(writer));
                    answer = new Answer(cmd, false, writer.toString());
                }

                answers[i] = answer;
                if (!answer.getResult() && request.stopOnError()) {
                    for (i++; i < cmds.length; i++) {
                        answers[i] = new Answer(cmds[i], false, "Stopped by previous failure");
                    }
                    break;
                }
            }
            response = new Response(request, answers);
        } finally {
            if (logger.isDebugEnabled()) {
                final String responseMsg = response.toString();
                if (responseMsg != null) {
                    logger.debug(response.toString());
                }
            }

            if (response != null) {
                try {
                    link.send(response.toBytes());
                } catch (final ClosedChannelException e) {
                    logger.warn("Unable to send response: {}", response.toString());
                }
            }
        }
    }

    public Answer setupAgentKeystore(final SetupKeyStoreCommand cmd) {
        final String keyStorePassword = cmd.getKeystorePassword();
        final long validityDays = cmd.getValidityDays();

        logger.debug("Setting up agent keystore file and generating CSR");

        final File agentFile = PropertiesUtil.findConfigFile("agent.properties");
        if (agentFile == null) {
            return new Answer(cmd, false, "Failed to find agent.properties file");
        }
        final String keyStoreFile = agentFile.getParent() + "/" + KeyStoreUtils.KS_FILENAME;
        final String csrFile = agentFile.getParent() + "/" + KeyStoreUtils.CSR_FILENAME;

        String storedPassword = shell.getPersistentProperty(null, KeyStoreUtils.KS_PASSPHRASE_PROPERTY);
        if (StringUtils.isEmpty(storedPassword)) {
            storedPassword = keyStorePassword;
            shell.setPersistentProperty(null, KeyStoreUtils.KS_PASSPHRASE_PROPERTY, storedPassword);
        }

        Script script = new Script(keystoreSetupSetupPath, 300000, logger);
        script.add(agentFile.getAbsolutePath());
        script.add(keyStoreFile);
        script.add(storedPassword);
        script.add(String.valueOf(validityDays));
        script.add(csrFile);
        String result = script.execute();
        if (result != null) {
            throw new CloudRuntimeException("Unable to setup keystore file");
        }

        final String csrString;
        try {
            csrString = FileUtils.readFileToString(new File(csrFile), Charset.defaultCharset());
        } catch (IOException e) {
            throw new CloudRuntimeException("Unable to read generated CSR file", e);
        }
        return new SetupKeystoreAnswer(csrString);
    }

    private Answer setupAgentCertificate(final SetupCertificateCommand cmd) {
        final String certificate = cmd.getCertificate();
        final String privateKey = cmd.getPrivateKey();
        final String caCertificates = cmd.getCaCertificates();

        logger.debug("Importing received certificate to agent's keystore");

        final File agentFile = PropertiesUtil.findConfigFile("agent.properties");
        if (agentFile == null) {
            return new Answer(cmd, false, "Failed to find agent.properties file");
        }
        final String keyStoreFile = agentFile.getParent() + "/" + KeyStoreUtils.KS_FILENAME;
        final String certFile = agentFile.getParent() + "/" + KeyStoreUtils.CERT_FILENAME;
        final String privateKeyFile = agentFile.getParent() + "/" + KeyStoreUtils.PKEY_FILENAME;
        final String caCertFile = agentFile.getParent() + "/" + KeyStoreUtils.CACERT_FILENAME;

        try {
            FileUtils.writeStringToFile(new File(certFile), certificate, Charset.defaultCharset());
            FileUtils.writeStringToFile(new File(caCertFile), caCertificates, Charset.defaultCharset());
            logger.debug("Saved received client certificate to: {}", certFile);
        } catch (IOException e) {
            throw new CloudRuntimeException("Unable to save received agent client and ca certificates", e);
        }

        String ksPassphrase = shell.getPersistentProperty(null, KeyStoreUtils.KS_PASSPHRASE_PROPERTY);
        Script script = new Script(keystoreCertImportScriptPath, 300000, logger);
        script.add(agentFile.getAbsolutePath());
        script.add(ksPassphrase);
        script.add(keyStoreFile);
        script.add(KeyStoreUtils.AGENT_MODE);
        script.add(certFile);
        script.add("");
        script.add(caCertFile);
        script.add("");
        script.add(privateKeyFile);
        script.add(privateKey);
        String result = script.execute();
        if (result != null) {
            throw new CloudRuntimeException("Unable to import certificate into keystore file");
        }
        return new SetupCertificateAnswer(true);
    }

    private void processManagementServerList(final List<String> msList, final String lbAlgorithm, final Long lbCheckInterval) {
        if (CollectionUtils.isNotEmpty(msList) && StringUtils.isNotEmpty(lbAlgorithm)) {
            try {
                final String newMSHosts = String.format("%s%s%s", com.cloud.utils.StringUtils.toCSVList(msList), IAgentShell.hostLbAlgorithmSeparator, lbAlgorithm);
                shell.setPersistentProperty(null, "host", newMSHosts);
                shell.setHosts(newMSHosts);
                shell.resetHostCounter();
                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();
        logger.debug("Received response: {}", response.toString());
        if (answer instanceof StartupAnswer) {
            processStartupAnswer(answer, response, link);
        } else if (answer instanceof AgentControlAnswer) {
            // Notice, we are doing callback while holding a lock!
            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 {
            updateLastPingResponseTime();
        }
    }

    public void processReadyCommand(final Command cmd) {
        final ReadyCommand ready = (ReadyCommand)cmd;
        // Set human readable sizes;
        Boolean humanReadable = ready.getEnableHumanReadableSizes();
        if (humanReadable != null){
            NumbersUtil.enableHumanReadableSizes = humanReadable;
        }

        logger.info("Processing agent ready command, agent id = {}, uuid = {}, name = {}", ready.getHostId(), ready.getHostUuid(), ready.getHostName());
        if (ready.getHostId() != null) {
            setId(ready.getHostId());
            setUuid(ready.getHostUuid());
            setName(ready.getHostName());
        }

        verifyAgentArch(ready.getArch());
        processManagementServerList(ready.getMsHostList(), ready.getLbAlgorithm(), ready.getLbCheckInterval());

        logger.info("Ready command is processed for agent [id: {}, uuid: {}, name: {}]", getId(), getUuid(), getName());
    }

    private void verifyAgentArch(String arch) {
        if (StringUtils.isNotBlank(arch)) {
            String agentArch = getAgentArch();
            if (!arch.equals(agentArch)) {
                logger.error("Unexpected arch {}, expected {}", agentArch, arch);
            }
        }
    }

    public void processOtherTask(final Task task) {
        final Object obj = task.get();
        if (obj instanceof Response) {
            if (System.currentTimeMillis() - lastPingResponseTime.get() > pingInterval * shell.getPingRetries()) {
                logger.error("Ping Interval has gone past {}. Won't reconnect to mgt server, as connection is still alive",
                        pingInterval * shell.getPingRetries());
                return;
            }

            final PingCommand ping = serverResource.getCurrentStatus(getId());
            final Request request = new Request(id, -1, ping, false);
            request.setSequence(getNextSequence());
            logger.debug("Sending ping: {}", request.toString());

            try {
                task.getLink().send(request.toBytes());
                //if i can send pingcommand out, means the link is ok
                updateLastPingResponseTime();
            } catch (final ClosedChannelException e) {
                logger.warn("Unable to send request to {} due to '{}', request: {}",
                        getLinkLog(task.getLink()), e.getMessage(), request);
            }

        } else if (obj instanceof Request) {
            final Request req = (Request)obj;
            final Command command = req.getCommand();
            if (command.getContextParam("logid") != null) {
                ThreadContext.put("logcontextid", command.getContextParam("logid"));
            }
            Answer answer = null;
            commandsInProgress.incrementAndGet();
            try {
                answer = serverResource.executeRequest(command);
            } finally {
                commandsInProgress.decrementAndGet();
            }
            if (answer != null) {
                final Response response = new Response(req, answer);

                logger.debug("Watch Sent: {}", response.toString());
                try {
                    task.getLink().send(response.toBytes());
                } catch (final ClosedChannelException e) {
                    logger.warn("Unable to send response: {}", response.toString());
                }
            }
        } else {
            logger.warn("Ignoring an unknown task");
        }
    }

    public void updateLastPingResponseTime() {
        lastPingResponseTime.set(System.currentTimeMillis());
    }

    protected long getNextSequence() {
        return sequence.getAndIncrement();
    }

    @Override
    public void registerControlListener(final IAgentControlListener listener) {
        controlListeners.add(listener);
    }

    @Override
    public void unregisterControlListener(final IAgentControlListener listener) {
        controlListeners.remove(listener);
    }

    @Override
    public AgentControlAnswer sendRequest(final AgentControlCommand cmd, final int timeoutInMilliseconds) throws AgentControlChannelException {
        final Request request = new Request(getId(), -1, new Command[] {cmd}, true, false);
        request.setSequence(getNextSequence());
        final AgentControlListener listener = new AgentControlListener(request);
        registerControlListener(listener);
        try {
            postRequest(request);
            synchronized (listener) {
                try {
                    listener.wait(timeoutInMilliseconds);
                } catch (final InterruptedException e) {
                    logger.warn("sendRequest is interrupted, exit waiting");
                }
            }
            return listener.getAnswer();
        } finally {
            unregisterControlListener(listener);
        }
    }

    @Override
    public void postRequest(final AgentControlCommand cmd) throws AgentControlChannelException {
        final Request request = new Request(getId(), -1, new Command[] {cmd}, true, false);
        request.setSequence(getNextSequence());
        postRequest(request);
    }

    private void postRequest(final Request request) throws AgentControlChannelException {
        if (link != null) {
            try {
                link.send(request.toBytes());
            } catch (final ClosedChannelException e) {
                logger.warn("Unable to post agent control request: {}", request.toString());
                throw new AgentControlChannelException("Unable to post agent control request due to " + e.getMessage());
            }
        } else {
            throw new AgentControlChannelException("Unable to post agent control request as link is not available");
        }
    }

    public class AgentControlListener implements IAgentControlListener {
        private AgentControlAnswer _answer;
        private final Request _request;

        public AgentControlListener(final Request request) {
            _request = request;
        }

        public AgentControlAnswer getAnswer() {
            return _answer;
        }

        @Override
        public Answer processControlRequest(final Request request, final AgentControlCommand cmd) {
            return null;
        }

        @Override
        public void processControlResponse(final Response response, final AgentControlAnswer answer) {
            if (_request.getSequence() == response.getSequence()) {
                _answer = answer;
                synchronized (this) {
                    notifyAll();
                }
            }
        }
    }

    protected class ShutdownThread extends Thread {
        Agent _agent;

        public ShutdownThread(final Agent agent) {
            super("AgentShutdownThread");
            _agent = agent;
        }

        @Override
        public void run() {
            _agent.stop(ShutdownCommand.Requested, null);
        }
    }

    public class WatchTask implements Runnable {
        protected Request _request;
        protected Agent _agent;
        protected Link _link;

        public WatchTask(final Link link, final Request request, final Agent agent) {
            super();
            _request = request;
            _link = link;
            _agent = agent;
        }

        @Override
        public void run() {
            logger.trace("Scheduling {}", (_request instanceof Response ? "Ping" : "Watch Task"));
            try {
                if (_request instanceof Response) {
                    outRequestHandler.submit(new ServerHandler(Task.Type.OTHER, _link, _request));
                } else {
                    _link.schedule(new ServerHandler(Task.Type.OTHER, _link, _request));
                }
            } catch (final ClosedChannelException e) {
                logger.warn("Unable to schedule task because channel is closed");
            }
        }
    }

    public class StartupTask implements Runnable {
        protected Link _link;
        private final AtomicBoolean cancelled = new AtomicBoolean(false);

        public StartupTask(final Link link) {
            logger.debug("Startup task created");
            _link = link;
        }

        public boolean cancel() {
            // TimerTask.cancel may fail depends on the calling context
            if (cancelled.compareAndSet(false, true)) {
                startupWait = DEFAULT_STARTUP_WAIT;
                logger.debug("Startup task cancelled");
            }
            return true;
        }

        @Override
        public void run() {
            if (cancelled.compareAndSet(false, true)) {
                logger.info("The running startup command is now invalid. Attempting reconnect");
                startupTask.set(null);
                startupWait = DEFAULT_STARTUP_WAIT * 2;
                logger.debug("Executing reconnect from task - {}", () -> getLinkLog(_link));
                reconnect(_link);
            }
        }
    }

    public class AgentRequestHandler extends Task {
        public AgentRequestHandler(final Task.Type type, final Link link, final Request req) {
            super(type, link, req);
        }

        @Override
        protected void doTask(final Task task) throws TaskExecutionException {
            final Request req = (Request)get();
            if (!(req instanceof Response)) {
                processRequest(req, task.getLink());
            }
        }
    }

    public class ServerHandler extends Task {
        public ServerHandler(final Task.Type type, final Link link, final byte[] data) {
            super(type, link, data);
        }

        public ServerHandler(final Task.Type type, final Link link, final Request req) {
            super(type, link, req);
        }

        @Override
        public void doTask(final Task task) throws TaskExecutionException {
            if (task.getType() == Task.Type.CONNECT) {
                shell.getBackoffAlgorithm().reset();
                setLink(task.getLink());
                sendStartup(task.getLink());
            } else if (task.getType() == Task.Type.DATA) {
                Request request;
                try {
                    request = Request.parse(task.getData());
                    if (request instanceof Response) {
                        //It's for pinganswer etc, should be processed immediately.
                        processResponse((Response)request, task.getLink());
                    } else {
                        //put the requests from mgt server into another thread pool, as the request may take a longer time to finish. Don't block the NIO main thread pool
                        //processRequest(request, task.getLink());
                        requestHandler.submit(new AgentRequestHandler(getType(), getLink(), request));
                    }
                } catch (final ClassNotFoundException e) {
                    logger.error("Unable to find this request ");
                } catch (final Exception e) {
                    logger.error("Error parsing task", e);
                }
            } else if (task.getType() == Task.Type.DISCONNECT) {
                logger.debug("Executing disconnect task - {}", () -> getLinkLog(task.getLink()));
                reconnect(task.getLink());
            } else if (task.getType() == Task.Type.OTHER) {
                processOtherTask(task);
            }
        }
    }

    /**
     * Task stops the current agent and launches a new agent
     * when there are no outstanding jobs in the agent's task queue
     */
    public class PostCertificateRenewalTask extends ManagedContextTimerTask {

        private Agent agent;

        public PostCertificateRenewalTask(final Agent agent) {
            this.agent = agent;
        }

        @Override
        protected void runInContext() {
            while (true) {
                try {
                    if (commandsInProgress.get() == 0) {
                        logger.debug("Running post certificate renewal task to restart services.");

                        // Let the resource perform any post certificate renewal cleanups
                        serverResource.executeRequest(new PostCertificateRenewalCommand());

                        IAgentShell shell = agent.shell;
                        ServerResource resource = agent.serverResource.getClass().getDeclaredConstructor().newInstance();

                        // Stop current agent
                        agent.cancelTasks();
                        agent.reconnectAllowed = false;
                        Runtime.getRuntime().removeShutdownHook(agent.shutdownThread);
                        agent.stop(ShutdownCommand.Requested, "Restarting due to new X509 certificates");

                        // Nullify references for GC
                        agent.shell = null;
                        agent.watchList = null;
                        agent.shutdownThread = null;
                        agent.controlListeners = null;
                        agent = null;

                        // Start a new agent instance
                        shell.launchNewAgent(resource);
                        return;
                    }
                    logger.debug("Other tasks are in progress, will retry post certificate renewal command after few seconds");
                    Thread.sleep(5000);
                } catch (final Exception e) {
                    logger.warn("Failed to execute post certificate renewal command:", e);
                    break;
                }
            }
        }
    }

    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();
                logger.debug("Running preferred host checker task, connected host={}, preferred host={}",
                        connectedHost, preferredHost);
                if (preferredHost == null || preferredHost.equals(connectedHost) || link == null) {
                    return;
                }
                boolean isHostUp = false;
                try (final Socket socket = new Socket()) {
                    socket.connect(new InetSocketAddress(preferredHost, shell.getPort()), 5000);
                    isHostUp = true;
                } catch (final IOException e) {
                    logger.debug("Host: {} is not reachable", preferredHost);
                }
                if (isHostUp && link != null && commandsInProgress.get() == 0) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Preferred host {} is found to be reachable, trying to reconnect", preferredHost);
                    }
                    shell.resetHostCounter();
                    reconnect(link);
                }
            } catch (Throwable t) {
                logger.error("Error caught while attempting to connect to preferred host", t);
            }
        }
    }
}