review comments

Author: sureshanaparti

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

@@ -23,6 +23,7 @@
 import java.net.ConnectException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.net.Socket;
 import java.net.UnknownHostException;
 import java.nio.channels.ClosedChannelException;
@@ -680,7 +681,7 @@ protected void reconnect(Link link, String preferredHost, boolean forceReconnect
                         try {
                             stopAndCleanupConnection();
                         } catch (Exception ex) {
-                            logger.warn("Got an exception during stop and cleanup connection", e);
+                            logger.warn("Got an exception during stop and cleanup connection", ex);
                         }
 
                         updateRefuseLoopState(e, host, firstRefuseLoopHostRef, ignoreAvoidMsHostListRef, skipTimeoutRef);
@@ -734,13 +735,13 @@ private void updateRefuseLoopState(Exception e, String host, AtomicReference<Str
         if (skipTimeout && firstRefuseLoopHost == null) {
             firstRefuseLoopHostRef.set(host);
             ignoreAvoidMsHostListRef.set(false);
-            logger.debug("Started refuse loop for host {}", firstRefuseLoopHost);
+            logger.debug("Started refuse loop for host {}", host);
             // closed "refuse loop"
         } else if (skipTimeout && firstRefuseLoopHost.equalsIgnoreCase(host)) {
             ignoreAvoidMsHostListRef.set(true);
             logger.debug("Closed refuse loop for host {}", firstRefuseLoopHost);
             // got non "refuse" related issue, break "refuse loop"
-        } else if (!skipTimeout && (firstRefuseLoopHostRef != null || ignoreAvoidMsHostListRef.get())) {
+        } else if (!skipTimeout && (firstRefuseLoopHost != null || ignoreAvoidMsHostListRef.get())) {
             logger.debug("Broke refuse loop for host {} by {}", firstRefuseLoopHost, host);
             firstRefuseLoopHostRef.set(null);
             ignoreAvoidMsHostListRef.set(false);
@@ -839,7 +840,8 @@ private boolean isReconnectStormDetected(Link link, String preferredHost, String
         logger.debug("Calling storm guard");
         boolean reconnectForCurrentLink = link == this.link;
         boolean currentLinkTerminated = this.link != null && this.link.isTerminated();
-        boolean reconnectForNewHost = this.hostname != null && this.hostname.equals(preferredHost);
+        String connectedHost = connection != null ? connection.getHost() : null;
+        boolean reconnectForNewHost = org.apache.commons.lang3.StringUtils.isNotBlank(preferredHost) && !preferredHost.equalsIgnoreCase(connectedHost);
         // if none of the above is true
         boolean stormDetected = !(reconnectForCurrentLink || currentLinkTerminated || reconnectForNewHost);
         // connection storm guard
@@ -1708,12 +1710,12 @@ public int getNewConnectionsCount() {
     }
 
     @Override
-    public void registerNewConnection(InetSocketAddress address) {
+    public void registerNewConnection(SocketAddress address) {
         logger.trace("Register new connection to {}", address);
     }
 
     @Override
-    public void unregisterNewConnection(InetSocketAddress address) {
+    public void unregisterNewConnection(SocketAddress address) {
         logger.trace("Unregister new connection to {}", address);
     }
 

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

@@ -101,7 +101,7 @@ public void setBackoffAlgorithm(BackoffAlgorithm backoffAlgorithm) {
         try {
             backoffAlgorithm.getConfiguration().forEach((key, value) -> setPersistentProperty(null, key, value));
         } catch (RuntimeException e) {
-            LOGGER.warn("Failed to persist backoff properties");
+            LOGGER.warn("Failed to persist backoff properties", e);
         }
     }
 
@@ -420,7 +420,7 @@ public void init(String[] args) throws ConfigurationException {
         if (LOGGER.isDebugEnabled()) {
             List<String> properties = Collections.list((Enumeration<String>)_properties.propertyNames());
             for (String property : properties) {
-                LOGGER.debug("Found property: {}, value: {}", property, _properties.getProperty(property));
+                LOGGER.debug("Found property: {}", property);
             }
         }
 
@@ -439,8 +439,7 @@ public void init(String[] args) throws ConfigurationException {
             setBackoffAlgorithm(BackoffFactory.create(_properties));
             LOGGER.info("Created {} delay algorithm implementation", getBackoffAlgorithm().getClass().getName());
         } catch (RuntimeException e) {
-            String msg = String.format("Failed to create backoff with provided properties %s, failing back to default", _properties);
-            LOGGER.warn(msg, e);
+            LOGGER.warn("Failed to create backoff algorithm from the provided properties, falling back to default", e);
             setBackoffAlgorithm(BackoffFactory.createDefault(_properties));
         }
     }
@@ -583,7 +582,6 @@ public static void main(String[] args) {
             shell.start();
         } catch (ConfigurationException e) {
             LOGGER.fatal(e.getMessage(), e);
-            System.out.println(e.getMessage());
         }
     }
 

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

@@ -125,30 +125,30 @@ public void updateHostStatusCheckDelay(int newDelaySec) {
     }
 
     /**
-     * Task wait for the Host to be available to connect to submit {@link StartupCommand}.
+     * Task waits for the Host to be available to connect to submit {@link StartupCommand}.
      * Checks Host status on Management Server cluster and submit {@link StartupCommand} only if there is no lock and
      * Host is not {@link Status#Connecting}.
      */
     public static class HostStatusTask implements Runnable, AsyncSend {
         private final Set<Status> operationalStatuses = Set.of(Status.Connecting, Status.Up, Status.Rebalancing);
 
-        private final Link _link;
-        private final boolean _forceConnect;
-        private final Agent _agent;
-        private final AtomicReference<? extends ScheduledFuture<?>> _futureRef;
+        private final Link link;
+        private final boolean forceConnect;
+        private final Agent agent;
+        private final AtomicReference<? extends ScheduledFuture<?>> futureRef;
 
         public HostStatusTask(Link link, boolean forceConnect, Agent agent,
                               AtomicReference<? extends ScheduledFuture<?>> futureRef) {
             logger.debug("{} created", this.getClass().getSimpleName());
-            _link = link;
-            _forceConnect = forceConnect;
-            _agent = agent;
-            _futureRef = futureRef;
+            this.link = link;
+            this.forceConnect = forceConnect;
+            this.agent = agent;
+            this.futureRef = futureRef;
         }
 
         private void cancel() {
             logger.debug("Cancelling future");
-            Optional.ofNullable(_futureRef.get())
+            Optional.ofNullable(futureRef.get())
                     .filter(Predicate.not(ScheduledFuture::isCancelled))
                     .ifPresent(future -> future.cancel(true));
             logger.debug("Cancelled future");
@@ -165,7 +165,7 @@ public void run() {
         }
 
         private void runInternal() {
-            ServerAttache attache = (ServerAttache) _link.attachment();
+            ServerAttache attache = (ServerAttache) link.attachment();
             if (attache == null || attache.getLink() == null) {
                 cancel();
                 return;
@@ -176,8 +176,8 @@ private void runInternal() {
                 answer = getAgentConnectStatusAnswer(attache);
             } catch (IOException e) {
                 cancel();
-                logger.error("The connection to {} interrupted, restarting the whole process", _link, e);
-                _agent.getRequestHandler().submit(() -> _agent.reconnect(_link, null, _forceConnect));
+                logger.error("The connection to {} interrupted, restarting the whole process", link, e);
+                agent.getRequestHandler().submit(() -> agent.reconnect(link, null, forceConnect));
                 return;
             }
             if (answer == null) {
@@ -190,24 +190,24 @@ private void runInternal() {
                 // send startup command here
                 logger.info("There is no lock and Host status is {}", status);
                 try {
-                    sendStartupCommand(_link, _forceConnect);
-                    logger.debug("Sending startup command to {} finished", _link);
+                    sendStartupCommand();
+                    logger.debug("Sending startup command to {} finished", link);
                     cancel();
                     logger.debug("Unscheduled {}", getClass().getSimpleName());
                 } catch (RuntimeException e) {
-                    logger.error("Failed to send startup command to {}", _link, e);
+                    logger.error("Failed to send startup command to {}", link, e);
                 } catch (IOException e) {
                     cancel();
-                    logger.error("The connection to {} interrupted, restarting the whole process", _link, e);
-                    _agent.getRequestHandler().submit(() -> _agent.reconnect(_link, null, _forceConnect));
+                    logger.error("The connection to {} interrupted, restarting the whole process", link, e);
+                    agent.getRequestHandler().submit(() -> agent.reconnect(link, null, forceConnect));
                 }
             } else {
                 logger.info("There is lock and Host status is {}, will retry later", status);
             }
         }
 
         private AgentConnectStatusAnswer getAgentConnectStatusAnswer(ServerAttache attache) throws IOException {
-            AgentConnectStatusCommand command = _agent.setupAgentConnectStatusCommand(new AgentConnectStatusCommand());
+            AgentConnectStatusCommand command = agent.setupAgentConnectStatusCommand(new AgentConnectStatusCommand());
             var commands = new Command[]{command};
             try {
                 return send(attache, commands, AgentConnectStatusAnswer.class, DEFAULT_ASYNC_COMMAND_TIMEOUT_SEC);
@@ -218,26 +218,26 @@ private AgentConnectStatusAnswer getAgentConnectStatusAnswer(ServerAttache attac
             }
         }
 
-        public void sendStartupCommand(Link link, boolean connectionTransfer) throws IOException {
+        public void sendStartupCommand() throws IOException {
             ServerAttache attache = (ServerAttache) link.attachment();
             if (attache == null || attache.getLink() == null) {
                 return;
             }
-            ServerResource serverResource = _agent.getResource();
+            ServerResource serverResource = agent.getResource();
             StartupCommand[] startup = serverResource.initialize();
             if (ArrayUtils.isEmpty(startup)) {
                 logger.warn("No startup commands returned from {}, Startup command sending skipped", serverResource.getName());
                 return;
             }
-            String msHostList = _agent.getPersistentProperty("host");
+            String msHostList = agent.getPersistentProperty("host");
             // need to downcast StartupCommand[] to Command[], otherwise logger will fail to decode JSON on MS side
             Command[] commands = new Command[startup.length];
             for (int i = 0; i < startup.length; i++) {
                 StartupCommand command = startup[i];
                 commands[i] = command;
-                _agent.setupStartupCommand(command);
+                agent.setupStartupCommand(command);
                 command.setMSHostList(msHostList);
-                command.setConnectionTransferred(connectionTransfer);
+                command.setConnectionTransferred(forceConnect);
                 ThreadContextCommandUtil.setContextInCommand(command);
             }
             String commandName = commands[0].getClass().getSimpleName();
@@ -263,29 +263,29 @@ public void sendStartupCommand(Link link, boolean connectionTransfer) throws IOE
                 AgentConnectStatusAnswer answer = getAgentConnectStatusAnswer(attache);
                 if (answer == null) {
                     logger.warn("Received empty agent connect status answer, reconnecting");
-                    _agent.getRequestHandler().submit(() -> _agent.reconnect(link, null, connectionTransfer));
+                    agent.getRequestHandler().submit(() -> agent.reconnect(link, null, forceConnect));
                     return;
                 }
                 Boolean lockAvailable = answer.isLockAvailable();
                 Status status = answer.getHostStatus();
                 if (Boolean.TRUE.equals(lockAvailable) && status != null && operationalStatuses.contains(status)) {
-                    logger.info("Host is in operational state {} on {}", status, _link);
+                    logger.info("Host is in operational state {} on {}", status, link);
                 } else if (Boolean.FALSE.equals(lockAvailable)) {
-                    logger.info("Host is locked and has state {} on {}", status, _link);
+                    logger.info("Host is locked and has state {} on {}", status, link);
                 } else {
-                    logger.info("Host is locked and has state {} on {}, reconnecting", status, _link);
-                    _agent.getRequestHandler().submit(() -> _agent.reconnect(link, null, connectionTransfer));
+                    logger.info("Host is locked and has state {} on {}, reconnecting", status, link);
+                    agent.getRequestHandler().submit(() -> agent.reconnect(link, null, forceConnect));
                     return;
                 }
             }
             if (serverResource instanceof ResourceStatusUpdater) {
-                ((ResourceStatusUpdater) serverResource).registerStatusUpdater(_agent);
+                ((ResourceStatusUpdater) serverResource).registerStatusUpdater(agent);
             }
 
         }
 
         public Agent getAgent() {
-            return _agent;
+            return agent;
         }
     }
 

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

@@ -124,7 +124,7 @@ protected void cancel(Request req) {
     protected synchronized void cancel(long seq) {
         logger.debug(log(seq, "Cancelling."));
 
-        ServerListener listener = _waitForList.remove(seq);
+        ServerListener listener = unregisterListener(seq);
         if (listener != null) {
             listener.processDisconnect();
         }
@@ -194,7 +194,7 @@ public boolean processAnswers(long seq, Response resp) {
                 logger.debug(log(seq, "Unable to find listener."));
             } else {
                 processed = monitor.processAnswers(seq, answers);
-                logger.trace(log(seq, (processed ? "" : " did not ") + " processed "));
+                logger.trace(log(seq, processed ? "Processed" : "Not processed"));
                 if (!monitor.isRecurring()) {
                     unregisterListener(seq);
                 }
@@ -486,7 +486,7 @@ protected void runInContext() {
                     listener.processTimeout(_seq);
                 }
             } catch (Exception e) {
-                ServerAttache.logger.warn("Exception ", e);
+                ServerAttache.logger.warn(log(_seq, "Failed to process listener timeout"), e);
             }
         }
     }

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

@@ -31,8 +31,9 @@
 public class AgentConnectStatusAnswer extends Answer {
 
     /**
-     * {@link Boolean#TRUE} means host has {@link GlobalLock#lock(int)} acquired, otherwise {@link Boolean#FALSE},
-     * and null if there is an error during executing {@link AgentConnectStatusCommand}.
+     * {@link Boolean#TRUE} means the host connect {@link GlobalLock} is available (no management server holds it),
+     * so the agent may send its {@link StartupCommand}; {@link Boolean#FALSE} means another connect process holds it;
+     * null if there is an error during executing {@link AgentConnectStatusCommand}.
      */
     private Boolean lockAvailable;
     /**

core/src/main/java/org/apache/cloudstack/threadcontext/ThreadContextCommandUtil.java

@@ -30,13 +30,24 @@ public class ThreadContextCommandUtil {
 
     /**
      * Propagate UUID and log context ID from Command trace context to MDC.
+     * A key absent from the command is removed from MDC so that a worker thread does not
+     * carry over the context of a previously processed command.
      *
      * @param cmd the command containing trace context parameters
      */
     public static void propagateContextFromCommand(Command cmd) {
-        if (cmd != null) {
-            ThreadContextUtil.setLogContextId(cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_LOG_ID_KEY));
-            ThreadContextUtil.setUuid(cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_UUID_KEY));
+        if (cmd == null) {
+            return;
+        }
+        putOrRemove(ThreadContextUtil.MDC_LOG_CONTEXT_ID_KEY, cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_LOG_ID_KEY));
+        putOrRemove(ThreadContextUtil.MDC_UUID_KEY, cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_UUID_KEY));
+    }
+
+    private static void putOrRemove(String mdcKey, String value) {
+        if (StringUtils.isNotEmpty(value)) {
+            ThreadContext.put(mdcKey, value);
+        } else {
+            ThreadContext.remove(mdcKey);
         }
     }