Skip to content

Commit e5dc90b

Browse files
mprokopchuksureshanaparti
mprokopchuk
authored and
sureshanaparti
committed
Propagate ThreadContext uuid value from MS to agent and to all async jobs
1 parent ff2b0a4 commit e5dc90b

10 files changed

Lines changed: 191 additions & 63 deletions

File tree

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

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@
6666
import org.apache.cloudstack.ca.SetupKeyStoreCommand;
6767
import org.apache.cloudstack.ca.SetupKeystoreAnswer;
6868
import org.apache.cloudstack.managed.context.ManagedContextTimerTask;
69+
import org.apache.cloudstack.threadcontext.ThreadContextCommandUtil;
6970
import org.apache.cloudstack.utils.security.KeyStoreUtils;
7071
import org.apache.commons.collections.CollectionUtils;
7172
import org.apache.commons.io.FileUtils;
@@ -947,9 +948,7 @@ protected void processRequest(final Request request, final Link link) {
947948
final Command cmd = cmds[i];
948949
Answer answer;
949950
try {
950-
if (cmd.getContextParam("logid") != null) {
951-
ThreadContext.put("logcontextid", cmd.getContextParam("logid"));
952-
}
951+
ThreadContextCommandUtil.propagateContextFromCommand(cmd);
953952
if (logger.isDebugEnabled()) {
954953
// ensures request is logged only once per method call
955954
if (!requestLogged)
@@ -1324,9 +1323,7 @@ public void processOtherTask(final Task task) {
13241323
} else if (obj instanceof Request) {
13251324
final Request req = (Request) obj;
13261325
final Command command = req.getCommand();
1327-
if (command.getContextParam("logid") != null) {
1328-
ThreadContext.put("logcontextid", command.getContextParam("logid"));
1329-
}
1326+
ThreadContextCommandUtil.propagateContextFromCommand(command);
13301327
Answer answer = null;
13311328
commandsInProgress.incrementAndGet();
13321329
try {

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

Lines changed: 4 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,11 @@
3232
import com.cloud.resource.ServerResource;
3333
import com.cloud.utils.concurrency.NamedThreadFactory;
3434
import com.cloud.utils.nio.Link;
35+
import org.apache.cloudstack.threadcontext.ThreadContextCommandUtil;
3536
import org.apache.cloudstack.threadcontext.ThreadContextUtil;
3637
import org.apache.commons.lang3.ArrayUtils;
3738
import org.apache.logging.log4j.Logger;
3839
import org.apache.logging.log4j.LogManager;
39-
import org.apache.logging.log4j.ThreadContext;
4040

4141
import java.io.IOException;
4242
import java.nio.channels.ClosedChannelException;
@@ -229,8 +229,6 @@ public void sendStartupCommand(Link link, boolean connectionTransfer) throws IOE
229229
logger.warn("No startup commands returned from {}, Startup command sending skipped", serverResource.getName());
230230
return;
231231
}
232-
String logId = Optional.ofNullable(ThreadContext.get("logcontextid"))
233-
.map(String.class::cast).orElse(null);
234232
String msHostList = _agent.getPersistentProperty("host");
235233
// need to downcast StartupCommand[] to Command[], otherwise logger will fail to decode JSON on MS side
236234
Command[] commands = new Command[startup.length];
@@ -240,9 +238,7 @@ public void sendStartupCommand(Link link, boolean connectionTransfer) throws IOE
240238
_agent.setupStartupCommand(command);
241239
command.setMSHostList(msHostList);
242240
command.setConnectionTransferred(connectionTransfer);
243-
if (logId != null) {
244-
command.setContextParam("logid", logId);
245-
}
241+
ThreadContextCommandUtil.setContextInCommand(command);
246242
}
247243
String commandName = commands[0].getClass().getSimpleName();
248244
boolean needAdditionalValidation = false;
@@ -298,16 +294,8 @@ interface AsyncSend {
298294

299295
default <T> T send(ServerAttache attache, Command[] commands, Class<T> answerType,
300296
int asyncCommandTimeoutSec) throws IOException {
301-
String logId = Optional.ofNullable(ThreadContext.get("logcontextid"))
302-
.filter(String.class::isInstance)
303-
.map(String.class::cast)
304-
.orElse(null);
305-
if (logId != null) {
306-
for (Command command : commands) {
307-
if (command.getContextParam("logid") == null) {
308-
command.setContextParam("logid", logId);
309-
}
310-
}
297+
for (Command command : commands) {
298+
ThreadContextCommandUtil.setContextInCommand(command);
311299
}
312300
Link link = attache.getLink();
313301
String commandName = commands[0].getClass().getSimpleName();

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.cloud.utils.concurrency.NamedThreadFactory;
2727
import com.cloud.utils.nio.Link;
2828
import org.apache.cloudstack.managed.context.ManagedContextRunnable;
29+
import org.apache.cloudstack.threadcontext.ThreadContextCommandUtil;
2930
import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
3031
import org.apache.commons.lang3.ArrayUtils;
3132
import org.apache.logging.log4j.LogManager;
@@ -422,6 +423,9 @@ public <T> T send(Long agentId, Command[] commands, Class<T> answerType, int asy
422423
}
423424
}
424425
}
426+
for (Command command : commands) {
427+
ThreadContextCommandUtil.setContextInCommand(command);
428+
}
425429
Link link = getLink();
426430
String commandName = commands[0].getClass().getSimpleName();
427431

‎api/src/main/java/com/cloud/agent/api/Command.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,9 @@ public enum State {
5858
// allow command to carry over hypervisor or other environment related context info
5959
@LogLevel(Log4jLevel.Trace)
6060
protected Map<String, String> contextMap = new HashMap<String, String>();
61+
62+
// separate map for logging/tracing related context info (ThreadContext parameters)
63+
protected Map<String, String> traceContextMap = new HashMap<>();
6164
private int wait; //in second
6265
private boolean bypassHostMaintenance = false;
6366
private transient long requestSequence = 0L;
@@ -106,6 +109,14 @@ public Map<String, String> getContextMap() {
106109
return contextMap;
107110
}
108111

112+
public void setTraceContextParam(String name, String value) {
113+
traceContextMap.put(name, value);
114+
}
115+
116+
public String getTraceContextParam(String name) {
117+
return traceContextMap.get(name);
118+
}
119+
109120
public boolean allowCaching() {
110121
return true;
111122
}
@@ -147,13 +158,15 @@ public boolean equals(Object o) {
147158

148159
if (wait != command.wait) return false;
149160
if (contextMap != null ? !contextMap.equals(command.contextMap) : command.contextMap != null) return false;
161+
if (traceContextMap != null ? !traceContextMap.equals(command.traceContextMap) : command.traceContextMap != null) return false;
150162

151163
return true;
152164
}
153165

154166
@Override
155167
public int hashCode() {
156168
int result = contextMap != null ? contextMap.hashCode() : 0;
169+
result = 31 * result + (traceContextMap != null ? traceContextMap.hashCode() : 0);
157170
result = 31 * result + wait;
158171
return result;
159172
}

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

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
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 org.apache.cloudstack.threadcontext;
18+
19+
import com.cloud.agent.api.Command;
20+
import com.cloud.utils.StringUtils;
21+
import org.apache.logging.log4j.ThreadContext;
22+
23+
/**
24+
* Utility class for Command-specific MDC operations.
25+
* This class handles propagation of MDC values to and from Command objects.
26+
*
27+
* @author mprokopchuk
28+
*/
29+
public class ThreadContextCommandUtil {
30+
31+
/**
32+
* Propagate UUID and log context ID from Command trace context to MDC.
33+
*
34+
* @param cmd the command containing trace context parameters
35+
*/
36+
public static void propagateContextFromCommand(Command cmd) {
37+
if (cmd != null) {
38+
ThreadContextUtil.setLogContextId(cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_LOG_ID_KEY));
39+
ThreadContextUtil.setUuid(cmd.getTraceContextParam(ThreadContextUtil.CONTEXT_UUID_KEY));
40+
}
41+
}
42+
43+
/**
44+
* Set UUID and log context ID in Command trace context from current MDC values.
45+
* MDC values are authoritative for the current thread; any pre-existing trace context
46+
* on the command is intentionally overwritten.
47+
*
48+
* @param cmd the command to set trace context parameters on
49+
*/
50+
public static void setContextInCommand(Command cmd) {
51+
if (cmd != null) {
52+
String logContextId = (String) ThreadContext.get(ThreadContextUtil.MDC_LOG_CONTEXT_ID_KEY);
53+
if (StringUtils.isNotEmpty(logContextId)) {
54+
cmd.setTraceContextParam(ThreadContextUtil.CONTEXT_LOG_ID_KEY, logContextId);
55+
}
56+
57+
String uuid = (String) ThreadContext.get(ThreadContextUtil.MDC_UUID_KEY);
58+
if (StringUtils.isNotEmpty(uuid)) {
59+
cmd.setTraceContextParam(ThreadContextUtil.CONTEXT_UUID_KEY, uuid);
60+
}
61+
}
62+
}
63+
}

‎engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java‎

Lines changed: 22 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -76,11 +76,11 @@
7676
import org.apache.cloudstack.managed.context.ManagedContextRunnable;
7777
import org.apache.cloudstack.management.ManagementServerHost;
7878
import org.apache.cloudstack.outofbandmanagement.dao.OutOfBandManagementDao;
79+
import org.apache.cloudstack.threadcontext.ThreadContextCommandUtil;
7980
import org.apache.cloudstack.utils.identity.ManagementServerNode;
8081
import org.apache.commons.collections.MapUtils;
8182
import org.apache.commons.lang3.BooleanUtils;
8283
import org.apache.commons.lang3.ObjectUtils;
83-
import org.apache.logging.log4j.ThreadContext;
8484

8585
import com.cloud.agent.AgentManager;
8686
import com.cloud.agent.Listener;
@@ -572,21 +572,6 @@ private void initAndScheduleAgentConnectionsMonitor() {
572572
newAgentConnectionsMonitor.scheduleAtFixedRate(new AgentNewConnectionsMonitorTask(), cleanupTimeInSecs, cleanupTimeInSecs, TimeUnit.SECONDS);
573573
}
574574

575-
private AgentControlAnswer handleControlCommand(final AgentAttache attache, final AgentControlCommand cmd) {
576-
AgentControlAnswer answer;
577-
578-
for (final Pair<Integer, Listener> listener : _cmdMonitors) {
579-
answer = listener.second().processControlCommand(attache.getId(), cmd);
580-
581-
if (answer != null) {
582-
return answer;
583-
}
584-
}
585-
586-
logger.warn("No handling of agent control command: {} sent from {}", cmd, attache);
587-
return new AgentControlAnswer(cmd);
588-
}
589-
590575
private AgentConnectStatusAnswer handleAgentConnectStatusCommand(AgentAttache attache, AgentConnectStatusCommand cmd) {
591576
HostVO hostVo = _hostDao.findById(attache.getId());
592577
if (hostVo == null) {
@@ -714,10 +699,7 @@ private static void tagCommand(final Command cmd) {
714699
cmd.setContextParam("job", "job-" + job.getId());
715700
}
716701
}
717-
String logcontextid = ThreadContext.get("logcontextid");
718-
if (StringUtils.isNotEmpty(logcontextid)) {
719-
cmd.setContextParam("logid", logcontextid);
720-
}
702+
ThreadContextCommandUtil.setContextInCommand(cmd);
721703
}
722704

723705
/**
@@ -2180,12 +2162,9 @@ private void processStartupRoutingCommand(StartupRoutingCommand startup, long ho
21802162
* and request to reconnect (instead of throwing away).
21812163
*/
21822164
private void processPingCommand(Link link, PingCommand cmd, Request request) {
2183-
request.logD("Processing:", true);
2184-
if (cmd.getContextParam("logid") != null) {
2185-
ThreadContext.put("logcontextid", cmd.getContextParam("logid"));
2186-
}
2187-
boolean requestStartupCommand = cmd instanceof PingRoutingCommand && ((PingRoutingCommand) cmd).isGatewayAccessible();
2188-
PingAnswer answer = new PingAnswer(cmd, getAvoidMsList(), requestStartupCommand);
2165+
ThreadContextCommandUtil.propagateContextFromCommand(cmd);
2166+
boolean sendStartup = cmd instanceof PingRoutingCommand && ((PingRoutingCommand) cmd).isGatewayAccessible();
2167+
PingAnswer answer = new PingAnswer(cmd, getAvoidMsList(), sendStartup);
21892168
Response response = new Response(request, new Answer[]{answer}, _nodeId, cmd.getHostId());
21902169
response.setSequence(request.getSequence());
21912170
response.logD("Sending:", true);
@@ -2211,9 +2190,7 @@ protected void processRequest(final Link link, final Request request) {
22112190
Command cmd = cmds[0];
22122191
boolean logD = true;
22132192

2214-
if (cmd !=null && cmd.getContextParam("logid") != null) {
2215-
ThreadContext.put("logcontextid", cmd.getContextParam("logid"));
2216-
}
2193+
ThreadContextCommandUtil.propagateContextFromCommand(cmd);
22172194

22182195
if (attache == null) {
22192196
// FIXME: if there are more than one command and first is not startup command,
@@ -2281,13 +2258,14 @@ protected void processRequest(final Link link, final Request request) {
22812258
}
22822259
return;
22832260
} else if (cmd instanceof AgentControlCommand) {
2284-
answer = handleControlCommand(attache, (AgentControlCommand) cmd);
2261+
answer = handleAgentControlCommand(attache, (AgentControlCommand) cmd);
22852262
} else if (cmd instanceof AgentConnectStatusCommand) {
22862263
answer = handleAgentConnectStatusCommand(attache, (AgentConnectStatusCommand) cmd);
22872264
} else {
22882265
handleCommands(attache, request.getSequence(), new Command[] {cmd});
22892266
if (cmd instanceof PingCommand) {
22902267
final long cmdHostId = ((PingCommand)cmd).getHostId();
2268+
boolean requestStartupCommand = false;
22912269

22922270
final HostVO host = _hostDao.findById(cmdHostId);
22932271
boolean gatewayAccessible = true;
@@ -2315,7 +2293,6 @@ protected void processRequest(final Link link, final Request request) {
23152293
}
23162294
}
23172295

2318-
boolean requestStartupCommand = false;
23192296
if (host != null && gatewayAccessible) {
23202297
requestStartupCommand = sendRequestStartupCommand(hostId, host);;
23212298
}
@@ -2397,6 +2374,20 @@ protected void processResponse(final Link link, final Response response) {
23972374
}
23982375
}
23992376

2377+
private AgentControlAnswer handleAgentControlCommand(final AgentAttache attache, final AgentControlCommand cmd) {
2378+
AgentControlAnswer answer;
2379+
2380+
for (final Pair<Integer, Listener> listener : _cmdMonitors) {
2381+
answer = listener.second().processControlCommand(attache.getId(), cmd);
2382+
if (answer != null) {
2383+
return answer;
2384+
}
2385+
}
2386+
2387+
logger.warn("No handling of agent control command: {} sent from {}", cmd, attache);
2388+
return new AgentControlAnswer(cmd);
2389+
}
2390+
24002391
private void processAgentConnectStatusCommand(Link link, AgentConnectStatusCommand cmd, Request request) {
24012392
request.logD("Processing:", true);
24022393
Long hostId = cmd.getHostId();

‎engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@
3838
import com.cloud.host.Status;
3939
import com.cloud.hypervisor.Hypervisor;
4040
import com.cloud.resource.ServerResource;
41-
import org.apache.logging.log4j.ThreadContext;
41+
import org.apache.cloudstack.threadcontext.ThreadContextCommandUtil;
4242

4343
public class DirectAgentAttache extends AgentAttache {
4444

@@ -181,9 +181,7 @@ protected synchronized void runInContext() {
181181
return;
182182
}
183183

184-
if (cmd.getContextParam("logid") != null) {
185-
ThreadContext.put("logcontextid", cmd.getContextParam("logid"));
186-
}
184+
ThreadContextCommandUtil.propagateContextFromCommand(cmd);
187185
logger.debug("Ping from [id: {}, uuid: {}, name: {}]", _id, _uuid, _name);
188186
long seq = _seq++;
189187

@@ -245,9 +243,7 @@ protected void runInContext() {
245243
for (int i = 0; i < cmds.length; i++) {
246244
Answer answer = null;
247245
Command currentCmd = cmds[i];
248-
if (currentCmd.getContextParam("logid") != null) {
249-
ThreadContext.put("logcontextid", currentCmd.getContextParam("logid"));
250-
}
246+
ThreadContextCommandUtil.propagateContextFromCommand(currentCmd);
251247
try {
252248
if (resource != null) {
253249
answer = resource.executeRequest(cmds[i]);
@@ -303,9 +299,7 @@ protected void runInContext() {
303299
for (int i = 0; i < cmds.length; i++) {
304300
Answer answer = null;
305301
Command currentCmd = cmds[i];
306-
if (currentCmd.getContextParam("logid") != null) {
307-
ThreadContext.put("logcontextid", currentCmd.getContextParam("logid"));
308-
}
302+
ThreadContextCommandUtil.propagateContextFromCommand(currentCmd);
309303
try {
310304
if (resource != null) {
311305
answer = resource.executeRequest(cmds[i]);

‎framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import org.apache.cloudstack.jobs.JobInfo.Status;
6565
import org.apache.cloudstack.managed.context.ManagedContextRunnable;
6666
import org.apache.cloudstack.management.ManagementServerHost;
67+
import org.apache.cloudstack.threadcontext.ThreadContextUtil;
6768
import org.apache.cloudstack.utils.identity.ManagementServerNode;
6869
import org.apache.logging.log4j.ThreadContext;
6970

@@ -667,6 +668,22 @@ protected void runInContext() {
667668
}
668669
ThreadContext.put("logcontextid", logContext);
669670

671+
if (StringUtils.isBlank(ThreadContext.get(ThreadContextUtil.MDC_UUID_KEY))) {
672+
AsyncJob jobToCheck = job;
673+
logger.debug("Updating UUID ThreadContext value");
674+
675+
// If current job has no cmdInfo and has a related parent job, check parent instead
676+
if (StringUtils.isNotBlank(related)) {
677+
AsyncJob parentJob = _jobDao.findByIdIncludingRemoved(Long.parseLong(related));
678+
if (parentJob != null && StringUtils.isNotBlank(parentJob.getCmdInfo())) {
679+
jobToCheck = parentJob;
680+
}
681+
}
682+
683+
// Extract entity UUID from the selected job
684+
ThreadContextUtil.extractAndSetUuidFromCmdInfo(jobToCheck.getCmdInfo());
685+
}
686+
670687
// execute the job
671688
if (logger.isDebugEnabled()) {
672689
logger.debug("Executing " + StringUtils.cleanString(job.toString()));

0 commit comments

Comments
 (0)