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 3 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 + ".successfulLaunchEventsOnActivationCount";
public static final String FLOW_FAILED_FORWARD_TO_DAG_MANAGER_COUNT = DAG_MANAGER_PREFIX + ".flowFailedForwardToDagManagerCount";

//Job status poll timer
Expand Down
11 changes: 6 additions & 5 deletions ...vice/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,12 @@ public void configure(Binder binder) {
ServiceConfigKeys.TOPOLOGYSPEC_FACTORY_KEY, ServiceConfigKeys.DEFAULT_TOPOLOGY_SPEC_FACTORY));
}

// Initialize flow catalog before DagManager is enabled so templates are loaded when initializing the DagManager
Copy link
Contributor

Choose a reason for hiding this comment

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

I'm not too familiar w/ Guice... but are you sure the call to binder.bind is guaranteed to synchronously carryout "initializing the flow catalog" (so that finishes before the later call to binder.bind(DagManager.class))?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The order does matter here but this is actually the wrong class initialization and won't fix the problem so leaving out of this PR.

OptionalBinder.newOptionalBinder(binder, FlowCatalog.class);
if (serviceConfig.isFlowCatalogEnabled()) {
binder.bind(FlowCatalog.class);
}

OptionalBinder.newOptionalBinder(binder, DagManager.class);
if (serviceConfig.isDagManagerEnabled()) {
binder.bind(DagManager.class);
Expand All @@ -214,11 +220,6 @@ public void configure(Binder binder) {
LOGGER.info("No ZooKeeper connection string. Running in single instance mode.");
}

OptionalBinder.newOptionalBinder(binder, FlowCatalog.class);
if (serviceConfig.isFlowCatalogEnabled()) {
binder.bind(FlowCatalog.class);
}

if (serviceConfig.isJobStatusMonitorEnabled()) {
binder.bind(KafkaJobStatusMonitor.class).toProvider(KafkaJobStatusMonitorFactory.class).in(Singleton.class);
}
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.incrementSuccessfulLaunchAttemptCount();
} else {
log.warn("Failed flow compilation of spec causing launch flow event to be skipped on startup. Flow {}", flowId);
this.dagManagerMetrics.incrementFailedLaunchCount();
Expand Down
13 changes: 12 additions & 1 deletion ...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
// Metric for unexpected flow handling failures
private ContextAwareCounter failedLaunchEventsOnActivationCount;
private ContextAwareCounter successfulLaunchEventsOnActivationCount;
umustafi marked this conversation as resolved.
Show resolved Hide resolved
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));
// TODO: remove duplicate use of 'GOBBLIN_SERVICE_PREFIX' once startup functionality is stable
failedLaunchEventsOnActivationCount = metricContext.contextAwareCounter(
MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX,
ServiceMetricNames.DAG_MANAGER_FAILED_LAUNCH_EVENTS_ON_STARTUP_COUNT));
successfulLaunchEventsOnActivationCount = metricContext.contextAwareCounter(
ServiceMetricNames.DAG_MANAGER_SUCCESSFUL_LAUNCH_EVENTS_ON_STARTUP_COUNT);
}
}

Expand Down Expand Up @@ -212,6 +216,13 @@ public void incrementFailedLaunchCount() {
}
}

// Increment the count for num of successful launches attempted during leader activation
public void incrementSuccessfulLaunchAttemptCount() {
umustafi marked this conversation as resolved.
Show resolved Hide resolved
if (this.metricContext != null) {
this.successfulLaunchEventsOnActivationCount.inc();
}
}

private List<ContextAwareCounter> getRunningJobsCounterForUser(Dag.DagNode<JobExecutionPlan> dagNode) {
Config configs = dagNode.getValue().getJobSpec().getConfig();
String proxy = ConfigUtils.getString(configs, AzkabanProjectConfig.USER_TO_PROXY, null);
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