Skip to content

[GOBBLIN-2183] Add COMPILED & RUNNING events to GaSJobObservabilityEventProducer #4086

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

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

[GOBBLIN-2183] Add COMPILED & RUNNING events to GaSJobObservabilityEventProducer #4086

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: 3 additions & 1 deletion gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSFlowObservabilityEvent.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,13 +50,15 @@
"type": "enum",
"name": "FlowStatus",
"symbols": [
"COMPILED",
"RUNNING",
"SUCCEEDED",
"COMPILATION_FAILURE",
"SUBMISSION_FAILURE",
"EXECUTION_FAILURE",
"CANCELLED"
],
"doc": "Final flow status for the GaaS flow",
"doc": "Flow status for the GaaS flow",
"compliance": "NONE"
}
},
Expand Down
4 changes: 3 additions & 1 deletion gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,13 +62,15 @@
"type": "enum",
"name": "JobStatus",
"symbols": [
"COMPILED",
"RUNNING",
"SUCCEEDED",
"COMPILATION_FAILURE",
"SUBMISSION_FAILURE",
"EXECUTION_FAILURE",
"CANCELLED"
],
"doc": "Final job status for this job in the GaaS flow",
"doc": "Job status for this job in the GaaS flow",
"compliance": "NONE"
}
},
Expand Down
51 changes: 38 additions & 13 deletions ...blin-kafka-09/src/test/java/org/apache/gobblin/runtime/KafkaAvroJobStatusMonitorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -627,6 +627,7 @@ public void testJobMonitorCreatesGaaSObservabilityEvent() throws IOException, Re
//Submit GobblinTrackingEvents to Kafka
ImmutableList.of(
createFlowCompiledEvent(),
createJobStartEvent(),
createWorkUnitTimingEvent(),
createJobSucceededEvent()
).forEach(event -> {
Expand All @@ -651,19 +652,31 @@ public void testJobMonitorCreatesGaaSObservabilityEvent() throws IOException, Re
State state = getNextJobStatusState(jobStatusMonitor, recordIterator, "NA", "NA");
Assert.assertEquals(state.getProp(JobStatusRetriever.EVENT_NAME_FIELD), ExecutionStatus.COMPILED.name());

state = getNextJobStatusState(jobStatusMonitor, recordIterator, this.jobGroup, this.jobName);
Assert.assertEquals(state.getProp(JobStatusRetriever.EVENT_NAME_FIELD), ExecutionStatus.RUNNING.name());

getNextJobStatusState(jobStatusMonitor, recordIterator, this.jobGroup, this.jobName);
state = getNextJobStatusState(jobStatusMonitor, recordIterator, this.jobGroup, this.jobName);
Assert.assertEquals(state.getProp(JobStatusRetriever.EVENT_NAME_FIELD), ExecutionStatus.COMPLETE.name());

// Only the COMPLETE event should create a GaaSJobObservabilityEvent
List<GaaSJobObservabilityEvent> emittedEvents = mockEventProducer.getTestEmittedJobEvents();
Assert.assertEquals(emittedEvents.size(), 3);
Iterator<GaaSJobObservabilityEvent> iterator = emittedEvents.iterator();
GaaSJobObservabilityEvent event1 = iterator.next();
Assert.assertEquals(event1.getJobStatus(), JobStatus.SUCCEEDED);
Assert.assertEquals(event1.getFlowName(), this.flowName);
Assert.assertEquals(event1.getFlowGroup(), this.flowGroup);
Assert.assertEquals(event1.getJobPlanningStartTimestamp(), Long.valueOf(2));
Assert.assertEquals(event1.getJobPlanningEndTimestamp(), Long.valueOf(3));
GaaSJobObservabilityEvent event = iterator.next();
Assert.assertEquals(event.getJobStatus(), JobStatus.COMPILED);
Assert.assertEquals(event.getFlowName(), this.flowName);
Assert.assertEquals(event.getFlowGroup(), this.flowGroup);
event = iterator.next();
Assert.assertEquals(event.getJobStatus(), JobStatus.RUNNING);
Assert.assertEquals(event.getFlowName(), this.flowName);
Assert.assertEquals(event.getFlowGroup(), this.flowGroup);
event = iterator.next();
Assert.assertEquals(event.getJobStatus(), JobStatus.SUCCEEDED);
Assert.assertEquals(event.getFlowName(), this.flowName);
Assert.assertEquals(event.getFlowGroup(), this.flowGroup);
Assert.assertEquals(event.getJobPlanningStartTimestamp(), Long.valueOf(2));
Assert.assertEquals(event.getJobPlanningEndTimestamp(), Long.valueOf(3));
jobStatusMonitor.shutDown();
}

Expand Down Expand Up @@ -706,12 +719,16 @@ public void testObservabilityEventSingleEmission() throws IOException, Reflectiv

// Only the COMPLETE event should create a GaaSJobObservabilityEvent
List<GaaSJobObservabilityEvent> emittedEvents = mockEventProducer.getTestEmittedJobEvents();
Assert.assertEquals(emittedEvents.size(), 1);
Assert.assertEquals(emittedEvents.size(), 2);
Iterator<GaaSJobObservabilityEvent> iterator = emittedEvents.iterator();
GaaSJobObservabilityEvent event1 = iterator.next();
Assert.assertEquals(event1.getJobStatus(), JobStatus.CANCELLED);
Assert.assertEquals(event1.getFlowName(), this.flowName);
Assert.assertEquals(event1.getFlowGroup(), this.flowGroup);
GaaSJobObservabilityEvent event = iterator.next();
Assert.assertEquals(event.getJobStatus(), JobStatus.COMPILED);
Assert.assertEquals(event.getFlowName(), this.flowName);
Assert.assertEquals(event.getFlowGroup(), this.flowGroup);
event = iterator.next();
Assert.assertEquals(event.getJobStatus(), JobStatus.CANCELLED);
Assert.assertEquals(event.getFlowName(), this.flowName);
Assert.assertEquals(event.getFlowGroup(), this.flowGroup);

jobStatusMonitor.shutDown();
}
Expand Down Expand Up @@ -766,9 +783,13 @@ public void testObservabilityEventFlowLevel() throws IOException, ReflectiveOper

// Only the COMPLETE event should create a GaaSFlowObservabilityEvent
List<GaaSFlowObservabilityEvent> emittedFlowEvents = mockEventProducer.getTestEmittedFlowEvents();
Assert.assertEquals(emittedFlowEvents.size(), 1);
Assert.assertEquals(emittedFlowEvents.size(), 2);
Iterator<GaaSFlowObservabilityEvent> flowIterator = emittedFlowEvents.iterator();
GaaSFlowObservabilityEvent flowEvent = flowIterator.next();
Assert.assertEquals(flowEvent.getFlowStatus(), FlowStatus.COMPILED);
Assert.assertEquals(flowEvent.getFlowName(), this.flowName);
Assert.assertEquals(flowEvent.getFlowGroup(), this.flowGroup);
flowEvent = flowIterator.next();
Assert.assertEquals(flowEvent.getFlowStatus(), FlowStatus.SUCCEEDED);
Assert.assertEquals(flowEvent.getFlowName(), this.flowName);
Assert.assertEquals(flowEvent.getFlowGroup(), this.flowGroup);
Expand Down Expand Up @@ -826,9 +847,13 @@ public void testObservabilityEventFlowFailed() throws IOException, ReflectiveOpe

// Only the COMPLETE event should create a GaaSFlowObservabilityEvent
List<GaaSFlowObservabilityEvent> emittedFlowEvents = mockEventProducer.getTestEmittedFlowEvents();
Assert.assertEquals(emittedFlowEvents.size(), 1);
Assert.assertEquals(emittedFlowEvents.size(), 2);
Iterator<GaaSFlowObservabilityEvent> flowIterator = emittedFlowEvents.iterator();
GaaSFlowObservabilityEvent flowEvent = flowIterator.next();
Assert.assertEquals(flowEvent.getFlowStatus(), FlowStatus.COMPILED);
Assert.assertEquals(flowEvent.getFlowName(), this.flowName);
Assert.assertEquals(flowEvent.getFlowGroup(), this.flowGroup);
flowEvent = flowIterator.next();
Assert.assertEquals(flowEvent.getFlowStatus(), FlowStatus.EXECUTION_FAILURE);
Assert.assertEquals(flowEvent.getFlowName(), this.flowName);
Assert.assertEquals(flowEvent.getFlowGroup(), this.flowGroup);
Expand Down
5 changes: 5 additions & 0 deletions ...rc/main/java/org/apache/gobblin/service/monitoring/GaaSJobObservabilityEventProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,11 @@ private GaaSJobObservabilityEvent createGaaSObservabilityEvent(final State jobSt

private static JobStatus convertExecutionStatusTojobState(State state, ExecutionStatus executionStatus) {
switch (executionStatus) {
// Add more mapping for new events rather than just terminal ones
case COMPILED:
return JobStatus.COMPILED;
case RUNNING:
return JobStatus.RUNNING;
case FAILED:
// TODO: Separate failure cases to SUBMISSION FAILURE and COMPILATION FAILURE, investigate events to populate these fields
if (state.contains(TimingEvent.JOB_END_TIME)) {
Expand Down
7 changes: 7 additions & 0 deletions ...in-service/src/main/java/org/apache/gobblin/service/monitoring/KafkaJobStatusMonitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
Expand Down Expand Up @@ -233,6 +234,12 @@ protected void processMessage(DecodeableKafkaRecord<byte[],byte[]> message) {
// as much as FAILED does if we chose to emit ObservabilityEvent for FAILED_PENDING_RETRY
boolean retryRequired = modifyStateIfRetryRequired(jobStatus);

if (Objects.equals(status, "COMPILED")) {
this.eventProducer.emitObservabilityEvent(jobStatus);
}
if (updatedJobStatus.getRight() == NewState.RUNNING) {
this.eventProducer.emitObservabilityEvent(jobStatus);
Comment on lines +237 to +241
Copy link
Member

Choose a reason for hiding this comment

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

Why this difference i.e. using status and updatedJobStatus.getRight() for COMPILED and RUNNING ?

}
if (updatedJobStatus.getRight() == NewState.FINISHED && !retryRequired) {
// do not send event if retry is required, because it can alert users to re-submit a job that is already set to be retried by GaaS
this.eventProducer.emitObservabilityEvent(jobStatus);
Expand Down