Skip to content

Commit fe39745

Browse files
mccullsmcculls
committed
WIP
1 parent 4709f62 commit fe39745

File tree

8 files changed

+96
-49
lines changed

8 files changed

+96
-49
lines changed

‎communication/src/main/java/datadog/communication/ddagent/SharedCommunicationObjects.java‎

Lines changed: 45 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
import datadog.remoteconfig.DefaultConfigurationPoller;
1212
import datadog.trace.api.Config;
1313
import datadog.trace.util.AgentTaskScheduler;
14+
import java.util.ArrayList;
15+
import java.util.List;
1416
import java.util.concurrent.TimeUnit;
1517
import java.util.function.Supplier;
1618
import okhttp3.HttpUrl;
@@ -21,12 +23,23 @@
2123
public class SharedCommunicationObjects {
2224
private static final Logger log = LoggerFactory.getLogger(SharedCommunicationObjects.class);
2325

26+
private final List<Runnable> pausedComponents = new ArrayList<>();
27+
private volatile boolean paused;
28+
2429
public OkHttpClient okHttpClient;
2530
public HttpUrl agentUrl;
2631
public Monitoring monitoring;
2732
private DDAgentFeaturesDiscovery featuresDiscovery;
2833
private ConfigurationPoller configurationPoller;
2934

35+
public SharedCommunicationObjects() {
36+
this(false);
37+
}
38+
39+
public SharedCommunicationObjects(boolean paused) {
40+
this.paused = paused;
41+
}
42+
3043
public void createRemaining(Config config) {
3144
if (monitoring == null) {
3245
monitoring = Monitoring.DISABLED;
@@ -46,6 +59,30 @@ public void createRemaining(Config config) {
4659
}
4760
}
4861

62+
public void whenReady(Runnable callback) {
63+
synchronized (pausedComponents) {
64+
if (paused) {
65+
pausedComponents.add(callback);
66+
} else {
67+
callback.run();
68+
}
69+
}
70+
}
71+
72+
public void resume() {
73+
paused = false;
74+
synchronized (pausedComponents) {
75+
for (Runnable callback : pausedComponents) {
76+
try {
77+
callback.run();
78+
} catch (Throwable e) {
79+
log.warn("Problem resuming remote component {}", callback, e);
80+
}
81+
}
82+
pausedComponents.clear();
83+
}
84+
}
85+
4986
private static HttpUrl parseAgentUrl(Config config) {
5087
String agentUrl = config.getAgentUrl();
5188
if (agentUrl.startsWith("unix:")) {
@@ -100,11 +137,14 @@ public DDAgentFeaturesDiscovery featuresDiscovery(Config config) {
100137
agentUrl,
101138
config.isTraceAgentV05Enabled(),
102139
config.isTracerMetricsEnabled());
103-
if (AGENT_THREAD_GROUP.equals(Thread.currentThread().getThreadGroup())) {
104-
featuresDiscovery.discover(); // safe to run on same thread
105-
} else {
106-
// avoid performing blocking I/O operation on application thread
107-
AgentTaskScheduler.INSTANCE.execute(featuresDiscovery::discover);
140+
141+
if (!paused) {
142+
if (AGENT_THREAD_GROUP.equals(Thread.currentThread().getThreadGroup())) {
143+
featuresDiscovery.discover(); // safe to run on same thread
144+
} else {
145+
// avoid performing blocking I/O operation on application thread
146+
AgentTaskScheduler.INSTANCE.execute(featuresDiscovery::discover);
147+
}
108148
}
109149
}
110150
return featuresDiscovery;

‎dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java‎

Lines changed: 28 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -347,7 +347,7 @@ public void run() {
347347
* logging facility. Likewise on IBM JDKs OkHttp may indirectly load 'IBMSASL' which in turn loads LogManager.
348348
*/
349349
InstallDatadogTracerCallback installDatadogTracerCallback =
350-
new InstallDatadogTracerCallback(initTelemetry, inst);
350+
new InstallDatadogTracerCallback(initTelemetry, inst, delayOkHttp);
351351
if (delayOkHttp) {
352352
log.debug("Custom logger detected. Delaying Datadog Tracer initialization.");
353353
registerLogManagerCallback(installDatadogTracerCallback);
@@ -496,26 +496,31 @@ public void execute() {
496496
}
497497

498498
protected static class InstallDatadogTracerCallback extends ClassLoadCallBack {
499-
private final InitializationTelemetry initTelemetry;
500499
private final Instrumentation instrumentation;
501500
private final Object sco;
502501
private final Class<?> scoClass;
502+
private final boolean delayOkHttp;
503503

504504
public InstallDatadogTracerCallback(
505-
InitializationTelemetry initTelemetry, Instrumentation instrumentation) {
506-
this.initTelemetry = initTelemetry;
505+
InitializationTelemetry initTelemetry,
506+
Instrumentation instrumentation,
507+
boolean delayOkHttp) {
508+
this.delayOkHttp = delayOkHttp;
507509
this.instrumentation = instrumentation;
508510
try {
509511
scoClass =
510512
AGENT_CLASSLOADER.loadClass("datadog.communication.ddagent.SharedCommunicationObjects");
511-
sco = scoClass.getConstructor().newInstance();
513+
sco = scoClass.getConstructor(boolean.class).newInstance(delayOkHttp);
512514
} catch (ClassNotFoundException
513515
| NoSuchMethodException
514516
| InstantiationException
515517
| IllegalAccessException
516518
| InvocationTargetException e) {
517519
throw new UndeclaredThrowableException(e);
518520
}
521+
522+
installDatadogTracer(initTelemetry, scoClass, sco);
523+
maybeInstallLogsIntake(scoClass, sco);
519524
}
520525

521526
@Override
@@ -525,11 +530,13 @@ public AgentThread agentThread() {
525530

526531
@Override
527532
public void execute() {
528-
installDatadogTracer(initTelemetry, scoClass, sco);
533+
if (delayOkHttp) {
534+
resumeRemoteComponents();
535+
}
536+
529537
maybeStartAppSec(scoClass, sco);
530538
maybeStartIast(instrumentation, scoClass, sco);
531539
maybeStartCiVisibility(instrumentation, scoClass, sco);
532-
maybeStartLogsIntake(scoClass, sco);
533540
// start debugger before remote config to subscribe to it before starting to poll
534541
maybeStartDebugger(instrumentation, scoClass, sco);
535542
maybeStartRemoteConfig(scoClass, sco);
@@ -538,6 +545,16 @@ public void execute() {
538545
startTelemetry(instrumentation, scoClass, sco);
539546
}
540547
}
548+
549+
private void resumeRemoteComponents() {
550+
try {
551+
Thread.sleep(1_000);
552+
scoClass.getMethod("resume").invoke(sco);
553+
} catch (InterruptedException ignore) {
554+
} catch (Exception e) {
555+
log.error("Error resuming remote components", e);
556+
}
557+
}
541558
}
542559

543560
protected static class StartProfilingAgentCallback extends ClassLoadCallBack {
@@ -864,17 +881,18 @@ private static void maybeStartCiVisibility(Instrumentation inst, Class<?> scoCla
864881
}
865882
}
866883

867-
private static void maybeStartLogsIntake(Class<?> scoClass, Object sco) {
884+
private static void maybeInstallLogsIntake(Class<?> scoClass, Object sco) {
868885
if (agentlessLogSubmissionEnabled) {
869886
StaticEventLogger.begin("Logs Intake");
870887

871888
try {
872889
final Class<?> logsIntakeSystemClass =
873890
AGENT_CLASSLOADER.loadClass("datadog.trace.logging.intake.LogsIntakeSystem");
874-
final Method logsIntakeInstallerMethod = logsIntakeSystemClass.getMethod("start", scoClass);
891+
final Method logsIntakeInstallerMethod =
892+
logsIntakeSystemClass.getMethod("install", scoClass);
875893
logsIntakeInstallerMethod.invoke(null, sco);
876894
} catch (final Throwable e) {
877-
log.warn("Not starting Logs Intake subsystem", e);
895+
log.warn("Not installing Logs Intake subsystem", e);
878896
}
879897

880898
StaticEventLogger.end("Logs Intake");

‎dd-java-agent/agent-logs-intake/src/main/java/datadog/trace/logging/intake/LogsIntakeSystem.java‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package datadog.trace.logging.intake;
22

3-
import datadog.communication.BackendApi;
43
import datadog.communication.BackendApiFactory;
54
import datadog.communication.ddagent.SharedCommunicationObjects;
65
import datadog.trace.api.Config;
@@ -12,18 +11,16 @@ public class LogsIntakeSystem {
1211

1312
private static final Logger LOGGER = LoggerFactory.getLogger(LogsIntakeSystem.class);
1413

15-
public static void start(SharedCommunicationObjects sco) {
14+
public static void install(SharedCommunicationObjects sco) {
1615
Config config = Config.get();
1716
if (!config.isAgentlessLogSubmissionEnabled()) {
1817
LOGGER.debug("Agentless logs intake is disabled");
1918
return;
2019
}
2120

2221
BackendApiFactory apiFactory = new BackendApiFactory(config, sco);
23-
BackendApi backendApi = apiFactory.createBackendApi(BackendApiFactory.Intake.LOGS);
24-
LogsDispatcher dispatcher = new LogsDispatcher(backendApi);
25-
LogsWriterImpl writer = new LogsWriterImpl(config, dispatcher);
26-
writer.start();
22+
LogsWriterImpl writer = new LogsWriterImpl(config, apiFactory);
23+
sco.whenReady(writer::start);
2724

2825
LogsIntake.registerWriter(writer);
2926
}

‎dd-java-agent/agent-logs-intake/src/main/java/datadog/trace/logging/intake/LogsWriterImpl.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import static datadog.trace.util.AgentThreadFactory.AGENT_THREAD_GROUP;
44

5+
import datadog.communication.BackendApiFactory;
56
import datadog.trace.api.Config;
67
import datadog.trace.api.logging.intake.LogsWriter;
78
import datadog.trace.util.AgentThreadFactory;
@@ -23,12 +24,12 @@ public class LogsWriterImpl implements LogsWriter {
2324
private static final int ENQUEUE_LOG_TIMEOUT_MILLIS = 1_000;
2425

2526
private final Map<String, Object> commonTags;
26-
private final LogsDispatcher logsDispatcher;
27+
private final BackendApiFactory apiFactory;
2728
private final BlockingQueue<Map<String, Object>> messageQueue;
2829
private final Thread messagePollingThread;
2930

30-
public LogsWriterImpl(Config config, LogsDispatcher logsDispatcher) {
31-
this.logsDispatcher = logsDispatcher;
31+
public LogsWriterImpl(Config config, BackendApiFactory apiFactory) {
32+
this.apiFactory = apiFactory;
3233

3334
commonTags = new HashMap<>();
3435
commonTags.put("ddsource", "java");
@@ -84,6 +85,9 @@ public void log(Map<String, Object> message) {
8485
}
8586

8687
private void logPollingLoop() {
88+
LogsDispatcher logsDispatcher =
89+
new LogsDispatcher(apiFactory.createBackendApi(BackendApiFactory.Intake.LOGS));
90+
8791
while (!Thread.currentThread().isInterrupted()) {
8892
try {
8993
List<Map<String, Object>> batch = new ArrayList<>();

‎dd-java-agent/src/test/groovy/datadog/trace/agent/CustomLogManagerTest.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ class CustomLogManagerTest extends Specification {
2727
, true) == 0
2828
}
2929

30-
def "agent services starts up in premain if configured log manager on system classpath"() {
30+
def "agent services startup is delayed even if configured log manager on system classpath"() {
3131
expect:
3232
IntegrationTestUtils.runOnSeparateJvm(LogManagerSetter.getName()
3333
, [

‎dd-java-agent/src/test/groovy/datadog/trace/agent/CustomMBeanServerBuilderTest.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ class CustomMBeanServerBuilderTest extends Specification {
2828
, true) == 0
2929
}
3030

31-
def "JMXFetch starts up in premain if configured MBeanServerBuilder on system classpath"() {
31+
def "JMXFetch startup is delayed even if configured MBeanServerBuilder on system classpath"() {
3232
expect:
3333
IntegrationTestUtils.runOnSeparateJvm(MBeanServerBuilderSetter.getName()
3434
, [

‎dd-java-agent/src/test/java/jvmbootstraptest/LogManagerSetter.java‎

Lines changed: 8 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,10 @@ public static void main(final String... args) throws Exception {
3939
} else if (System.getProperty("java.util.logging.manager") != null) {
4040
System.out.println("java.util.logging.manager != null");
4141

42-
assertTraceInstallationDelayed(
43-
"tracer install must be delayed when log manager system property is present.");
42+
customAssert(
43+
isTracerInstalled(false),
44+
true,
45+
"tracer install is not delayed when log manager system property is present.");
4446
customAssert(
4547
isJmxfetchStarted(false),
4648
false,
@@ -57,8 +59,6 @@ public static void main(final String... args) throws Exception {
5759
.getClassLoader()
5860
.loadClass(System.getProperty("java.util.logging.manager")),
5961
"Javaagent should not prevent setting a custom log manager");
60-
customAssert(
61-
isTracerInstalled(true), true, "tracer should be installed after loading LogManager.");
6262
customAssert(
6363
isJmxfetchStarted(true), true, "jmxfetch should start after loading LogManager.");
6464
if (isJFRSupported()) {
@@ -67,8 +67,10 @@ public static void main(final String... args) throws Exception {
6767
}
6868
} else if (System.getenv("JBOSS_HOME") != null) {
6969
System.out.println("JBOSS_HOME != null");
70-
assertTraceInstallationDelayed(
71-
"tracer install must be delayed when JBOSS_HOME property is present.");
70+
customAssert(
71+
isTracerInstalled(false),
72+
true,
73+
"tracer install is not delayed when JBOSS_HOME property is present.");
7274
customAssert(
7375
isJmxfetchStarted(false),
7476
false,
@@ -85,10 +87,6 @@ public static void main(final String... args) throws Exception {
8587
.getClassLoader()
8688
.loadClass(System.getProperty("java.util.logging.manager")),
8789
"Javaagent should not prevent setting a custom log manager");
88-
customAssert(
89-
isTracerInstalled(true),
90-
true,
91-
"tracer should be installed after loading with JBOSS_HOME set.");
9290
customAssert(
9391
isJmxfetchStarted(true),
9492
true,
@@ -128,17 +126,6 @@ private static void customAssert(
128126
}
129127
}
130128

131-
private static void assertTraceInstallationDelayed(final String message) {
132-
if (okHttpMayIndirectlyLoadJUL()) {
133-
customAssert(isTracerInstalled(false), false, message);
134-
} else {
135-
customAssert(
136-
isTracerInstalled(false),
137-
true,
138-
"We can safely install tracer on java9+ since it doesn't indirectly trigger logger manager init");
139-
}
140-
}
141-
142129
private static void assertProfilingStartupDelayed(final String message) {
143130
if (okHttpMayIndirectlyLoadJUL()) {
144131
customAssert(isProfilingStarted(false), false, message);

‎dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -689,7 +689,7 @@ private CoreTracer(
689689
}
690690
pendingTraceBuffer.start();
691691

692-
this.writer.start();
692+
sharedCommunicationObjects.whenReady(this.writer::start);
693693

694694
metricsAggregator = createMetricsAggregator(config, sharedCommunicationObjects);
695695
// Schedule the metrics aggregator to begin reporting after a random delay of 1 to 10 seconds
@@ -705,7 +705,8 @@ private CoreTracer(
705705
} else {
706706
this.dataStreamsMonitoring = dataStreamsMonitoring;
707707
}
708-
this.dataStreamsMonitoring.start();
708+
709+
sharedCommunicationObjects.whenReady(this.dataStreamsMonitoring::start);
709710

710711
// Create default extractor from config if not provided and decorate it with DSM extractor
711712
HttpCodec.Extractor builtExtractor =

0 commit comments

Comments
 (0)