Skip to content

Commit 21e0d03

Browse files
mobuchowskimobuchowski
committed
try different CLs
Signed-off-by: Maciej Obuchowski <[email protected]>
1 parent e598da6 commit 21e0d03

File tree

6 files changed

+151
-71
lines changed

6 files changed

+151
-71
lines changed

‎dd-java-agent/instrumentation/spark/spark_2.12/src/main/java/datadog/trace/instrumentation/spark/Spark212Instrumentation.java

+7
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import net.bytebuddy.asm.Advice;
1010
import org.apache.spark.SparkContext;
1111
import org.apache.spark.util.Utils;
12+
import org.slf4j.Logger;
13+
import org.slf4j.LoggerFactory;
1214

1315
@AutoService(InstrumenterModule.class)
1416
public class Spark212Instrumentation extends AbstractSparkInstrumentation {
@@ -45,6 +47,11 @@ public static class InjectListener {
4547
@Advice.OnMethodEnter(suppress = Throwable.class)
4648
public static void enter(@Advice.This SparkContext sparkContext) {
4749
// checking whether OpenLineage integration is enabled, available and that it supports tags
50+
Logger log = LoggerFactory.getLogger(Config.class);
51+
log.info(
52+
"IL: ADSL classloader: ({}) {}",
53+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
54+
AbstractDatadogSparkListener.class.getClassLoader());
4855

4956
if (Config.get().isDataJobsOpenLineageEnabled()
5057
&& Utils.classIsLoadable("io.openlineage.spark.agent.OpenLineageSparkListener")

‎dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java

+35-37
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import java.util.Optional;
3737
import java.util.Properties;
3838
import java.util.UUID;
39-
import java.util.concurrent.atomic.AtomicReference;
4039
import java.util.function.Consumer;
4140
import org.apache.spark.ExceptionFailure;
4241
import org.apache.spark.SparkConf;
@@ -70,9 +69,6 @@ public abstract class AbstractDatadogSparkListener extends SparkListener {
7069
private static final Logger log = LoggerFactory.getLogger(AbstractDatadogSparkListener.class);
7170
private static final ObjectMapper objectMapper = new ObjectMapper();
7271
public static volatile AbstractDatadogSparkListener listener = null;
73-
public static volatile AtomicReference<SparkListenerInterface> openLineageSparkListener =
74-
new AtomicReference<>(null);
75-
public static volatile SparkConf openLineageSparkConf = null;
7672

7773
public static volatile boolean finishTraceOnApplicationEnd = true;
7874
public static volatile boolean isPysparkShell = false;
@@ -86,6 +82,9 @@ public abstract class AbstractDatadogSparkListener extends SparkListener {
8682
private final String sparkVersion;
8783
private final String appId;
8884

85+
public static volatile SparkListenerInterface openLineageSparkListener = null;
86+
public static volatile SparkConf openLineageSparkConf = null;
87+
8988
private final AgentTracer.TracerAPI tracer;
9089

9190
// This is created by constructor, and used if we're not in other known
@@ -168,33 +167,31 @@ public AbstractDatadogSparkListener(SparkConf sparkConf, String appId, String sp
168167
}
169168

170169
public void setupOpenLineage(DDTraceId traceId) {
171-
log.debug("Setting up OpenLineage tags");
170+
log.debug("Setting up OpenLineage: {} {}", openLineageSparkListener, openLineageSparkConf);
171+
172+
log.info(
173+
"Classloader for SL: ({}) {}",
174+
System.identityHashCode(openLineageSparkListener.getClass().getClassLoader()),
175+
openLineageSparkListener.getClass().getClassLoader());
172176
log.info(
173-
"Classloader for current variable: ({}) {}",
174-
System.identityHashCode(
175-
AbstractDatadogSparkListener.openLineageSparkListener
176-
.get()
177-
.getClass()
178-
.getClassLoader()),
179-
AbstractDatadogSparkListener.openLineageSparkListener.get().getClass().getClassLoader());
177+
"Classloader for ADSL: ({}) {}",
178+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
179+
AbstractDatadogSparkListener.class.getClassLoader());
180180
log.info(
181181
"Current thread class loader: ({}) {}",
182182
System.identityHashCode(Thread.currentThread().getContextClassLoader()),
183183
Thread.currentThread().getContextClassLoader());
184-
if (AbstractDatadogSparkListener.openLineageSparkListener.get() != null) {
185-
AbstractDatadogSparkListener.openLineageSparkConf.set(
186-
"spark.openlineage.transport.type", "composite");
187-
AbstractDatadogSparkListener.openLineageSparkConf.set(
188-
"spark.openlineage.transport.continueOnFailure", "true");
189-
AbstractDatadogSparkListener.openLineageSparkConf.set(
190-
"spark.openlineage.transport.transports.agent.type", "http");
191-
AbstractDatadogSparkListener.openLineageSparkConf.set(
184+
185+
if (openLineageSparkListener != null) {
186+
openLineageSparkConf.set("spark.openlineage.transport.type", "composite");
187+
openLineageSparkConf.set("spark.openlineage.transport.continueOnFailure", "true");
188+
openLineageSparkConf.set("spark.openlineage.transport.transports.agent.type", "http");
189+
openLineageSparkConf.set(
192190
"spark.openlineage.transport.transports.agent.url", getAgentHttpUrl());
193-
AbstractDatadogSparkListener.openLineageSparkConf.set(
191+
openLineageSparkConf.set(
194192
"spark.openlineage.transport.transports.agent.endpoint", AGENT_OL_ENDPOINT);
195-
AbstractDatadogSparkListener.openLineageSparkConf.set(
196-
"spark.openlineage.transport.transports.agent.compression", "gzip");
197-
AbstractDatadogSparkListener.openLineageSparkConf.set(
193+
openLineageSparkConf.set("spark.openlineage.transport.transports.agent.compression", "gzip");
194+
openLineageSparkConf.set(
198195
"spark.openlineage.run.tags",
199196
"_dd.trace_id:"
200197
+ traceId.toString()
@@ -204,10 +201,10 @@ public void setupOpenLineage(DDTraceId traceId) {
204201
}
205202
log.info(
206203
"There is no OpenLineage Spark Listener in the context. Skipping setting tags. {}",
207-
AbstractDatadogSparkListener.openLineageSparkListener.get());
204+
openLineageSparkListener);
208205
log.info(
209206
"There is no OpenLineage SparkConf in the context. Skipping setting tags. {}",
210-
AbstractDatadogSparkListener.openLineageSparkConf);
207+
openLineageSparkConf);
211208
}
212209

213210
/** Resource name of the spark job. Provide an implementation based on a specific scala version */
@@ -232,13 +229,14 @@ public void setupOpenLineage(DDTraceId traceId) {
232229
public synchronized void onApplicationStart(SparkListenerApplicationStart applicationStart) {
233230
this.applicationStart = applicationStart;
234231

235-
if (Config.get().isDataJobsOpenLineageEnabled()) {
236-
setupOpenLineage(
232+
if (openLineageSparkListener != null) {
233+
DDTraceId traceId =
237234
OpenlineageParentContext.from(sparkConf)
238235
.map(context -> context.getTraceId())
239-
.orElse(predeterminedTraceIdContext.getTraceId()));
236+
.orElse(predeterminedTraceIdContext.getTraceId());
237+
setupOpenLineage(traceId);
240238
}
241-
notifyOl(x -> openLineageSparkListener.get().onApplicationStart(x), applicationStart);
239+
notifyOl(x -> openLineageSparkListener.onApplicationStart(x), applicationStart);
242240
}
243241

244242
private void initApplicationSpanIfNotInitialized() {
@@ -299,7 +297,7 @@ public void onApplicationEnd(SparkListenerApplicationEnd applicationEnd) {
299297
log.info(
300298
"Received spark application end event, finish trace on this event: {}",
301299
finishTraceOnApplicationEnd);
302-
notifyOl(x -> openLineageSparkListener.get().onApplicationEnd(x), applicationEnd);
300+
notifyOl(x -> openLineageSparkListener.onApplicationEnd(x), applicationEnd);
303301

304302
if (finishTraceOnApplicationEnd) {
305303
finishApplication(applicationEnd.time(), null, 0, null);
@@ -494,7 +492,7 @@ public synchronized void onJobStart(SparkListenerJobStart jobStart) {
494492
stageToJob.put(stageId, jobStart.jobId());
495493
}
496494
jobSpans.put(jobStart.jobId(), jobSpan);
497-
notifyOl(x -> openLineageSparkListener.get().onJobStart(x), jobStart);
495+
notifyOl(x -> openLineageSparkListener.onJobStart(x), jobStart);
498496
}
499497

500498
@Override
@@ -525,7 +523,7 @@ public synchronized void onJobEnd(SparkListenerJobEnd jobEnd) {
525523
if (metrics != null) {
526524
metrics.setSpanMetrics(jobSpan);
527525
}
528-
notifyOl(x -> openLineageSparkListener.get().onJobEnd(x), jobEnd);
526+
notifyOl(x -> openLineageSparkListener.onJobEnd(x), jobEnd);
529527

530528
jobSpan.finish(jobEnd.time() * 1000);
531529
}
@@ -695,7 +693,7 @@ public void onTaskEnd(SparkListenerTaskEnd taskEnd) {
695693
Properties props = stageProperties.get(stageSpanKey);
696694
sendTaskSpan(stageSpan, taskEnd, props);
697695

698-
notifyOl(x -> openLineageSparkListener.get().onTaskEnd(x), taskEnd);
696+
notifyOl(x -> openLineageSparkListener.onTaskEnd(x), taskEnd);
699697
}
700698

701699
private void sendTaskSpan(
@@ -787,7 +785,7 @@ private <T extends SparkListenerEvent> void notifyOl(Consumer<T> ol, T event) {
787785
log.debug("Not emitting event when running on databricks or on streaming jobs");
788786
return;
789787
}
790-
if (AbstractDatadogSparkListener.openLineageSparkListener.get() != null) {
788+
if (openLineageSparkListener != null) {
791789
log.debug(
792790
"Passing event `{}` to OpenLineageSparkListener", event.getClass().getCanonicalName());
793791
ol.accept(event);
@@ -844,7 +842,7 @@ private synchronized void updateAdaptiveSQLPlan(SparkListenerEvent event) {
844842
private synchronized void onSQLExecutionStart(SparkListenerSQLExecutionStart sqlStart) {
845843
sqlPlans.put(sqlStart.executionId(), sqlStart.sparkPlanInfo());
846844
sqlQueries.put(sqlStart.executionId(), sqlStart);
847-
notifyOl(x -> openLineageSparkListener.get().onOtherEvent(x), sqlStart);
845+
notifyOl(x -> openLineageSparkListener.onOtherEvent(x), sqlStart);
848846
}
849847

850848
private synchronized void onSQLExecutionEnd(SparkListenerSQLExecutionEnd sqlEnd) {
@@ -857,7 +855,7 @@ private synchronized void onSQLExecutionEnd(SparkListenerSQLExecutionEnd sqlEnd)
857855
if (metrics != null) {
858856
metrics.setSpanMetrics(span);
859857
}
860-
notifyOl(x -> openLineageSparkListener.get().onOtherEvent(x), sqlEnd);
858+
notifyOl(x -> openLineageSparkListener.onOtherEvent(x), sqlEnd);
861859

862860
span.finish(sqlEnd.time() * 1000);
863861
}

‎dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/AbstractSparkInstrumentation.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import net.bytebuddy.asm.Advice;
1313
import org.apache.spark.deploy.SparkSubmitArguments;
1414
import org.apache.spark.scheduler.SparkListenerInterface;
15+
import org.slf4j.Logger;
1516
import org.slf4j.LoggerFactory;
1617

1718
public abstract class AbstractSparkInstrumentation extends InstrumenterModule.Tracing
@@ -119,6 +120,11 @@ public static class LiveListenerBusAdvice {
119120
@Advice.OnMethodEnter(suppress = Throwable.class, skipOn = Advice.OnNonDefaultValue.class)
120121
// If OL is disabled in tracer config but user set it up manually don't interfere
121122
public static boolean enter(@Advice.Argument(0) SparkListenerInterface listener) {
123+
Logger log = LoggerFactory.getLogger(Config.class);
124+
log.info(
125+
"LLBA: ADSL classloader: ({}) {}",
126+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
127+
AbstractDatadogSparkListener.class.getClassLoader());
122128
if (Config.get().isDataJobsOpenLineageEnabled()
123129
&& listener != null
124130
&& "io.openlineage.spark.agent.OpenLineageSparkListener"

‎dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/ClassLoaderUtils.java

+61
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
package datadog.trace.instrumentation.spark;
2+
3+
import org.slf4j.Logger;
4+
import org.slf4j.LoggerFactory;
5+
6+
public class ClassLoaderUtils {
7+
public static final Logger log = LoggerFactory.getLogger(ClassLoaderUtils.class);
8+
9+
/** Print the classloader hierarchy for the current thread */
10+
public static void printClassLoaderHierarchy() {
11+
ClassLoader loader = Thread.currentThread().getContextClassLoader();
12+
printClassLoaderHierarchy(loader);
13+
}
14+
15+
/** Print the classloader hierarchy for a specific class */
16+
public static void printClassLoaderHierarchy(Class<?> clazz) {
17+
log.info("ClassLoader hierarchy for " + clazz.getName() + ":");
18+
printClassLoaderHierarchy(clazz.getClassLoader());
19+
}
20+
21+
/** Print the classloader hierarchy starting from a specific classloader */
22+
public static void printClassLoaderHierarchy(ClassLoader classLoader) {
23+
if (classLoader == null) {
24+
log.info("Bootstrap ClassLoader (null)");
25+
return;
26+
}
27+
28+
int level = 0;
29+
ClassLoader current = classLoader;
30+
31+
while (current != null) {
32+
log.info(
33+
getIndent(level)
34+
+ "→ "
35+
+ current.getClass().getName()
36+
+ "@"
37+
+ Integer.toHexString(System.identityHashCode(current)));
38+
39+
// Print URLs for URLClassLoaders
40+
// if (current instanceof URLClassLoader) {
41+
// URLClassLoader urlLoader = (URLClassLoader) current;
42+
// for (URL url : urlLoader.getURLs()) {
43+
// log.info(getIndent(level + 1) + "- " + url);
44+
// }
45+
// }
46+
47+
current = current.getParent();
48+
level++;
49+
}
50+
51+
log.info(getIndent(level) + "→ Bootstrap ClassLoader (null)");
52+
}
53+
54+
private static String getIndent(int level) {
55+
StringBuilder indent = new StringBuilder();
56+
for (int i = 0; i < level; i++) {
57+
indent.append(" ");
58+
}
59+
return indent.toString();
60+
}
61+
}

‎dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/OpenLineageInstrumentation.java

+41-33
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import datadog.trace.agent.tooling.Instrumenter;
1010
import datadog.trace.agent.tooling.InstrumenterModule;
1111
import datadog.trace.api.Config;
12+
import java.lang.reflect.Field;
1213
import net.bytebuddy.asm.Advice;
1314
import org.apache.spark.SparkConf;
1415
import org.apache.spark.scheduler.SparkListenerInterface;
@@ -36,6 +37,7 @@ public String[] helperClassNames() {
3637
packageName + ".SparkSQLUtils",
3738
packageName + ".SparkSQLUtils$SparkPlanInfoForStage",
3839
packageName + ".SparkSQLUtils$AccumulatorWithStage",
40+
packageName + ".ClassLoaderUtils",
3941
};
4042
}
4143

@@ -62,43 +64,49 @@ public void methodAdvice(MethodTransformer transformer) {
6264
public static class OpenLineageSparkListenerAdvice {
6365
@Advice.OnMethodExit(suppress = Throwable.class)
6466
public static void exit(@Advice.This Object self, @Advice.FieldValue("conf") SparkConf conf) {
65-
try {
66-
Logger log = LoggerFactory.getLogger(Config.class);
67-
AbstractDatadogSparkListener.openLineageSparkConf = conf;
67+
// try {
68+
Logger log = LoggerFactory.getLogger(Config.class);
69+
ClassLoaderUtils.printClassLoaderHierarchy(
70+
AbstractDatadogSparkListener.class.getClassLoader());
71+
log.info(
72+
"OLSLA: ADSL classloader: ({}) {}",
73+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
74+
AbstractDatadogSparkListener.class.getClassLoader());
75+
log.info(
76+
"OLSLA: thread class loader ({}) {}",
77+
System.identityHashCode(Thread.currentThread().getContextClassLoader()),
78+
Thread.currentThread().getContextClassLoader());
79+
log.info(
80+
"OLSLA: parent thread class loader ({}) {}",
81+
System.identityHashCode(Thread.currentThread().getContextClassLoader().getParent()),
82+
Thread.currentThread().getContextClassLoader().getParent());
6883

84+
ClassLoader cl = Thread.currentThread().getContextClassLoader();
85+
if (cl.getClass().getName().contains("MutableURLClassLoader")
86+
|| cl.getClass().getName().contains("ChildFirstURLClassLoader")) {
6987
log.debug(
70-
"ADSL classloader ({}) {}",
71-
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
72-
AbstractDatadogSparkListener.class.getClassLoader());
73-
log.debug(
74-
"Current classloader ({}) {}",
75-
System.identityHashCode(Thread.currentThread().getContextClassLoader()),
76-
Thread.currentThread().getContextClassLoader().toString());
77-
log.debug("Got OpenLineageSparkListener: {}", self);
78-
AbstractDatadogSparkListener.openLineageSparkListener.set((SparkListenerInterface) self);
79-
log.debug(
80-
"Set OpenLineageSparkListener: {}",
81-
AbstractDatadogSparkListener.openLineageSparkListener.get());
82-
log.debug("Check for self: {}", self);
83-
if (AbstractDatadogSparkListener.openLineageSparkListener.get() != null) {
88+
"Detected MutableURLClassLoader. Setting OpenLineage on AbstractDatadogSparkListener.class of parent classloader");
89+
try {
8490
log.debug(
85-
"Detected OpenLineageSparkListener, passed to DatadogSparkListener {} on ClassLoader ({}) {}",
86-
AbstractDatadogSparkListener.openLineageSparkListener.get().getClass().getName(),
87-
System.identityHashCode(
88-
AbstractDatadogSparkListener.openLineageSparkListener
89-
.get()
90-
.getClass()
91-
.getClassLoader()),
92-
AbstractDatadogSparkListener.openLineageSparkListener
93-
.get()
94-
.getClass()
95-
.getClassLoader());
96-
} else {
97-
log.debug("WTF it's null");
91+
"Parent classloader: ({}) {}",
92+
System.identityHashCode(cl.getParent()),
93+
cl.getParent());
94+
Class clazz = cl.getParent().loadClass(AbstractDatadogSparkListener.class.getName());
95+
Field openLineageSparkListener = clazz.getDeclaredField("openLineageSparkListener");
96+
openLineageSparkListener.setAccessible(true);
97+
openLineageSparkListener.set(null, self);
98+
99+
Field openLineageSparkConf = clazz.getDeclaredField("openLineageSparkConf");
100+
openLineageSparkConf.setAccessible(true);
101+
openLineageSparkConf.set(null, conf);
102+
} catch (Throwable e) {
103+
log.info("Failed to setup OpenLineage", e);
98104
}
99-
} catch (Exception e) {
100-
LoggerFactory.getLogger(Config.class)
101-
.error("Failed to set OpenLineageSparkListener: {}", e.getMessage());
105+
} else {
106+
log.debug(
107+
"Detected other classloader than MutableURLClassLoader. Setting OpenLineage on AbstractDatadogSparkListener.class");
108+
AbstractDatadogSparkListener.openLineageSparkListener = (SparkListenerInterface) self;
109+
AbstractDatadogSparkListener.openLineageSparkConf = conf;
102110
}
103111
}
104112
}

‎dd-java-agent/instrumentation/spark/src/main/java/datadog/trace/instrumentation/spark/OpenlineageParentContext.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,7 @@ public PathwayContext getPathwayContext() {
145145

146146
@Override
147147
public boolean isRemote() {
148-
return false;
148+
return true;
149149
}
150150

151151
public String getParentJobNamespace() {

0 commit comments

Comments
 (0)