Skip to content

Commit debdd10

Browse files
review changes
1 parent d9b60bc commit debdd10

8 files changed

Lines changed: 116 additions & 34 deletions

File tree

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

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -103,10 +103,11 @@ public void scheduleConnectProcess(Link link, boolean connectionTransfer) {
103103
logger.debug("Scheduling connect process for {}", link);
104104
initExecutors();
105105

106-
var task = new HostStatusTask(link, connectionTransfer, agent, hostStatusFutureRef);
106+
var task = new HostStatusTask(link, connectionTransfer, agent);
107107
var future = hostStatusExecutor.scheduleWithFixedDelay(ThreadContextUtil.wrapThreadContext(task),
108108
HOST_STATUS_CHECK_INITIAL_DELAY_SEC,
109109
hostStatusCheckDelaySec, TimeUnit.SECONDS);
110+
task.setFuture(future);
110111
hostStatusFutureRef.set(future);
111112
}
112113

@@ -135,20 +136,23 @@ public static class HostStatusTask implements Runnable, AsyncSend {
135136
private final Link link;
136137
private final boolean forceConnect;
137138
private final Agent agent;
138-
private final AtomicReference<? extends ScheduledFuture<?>> futureRef;
139+
// each task generation cancels only its own schedule, never a newer connect process
140+
private final AtomicReference<ScheduledFuture<?>> ownFutureRef = new AtomicReference<>();
139141

140-
public HostStatusTask(Link link, boolean forceConnect, Agent agent,
141-
AtomicReference<? extends ScheduledFuture<?>> futureRef) {
142+
public HostStatusTask(Link link, boolean forceConnect, Agent agent) {
142143
logger.debug("{} created", this.getClass().getSimpleName());
143144
this.link = link;
144145
this.forceConnect = forceConnect;
145146
this.agent = agent;
146-
this.futureRef = futureRef;
147+
}
148+
149+
public void setFuture(ScheduledFuture<?> future) {
150+
ownFutureRef.set(future);
147151
}
148152

149153
private void cancel() {
150154
logger.debug("Cancelling future");
151-
Optional.ofNullable(futureRef.get())
155+
Optional.ofNullable(ownFutureRef.get())
152156
.filter(Predicate.not(ScheduledFuture::isCancelled))
153157
.ifPresent(future -> future.cancel(true));
154158
logger.debug("Cancelled future");

‎framework/cluster/src/main/java/com/cloud/cluster/ClusterManagerImpl.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1256,19 +1256,24 @@ private boolean pingManagementNode(final ManagementServerHostVO mshost) {
12561256
final String targetHostName = mshost.getName();
12571257
final String currentHostName = NetUtils.getCanonicalHostName();
12581258

1259-
// Prefer hostname (CNAME support) but fall back to the non-null service IP when name is unset.
1260-
final String connectTarget = StringUtils.isNotBlank(targetHostName) ? targetHostName : targetIp;
1261-
1262-
// Check for self-detection using hostname comparison (for CNAME support)
1263-
if (currentHostName != null && currentHostName.equals(targetHostName)) {
1264-
logger.info("ping management node cluster service can not be performed on self (hostname match: {})", currentHostName);
1259+
if (mshost.getMsid() == _msId) {
1260+
logger.info("ping management node cluster service can not be performed on self (msid: {})", _msId);
12651261
return false;
12661262
}
12671263
if (LOCALHOST_IP.equals(targetIp) || ANY_IP.equals(targetIp)) {
12681264
logger.info("ping management node cluster service can not be performed on self");
12691265
return false;
12701266
}
12711267

1268+
// Prefer hostname (CNAME support), but a peer sharing this node's hostname can only be reached by its IP
1269+
// because the name resolves to this node.
1270+
final boolean sameHostName = currentHostName != null && currentHostName.equalsIgnoreCase(targetHostName);
1271+
final String connectTarget = StringUtils.isNotBlank(targetHostName) && !sameHostName ? targetHostName : targetIp;
1272+
if (StringUtils.isBlank(connectTarget)) {
1273+
logger.warn("Unable to ping management node {} (msid: {}), it has neither hostname nor service IP", mshost.getId(), mshost.getMsid());
1274+
return false;
1275+
}
1276+
12721277
int retry = 10;
12731278
while (--retry > 0) {
12741279
SocketChannel sch = null;

‎framework/cluster/src/main/java/com/cloud/cluster/ClusterServiceServletAdapter.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import javax.inject.Inject;
2424
import javax.naming.ConfigurationException;
2525

26+
import com.cloud.utils.StringUtils;
2627
import com.cloud.utils.exception.CloudRuntimeException;
2728
import org.apache.cloudstack.ca.CAManager;
2829
import org.apache.cloudstack.framework.config.ConfigDepot;
@@ -95,9 +96,9 @@ public int getServicePort() {
9596

9697
private String composeEndpointName(ManagementServerHostVO msHost, int port) {
9798
final String hostName = msHost.getName();
98-
final String identifier = hostName != null ? hostName : msHost.getServiceIP();
99+
final String identifier = StringUtils.isNotBlank(hostName) ? hostName : msHost.getServiceIP();
99100

100-
if (identifier == null) {
101+
if (StringUtils.isBlank(identifier)) {
101102
throw new CloudRuntimeException("Management server host has neither hostname nor IP address: " + msHost);
102103
}
103104

‎utils/src/main/java/com/cloud/utils/backoff/impl/ExponentialWithJitterBackoff.java‎

Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -92,25 +92,23 @@ public void waitBeforeRetry() {
9292
/**
9393
* Calculates next attempt and direction.
9494
*/
95-
private void calculateNextAttempt() {
95+
private synchronized void calculateNextAttempt() {
9696
if (increasing) {
9797
int nextAttemptNumber = attemptNumber + 1;
98-
increasing = getNextDelay() <= maxDelayMs && nextAttemptNumber <= maxAttempts;
98+
increasing = nextAttemptNumber <= maxAttempts;
9999
if (increasing) {
100100
attemptNumber = nextAttemptNumber;
101101
}
102102
} else {
103-
int nextAttemptNumber = Math.max(attemptNumber - 1, 0);
104-
increasing = nextAttemptNumber == 0;
105-
if (!increasing) {
106-
attemptNumber = nextAttemptNumber;
107-
}
103+
attemptNumber = Math.max(attemptNumber - 1, 0);
104+
increasing = attemptNumber == 0;
108105
}
109106
}
110107

111108
@Override
112-
public void reset() {
109+
public synchronized void reset() {
113110
attemptNumber = 0;
111+
increasing = true;
114112
}
115113

116114
@Override
@@ -123,7 +121,7 @@ public Map<String, String> getConfiguration() {
123121
}
124122

125123
@Override
126-
public boolean configure(String name, Map<String, Object> params) {
124+
public synchronized boolean configure(String name, Map<String, Object> params) {
127125
minDelayMs = NumbersUtil.parseInt((String) params.get(MIN_DELAY_MS_CONFIG_KEY), MIN_DELAY_MS_DEFAULT);
128126
maxDelayMs = NumbersUtil.parseInt((String) params.get(MAX_DELAY_MS_CONFIG_KEY), MAX_DELAY_MS_DEFAULT);
129127
maxAttempts = (int) Math.round(Math.log((double) maxDelayMs / minDelayMs) / Math.log(2));
@@ -154,15 +152,10 @@ private long getNextDelay() {
154152
}
155153

156154
@Override
157-
public long getTimeToWait() {
155+
public synchronized long getTimeToWait() {
158156
long delay = Math.max(1L, getNextDelay());
159157
int jitterBound = (int) Math.max(1L, delay / 2);
160158
int jitter = random.nextInt(jitterBound);
161159
return delay + jitter;
162160
}
163-
164-
@Override
165-
public void setTimeToWait(long seconds) {
166-
// ignore
167-
}
168161
}

‎utils/src/main/java/com/cloud/utils/backoff/impl/ExponentialWithJitterBackoffMBean.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@
2323
public interface ExponentialWithJitterBackoffMBean extends ManagementBean {
2424
long getTimeToWait();
2525

26-
void setTimeToWait(long seconds);
27-
2826
Collection<String> getWaiters();
2927

3028
boolean wakeup(String threadName);

‎utils/src/main/java/com/cloud/utils/nio/NioConnection.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -251,7 +251,7 @@ protected boolean rejectConnectionIfBlocked(final SocketChannel socketChannel) t
251251
} catch (IOException ignore) {
252252
// ignore exception
253253
}
254-
logger.warn("Rejecting new connection from {} as the server is blocked from accepting new connections", socketChannel.getRemoteAddress());
254+
logger.warn("Rejecting new connection from {} as the server is blocked from accepting new connections", socketChannel.socket().getRemoteSocketAddress());
255255
socketChannel.close();
256256
_selector.wakeup();
257257
return true;
@@ -273,7 +273,7 @@ protected boolean rejectConnectionIfBusy(final SocketChannel socketChannel) thro
273273
}
274274
// Reject new connection if the server is busy
275275
logger.warn("{} Rejecting new connection from {}. {} active connections currently",
276-
SERVER_BUSY_MESSAGE, socketChannel.getRemoteAddress(), factoryMaxNewConnectionsCount);
276+
SERVER_BUSY_MESSAGE, socketChannel.socket().getRemoteSocketAddress(), factoryMaxNewConnectionsCount);
277277
socketChannel.close();
278278
_selector.wakeup();
279279
return true;

‎utils/src/main/java/org/apache/cloudstack/threadcontext/ThreadContextUtil.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,12 +92,18 @@ public static void setLogContextId(String logContextId) {
9292

9393
/**
9494
* Extract UUID from JSON cmdInfo string and set it in MDC if UUID is not already present.
95-
* This is specifically used for async job processing.
95+
* This is specifically used for async job processing. Only API jobs persist cmdInfo as a JSON map;
96+
* VM work jobs persist a Java-serialized {@code VmWork} (see {@code VmWorkSerializer}), which carries
97+
* no UUID, so such values are skipped.
9698
*
9799
* @param cmdInfo the JSON string containing command info
98100
*/
99101
public static void extractAndSetUuidFromCmdInfo(String cmdInfo) {
100102
if (StringUtils.isBlank((String) ThreadContext.get(MDC_UUID_KEY)) && StringUtils.isNotBlank(cmdInfo)) {
103+
if (!cmdInfo.trim().startsWith("{")) {
104+
logger.trace("cmdInfo is not a JSON map, skipping UUID extraction");
105+
return;
106+
}
101107
try {
102108
Type mapType = new TypeToken<Map<String, String>>() {}.getType();
103109
Gson gson = new Gson();
Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
package com.cloud.utils.backoff.impl;
18+
19+
import org.apache.commons.lang3.reflect.FieldUtils;
20+
import org.junit.Assert;
21+
import org.junit.Before;
22+
import org.junit.Test;
23+
24+
import java.util.HashMap;
25+
import java.util.Map;
26+
27+
public class ExponentialWithJitterBackoffTest {
28+
29+
private ExponentialWithJitterBackoff backoff;
30+
31+
@Before
32+
public void setUp() {
33+
backoff = new ExponentialWithJitterBackoff();
34+
Map<String, Object> params = new HashMap<>();
35+
params.put(ExponentialWithJitterBackoff.MIN_DELAY_MS_CONFIG_KEY, "1");
36+
params.put(ExponentialWithJitterBackoff.MAX_DELAY_MS_CONFIG_KEY, "8");
37+
backoff.configure("test", params);
38+
backoff.reset();
39+
}
40+
41+
private int attemptNumber() throws IllegalAccessException {
42+
return (Integer) FieldUtils.readField(backoff, "attemptNumber", true);
43+
}
44+
45+
@Test
46+
public void testAttemptCyclesUpToMaxAndBackToBase() throws IllegalAccessException {
47+
// the maximum delay is used for two consecutive waits before the delay decreases back to base
48+
int[] expected = {1, 2, 3, 3, 2, 1, 0, 1, 2, 3, 3, 2, 1, 0};
49+
for (int i = 0; i < expected.length; i++) {
50+
backoff.waitBeforeRetry();
51+
Assert.assertEquals("attempt number after wait #" + (i + 1), expected[i], attemptNumber());
52+
}
53+
}
54+
55+
@Test
56+
public void testTimeToWaitStaysWithinBounds() throws IllegalAccessException {
57+
for (int i = 0; i < 20; i++) {
58+
long wait = backoff.getTimeToWait();
59+
long delay = (long) Math.min(Math.pow(2, attemptNumber()), 8);
60+
Assert.assertTrue("wait " + wait + " below delay " + delay, wait >= delay);
61+
Assert.assertTrue("wait " + wait + " above delay plus jitter", wait < delay + Math.max(1, delay / 2));
62+
backoff.waitBeforeRetry();
63+
}
64+
}
65+
66+
@Test
67+
public void testResetRestartsIncreasingFromBase() throws IllegalAccessException {
68+
backoff.waitBeforeRetry();
69+
backoff.waitBeforeRetry();
70+
backoff.reset();
71+
Assert.assertEquals(0, attemptNumber());
72+
backoff.waitBeforeRetry();
73+
Assert.assertEquals(1, attemptNumber());
74+
}
75+
}

0 commit comments

Comments
 (0)