Copilot commented on code in PR #13345:
URL: https://github.com/apache/cloudstack/pull/13345#discussion_r3864133621
##########
agent/src/main/java/com/cloud/agent/Agent.java:
##########
@@ -623,98 +559,363 @@ public Task create(final Task.Type type, final Link
link, final byte[] data) {
return new ServerHandler(type, link, data);
}
- protected void reconnect(final Link link) {
- reconnect(link, null, false);
+ protected void closeAndTerminateLink(Link link) {
+ Optional.ofNullable(link)
+ .map(Link::attachment)
+ .filter(ServerAttache.class::isInstance)
+ .map(ServerAttache.class::cast)
+ .ifPresentOrElse(ServerAttache::disconnect, () -> {
+ if (link != null) {
+ link.close();
+ link.terminated();
+ }
+ });
}
- protected void reconnect(final Link link, String preferredMSHost, boolean
forTransfer) {
- if (!(forTransfer || reconnectAllowed)) {
- logger.debug("Reconnect requested but it is not allowed {}", () ->
getLinkLog(link));
+ protected void stopAndCleanupConnection() {
+ if (connection == null) {
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);
- String host = preferredMSHost;
- if (org.apache.commons.lang3.StringUtils.isBlank(host)) {
- host = shell.getNextHost();
- }
- List<String> avoidMSHostList = shell.getAvoidHosts();
- do {
- if (CollectionUtils.isEmpty(avoidMSHostList) ||
!avoidMSHostList.contains(host)) {
- 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);
- }
+ NioConnection connection = this.connection;
+ connection.stop();
+ try {
+ connection.cleanUp();
+ } catch (final IOException e) {
+ logger.warn("Fail to clean up old connection", e);
+ }
+
+ try {
+ while (connection.isStartup()) {
+ logger.debug("Waiting for connection graceful stop");
+ shell.getBackoffAlgorithm().waitBeforeRetry();
+ connection.stop();
}
- shell.getBackoffAlgorithm().waitBeforeRetry();
- host = shell.getNextHost();
- } while (!connection.isStartup());
- shell.updateConnectedHost(((NioClient)connection).getHost());
- logger.info("Connected to the host: {}", shell.getConnectedHost());
+ } catch (Exception e) {
+ logger.warn("Failed to gracefully stop connection", e);
+ }
+ logger.debug("Connection stopped");
+ }
+
+ /**
+ * Select the host to reconnect to based on priority:
+ * 1. preferredHost if defined and not blank
+ * 2. Link's socket address IP if available and not null
+ * 3. shell.getNextHost() if the above two options are not met
+ *
+ * @param preferredHost the preferred host to connect to
+ * @param link the current link which may contain socket address
information
+ * @return the host to connect to
+ */
+ protected String selectReconnectionHost(String preferredHost, Link link) {
+ return Optional.ofNullable(preferredHost)
+ .filter(org.apache.commons.lang3.StringUtils::isNotBlank)
+ .orElseGet(() -> Optional.ofNullable(link)
+ .map(Link::getSocketAddress)
+ .map(InetSocketAddress::getAddress)
+ .map(InetAddress::getHostAddress)
+ .orElseGet(shell::getNextHost));
}
- protected void closeAndTerminateLink(final Link link) {
- if (link == null) {
+ /**
+ * Reconnect to Management Server.
+ *
+ * @param link - connection holder
+ * @param preferredHost - if defined, reconnect will be performed to this
Host first,
+ * otherwise will be used {@link
IAgentShell#getNextHost()}
+ * @param forceReconnect - expected to be true if called by {@link
MigrateAgentConnectionCommand},
+ * this is only "switch Management Server", it does
not perform full Host Connect process.
+ */
+ protected void reconnect(Link link, String preferredHost, boolean
forceReconnect) {
+ if (!reconnectLock.compareAndSet(false, true)) {
+ logger.warn("Reconnect is already running, exiting");
return;
}
- link.close();
- link.terminated();
+ String requestedLink =
Optional.ofNullable(link).map(Link::toString).orElse("N/A");
+ String currentLink =
Optional.ofNullable(this.link).map(Link::toString).orElse("N/A");
+ logger.info("Reconnect info: provided link: {}, agent link: {},
preferred host: {}, force" +
+ " reconnect: {}", requestedLink, currentLink, preferredHost,
forceReconnect);
+
+ try {
+ logger.debug("Obtained reconnect lock");
+ if (!(forceReconnect || reconnectAllowed)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Reconnect requested but it is not allowed
{}", link);
+ }
+ return;
+ }
+
+ if (isReconnectStormDetected(link, preferredHost, requestedLink,
currentLink)) {
+ return;
+ }
+
+ cleanupConnectionBeforeReconnect(link);
+ // start with preferred host
+ String host = selectReconnectionHost(preferredHost, link);
+
+ String hostLog = LogUtils.getHostLog(host, shell.getPort());
+ List<String> avoidMsHostList =
Optional.ofNullable(shell.getAvoidHosts()).orElseGet(List::of);
+ // pointer to the first element of "refuse loop"
+ AtomicReference<String> firstRefuseLoopHostRef = new
AtomicReference<>(null);
+ // to break deadlock where "non-avoid" MS Hosts are down and only
"avoid" are up
+ AtomicBoolean ignoreAvoidMsHostListRef = new AtomicBoolean(false);
+ do {
+ AtomicBoolean skipTimeoutRef = new AtomicBoolean(false);
+ String parentLogContextId = (String)
ThreadContext.get("logcontextid");
+ if (parentLogContextId != null) {
+ ThreadContext.put("logcontextid-parent",
parentLogContextId);
+ }
+ ThreadContext.put("logcontextid",
UuidUtils.first(UUID.randomUUID().toString()));
+ if (ignoreAvoidMsHostListRef.get() ||
!avoidMsHostList.contains(host)) {
+ connection = new NioClient(getAgentName(), host,
shell.getPort(), shell.getWorkers(),
+ shell.getSslHandshakeTimeout(), this);
+ logger.info("Reconnecting to host: {}", hostLog);
+ try {
+ connection.start();
+ // successfully connected, skip the rest
+ continue;
+ } catch (Exception e) {
+ logReconnectionFailure(e, hostLog);
+
+ try {
+ stopAndCleanupConnection();
+ } catch (Exception ex) {
+ logger.warn("Got an exception during stop and
cleanup connection", e);
+ }
+
+ updateRefuseLoopState(e, host, firstRefuseLoopHostRef,
ignoreAvoidMsHostListRef, skipTimeoutRef);
+ }
+ } else {
+ logger.debug("Next host {} is in avoid list, skipped",
hostLog);
+ if
(org.apache.commons.lang3.StringUtils.isBlank(preferredHost)) {
+ logHostLists(avoidMsHostList);
+ skipTimeoutRef.set(true);
+ }
+ }
+ if (!skipTimeoutRef.get()) {
+ shell.getBackoffAlgorithm().waitBeforeRetry();
+ }
+ host = shell.getNextHost();
+ hostLog = LogUtils.getHostLog(host, shell.getPort());
+ logger.debug("Next host to connect: {}", hostLog);
+ } while (!connection.isStartup());
+ // successfully connected
+ shell.updateConnectedHost(((NioClient) connection).getHost());
+ String msg = String.format("Connected to the host: %s (%s)",
shell.getConnectedHost(), this.link);
+ logger.info(msg);
+ } finally {
+ reconnectLock.set(false);
+ logger.debug("Removed reconnect lock");
+ }
}
- protected void stopAndCleanupConnection(boolean waitForStop) {
- if (connection == null) {
- return;
+ /**
+ * Handles "Connection refused" loop detection and determines if backoff
timeout should be skipped.
+ * Manages refuse loop state to detect when all management servers have
been tried and need
+ * to ignore avoid list to prevent deadlock.
+ *
+ * @param e the exception from connection attempt
+ * @param host the current host being attempted
+ * @param firstRefuseLoopHostRef reference to first host in refuse loop
(modified by this method)
+ * @param ignoreAvoidMsHostListRef flag to ignore avoid list (modified by
this method)
+ * @return true if timeout should be skipped (connection refused), false
otherwise
+ */
+ private void updateRefuseLoopState(Exception e, String host,
AtomicReference<String> firstRefuseLoopHostRef, AtomicBoolean
ignoreAvoidMsHostListRef, AtomicBoolean skipTimeoutRef) {
+ // we are skipping timeout for "Connection refused" to not waste time
on down MS
+ boolean skipTimeout = Optional.ofNullable(e.getCause())
+ .filter(ConnectException.class::isInstance)
+ .map(Throwable::getMessage)
+ .filter(CONNECTION_REFUSED_MSG::equalsIgnoreCase)
+ .isPresent();
+ skipTimeoutRef.set(skipTimeout);
+ String firstRefuseLoopHost = firstRefuseLoopHostRef.get();
+ // for each "Connection refused" (maybe need to have a copy of
variable with better name)
+ // start "refuse loop"
+ if (skipTimeout && firstRefuseLoopHost == null) {
+ firstRefuseLoopHostRef.set(host);
+ ignoreAvoidMsHostListRef.set(false);
+ logger.debug("Started refuse loop for host {}",
firstRefuseLoopHost);
+ // 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())) {
+ logger.debug("Broke refuse loop for host {} by {}",
firstRefuseLoopHost, host);
+ firstRefuseLoopHostRef.set(null);
+ ignoreAvoidMsHostListRef.set(false);
}
- connection.stop();
+ }
+
+ /**
+ * Logs reconnection failure with appropriate level based on rejection
reason.
+ * If connection was rejected due to max concurrent connections limit
(Broken pipe),
+ * logs as warning. Otherwise logs as info.
+ *
+ * @param e the exception that occurred during reconnection attempt
+ * @param hostLog the formatted host log string for logging
+ */
+ private void logReconnectionFailure(Exception e, String hostLog) {
+ // check if got NIO Connection exception, caused by IO Exception
"Broken pipe"
+ boolean rejectedByMs =
Optional.of(e).filter(NioConnectionException.class::isInstance)
+ .map(Exception::getCause)
+ .filter(IOException.class::isInstance)
+ .map(IOException.class::cast)
+ .map(IOException::getMessage)
+ .filter(BROKEN_PIPE_MSG::equalsIgnoreCase)
+ .isPresent();
+ if (rejectedByMs) {
+ logger.warn("Attempted to re-connect to {}, but rejected" +
+ " due to 'agent.max.concurrent.new.connections' reached
limit," +
+ " will try again", hostLog, e);
+ } else {
+ logger.info("Attempted to re-connect to {}, but got exception," +
+ " will try again", hostLog, e);
+ }
+ }
+
+ /**
+ * Logs all management server host lists for debugging reconnection logic.
+ * Outputs defined hosts, hosts to avoid, and calculated available hosts.
+ *
+ * @param avoidMsHostList list of management server hosts to avoid during
reconnection
+ */
+ private void logHostLists(List<String> avoidMsHostList) {
+ logger.debug("Preferred host is not defined");
try {
- connection.cleanUp();
- } catch (final IOException e) {
- logger.warn("Fail to clean up old connection. {}", e);
+ List<String> hostsList = Optional.ofNullable(shell.getHosts())
+ .map(Arrays::asList)
+ .orElseGet(List::of);
+
+ List<String> hostsShortList = new ArrayList<>(hostsList);
+ hostsShortList.removeAll(avoidMsHostList);
+
+ logger.info("Defined hosts: {} Avoid hosts: {} Available hosts:
{}",
+ String.join(", ", hostsList), String.join(", ",
avoidMsHostList), String.join(", ", hostsShortList));
+ } catch (Exception e) {
+ logger.warn("Failed to calculate next host logic", e);
}
- if (!waitForStop) {
- return;
+ }
+
+ /**
+ * Cleans up current connection state before attempting reconnection.
+ * Stops host connect process, terminates links, cancels scheduled tasks,
+ * notifies server resource about disconnection, and resets connection
tracking.
+ *
+ * @param link the link that triggered reconnection
+ */
+ private void cleanupConnectionBeforeReconnect(Link link) {
+ String lastConnectedHost = shell.getConnectedHost();
+ try {
+ // reset Host status track and Startup process initiating
+ logger.debug("Stopping Host Connect process");
+ hostConnectProcess.stop();
+ closeAndTerminateLink(link);
+ closeAndTerminateLink(this.link);
+ setLink(null);
+ cancelTasks();
+ serverResource.disconnected();
+ stopAndCleanupConnection();
+ shell.updateConnectedHost(null);
+ } catch (Exception ex) {
+ logger.error("Failed to cleanup previous connection", ex);
+ }
+ logger.info("Lost connection to host: {}. Attempting reconnection
while we still have" +
+ " {} commands in progress.", lastConnectedHost,
commandsInProgress.get());
+ }
+
+ /**
+ * Detects reconnection storm by checking if the reconnect request is
redundant.
+ * This prevents processing stale reconnection requests for old links when
+ * agent has already established a new connection.
+ *
+ * @param link the link requesting reconnection
+ * @param preferredHost the preferred host to reconnect to (may be null)
+ * @param requestedLink string representation of the requested link for
logging
+ * @param currentLink string representation of the current agent link for
logging
+ * @return true if reconnection storm is detected and request should be
skipped, false otherwise
+ */
+ private boolean isReconnectStormDetected(Link link, String preferredHost,
String requestedLink, String currentLink) {
+ 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);
+ // if none of the above is true
Review Comment:
`reconnectForNewHost` compares `preferredHost` to `this.hostname` (the
agent’s own hostname), which is unrelated to the management server host
selection and can cause the storm-guard to behave incorrectly. Consider basing
this on the currently connected management server host instead (or drop this
condition entirely if link/termination checks are sufficient).
##########
server/src/main/java/org/apache/cloudstack/agent/lb/IndirectAgentLBServiceImpl.java:
##########
@@ -446,6 +446,7 @@ protected boolean migrateNonRoutingHostAgentsInZone(String
fromMsUuid, long from
break;
}
+ // FIXME: it is fire and forget, Management Server will never know
if task failed
migrateAgentsExecutorService.submit(new
MigrateAgentConnectionTask(fromMsId, hostId, dc.getId(), orderedHostIdList,
avoidMsList, lbCheckInterval, lbAlgorithm, lbAlgorithmChanged));
}
Review Comment:
There’s an explicit `FIXME` noting the migration tasks are fire-and-forget
and failures are not observable by the management server. Leaving this as-is
can lead to silent partial migrations. Consider tracking submitted futures and
reporting failures (or propagating them via status/metrics) instead of ignoring
task outcomes.
##########
agent/src/main/java/com/cloud/agent/Agent.java:
##########
@@ -623,98 +559,363 @@ public Task create(final Task.Type type, final Link
link, final byte[] data) {
return new ServerHandler(type, link, data);
}
- protected void reconnect(final Link link) {
- reconnect(link, null, false);
+ protected void closeAndTerminateLink(Link link) {
+ Optional.ofNullable(link)
+ .map(Link::attachment)
+ .filter(ServerAttache.class::isInstance)
+ .map(ServerAttache.class::cast)
+ .ifPresentOrElse(ServerAttache::disconnect, () -> {
+ if (link != null) {
+ link.close();
+ link.terminated();
+ }
+ });
}
- protected void reconnect(final Link link, String preferredMSHost, boolean
forTransfer) {
- if (!(forTransfer || reconnectAllowed)) {
- logger.debug("Reconnect requested but it is not allowed {}", () ->
getLinkLog(link));
+ protected void stopAndCleanupConnection() {
+ if (connection == null) {
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);
- String host = preferredMSHost;
- if (org.apache.commons.lang3.StringUtils.isBlank(host)) {
- host = shell.getNextHost();
- }
- List<String> avoidMSHostList = shell.getAvoidHosts();
- do {
- if (CollectionUtils.isEmpty(avoidMSHostList) ||
!avoidMSHostList.contains(host)) {
- 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);
- }
+ NioConnection connection = this.connection;
+ connection.stop();
+ try {
+ connection.cleanUp();
+ } catch (final IOException e) {
+ logger.warn("Fail to clean up old connection", e);
+ }
+
+ try {
+ while (connection.isStartup()) {
+ logger.debug("Waiting for connection graceful stop");
+ shell.getBackoffAlgorithm().waitBeforeRetry();
+ connection.stop();
}
- shell.getBackoffAlgorithm().waitBeforeRetry();
- host = shell.getNextHost();
- } while (!connection.isStartup());
- shell.updateConnectedHost(((NioClient)connection).getHost());
- logger.info("Connected to the host: {}", shell.getConnectedHost());
+ } catch (Exception e) {
+ logger.warn("Failed to gracefully stop connection", e);
+ }
+ logger.debug("Connection stopped");
+ }
+
+ /**
+ * Select the host to reconnect to based on priority:
+ * 1. preferredHost if defined and not blank
+ * 2. Link's socket address IP if available and not null
+ * 3. shell.getNextHost() if the above two options are not met
+ *
+ * @param preferredHost the preferred host to connect to
+ * @param link the current link which may contain socket address
information
+ * @return the host to connect to
+ */
+ protected String selectReconnectionHost(String preferredHost, Link link) {
+ return Optional.ofNullable(preferredHost)
+ .filter(org.apache.commons.lang3.StringUtils::isNotBlank)
+ .orElseGet(() -> Optional.ofNullable(link)
+ .map(Link::getSocketAddress)
+ .map(InetSocketAddress::getAddress)
+ .map(InetAddress::getHostAddress)
+ .orElseGet(shell::getNextHost));
}
- protected void closeAndTerminateLink(final Link link) {
- if (link == null) {
+ /**
+ * Reconnect to Management Server.
+ *
+ * @param link - connection holder
+ * @param preferredHost - if defined, reconnect will be performed to this
Host first,
+ * otherwise will be used {@link
IAgentShell#getNextHost()}
+ * @param forceReconnect - expected to be true if called by {@link
MigrateAgentConnectionCommand},
+ * this is only "switch Management Server", it does
not perform full Host Connect process.
+ */
+ protected void reconnect(Link link, String preferredHost, boolean
forceReconnect) {
+ if (!reconnectLock.compareAndSet(false, true)) {
+ logger.warn("Reconnect is already running, exiting");
return;
}
- link.close();
- link.terminated();
+ String requestedLink =
Optional.ofNullable(link).map(Link::toString).orElse("N/A");
+ String currentLink =
Optional.ofNullable(this.link).map(Link::toString).orElse("N/A");
+ logger.info("Reconnect info: provided link: {}, agent link: {},
preferred host: {}, force" +
+ " reconnect: {}", requestedLink, currentLink, preferredHost,
forceReconnect);
+
+ try {
+ logger.debug("Obtained reconnect lock");
+ if (!(forceReconnect || reconnectAllowed)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Reconnect requested but it is not allowed
{}", link);
+ }
+ return;
+ }
+
+ if (isReconnectStormDetected(link, preferredHost, requestedLink,
currentLink)) {
+ return;
+ }
+
+ cleanupConnectionBeforeReconnect(link);
+ // start with preferred host
+ String host = selectReconnectionHost(preferredHost, link);
+
+ String hostLog = LogUtils.getHostLog(host, shell.getPort());
+ List<String> avoidMsHostList =
Optional.ofNullable(shell.getAvoidHosts()).orElseGet(List::of);
+ // pointer to the first element of "refuse loop"
+ AtomicReference<String> firstRefuseLoopHostRef = new
AtomicReference<>(null);
+ // to break deadlock where "non-avoid" MS Hosts are down and only
"avoid" are up
+ AtomicBoolean ignoreAvoidMsHostListRef = new AtomicBoolean(false);
+ do {
+ AtomicBoolean skipTimeoutRef = new AtomicBoolean(false);
+ String parentLogContextId = (String)
ThreadContext.get("logcontextid");
+ if (parentLogContextId != null) {
+ ThreadContext.put("logcontextid-parent",
parentLogContextId);
+ }
+ ThreadContext.put("logcontextid",
UuidUtils.first(UUID.randomUUID().toString()));
+ if (ignoreAvoidMsHostListRef.get() ||
!avoidMsHostList.contains(host)) {
+ connection = new NioClient(getAgentName(), host,
shell.getPort(), shell.getWorkers(),
+ shell.getSslHandshakeTimeout(), this);
+ logger.info("Reconnecting to host: {}", hostLog);
+ try {
+ connection.start();
+ // successfully connected, skip the rest
+ continue;
+ } catch (Exception e) {
+ logReconnectionFailure(e, hostLog);
+
+ try {
+ stopAndCleanupConnection();
+ } catch (Exception ex) {
+ logger.warn("Got an exception during stop and
cleanup connection", e);
Review Comment:
The catch block logs the wrong exception variable: it catches `ex` but logs
`e`, which will hide the actual failure during cleanup and can mislead
troubleshooting.
##########
framework/cluster/src/main/java/com/cloud/cluster/ClusterServiceServletHttpHandler.java:
##########
@@ -113,6 +117,14 @@ private void writeResponse(HttpResponse response, int
statusCode, String content
response.setEntity(body);
}
+ private void logRequest(HttpRequest request, String requestBody) {
+ Optional<HttpRequest> requestOpt = Optional.ofNullable(request);
+ Optional<RequestLine> requestLineOpt =
requestOpt.map(HttpRequest::getRequestLine);
+ String method =
requestLineOpt.map(RequestLine::getMethod).orElse(null);
+ String uri = requestLineOpt.map(RequestLine::getUri).orElse(null);
+ logger.debug("{} {} {}", method, uri, requestBody);
+ }
Review Comment:
This debug log prints the full raw request body (all key/value parameters).
Cluster PDUs can be large and may include sensitive data; logging the entire
payload can create operational/security risk. Consider logging only method/URI
plus size (or redact selected keys).
##########
agent/src/main/java/com/cloud/agent/AgentShell.java:
##########
@@ -95,6 +95,16 @@ public BackoffAlgorithm getBackoffAlgorithm() {
return _backoff;
}
+ @Override
+ public void setBackoffAlgorithm(BackoffAlgorithm backoffAlgorithm) {
+ this._backoff = backoffAlgorithm;
+ try {
+ backoffAlgorithm.getConfiguration().forEach((key, value) ->
setPersistentProperty(null, key, value));
+ } catch (RuntimeException e) {
+ LOGGER.warn("Failed to persist backoff properties");
+ }
Review Comment:
The caught exception is swallowed in this log statement. Include the
exception so failures to persist backoff settings are diagnosable.
##########
server/src/main/java/org/apache/cloudstack/agent/lb/IndirectAgentLBServiceImpl.java:
##########
@@ -595,7 +597,9 @@ protected void runInContext() {
msList = getManagementServerList(hostId, dcId,
orderedHostIdList, lbAlgorithm);
}
+ // ask Host to reconnect to another Management Server
final MigrateAgentConnectionCommand cmd = new
MigrateAgentConnectionCommand(msList, avoidMsList, lbAlgorithm,
lbCheckInterval);
+ // timeout 1 minute (FIXME: should it be configurable?)
cmd.setWait(60);
final Answer answer = agentManager.easySend(hostId, cmd);
//may not receive answer when the agent disconnects immediately and try
reconnecting to other ms host
Review Comment:
The migration command timeout is hard-coded to 60 seconds and marked as a
`FIXME`. This contradicts the PR goal of parameterized timeouts and makes
tuning difficult in different environments. Consider using an existing config
key or introducing one for this wait value.
##########
core/src/main/java/com/cloud/agent/api/AgentConnectStatusAnswer.java:
##########
@@ -0,0 +1,65 @@
+//
+// 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.api;
+
+import com.cloud.host.HostVO;
+import com.cloud.host.Status;
+import com.cloud.utils.db.GlobalLock;
+
+/**
+ * Answer for {@link AgentConnectStatusCommand}.
+ *
+ * @author mprokopchuk
+ */
+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}.
+ */
Review Comment:
The `lockAvailable` Javadoc contradicts the field name and the consuming
logic (e.g., HostConnectProcess treats TRUE as “lock is free/available”).
Please clarify the meaning to avoid misinterpretation by future maintainers.
##########
agent/src/main/java/com/cloud/agent/Agent.java:
##########
@@ -623,98 +559,363 @@ public Task create(final Task.Type type, final Link
link, final byte[] data) {
return new ServerHandler(type, link, data);
}
- protected void reconnect(final Link link) {
- reconnect(link, null, false);
+ protected void closeAndTerminateLink(Link link) {
+ Optional.ofNullable(link)
+ .map(Link::attachment)
+ .filter(ServerAttache.class::isInstance)
+ .map(ServerAttache.class::cast)
+ .ifPresentOrElse(ServerAttache::disconnect, () -> {
+ if (link != null) {
+ link.close();
+ link.terminated();
+ }
+ });
}
- protected void reconnect(final Link link, String preferredMSHost, boolean
forTransfer) {
- if (!(forTransfer || reconnectAllowed)) {
- logger.debug("Reconnect requested but it is not allowed {}", () ->
getLinkLog(link));
+ protected void stopAndCleanupConnection() {
+ if (connection == null) {
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);
- String host = preferredMSHost;
- if (org.apache.commons.lang3.StringUtils.isBlank(host)) {
- host = shell.getNextHost();
- }
- List<String> avoidMSHostList = shell.getAvoidHosts();
- do {
- if (CollectionUtils.isEmpty(avoidMSHostList) ||
!avoidMSHostList.contains(host)) {
- 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);
- }
+ NioConnection connection = this.connection;
+ connection.stop();
+ try {
+ connection.cleanUp();
+ } catch (final IOException e) {
+ logger.warn("Fail to clean up old connection", e);
+ }
+
+ try {
+ while (connection.isStartup()) {
+ logger.debug("Waiting for connection graceful stop");
+ shell.getBackoffAlgorithm().waitBeforeRetry();
+ connection.stop();
}
- shell.getBackoffAlgorithm().waitBeforeRetry();
- host = shell.getNextHost();
- } while (!connection.isStartup());
- shell.updateConnectedHost(((NioClient)connection).getHost());
- logger.info("Connected to the host: {}", shell.getConnectedHost());
+ } catch (Exception e) {
+ logger.warn("Failed to gracefully stop connection", e);
+ }
+ logger.debug("Connection stopped");
+ }
+
+ /**
+ * Select the host to reconnect to based on priority:
+ * 1. preferredHost if defined and not blank
+ * 2. Link's socket address IP if available and not null
+ * 3. shell.getNextHost() if the above two options are not met
+ *
+ * @param preferredHost the preferred host to connect to
+ * @param link the current link which may contain socket address
information
+ * @return the host to connect to
+ */
+ protected String selectReconnectionHost(String preferredHost, Link link) {
+ return Optional.ofNullable(preferredHost)
+ .filter(org.apache.commons.lang3.StringUtils::isNotBlank)
+ .orElseGet(() -> Optional.ofNullable(link)
+ .map(Link::getSocketAddress)
+ .map(InetSocketAddress::getAddress)
+ .map(InetAddress::getHostAddress)
+ .orElseGet(shell::getNextHost));
}
- protected void closeAndTerminateLink(final Link link) {
- if (link == null) {
+ /**
+ * Reconnect to Management Server.
+ *
+ * @param link - connection holder
+ * @param preferredHost - if defined, reconnect will be performed to this
Host first,
+ * otherwise will be used {@link
IAgentShell#getNextHost()}
+ * @param forceReconnect - expected to be true if called by {@link
MigrateAgentConnectionCommand},
+ * this is only "switch Management Server", it does
not perform full Host Connect process.
+ */
+ protected void reconnect(Link link, String preferredHost, boolean
forceReconnect) {
+ if (!reconnectLock.compareAndSet(false, true)) {
+ logger.warn("Reconnect is already running, exiting");
return;
}
- link.close();
- link.terminated();
+ String requestedLink =
Optional.ofNullable(link).map(Link::toString).orElse("N/A");
+ String currentLink =
Optional.ofNullable(this.link).map(Link::toString).orElse("N/A");
+ logger.info("Reconnect info: provided link: {}, agent link: {},
preferred host: {}, force" +
+ " reconnect: {}", requestedLink, currentLink, preferredHost,
forceReconnect);
+
+ try {
+ logger.debug("Obtained reconnect lock");
+ if (!(forceReconnect || reconnectAllowed)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Reconnect requested but it is not allowed
{}", link);
+ }
+ return;
+ }
+
+ if (isReconnectStormDetected(link, preferredHost, requestedLink,
currentLink)) {
+ return;
+ }
+
+ cleanupConnectionBeforeReconnect(link);
+ // start with preferred host
+ String host = selectReconnectionHost(preferredHost, link);
+
+ String hostLog = LogUtils.getHostLog(host, shell.getPort());
+ List<String> avoidMsHostList =
Optional.ofNullable(shell.getAvoidHosts()).orElseGet(List::of);
+ // pointer to the first element of "refuse loop"
+ AtomicReference<String> firstRefuseLoopHostRef = new
AtomicReference<>(null);
+ // to break deadlock where "non-avoid" MS Hosts are down and only
"avoid" are up
+ AtomicBoolean ignoreAvoidMsHostListRef = new AtomicBoolean(false);
+ do {
+ AtomicBoolean skipTimeoutRef = new AtomicBoolean(false);
+ String parentLogContextId = (String)
ThreadContext.get("logcontextid");
+ if (parentLogContextId != null) {
+ ThreadContext.put("logcontextid-parent",
parentLogContextId);
+ }
+ ThreadContext.put("logcontextid",
UuidUtils.first(UUID.randomUUID().toString()));
+ if (ignoreAvoidMsHostListRef.get() ||
!avoidMsHostList.contains(host)) {
+ connection = new NioClient(getAgentName(), host,
shell.getPort(), shell.getWorkers(),
+ shell.getSslHandshakeTimeout(), this);
+ logger.info("Reconnecting to host: {}", hostLog);
+ try {
+ connection.start();
+ // successfully connected, skip the rest
+ continue;
+ } catch (Exception e) {
+ logReconnectionFailure(e, hostLog);
+
+ try {
+ stopAndCleanupConnection();
+ } catch (Exception ex) {
+ logger.warn("Got an exception during stop and
cleanup connection", e);
+ }
+
+ updateRefuseLoopState(e, host, firstRefuseLoopHostRef,
ignoreAvoidMsHostListRef, skipTimeoutRef);
+ }
+ } else {
+ logger.debug("Next host {} is in avoid list, skipped",
hostLog);
+ if
(org.apache.commons.lang3.StringUtils.isBlank(preferredHost)) {
+ logHostLists(avoidMsHostList);
+ skipTimeoutRef.set(true);
+ }
+ }
+ if (!skipTimeoutRef.get()) {
+ shell.getBackoffAlgorithm().waitBeforeRetry();
+ }
+ host = shell.getNextHost();
+ hostLog = LogUtils.getHostLog(host, shell.getPort());
+ logger.debug("Next host to connect: {}", hostLog);
+ } while (!connection.isStartup());
+ // successfully connected
+ shell.updateConnectedHost(((NioClient) connection).getHost());
+ String msg = String.format("Connected to the host: %s (%s)",
shell.getConnectedHost(), this.link);
+ logger.info(msg);
+ } finally {
+ reconnectLock.set(false);
+ logger.debug("Removed reconnect lock");
+ }
}
- protected void stopAndCleanupConnection(boolean waitForStop) {
- if (connection == null) {
- return;
+ /**
+ * Handles "Connection refused" loop detection and determines if backoff
timeout should be skipped.
+ * Manages refuse loop state to detect when all management servers have
been tried and need
+ * to ignore avoid list to prevent deadlock.
+ *
+ * @param e the exception from connection attempt
+ * @param host the current host being attempted
+ * @param firstRefuseLoopHostRef reference to first host in refuse loop
(modified by this method)
+ * @param ignoreAvoidMsHostListRef flag to ignore avoid list (modified by
this method)
+ * @return true if timeout should be skipped (connection refused), false
otherwise
+ */
+ private void updateRefuseLoopState(Exception e, String host,
AtomicReference<String> firstRefuseLoopHostRef, AtomicBoolean
ignoreAvoidMsHostListRef, AtomicBoolean skipTimeoutRef) {
+ // we are skipping timeout for "Connection refused" to not waste time
on down MS
+ boolean skipTimeout = Optional.ofNullable(e.getCause())
+ .filter(ConnectException.class::isInstance)
+ .map(Throwable::getMessage)
+ .filter(CONNECTION_REFUSED_MSG::equalsIgnoreCase)
+ .isPresent();
+ skipTimeoutRef.set(skipTimeout);
+ String firstRefuseLoopHost = firstRefuseLoopHostRef.get();
+ // for each "Connection refused" (maybe need to have a copy of
variable with better name)
+ // start "refuse loop"
+ if (skipTimeout && firstRefuseLoopHost == null) {
+ firstRefuseLoopHostRef.set(host);
+ ignoreAvoidMsHostListRef.set(false);
+ logger.debug("Started refuse loop for host {}",
firstRefuseLoopHost);
Review Comment:
This log line uses `firstRefuseLoopHost` which is still null in this branch
(it’s read before `firstRefuseLoopHostRef.set(host)`), so the message will
always log `null`. Log the current `host` instead.
This issue also appears on line 742 of the same file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
