-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
TSPS-140 Send email notifications for succeeded and failed jobs (#181)
- Loading branch information
1 parent
9567a35
commit aa748b5
Showing
33 changed files
with
1,209 additions
and
20 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
6 changes: 6 additions & 0 deletions
6
...c/main/java/bio/terra/pipelines/app/configuration/internal/NotificationConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| package bio.terra.pipelines.app.configuration.internal; | ||
|
|
||
| import org.springframework.boot.context.properties.ConfigurationProperties; | ||
|
|
||
| @ConfigurationProperties("pipelines.notifications") | ||
| public record NotificationConfiguration(String projectId, String topicId) {} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
56 changes: 56 additions & 0 deletions
56
...src/main/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHook.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package bio.terra.pipelines.common.utils; | ||
|
|
||
| import static bio.terra.pipelines.common.utils.FlightUtils.flightMapKeyIsTrue; | ||
|
|
||
| import bio.terra.pipelines.dependencies.stairway.JobMapKeys; | ||
| import bio.terra.pipelines.notifications.NotificationService; | ||
| import bio.terra.stairway.FlightContext; | ||
| import bio.terra.stairway.FlightMap; | ||
| import bio.terra.stairway.FlightStatus; | ||
| import bio.terra.stairway.HookAction; | ||
| import bio.terra.stairway.StairwayHook; | ||
| import java.util.UUID; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| /** | ||
| * A {@link StairwayHook} that sends a Job Failed Notification email via pubsub/Thurloe upon flight | ||
| * failure. | ||
| * | ||
| * <p>This hook action will only run if the flight's input parameters contain the JobMapKeys key for | ||
| * DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK and the flight's status is not SUCCESS. | ||
| * | ||
| * <p>The JobMapKeys key for PIPELINE_NAME is required to send the notification. | ||
| */ | ||
| @Component | ||
| public class StairwaySendFailedJobNotificationHook implements StairwayHook { | ||
| private final NotificationService notificationService; | ||
| private static final Logger logger = | ||
| LoggerFactory.getLogger(StairwaySendFailedJobNotificationHook.class); | ||
|
|
||
| public StairwaySendFailedJobNotificationHook(NotificationService notificationService) { | ||
| this.notificationService = notificationService; | ||
| } | ||
|
|
||
| @Override | ||
| public HookAction endFlight(FlightContext context) { | ||
|
|
||
| FlightMap inputParameters = context.getInputParameters(); | ||
|
|
||
| if (flightMapKeyIsTrue(inputParameters, JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK) | ||
| && context.getFlightStatus() != FlightStatus.SUCCESS) { | ||
| logger.info( | ||
| "Flight has status {}, sending failed job notification email", context.getFlightStatus()); | ||
|
|
||
| FlightUtils.validateRequiredEntries(inputParameters, JobMapKeys.USER_ID); | ||
|
|
||
| UUID jobId = UUID.fromString(context.getFlightId()); | ||
| String userId = inputParameters.get(JobMapKeys.USER_ID, String.class); | ||
|
|
||
| // send email notification | ||
| notificationService.configureAndSendPipelineRunFailedNotification(jobId, userId, context); | ||
| } | ||
| return HookAction.CONTINUE; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
34 changes: 34 additions & 0 deletions
34
service/src/main/java/bio/terra/pipelines/notifications/BaseTeaspoonsJobNotification.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package bio.terra.pipelines.notifications; | ||
|
|
||
| import lombok.Getter; | ||
|
|
||
| /** Base class for Teaspoons job notifications. Contains common fields for all job notifications. */ | ||
| @Getter | ||
| public abstract class BaseTeaspoonsJobNotification { | ||
| protected String notificationType; | ||
| protected String recipientUserId; | ||
| protected String pipelineDisplayName; | ||
| protected String jobId; | ||
| protected String timeSubmitted; | ||
| protected String timeCompleted; | ||
| protected String quotaRemaining; | ||
| protected String quotaConsumedByJob; | ||
| protected String userDescription; | ||
|
|
||
| protected BaseTeaspoonsJobNotification( | ||
| String recipientUserId, | ||
| String pipelineDisplayName, | ||
| String jobId, | ||
| String timeSubmitted, | ||
| String timeCompleted, | ||
| String quotaRemaining, | ||
| String userDescription) { | ||
| this.recipientUserId = recipientUserId; | ||
| this.pipelineDisplayName = pipelineDisplayName; | ||
| this.jobId = jobId; | ||
| this.timeSubmitted = timeSubmitted; | ||
| this.timeCompleted = timeCompleted; | ||
| this.quotaRemaining = quotaRemaining; | ||
| this.userDescription = userDescription; | ||
| } | ||
| } |
162 changes: 162 additions & 0 deletions
162
service/src/main/java/bio/terra/pipelines/notifications/NotificationService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,162 @@ | ||
| package bio.terra.pipelines.notifications; | ||
|
|
||
| import static bio.terra.pipelines.app.controller.JobApiUtils.buildApiErrorReport; | ||
|
|
||
| import bio.terra.pipelines.app.configuration.internal.NotificationConfiguration; | ||
| import bio.terra.pipelines.db.entities.Pipeline; | ||
| import bio.terra.pipelines.db.entities.PipelineRun; | ||
| import bio.terra.pipelines.db.entities.UserQuota; | ||
| import bio.terra.pipelines.generated.model.ApiErrorReport; | ||
| import bio.terra.pipelines.service.PipelineRunsService; | ||
| import bio.terra.pipelines.service.PipelinesService; | ||
| import bio.terra.pipelines.service.QuotasService; | ||
| import bio.terra.stairway.FlightContext; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import java.io.IOException; | ||
| import java.time.Instant; | ||
| import java.time.ZoneId; | ||
| import java.time.format.DateTimeFormatter; | ||
| import java.util.Optional; | ||
| import java.util.UUID; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| /** | ||
| * Service to encapsulate the logic for composing and sending email notifications to users about | ||
| * completed pipeline runs. Works with the Terra Thurloe service via PubSub messages. | ||
| */ | ||
| @Service | ||
| public class NotificationService { | ||
| private static final Logger logger = LoggerFactory.getLogger(NotificationService.class); | ||
|
|
||
| private final PipelineRunsService pipelineRunsService; | ||
| private final PipelinesService pipelinesService; | ||
| private final QuotasService quotasService; | ||
| private final PubsubService pubsubService; | ||
| private final NotificationConfiguration notificationConfiguration; | ||
| private final ObjectMapper objectMapper; | ||
|
|
||
| public NotificationService( | ||
| PipelineRunsService pipelineRunsService, | ||
| PipelinesService pipelinesService, | ||
| QuotasService quotasService, | ||
| PubsubService pubsubService, | ||
| NotificationConfiguration notificationConfiguration, | ||
| ObjectMapper objectMapper) { | ||
| this.pipelineRunsService = pipelineRunsService; | ||
| this.pipelinesService = pipelinesService; | ||
| this.quotasService = quotasService; | ||
| this.pubsubService = pubsubService; | ||
| this.notificationConfiguration = notificationConfiguration; | ||
| this.objectMapper = objectMapper; | ||
| } | ||
|
|
||
| /** | ||
| * Pull together the common fields for a notification. | ||
| * | ||
| * @param jobId the job id | ||
| * @param userId the user id | ||
| * @param context the flight context (only needed for failed notifications) | ||
| * @param isSuccess whether the notification is for a succeeded job; if false, creates a failed | ||
| * notification | ||
| * @return the base notification object | ||
| */ | ||
| private BaseTeaspoonsJobNotification createTeaspoonsJobNotification( | ||
| UUID jobId, String userId, FlightContext context, boolean isSuccess) { | ||
| PipelineRun pipelineRun = pipelineRunsService.getPipelineRun(jobId, userId); | ||
| Pipeline pipeline = pipelinesService.getPipelineById(pipelineRun.getPipelineId()); | ||
| String pipelineDisplayName = pipeline.getDisplayName(); | ||
|
|
||
| // if flight fails before quota steps on user's first run, there won't be a row for them yet | ||
| // in the quotas table | ||
| UserQuota userQuota = | ||
| quotasService.getOrCreateQuotaForUserAndPipeline(userId, pipeline.getName()); | ||
| String quotaRemaining = String.valueOf(userQuota.getQuota() - userQuota.getQuotaConsumed()); | ||
|
|
||
| if (isSuccess) { // succeeded | ||
| return new TeaspoonsJobSucceededNotification( | ||
| userId, | ||
| pipelineDisplayName, | ||
| jobId.toString(), | ||
| formatInstantToReadableString(pipelineRun.getCreated()), | ||
| formatInstantToReadableString(pipelineRun.getUpdated()), | ||
| pipelineRun.getQuotaConsumed().toString(), | ||
| quotaRemaining, | ||
| pipelineRun.getDescription()); | ||
| } else { // failed | ||
| // get exception | ||
| Optional<Exception> exception = context.getResult().getException(); | ||
| String errorMessage; | ||
| if (exception.isPresent()) { | ||
| ApiErrorReport errorReport = | ||
| buildApiErrorReport(exception.get()); // use same logic that the status endpoint uses | ||
| errorMessage = errorReport.getMessage(); | ||
| } else { | ||
| logger.error( | ||
| "No exception found in flight result for flight {} with status {}", | ||
| context.getFlightId(), | ||
| context.getFlightStatus()); | ||
| errorMessage = "Unknown error"; | ||
| } | ||
| return new TeaspoonsJobFailedNotification( | ||
| userId, | ||
| pipelineDisplayName, | ||
| jobId.toString(), | ||
| errorMessage, | ||
| formatInstantToReadableString(pipelineRun.getCreated()), | ||
| formatInstantToReadableString(pipelineRun.getUpdated()), | ||
| quotaRemaining, | ||
| pipelineRun.getDescription()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Format an Instant as a date time string in UTC using the RFC-1123 date-time formatter, such as | ||
| * 'Tue, 3 Jun 2008 11:05:30 GMT'. | ||
| * | ||
| * @param dateTime the Instant to format | ||
| * @return the formatted date time string | ||
| */ | ||
| protected String formatInstantToReadableString(Instant dateTime) { | ||
| return dateTime.atZone(ZoneId.of("UTC")).format(DateTimeFormatter.RFC_1123_DATE_TIME); | ||
| } | ||
|
|
||
| /** | ||
| * Configure and send a notification that a job has succeeded. | ||
| * | ||
| * @param jobId the job id | ||
| * @param userId the user id | ||
| */ | ||
| public void configureAndSendPipelineRunSucceededNotification(UUID jobId, String userId) { | ||
| try { | ||
| pubsubService.publishMessage( | ||
| notificationConfiguration.projectId(), | ||
| notificationConfiguration.topicId(), | ||
| objectMapper.writeValueAsString( | ||
| createTeaspoonsJobNotification(jobId, userId, null, true))); | ||
| } catch (IOException e) { | ||
| logger.error("Error sending pipelineRunSucceeded notification", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Configure and send a notification that a job has failed. | ||
| * | ||
| * @param jobId the job id | ||
| * @param userId the user id | ||
| * @param context the flight context | ||
| */ | ||
| public void configureAndSendPipelineRunFailedNotification( | ||
| UUID jobId, String userId, FlightContext context) { | ||
| try { | ||
| pubsubService.publishMessage( | ||
| notificationConfiguration.projectId(), | ||
| notificationConfiguration.topicId(), | ||
| objectMapper.writeValueAsString( | ||
| createTeaspoonsJobNotification(jobId, userId, context, false))); | ||
| } catch (IOException e) { | ||
| logger.error("Error sending pipelineRunFailed notification", e); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.