You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2022/08/01 23:39:45 UTC
[GitHub] [gobblin] Will-Lo opened a new pull request, #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Will-Lo opened a new pull request, #3532:
URL: https://github.com/apache/gobblin/pull/3532
…executor for SUCCESS, FAILED, SLA_EXCEEDED, START_SLA_EXCEEDED
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [x] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1672
### Description
- [x] Here are some details about my PR, including screenshots (if applicable):
- Add more metrics related to flow management. Since a good chunk of the code in the DagManager was related to metrics emission, I tried to contain it within a class.
- Adds the following metrics:
1. Successes per executor
2. Fails per executor
3. SLA Exceeded per executor
4. SLA Exceeded per flowgroup
5. Total SLA Exceeded
6. Jobs sent to spec executors
7. Start SLA exceeded per executor
### Tests
- [x] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [x] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
5. Subject uses the imperative mood ("add", not "adding")
6. Body wraps at 72 characters
7. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936031836
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerTest.java:
##########
@@ -779,15 +785,18 @@ public void testJobKilledSLAMetricsArePerExecutor() throws URISyntaxException, I
thenReturn(jobStatusIterator2).
thenReturn(jobStatusIterator3);
- // Run the thread once. All 3 jobs should be emitted an sla exceeded event
+ // Run the thread once. All 3 jobs should be emitted a sla exceeded event
Review Comment:
Actually, it should be in uppercase SLA, then intellij recognizes the 'As' sound :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936056633
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -1132,24 +1035,39 @@ private void cleanUp() {
DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
deleteJobState(dagId, dagNode);
}
- log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
- onFlowFailure(dagId);
+ Dag<JobExecutionPlan> dag = this.dags.get(dagId);
+ String status = TimingEvent.FlowTimings.FLOW_FAILED;
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
+ addFailedDag(dagId);
+ log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
// send an event before cleaning up dag
- DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), TimingEvent.FlowTimings.FLOW_FAILED);
+ DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), status);
dagIdstoClean.add(dagId);
}
- //Clean up completed dags
- for (String dagId : this.dags.keySet()) {
+ // Remove dags that are finished and emit their appropriate metrics
+ for (Map.Entry<String, Dag<JobExecutionPlan>> dagIdKeyPair : this.dags.entrySet()) {
+ String dagId = dagIdKeyPair.getKey();
+ Dag<JobExecutionPlan> dag = dagIdKeyPair.getValue();
if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
String status = TimingEvent.FlowTimings.FLOW_SUCCEEDED;
if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
- onFlowFailure(dagId);
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
Review Comment:
Are we sure we want to decouple `addFailedDag` and metrics emission?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r935991914
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerTest.java:
##########
@@ -779,15 +785,18 @@ public void testJobKilledSLAMetricsArePerExecutor() throws URISyntaxException, I
thenReturn(jobStatusIterator2).
thenReturn(jobStatusIterator3);
- // Run the thread once. All 3 jobs should be emitted an sla exceeded event
+ // Run the thread once. All 3 jobs should be emitted a sla exceeded event
Review Comment:
"an" might be right, https://www.reddit.com/r/ENGLISH/comments/433x3y/do_you_use_a_or_an_before_the_acronym_sla_sla_is/
I will let you decide on this :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#issuecomment-1209781526
+1
LGTM. Thank you for doing the much needed refactoring of DagManager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] codecov-commenter commented on pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#issuecomment-1201852581
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3532?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3532](https://codecov.io/gh/apache/gobblin/pull/3532?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (3117bdb) into [master](https://codecov.io/gh/apache/gobblin/commit/9ce1e65aa36674742877b5aa2083412d85b5764f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (9ce1e65) will **decrease** coverage by `3.29%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3532 +/- ##
============================================
- Coverage 46.69% 43.39% -3.30%
+ Complexity 10432 2038 -8394
============================================
Files 2083 406 -1677
Lines 81516 17526 -63990
Branches 9100 2141 -6959
============================================
- Hits 38061 7605 -30456
+ Misses 39944 9072 -30872
+ Partials 3511 849 -2662
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3532?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...a/org/apache/gobblin/cluster/GobblinHelixTask.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpbkhlbGl4VGFzay5qYXZh) | `60.21% <0.00%> (-2.16%)` | :arrow_down: |
| [...org/apache/gobblin/metrics/ServiceMetricNames.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tZXRyaWNzLWxpYnMvZ29iYmxpbi1tZXRyaWNzL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL21ldHJpY3MvU2VydmljZU1ldHJpY05hbWVzLmphdmE=) | | |
| [...blin/service/modules/orchestration/DagManager.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9vcmNoZXN0cmF0aW9uL0RhZ01hbmFnZXIuamF2YQ==) | | |
| [...hive/filter/DateRangePartitionFilterGenerator.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvaGl2ZS9maWx0ZXIvRGF0ZVJhbmdlUGFydGl0aW9uRmlsdGVyR2VuZXJhdG9yLmphdmE=) | | |
| [...apache/gobblin/writer/SimpleDataWriterBuilder.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3dyaXRlci9TaW1wbGVEYXRhV3JpdGVyQnVpbGRlci5qYXZh) | | |
| [...n/yarn/event/GetApplicationReportFailureEvent.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vZXZlbnQvR2V0QXBwbGljYXRpb25SZXBvcnRGYWlsdXJlRXZlbnQuamF2YQ==) | | |
| [...org/apache/gobblin/writer/PartitionIdentifier.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vd3JpdGVyL1BhcnRpdGlvbklkZW50aWZpZXIuamF2YQ==) | | |
| [...he/gobblin/runtime/job\_catalog/JobCatalogBase.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvam9iX2NhdGFsb2cvSm9iQ2F0YWxvZ0Jhc2UuamF2YQ==) | | |
| [...lin/compaction/mapreduce/orc/OrcKeyComparator.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1jb21wYWN0aW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbXBhY3Rpb24vbWFwcmVkdWNlL29yYy9PcmNLZXlDb21wYXJhdG9yLmphdmE=) | | |
| [...ermark/FullPathCopyableFileWatermarkGenerator.java](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvd2F0ZXJtYXJrL0Z1bGxQYXRoQ29weWFibGVGaWxlV2F0ZXJtYXJrR2VuZXJhdG9yLmphdmE=) | | |
| ... and [1670 more](https://codecov.io/gh/apache/gobblin/pull/3532/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r935986637
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerTest.java:
##########
@@ -1053,6 +1062,150 @@ public void testEmitFlowMetricOnlyIfNotAdhoc() throws URISyntaxException, IOExce
Assert.assertEquals(metricContext.getParent().get().getGauges().get(flowStateGaugeName1).getValue(), DagManager.FlowState.SUCCESSFUL.value);
}
+ @Test (dependsOnMethods = "testEmitFlowMetricOnlyIfNotAdhoc")
+ public void testJobSlaKilledMetrics() throws URISyntaxException, IOException {
+ long flowExecutionId = System.currentTimeMillis() - 10 * 60 * 1000;
Review Comment:
*nit
I use Duration.ofMinutes(10).toMillis() to keep it more readable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] ZihanLi58 merged pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
ZihanLi58 merged PR #3532:
URL: https://github.com/apache/gobblin/pull/3532
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936074643
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -1132,24 +1035,39 @@ private void cleanUp() {
DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
deleteJobState(dagId, dagNode);
}
- log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
- onFlowFailure(dagId);
+ Dag<JobExecutionPlan> dag = this.dags.get(dagId);
+ String status = TimingEvent.FlowTimings.FLOW_FAILED;
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
+ addFailedDag(dagId);
+ log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
// send an event before cleaning up dag
- DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), TimingEvent.FlowTimings.FLOW_FAILED);
+ DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), status);
dagIdstoClean.add(dagId);
}
- //Clean up completed dags
- for (String dagId : this.dags.keySet()) {
+ // Remove dags that are finished and emit their appropriate metrics
+ for (Map.Entry<String, Dag<JobExecutionPlan>> dagIdKeyPair : this.dags.entrySet()) {
+ String dagId = dagIdKeyPair.getKey();
+ Dag<JobExecutionPlan> dag = dagIdKeyPair.getValue();
if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
String status = TimingEvent.FlowTimings.FLOW_SUCCEEDED;
if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
- onFlowFailure(dagId);
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.conditionallyMarkFlowAsState(DagManagerUtils.getFlowId(this.dags.get(dagId)),
+ DagManager.FlowState.FAILED);
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
status = TimingEvent.FlowTimings.FLOW_FAILED;
+ addFailedDag(dagId);
this.failedDagIdsFinishAllPossible.remove(dagId);
- conditionallyUpdateFlowGaugeExecutionState(flowGauges, DagManagerUtils.getFlowId(this.dags.get(dagId)), FlowState.FAILED);
Review Comment:
This line is intentionally removed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936005023
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerTest.java:
##########
@@ -779,15 +785,18 @@ public void testJobKilledSLAMetricsArePerExecutor() throws URISyntaxException, I
thenReturn(jobStatusIterator2).
thenReturn(jobStatusIterator3);
- // Run the thread once. All 3 jobs should be emitted an sla exceeded event
+ // Run the thread once. All 3 jobs should be emitted a sla exceeded event
Review Comment:
Damn today I learned 😅 I will fix that, Intellij's syntax checker flagged it for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936031836
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagManagerTest.java:
##########
@@ -779,15 +785,18 @@ public void testJobKilledSLAMetricsArePerExecutor() throws URISyntaxException, I
thenReturn(jobStatusIterator2).
thenReturn(jobStatusIterator3);
- // Run the thread once. All 3 jobs should be emitted an sla exceeded event
+ // Run the thread once. All 3 jobs should be emitted a sla exceeded event
Review Comment:
Actually, it should be in uppercase SLA, then intellij recognize the 'As' sound
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936061609
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -1132,24 +1035,39 @@ private void cleanUp() {
DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
deleteJobState(dagId, dagNode);
}
- log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
- onFlowFailure(dagId);
+ Dag<JobExecutionPlan> dag = this.dags.get(dagId);
+ String status = TimingEvent.FlowTimings.FLOW_FAILED;
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
+ addFailedDag(dagId);
+ log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
// send an event before cleaning up dag
- DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), TimingEvent.FlowTimings.FLOW_FAILED);
+ DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), status);
dagIdstoClean.add(dagId);
}
- //Clean up completed dags
- for (String dagId : this.dags.keySet()) {
+ // Remove dags that are finished and emit their appropriate metrics
+ for (Map.Entry<String, Dag<JobExecutionPlan>> dagIdKeyPair : this.dags.entrySet()) {
+ String dagId = dagIdKeyPair.getKey();
+ Dag<JobExecutionPlan> dag = dagIdKeyPair.getValue();
if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
String status = TimingEvent.FlowTimings.FLOW_SUCCEEDED;
if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
- onFlowFailure(dagId);
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
Review Comment:
Maybe, we can move this if block inside `addFailedDag` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936076417
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -1132,24 +1035,39 @@ private void cleanUp() {
DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
deleteJobState(dagId, dagNode);
}
- log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
- onFlowFailure(dagId);
+ Dag<JobExecutionPlan> dag = this.dags.get(dagId);
+ String status = TimingEvent.FlowTimings.FLOW_FAILED;
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
+ addFailedDag(dagId);
+ log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
// send an event before cleaning up dag
- DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), TimingEvent.FlowTimings.FLOW_FAILED);
+ DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), status);
dagIdstoClean.add(dagId);
}
- //Clean up completed dags
- for (String dagId : this.dags.keySet()) {
+ // Remove dags that are finished and emit their appropriate metrics
+ for (Map.Entry<String, Dag<JobExecutionPlan>> dagIdKeyPair : this.dags.entrySet()) {
+ String dagId = dagIdKeyPair.getKey();
+ Dag<JobExecutionPlan> dag = dagIdKeyPair.getValue();
if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
String status = TimingEvent.FlowTimings.FLOW_SUCCEEDED;
if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
- onFlowFailure(dagId);
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.conditionallyMarkFlowAsState(DagManagerUtils.getFlowId(this.dags.get(dagId)),
+ DagManager.FlowState.FAILED);
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
status = TimingEvent.FlowTimings.FLOW_FAILED;
+ addFailedDag(dagId);
this.failedDagIdsFinishAllPossible.remove(dagId);
- conditionallyUpdateFlowGaugeExecutionState(flowGauges, DagManagerUtils.getFlowId(this.dags.get(dagId)), FlowState.FAILED);
Review Comment:
Why are we removing flowGauges ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r936001512
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagerMetrics.java:
##########
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.modules.orchestration;
+
+import com.codahale.metrics.MetricRegistry;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import com.typesafe.config.Config;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metrics.ContextAwareCounter;
+import org.apache.gobblin.metrics.ContextAwareGauge;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.RootMetricContext;
+import org.apache.gobblin.metrics.ServiceMetricNames;
+import org.apache.gobblin.metrics.metric.filter.MetricNameRegexFilter;
+import org.apache.gobblin.service.FlowId;
+import org.apache.gobblin.service.RequesterService;
+import org.apache.gobblin.service.ServiceRequester;
+import org.apache.gobblin.service.modules.flowgraph.Dag;
+import org.apache.gobblin.service.modules.spec.JobExecutionPlan;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+@Slf4j
+public class DagManagerMetrics {
+ private static final Map<String, DagManager.FlowState> flowGauges = Maps.newConcurrentMap();
+ // Meters representing the total number of flows in a given state
+ private ContextAwareMeter allSuccessfulMeter;
+ private ContextAwareMeter allFailedMeter;
+ private ContextAwareMeter allRunningMeter;
+ private ContextAwareMeter allSlaExceededMeter;
+ private ContextAwareMeter allStartSlaExceededMeter;
+ // Meters representing the flows in a given state per flowgroup
+ private final Map<String, ContextAwareMeter> groupSuccessfulMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> groupFailureMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> groupStartSlaExceededMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> groupSlaExceededMeters = Maps.newConcurrentMap();
+
+ // Meters representing the jobs in a given state per executor
+ // These metrics need to be invoked differently to account for automated retries and multihop scenarios.
+ private final Map<String, ContextAwareMeter> executorSuccessMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> executorFailureMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> executorStartSlaExceededMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> executorSlaExceededMeters = Maps.newConcurrentMap();
+ private final Map<String, ContextAwareMeter> executorJobSentMeters = Maps.newConcurrentMap();
+ private boolean instrumentationEnabled;
+ MetricContext metricContext;
+
+ public DagManagerMetrics(MetricContext metricContext, boolean instrumentationEnabled) {
Review Comment:
should we just have metricContext in this signature?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3532: [GOBBLIN-1672] Refactor metrics from DagManager into its own class, add metrics per …
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3532:
URL: https://github.com/apache/gobblin/pull/3532#discussion_r937128646
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -1132,24 +1035,39 @@ private void cleanUp() {
DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
deleteJobState(dagId, dagNode);
}
- log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
- onFlowFailure(dagId);
+ Dag<JobExecutionPlan> dag = this.dags.get(dagId);
+ String status = TimingEvent.FlowTimings.FLOW_FAILED;
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
+ addFailedDag(dagId);
+ log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
// send an event before cleaning up dag
- DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), TimingEvent.FlowTimings.FLOW_FAILED);
+ DagManagerUtils.emitFlowEvent(this.eventSubmitter, this.dags.get(dagId), status);
dagIdstoClean.add(dagId);
}
- //Clean up completed dags
- for (String dagId : this.dags.keySet()) {
+ // Remove dags that are finished and emit their appropriate metrics
+ for (Map.Entry<String, Dag<JobExecutionPlan>> dagIdKeyPair : this.dags.entrySet()) {
+ String dagId = dagIdKeyPair.getKey();
+ Dag<JobExecutionPlan> dag = dagIdKeyPair.getValue();
if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
String status = TimingEvent.FlowTimings.FLOW_SUCCEEDED;
if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
- onFlowFailure(dagId);
+ if (TimingEvent.FlowTimings.FLOW_RUN_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.emitFlowSlaExceededMetrics(DagManagerUtils.getFlowId(dag));
+ } else if (!TimingEvent.FlowTimings.FLOW_START_DEADLINE_EXCEEDED.equals(dag.getFlowEvent())) {
+ this.dagManagerMetrics.conditionallyMarkFlowAsState(DagManagerUtils.getFlowId(this.dags.get(dagId)),
+ DagManager.FlowState.FAILED);
+ dagManagerMetrics.emitFlowFailedMetrics(DagManagerUtils.getFlowId(this.dags.get(dagId)));
+ }
status = TimingEvent.FlowTimings.FLOW_FAILED;
+ addFailedDag(dagId);
this.failedDagIdsFinishAllPossible.remove(dagId);
- conditionallyUpdateFlowGaugeExecutionState(flowGauges, DagManagerUtils.getFlowId(this.dags.get(dagId)), FlowState.FAILED);
Review Comment:
I think it was unintentional, I went back and forth on where to add the flow gauge removal and mistakenly left it out of addFailedDag(), will add it there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org