Skip to content

[DJM-0000] Debug Databricks #10328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
charlesmyu wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

[DJM-0000] Debug Databricks #10328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions ...ommon/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -399,6 +399,7 @@ private void addDatabricksSpecificTags(

// ids to link those spans to databricks job/task traces
builder.withTag("databricks_job_id", databricksJobId);
log.error("[CHARLES] addDatabricksSpecificTags databricksJobId: '{}'", databricksJobId);
builder.withTag("databricks_job_run_id", databricksJobRunId);
builder.withTag("databricks_task_run_id", databricksTaskRunId);

Expand Down Expand Up @@ -1085,13 +1086,15 @@ private long stageSpanKey(int stageId, int attemptId) {
@SuppressForbidden // split with one-char String use a fast-path without regex usage
private static String getDatabricksJobId(Properties properties) {
String jobId = properties.getProperty("spark.databricks.job.id");
log.error("[CHARLES] getDatabricksJobId spark.databricks.job.id: '{}'", jobId);
if (jobId != null) {
return jobId;
}

// First fallback, use spark.jobGroup.id with the pattern
// <scheduler_id>_job-<job_id>-run-<task_run_id>-action-<action_id>
String jobGroupId = properties.getProperty("spark.jobGroup.id");
log.error("[CHARLES] getDatabricksJobId spark.jobGroup.id: '{}'", jobGroupId);
if (jobGroupId != null) {
int startIndex = jobGroupId.indexOf("job-");
int endIndex = jobGroupId.indexOf("-run", startIndex);
Expand All @@ -1103,6 +1106,7 @@ private static String getDatabricksJobId(Properties properties) {
// Second fallback, use spark.databricks.workload.id with pattern
// <org_id>-<job_id>-<task_run_id>
String workloadId = properties.getProperty("spark.databricks.workload.id");
log.error("[CHARLES] getDatabricksJobId spark.databricks.workload.id: '{}'", workloadId);
if (workloadId != null) {
String[] parts = workloadId.split("-");
if (parts.length > 1) {
Expand Down
18 changes: 9 additions & 9 deletions dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ static DDSpan create(
@Nonnull DDSpanContext context,
final List<AgentSpanLink> links) {
final DDSpan span = new DDSpan(instrumentationName, timestampMicro, context, links);
log.debug("Started span: {}", span);
log.error("Started span: {}", span);
context.getTraceCollector().registerSpan(span);
return span;
}
Expand Down Expand Up @@ -160,9 +160,9 @@ private void finishAndAddToTrace(final long durationNano) {
}
this.metrics.onSpanFinished();
TraceCollector.PublishState publishState = context.getTraceCollector().onPublish(this);
log.debug("Finished span ({}): {}", publishState, this);
log.error("Finished span ({}): {}", publishState, this);
} else {
log.debug("Already finished: {}", this);
log.error("Already finished: {}", this);
}
}

Expand Down Expand Up @@ -224,7 +224,7 @@ public void finishWithEndToEnd() {
try {
e2eStart = null != value ? MILLISECONDS.toNanos(Long.parseLong(value)) : 0;
} catch (RuntimeException e) {
log.debug("Ignoring invalid end-to-end start time {}", value, e);
log.error("Ignoring invalid end-to-end start time {}", value, e);
e2eStart = 0;
}
} else {
Expand Down Expand Up @@ -254,10 +254,10 @@ public final boolean phasedFinish() {
}
// Flip the negative bit of the result to allow verifying that publish() is only called once.
if (DURATION_NANO_UPDATER.compareAndSet(this, 0, Math.max(1, durationNano) | Long.MIN_VALUE)) {
log.debug("Finished span (PHASED): {}", this);
log.error("Finished span (PHASED): {}", this);
return true;
} else {
log.debug("Already finished: {}", this);
log.error("Already finished: {}", this);
return false;
}
}
Expand All @@ -266,13 +266,13 @@ public final boolean phasedFinish() {
public final void publish() {
long durationNano = this.durationNano;
if (durationNano == 0) {
log.debug("Can't publish unfinished span: {}", this);
log.error("Can't publish unfinished span: {}", this);
} else if (durationNano > 0) {
log.debug("Already published: {}", this);
log.error("Already published: {}", this);
} else if (DURATION_NANO_UPDATER.compareAndSet(
this, durationNano, durationNano & Long.MAX_VALUE)) {
TraceCollector.PublishState publishState = context.getTraceCollector().onPublish(this);
log.debug("Published span ({}): {}", publishState, this);
log.error("Published span ({}): {}", publishState, this);
}
}

Expand Down
Loading