Skip to content

Commit c7c75ac

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

File tree

4 files changed

+100
-70
lines changed

4 files changed

+100
-70
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

+32-34
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,6 @@ public abstract class AbstractDatadogSparkListener extends SparkListener {
7070
private static final Logger log = LoggerFactory.getLogger(AbstractDatadogSparkListener.class);
7171
private static final ObjectMapper objectMapper = new ObjectMapper();
7272
public static volatile AbstractDatadogSparkListener listener = null;
73-
public static volatile AtomicReference<SparkListenerInterface> openLineageSparkListener =
74-
new AtomicReference<>(null);
75-
public static volatile SparkConf openLineageSparkConf = null;
7673

7774
public static volatile boolean finishTraceOnApplicationEnd = true;
7875
public static volatile boolean isPysparkShell = false;
@@ -86,6 +83,10 @@ public abstract class AbstractDatadogSparkListener extends SparkListener {
8683
private final String sparkVersion;
8784
private final String appId;
8885

86+
public volatile AtomicReference<SparkListenerInterface> openLineageSparkListener =
87+
new AtomicReference<>(null);
88+
public volatile SparkConf openLineageSparkConf = null;
89+
8990
private final AgentTracer.TracerAPI tracer;
9091

9192
// This is created by constructor, and used if we're not in other known
@@ -167,34 +168,38 @@ public AbstractDatadogSparkListener(SparkConf sparkConf, String appId, String sp
167168
}));
168169
}
169170

170-
public void setupOpenLineage(DDTraceId traceId) {
171-
log.debug("Setting up OpenLineage tags");
171+
public void setupOpenLineage(SparkConf olSparkconf, SparkListenerInterface olSparkListener) {
172+
log.debug("Setting up OpenLineage: {} {}", olSparkconf, olSparkListener);
173+
DDTraceId traceId =
174+
OpenlineageParentContext.from(sparkConf)
175+
.map(context -> context.getTraceId())
176+
.orElse(predeterminedTraceIdContext.getTraceId());
177+
178+
openLineageSparkListener.set(olSparkListener);
179+
openLineageSparkConf = olSparkconf;
180+
181+
log.info(
182+
"Classloader for SL: ({}) {}",
183+
System.identityHashCode(openLineageSparkListener.getClass().getClassLoader()),
184+
openLineageSparkListener.getClass().getClassLoader());
172185
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());
186+
"Classloader for ADSL: ({}) {}",
187+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
188+
AbstractDatadogSparkListener.class.getClassLoader());
180189
log.info(
181190
"Current thread class loader: ({}) {}",
182191
System.identityHashCode(Thread.currentThread().getContextClassLoader()),
183192
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(
193+
if (openLineageSparkListener.get() != null) {
194+
openLineageSparkConf.set("spark.openlineage.transport.type", "composite");
195+
openLineageSparkConf.set("spark.openlineage.transport.continueOnFailure", "true");
196+
openLineageSparkConf.set("spark.openlineage.transport.transports.agent.type", "http");
197+
openLineageSparkConf.set(
192198
"spark.openlineage.transport.transports.agent.url", getAgentHttpUrl());
193-
AbstractDatadogSparkListener.openLineageSparkConf.set(
199+
openLineageSparkConf.set(
194200
"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(
201+
openLineageSparkConf.set("spark.openlineage.transport.transports.agent.compression", "gzip");
202+
openLineageSparkConf.set(
198203
"spark.openlineage.run.tags",
199204
"_dd.trace_id:"
200205
+ traceId.toString()
@@ -204,10 +209,10 @@ public void setupOpenLineage(DDTraceId traceId) {
204209
}
205210
log.info(
206211
"There is no OpenLineage Spark Listener in the context. Skipping setting tags. {}",
207-
AbstractDatadogSparkListener.openLineageSparkListener.get());
212+
openLineageSparkListener.get());
208213
log.info(
209214
"There is no OpenLineage SparkConf in the context. Skipping setting tags. {}",
210-
AbstractDatadogSparkListener.openLineageSparkConf);
215+
openLineageSparkConf);
211216
}
212217

213218
/** Resource name of the spark job. Provide an implementation based on a specific scala version */
@@ -231,13 +236,6 @@ public void setupOpenLineage(DDTraceId traceId) {
231236
@Override
232237
public synchronized void onApplicationStart(SparkListenerApplicationStart applicationStart) {
233238
this.applicationStart = applicationStart;
234-
235-
if (Config.get().isDataJobsOpenLineageEnabled()) {
236-
setupOpenLineage(
237-
OpenlineageParentContext.from(sparkConf)
238-
.map(context -> context.getTraceId())
239-
.orElse(predeterminedTraceIdContext.getTraceId()));
240-
}
241239
notifyOl(x -> openLineageSparkListener.get().onApplicationStart(x), applicationStart);
242240
}
243241

@@ -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.get() != null) {
791789
log.debug(
792790
"Passing event `{}` to OpenLineageSparkListener", event.getClass().getCanonicalName());
793791
ol.accept(event);

‎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/OpenLineageInstrumentation.java

+55-36
Original file line numberDiff line numberDiff line change
@@ -62,44 +62,63 @@ public void methodAdvice(MethodTransformer transformer) {
6262
public static class OpenLineageSparkListenerAdvice {
6363
@Advice.OnMethodExit(suppress = Throwable.class)
6464
public static void exit(@Advice.This Object self, @Advice.FieldValue("conf") SparkConf conf) {
65+
// try {
66+
Logger log = LoggerFactory.getLogger(Config.class);
67+
log.info(
68+
"OLSLA: ADSL classloader: ({}) {}",
69+
System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
70+
AbstractDatadogSparkListener.class.getClassLoader());
6571
try {
66-
Logger log = LoggerFactory.getLogger(Config.class);
67-
AbstractDatadogSparkListener.openLineageSparkConf = conf;
68-
69-
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) {
84-
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");
98-
}
99-
} catch (Exception e) {
100-
LoggerFactory.getLogger(Config.class)
101-
.error("Failed to set OpenLineageSparkListener: {}", e.getMessage());
72+
AbstractDatadogSparkListener.listener.setupOpenLineage(conf, (SparkListenerInterface) self);
73+
} catch (Throwable t) {
74+
}
75+
try {
76+
ClassLoader cl = Thread.currentThread().getContextClassLoader();
77+
Thread.currentThread()
78+
.setContextClassLoader(AbstractDatadogSparkListener.class.getClassLoader());
79+
AbstractDatadogSparkListener.listener.setupOpenLineage(conf, (SparkListenerInterface) self);
80+
Thread.currentThread().setContextClassLoader(cl);
81+
} catch (Throwable t) {
10282
}
83+
84+
// log.debug(
85+
// "ADSL classloader ({}) {}",
86+
// System.identityHashCode(AbstractDatadogSparkListener.class.getClassLoader()),
87+
// AbstractDatadogSparkListener.class.getClassLoader());
88+
// log.debug(
89+
// "Current classloader ({}) {}",
90+
// System.identityHashCode(Thread.currentThread().getContextClassLoader()),
91+
// Thread.currentThread().getContextClassLoader().toString());
92+
// log.debug("Got OpenLineageSparkListener: {}", self);
93+
// AbstractDatadogSparkListener.openLineageSparkListener.set((SparkListenerInterface)
94+
// self);
95+
// log.debug(
96+
// "Set OpenLineageSparkListener: {}",
97+
// AbstractDatadogSparkListener.openLineageSparkListener.get());
98+
// log.debug("Check for self: {}", self);
99+
// if (AbstractDatadogSparkListener.openLineageSparkListener.get() != null) {
100+
// log.debug(
101+
// "Detected OpenLineageSparkListener, passed to DatadogSparkListener {} on
102+
// ClassLoader ({}) {}",
103+
//
104+
// AbstractDatadogSparkListener.openLineageSparkListener.get().getClass().getName(),
105+
// System.identityHashCode(
106+
// AbstractDatadogSparkListener.openLineageSparkListener
107+
// .get()
108+
// .getClass()
109+
// .getClassLoader()),
110+
// AbstractDatadogSparkListener.openLineageSparkListener
111+
// .get()
112+
// .getClass()
113+
// .getClassLoader());
114+
// } else {
115+
// log.debug("WTF it's null");
116+
// }
117+
// } catch (Exception e) {
118+
// LoggerFactory.getLogger(Config.class)
119+
// .error("Failed to set OpenLineageSparkListener: {}", e.getMessage());
120+
// }
121+
103122
}
104123
}
105124
}

0 commit comments

Comments
 (0)