Skip to content
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

Jump to bottom

[GOBBLIN-1969] Increase Visibility for Flow Compilation Failures #3840

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open

[GOBBLIN-1969] Increase Visibility for Flow Compilation Failures #3840

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
99769c4
Add logging and metrics for flow compilation validation
Dec 1, 2023
37654ff
Change metric to count successful flow compilations
Dec 1, 2023
33ada42
Initialize FlowCatalog before DagManager started
Dec 2, 2023
028e21b
Revert "Initialize FlowCatalog before DagManager started"
Dec 2, 2023
ebbdad3
Rename metrics to be uniform
Dec 2, 2023
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
1 change: 1 addition & 0 deletions ...ics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class ServiceMetricNames {
public static final String DAG_MANAGER_PREFIX = GOBBLIN_SERVICE_PREFIX_WITH_DELIMITER + "dagManager";
public static final String
DAG_MANAGER_FAILED_LAUNCH_EVENTS_ON_STARTUP_COUNT = DAG_MANAGER_PREFIX + ".failedLaunchEventsOnStartupCount";
public static final String DAG_MANAGER_SUCCESSFUL_LAUNCH_EVENTS_ON_STARTUP_COUNT = DAG_MANAGER_PREFIX + ".successfulLaunchEventsOnStartupCount";
public static final String FLOW_FAILED_FORWARD_TO_DAG_MANAGER_COUNT = DAG_MANAGER_PREFIX + ".flowFailedForwardToDagManagerCount";

//Job status poll timer
Expand Down
4 changes: 4 additions & 0 deletions .../src/main/java/org/apache/gobblin/service/modules/flow/IdentityFlowToJobSpecCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,10 @@ public Dag<JobExecutionPlan> compileFlow(Spec spec) {
List<JobExecutionPlan> jobExecutionPlans;
try {
jobExecutionPlans = getJobExecutionPlans(source, destination, jobSpec);
if (jobExecutionPlans.isEmpty()) {
flowSpec.addCompilationError(source, destination,
String.format("Could not find path between source: %s and destination: %s", source, destination));
}
} catch (InterruptedException | ExecutionException e) {
Instrumented.markMeter(this.flowCompilationFailedMeter);
throw new RuntimeException("Cannot determine topology capabilities", e);
Expand Down
1 change: 1 addition & 0 deletions ...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -507,6 +507,7 @@ public void handleLaunchFlowEvent(DagActionStore.DagAction launchAction) {
this.flowCompilationValidationHelper.createExecutionPlanIfValid(spec, Optional.absent());
if (optionalJobExecutionPlanDag.isPresent()) {
addDag(optionalJobExecutionPlanDag.get(), true, true);
this.dagManagerMetrics.incrementSuccessfulLaunchCount();
} else {
log.warn("Failed flow compilation of spec causing launch flow event to be skipped on startup. Flow {}", flowId);
this.dagManagerMetrics.incrementFailedLaunchCount();
Expand Down
25 changes: 20 additions & 5 deletions ...ice/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,8 +76,9 @@ public class DagManagerMetrics {
private final Map<String, ContextAwareMeter> executorSlaExceededMeters = Maps.newConcurrentMap();
private final Map<String, ContextAwareMeter> executorJobSentMeters = Maps.newConcurrentMap();

// Metrics for unexpected flow handling failures
private ContextAwareCounter failedLaunchEventsOnActivationCount;
// Metric for unexpected flow handling outcomes
private ContextAwareCounter failedLaunchEventsOnStartupCount;
private ContextAwareCounter successfulLaunchEventsOnStartupCount;
MetricContext metricContext;

public DagManagerMetrics(MetricContext metricContext) {
Expand All @@ -103,9 +104,12 @@ public void activate() {
ServiceMetricNames.SLA_EXCEEDED_FLOWS_METER));
allRunningMeter = metricContext.contextAwareMeter(MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
ServiceMetricNames.JOBS_SENT_TO_SPEC_EXECUTOR));
failedLaunchEventsOnActivationCount = metricContext.contextAwareCounter(
// TODO: remove duplicate use of 'GOBBLIN_SERVICE_PREFIX' once startup functionality is stable
failedLaunchEventsOnStartupCount = metricContext.contextAwareCounter(
MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
ServiceMetricNames.DAG_MANAGER_FAILED_LAUNCH_EVENTS_ON_STARTUP_COUNT));
successfulLaunchEventsOnStartupCount = metricContext.contextAwareCounter(
ServiceMetricNames.DAG_MANAGER_SUCCESSFUL_LAUNCH_EVENTS_ON_STARTUP_COUNT);
}
}

Expand Down Expand Up @@ -205,10 +209,21 @@ public void incrementCountsStartSlaExceeded(Dag.DagNode<JobExecutionPlan> node)
}
}

// Increment the count for num of failed launches during leader activation
/**
* Increment the count for num of failed launches during system startup
*/
public void incrementFailedLaunchCount() {
if (this.metricContext != null) {
this.failedLaunchEventsOnActivationCount.inc();
this.failedLaunchEventsOnStartupCount.inc();
}
}

/**
* Increment the count for num of successful launches during system startup
*/
public void incrementSuccessfulLaunchCount() {
if (this.metricContext != null) {
this.successfulLaunchEventsOnStartupCount.inc();
}
}

Expand Down
6 changes: 6 additions & 0 deletions ...c/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,10 +86,16 @@ public Optional<Dag<JobExecutionPlan>> createExecutionPlanIfValid(FlowSpec flowS
Map<String, String> flowMetadata = TimingEventUtils.getFlowMetadata(flowSpec);

if (!jobExecutionPlanDagOptional.isPresent()) {
log.warn("No dag execution plan created for flowGroup: {} flowName: {}",
flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD),
flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD));
Comment on lines +90 to +91
Copy link
Contributor

@phet phet Dec 4, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if we do anything other than logging the value, I recommend a default (even when it should always be set). for logging only, it's safe enough to print null, when that's what's actually returned.

return Optional.absent();
}

if (jobExecutionPlanDagOptional.get() == null || jobExecutionPlanDagOptional.get().isEmpty()) {
log.warn("Null or empty dag execution plan created for flowGroup: {} flowName: {}",
flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD),
flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD));
populateFlowCompilationFailedEventMessage(eventSubmitter, flowSpec, flowMetadata);
return Optional.absent();
}
Expand Down