diff --git a/.github/workflows/tag-publish.yml b/.github/workflows/tag-publish.yml index 52650edb..1a0deddf 100644 --- a/.github/workflows/tag-publish.yml +++ b/.github/workflows/tag-publish.yml @@ -176,7 +176,7 @@ jobs: - name: Auth to GCP id: 'auth' - uses: google-github-actions/auth@v0 + uses: google-github-actions/auth@v2 with: token_format: 'access_token' workload_identity_provider: 'projects/1038484894585/locations/global/workloadIdentityPools/github-wi-pool/providers/github-wi-provider' @@ -184,7 +184,7 @@ jobs: # Install gcloud, `setup-gcloud` automatically picks up authentication from `auth`. - name: 'Set up Cloud SDK' - uses: 'google-github-actions/setup-gcloud@v0' + uses: google-github-actions/setup-gcloud@v2 - name: Explicitly auth Docker for Artifact Registry run: gcloud auth configure-docker $GOOGLE_DOCKER_REPOSITORY --quiet diff --git a/service/build.gradle b/service/build.gradle index 47ad118c..ab1df168 100644 --- a/service/build.gradle +++ b/service/build.gradle @@ -89,6 +89,7 @@ dependencies { // gcs implementation platform('com.google.cloud:libraries-bom:26.44.0') implementation 'com.google.cloud:google-cloud-storage' + implementation 'com.google.cloud:google-cloud-pubsub' liquibaseRuntime 'info.picocli:picocli:4.6.1' liquibaseRuntime 'org.postgresql:postgresql:42.6.1' diff --git a/service/src/main/java/bio/terra/pipelines/app/configuration/internal/NotificationConfiguration.java b/service/src/main/java/bio/terra/pipelines/app/configuration/internal/NotificationConfiguration.java new file mode 100644 index 00000000..32da2ce4 --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/app/configuration/internal/NotificationConfiguration.java @@ -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) {} diff --git a/service/src/main/java/bio/terra/pipelines/common/utils/FlightBeanBag.java b/service/src/main/java/bio/terra/pipelines/common/utils/FlightBeanBag.java index 163a88a2..d6d68c9a 100644 --- a/service/src/main/java/bio/terra/pipelines/common/utils/FlightBeanBag.java +++ b/service/src/main/java/bio/terra/pipelines/common/utils/FlightBeanBag.java @@ -9,6 +9,7 @@ import bio.terra.pipelines.dependencies.sam.SamService; import bio.terra.pipelines.dependencies.wds.WdsService; import bio.terra.pipelines.dependencies.workspacemanager.WorkspaceManagerService; +import bio.terra.pipelines.notifications.NotificationService; import bio.terra.pipelines.service.PipelineInputsOutputsService; import bio.terra.pipelines.service.PipelineRunsService; import bio.terra.pipelines.service.PipelinesService; @@ -38,6 +39,7 @@ public class FlightBeanBag { private final WorkspaceManagerService workspaceManagerService; private final RawlsService rawlsService; private final QuotasService quotasService; + private final NotificationService notificationService; private final ImputationConfiguration imputationConfiguration; private final CbasConfiguration cbasConfiguration; private final WdlPipelineConfiguration wdlPipelineConfiguration; @@ -54,6 +56,7 @@ public FlightBeanBag( CbasService cbasService, RawlsService rawlsService, QuotasService quotasService, + NotificationService notificationService, WorkspaceManagerService workspaceManagerService, ImputationConfiguration imputationConfiguration, CbasConfiguration cbasConfiguration, @@ -68,6 +71,7 @@ public FlightBeanBag( this.workspaceManagerService = workspaceManagerService; this.rawlsService = rawlsService; this.quotasService = quotasService; + this.notificationService = notificationService; this.imputationConfiguration = imputationConfiguration; this.cbasConfiguration = cbasConfiguration; this.wdlPipelineConfiguration = wdlPipelineConfiguration; diff --git a/service/src/main/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHook.java b/service/src/main/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHook.java new file mode 100644 index 00000000..14ccf7c6 --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHook.java @@ -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. + * + *

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. + * + *

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; + } +} diff --git a/service/src/main/java/bio/terra/pipelines/db/entities/PipelineRun.java b/service/src/main/java/bio/terra/pipelines/db/entities/PipelineRun.java index c5513bf6..5aa49472 100644 --- a/service/src/main/java/bio/terra/pipelines/db/entities/PipelineRun.java +++ b/service/src/main/java/bio/terra/pipelines/db/entities/PipelineRun.java @@ -69,6 +69,9 @@ public class PipelineRun { @Column(name = "description") private String description; + @Column(name = "quota_consumed") + private Integer quotaConsumed; + /** Constructor for in progress or complete PipelineRun. */ public PipelineRun( UUID jobId, @@ -83,7 +86,8 @@ public PipelineRun( Instant created, Instant updated, CommonPipelineRunStatusEnum status, - String description) { + String description, + Integer quotaConsumed) { this.jobId = jobId; this.userId = userId; this.pipelineId = pipelineId; @@ -97,6 +101,7 @@ public PipelineRun( this.updated = updated; this.status = status; this.description = description; + this.quotaConsumed = quotaConsumed; } /** Constructor for creating a new GCP pipeline run. Timestamps are auto-generated. */ diff --git a/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobMapKeys.java b/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobMapKeys.java index 7ff217c6..7ee108b1 100644 --- a/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobMapKeys.java +++ b/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobMapKeys.java @@ -19,6 +19,8 @@ public class JobMapKeys { "do_set_pipeline_run_status_failed_hook"; public static final String DO_INCREMENT_METRICS_FAILED_COUNTER_HOOK = "do_increment_metrics_failed_counter_hook"; + public static final String DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK = + "do_send_job_failure_notification_hook"; JobMapKeys() { throw new IllegalStateException("Attempted to instantiate utility class JobMapKeys"); diff --git a/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobService.java b/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobService.java index 76ca289a..36a0d628 100644 --- a/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobService.java +++ b/service/src/main/java/bio/terra/pipelines/dependencies/stairway/JobService.java @@ -13,6 +13,7 @@ import bio.terra.pipelines.common.utils.FlightBeanBag; import bio.terra.pipelines.common.utils.PipelinesEnum; import bio.terra.pipelines.common.utils.StairwayFailedMetricsCounterHook; +import bio.terra.pipelines.common.utils.StairwaySendFailedJobNotificationHook; import bio.terra.pipelines.common.utils.StairwaySetPipelineRunStatusHook; import bio.terra.pipelines.dependencies.stairway.exception.*; import bio.terra.pipelines.dependencies.stairway.model.EnumeratedJob; @@ -112,6 +113,8 @@ public void initialize() { .addHook(new StairwayLoggingHook()) .addHook(new MonitoringHook(openTelemetry)) .addHook(new StairwayFailedMetricsCounterHook()) + .addHook( + new StairwaySendFailedJobNotificationHook(flightBeanBag.getNotificationService())) .addHook(new StairwaySetPipelineRunStatusHook(flightBeanBag.getPipelineRunsService())) .exceptionSerializer(new StairwayExceptionSerializer(objectMapper))); } diff --git a/service/src/main/java/bio/terra/pipelines/notifications/BaseTeaspoonsJobNotification.java b/service/src/main/java/bio/terra/pipelines/notifications/BaseTeaspoonsJobNotification.java new file mode 100644 index 00000000..f91bba07 --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/notifications/BaseTeaspoonsJobNotification.java @@ -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; + } +} diff --git a/service/src/main/java/bio/terra/pipelines/notifications/NotificationService.java b/service/src/main/java/bio/terra/pipelines/notifications/NotificationService.java new file mode 100644 index 00000000..26a8279f --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/notifications/NotificationService.java @@ -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 = 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); + } + } +} diff --git a/service/src/main/java/bio/terra/pipelines/notifications/PubsubService.java b/service/src/main/java/bio/terra/pipelines/notifications/PubsubService.java new file mode 100644 index 00000000..c07e8512 --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/notifications/PubsubService.java @@ -0,0 +1,80 @@ +package bio.terra.pipelines.notifications; + +import com.google.api.core.ApiFuture; +import com.google.api.gax.rpc.ApiException; +import com.google.cloud.pubsub.v1.Publisher; +import com.google.protobuf.ByteString; +import com.google.pubsub.v1.PubsubMessage; +import com.google.pubsub.v1.TopicName; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +/** Service to interact with Pubsub. Used by NotificationService. */ +@Service +public class PubsubService { + private static final Logger logger = LoggerFactory.getLogger(PubsubService.class); + + protected static Publisher publisher; + + /** + * Initialize a publisher for a Google PubSub topic. Does nothing if the publisher already exists. + */ + protected static void initPublisher(TopicName topicName) throws IOException { + if (publisher != null) { + logger.info("Publisher already exists for Google PubSub topicName {}", topicName); + return; + } + try { + logger.info("Creating publisher for Google PubSub topicName {}", topicName); + // Create a publisher instance with default settings bound to the topic + publisher = Publisher.newBuilder(topicName).build(); + } catch (IOException e) { + logger.error("Error creating publisher for Google PubSub topicName {}", topicName); + throw e; + } + } + + /** Publish a message to a Google PubSub topic. */ + public void publishMessage(String projectId, String topicId, String message) throws IOException { + TopicName topicName = TopicName.of(projectId, topicId); + logger.info( + "Publishing message to Google PubSub projectId {}, topicId {}: {}", + projectId, + topicId, + message); + + initPublisher(topicName); + + ByteString data = ByteString.copyFromUtf8(message); + PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build(); + + // Once published, returns a server-assigned message id (unique within the topic) + ApiFuture future = publisher.publish(pubsubMessage); + + try { + // Wait on any pending publish requests with a 30-second timeout + String messageId = future.get(30, TimeUnit.SECONDS); + logger.info("Published message ID: {}", messageId); + } catch (Exception e) { + String errorMessage; + if (e instanceof ApiException apiException) { + // details on the API exception + errorMessage = + "Google API exception: status code %s, is retryable: %s" + .formatted(apiException.getStatusCode().getCode(), apiException.isRetryable()); + logger.error( + "Error publishing message to Google PubSub: {}; {}", errorMessage, e.getMessage()); + } else if (e instanceof InterruptedException) { + errorMessage = "Thread was interrupted"; + logger.error( + "Error publishing message to Google PubSub: {}; {}", errorMessage, e.getMessage()); + Thread.currentThread().interrupt(); + } else { + logger.error("Error publishing message to Google PubSub: {}", e.getMessage()); + } + } + } +} diff --git a/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotification.java b/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotification.java new file mode 100644 index 00000000..712d390f --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotification.java @@ -0,0 +1,33 @@ +package bio.terra.pipelines.notifications; + +import lombok.Getter; + +@Getter +@SuppressWarnings({"java:S107"}) // Disable "Methods should not have too many parameters" +public class TeaspoonsJobFailedNotification extends BaseTeaspoonsJobNotification { + private static final String NOTIFICATION_TYPE = "TeaspoonsJobFailedNotification"; + private static final String QUOTA_CONSUMED_BY_FAILED_JOB = "0"; + private final String errorMessage; + + public TeaspoonsJobFailedNotification( + String recipientUserId, + String pipelineDisplayName, + String jobId, + String errorMessage, + String timeSubmitted, + String timeCompleted, + String quotaRemaining, + String userDescription) { + super( + recipientUserId, + pipelineDisplayName, + jobId, + timeSubmitted, + timeCompleted, + quotaRemaining, + userDescription); + this.notificationType = NOTIFICATION_TYPE; + this.quotaConsumedByJob = QUOTA_CONSUMED_BY_FAILED_JOB; + this.errorMessage = errorMessage; + } +} diff --git a/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotification.java b/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotification.java new file mode 100644 index 00000000..da41ccba --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotification.java @@ -0,0 +1,30 @@ +package bio.terra.pipelines.notifications; + +import lombok.Getter; + +@Getter +@SuppressWarnings({"java:S107"}) // Disable "Methods should not have too many parameters" +public class TeaspoonsJobSucceededNotification extends BaseTeaspoonsJobNotification { + private static final String NOTIFICATION_TYPE = "TeaspoonsJobSucceededNotification"; + + public TeaspoonsJobSucceededNotification( + String recipientUserId, + String pipelineDisplayName, + String jobId, + String timeSubmitted, + String timeCompleted, + String quotaConsumedByJob, + String quotaRemaining, + String userDescription) { + super( + recipientUserId, + pipelineDisplayName, + jobId, + timeSubmitted, + timeCompleted, + quotaRemaining, + userDescription); + this.notificationType = NOTIFICATION_TYPE; + this.quotaConsumedByJob = quotaConsumedByJob; + } +} diff --git a/service/src/main/java/bio/terra/pipelines/service/PipelineRunsService.java b/service/src/main/java/bio/terra/pipelines/service/PipelineRunsService.java index 606a8baa..9073257a 100644 --- a/service/src/main/java/bio/terra/pipelines/service/PipelineRunsService.java +++ b/service/src/main/java/bio/terra/pipelines/service/PipelineRunsService.java @@ -163,6 +163,7 @@ public PipelineRun startPipelineRun(Pipeline pipeline, UUID jobId, String userId .addParameter(JobMapKeys.PIPELINE_ID, pipeline.getId()) .addParameter(JobMapKeys.DOMAIN_NAME, ingressConfiguration.getDomainName()) .addParameter(JobMapKeys.DO_SET_PIPELINE_RUN_STATUS_FAILED_HOOK, true) + .addParameter(JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK, true) .addParameter(JobMapKeys.DO_INCREMENT_METRICS_FAILED_COUNTER_HOOK, true) .addParameter( ImputationJobMapKeys.PIPELINE_INPUT_DEFINITIONS, @@ -287,7 +288,8 @@ public PipelineRun startPipelineRunInDb(UUID jobId, String userId) { } /** - * Mark a pipeline run as successful (status = SUCCEEDED) in our database. + * Mark a pipeline run as successful (status = SUCCEEDED) in our database and store the quota + * consumed by the job. * *

We expect this method to be called by the final step of a flight, at which point we assume * that the pipeline_run has completed successfully. Therefore, we do not do any checks on the @@ -296,12 +298,13 @@ public PipelineRun startPipelineRunInDb(UUID jobId, String userId) { */ @WriteTransaction public PipelineRun markPipelineRunSuccessAndWriteOutputs( - UUID jobId, String userId, Map outputs) { + UUID jobId, String userId, Map outputs, int quotaConsumed) { PipelineRun pipelineRun = getPipelineRun(jobId, userId); pipelineInputsOutputsService.savePipelineOutputs(pipelineRun.getId(), outputs); pipelineRun.setStatus(CommonPipelineRunStatusEnum.SUCCEEDED); + pipelineRun.setQuotaConsumed(quotaConsumed); return pipelineRunsRepository.save(pipelineRun); } diff --git a/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationAzureJobFlight.java b/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationAzureJobFlight.java index 321c3982..5727163b 100644 --- a/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationAzureJobFlight.java +++ b/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationAzureJobFlight.java @@ -37,6 +37,7 @@ public RunImputationAzureJobFlight(FlightMap inputParameters, Object beanBag) { JobMapKeys.PIPELINE_ID, JobMapKeys.DOMAIN_NAME, JobMapKeys.DO_SET_PIPELINE_RUN_STATUS_FAILED_HOOK, + JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK, JobMapKeys.DO_INCREMENT_METRICS_FAILED_COUNTER_HOOK, ImputationJobMapKeys.PIPELINE_INPUT_DEFINITIONS, ImputationJobMapKeys.PIPELINE_OUTPUT_DEFINITIONS, diff --git a/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpJobFlight.java b/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpJobFlight.java index ccd4db7d..bf03c925 100644 --- a/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpJobFlight.java +++ b/service/src/main/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpJobFlight.java @@ -9,6 +9,7 @@ import bio.terra.pipelines.stairway.steps.common.FetchQuotaConsumedFromDataTableStep; import bio.terra.pipelines.stairway.steps.common.PollQuotaConsumedSubmissionStatusStep; import bio.terra.pipelines.stairway.steps.common.QuotaConsumedValidationStep; +import bio.terra.pipelines.stairway.steps.common.SendJobSucceededNotificationStep; import bio.terra.pipelines.stairway.steps.common.SubmitQuotaConsumedSubmissionStep; import bio.terra.pipelines.stairway.steps.imputation.PrepareImputationInputsStep; import bio.terra.pipelines.stairway.steps.imputation.gcp.AddDataTableRowStep; @@ -55,6 +56,7 @@ public RunImputationGcpJobFlight(FlightMap inputParameters, Object beanBag) { JobMapKeys.PIPELINE_ID, JobMapKeys.DOMAIN_NAME, JobMapKeys.DO_SET_PIPELINE_RUN_STATUS_FAILED_HOOK, + JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK, JobMapKeys.DO_INCREMENT_METRICS_FAILED_COUNTER_HOOK, ImputationJobMapKeys.PIPELINE_INPUT_DEFINITIONS, ImputationJobMapKeys.PIPELINE_OUTPUT_DEFINITIONS, @@ -122,5 +124,8 @@ public RunImputationGcpJobFlight(FlightMap inputParameters, Object beanBag) { externalServiceRetryRule); addStep(new CompletePipelineRunStep(flightBeanBag.getPipelineRunsService()), dbRetryRule); + + addStep( + new SendJobSucceededNotificationStep(flightBeanBag.getNotificationService()), dbRetryRule); } } diff --git a/service/src/main/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStep.java b/service/src/main/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStep.java index 42c85730..87164de9 100644 --- a/service/src/main/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStep.java +++ b/service/src/main/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStep.java @@ -12,6 +12,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +/** + * Step to mark a pipeline run as a success and write the outputs and quota consumed to the database + * + *

This step expects the JobMapKeys.USER_ID in the input parameters and + * ImputationJobMapKeys.PIPELINE_RUN_OUTPUTS and ImputationJobMapKeys.EFFECTIVE_QUOTA_CONSUMED in + * the working map. + */ public class CompletePipelineRunStep implements Step { private final PipelineRunsService pipelineRunsService; private final Logger logger = LoggerFactory.getLogger(CompletePipelineRunStep.class); @@ -21,6 +28,8 @@ public CompletePipelineRunStep(PipelineRunsService pipelineRunsService) { } @Override + @SuppressWarnings("java:S2259") // suppress warning for possible NPE when unboxing quotaConsumed, + // since we do validate that quotaConsumed is not null in `validateRequiredEntries` public StepResult doStep(FlightContext flightContext) { // validate and extract parameters from input map var inputParameters = flightContext.getInputParameters(); @@ -31,11 +40,17 @@ public StepResult doStep(FlightContext flightContext) { // validate and extract parameters from working map var workingMap = flightContext.getWorkingMap(); - FlightUtils.validateRequiredEntries(workingMap, ImputationJobMapKeys.PIPELINE_RUN_OUTPUTS); + FlightUtils.validateRequiredEntries( + workingMap, + ImputationJobMapKeys.PIPELINE_RUN_OUTPUTS, + ImputationJobMapKeys.EFFECTIVE_QUOTA_CONSUMED); Map outputsMap = workingMap.get(ImputationJobMapKeys.PIPELINE_RUN_OUTPUTS, Map.class); + int quotaConsumed = + workingMap.get(ImputationJobMapKeys.EFFECTIVE_QUOTA_CONSUMED, Integer.class); - pipelineRunsService.markPipelineRunSuccessAndWriteOutputs(jobId, userId, outputsMap); + pipelineRunsService.markPipelineRunSuccessAndWriteOutputs( + jobId, userId, outputsMap, quotaConsumed); logger.info("Marked run {} as a success and wrote outputs to the db", jobId); diff --git a/service/src/main/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStep.java b/service/src/main/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStep.java new file mode 100644 index 00000000..9959197a --- /dev/null +++ b/service/src/main/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStep.java @@ -0,0 +1,51 @@ +package bio.terra.pipelines.stairway.steps.common; + +import bio.terra.pipelines.common.utils.FlightUtils; +import bio.terra.pipelines.dependencies.stairway.JobMapKeys; +import bio.terra.pipelines.notifications.NotificationService; +import bio.terra.stairway.FlightContext; +import bio.terra.stairway.Step; +import bio.terra.stairway.StepResult; +import java.util.UUID; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Step to send an email notification that a job has succeeded. This step cannot fail (we catch all + * exceptions). + * + *

This step expects JobMapKeys.USER_ID in the input parameters. + */ +public class SendJobSucceededNotificationStep implements Step { + private final NotificationService notificationService; + private final Logger logger = LoggerFactory.getLogger(SendJobSucceededNotificationStep.class); + + public SendJobSucceededNotificationStep(NotificationService notificationService) { + this.notificationService = notificationService; + } + + @Override + public StepResult doStep(FlightContext flightContext) { + // we place the entire logic of this step in a try-catch so that it cannot fail + try { + // validate and extract parameters from input map + var inputParameters = flightContext.getInputParameters(); + FlightUtils.validateRequiredEntries(inputParameters, JobMapKeys.USER_ID); + + UUID jobId = UUID.fromString(flightContext.getFlightId()); + String userId = inputParameters.get(JobMapKeys.USER_ID, String.class); + + // send email notification + notificationService.configureAndSendPipelineRunSucceededNotification(jobId, userId); + } catch (Exception e) { + logger.error("Failed to send email notification", e); + } + return StepResult.getStepResultSuccess(); + } + + @Override + public StepResult undoStep(FlightContext flightContext) { + // nothing to undo + return StepResult.getStepResultSuccess(); + } +} diff --git a/service/src/main/resources/application.yml b/service/src/main/resources/application.yml index 26f682fc..2bd58604 100644 --- a/service/src/main/resources/application.yml +++ b/service/src/main/resources/application.yml @@ -34,6 +34,9 @@ env: imputation: # the default currently points to the GCP dev workspace teaspoons-imputation-dev/teaspoons_imputation_dev_storage_workspace_20240726 storageWorkspaceStorageUrl: ${IMPUTATION_STORAGE_WORKSPACE_STORAGE_URL:gs://fc-secure-10efd4d7-392a-4e9e-89ea-d6629fbb06cc} + notifications: + projectId: ${NOTIFICATION_PROJECT_ID:broad-dsde-dev} + topicId: ${NOTIFICATION_TOPIC_ID:workbench-notifications-dev} # Below here is non-deployment-specific @@ -155,6 +158,10 @@ pipelines: quotaConsumedPollingIntervalSeconds: 60 quotaConsumedUseCallCaching: false + notifications: + projectId: ${env.pipelines.notifications.projectId} + topicId: ${env.pipelines.notifications.topicId} + terra.common: kubernetes: in-kubernetes: ${env.kubernetes.in-kubernetes} # whether to use a pubsub queue for Stairway; if false, use a local queue diff --git a/service/src/main/resources/db/changelog.xml b/service/src/main/resources/db/changelog.xml index 0d3316ec..0e0a32e7 100644 --- a/service/src/main/resources/db/changelog.xml +++ b/service/src/main/resources/db/changelog.xml @@ -8,6 +8,7 @@ + diff --git a/service/src/main/resources/db/changesets/20241212_add_quota_consumed_update_pipeline_display_name.yaml b/service/src/main/resources/db/changesets/20241212_add_quota_consumed_update_pipeline_display_name.yaml new file mode 100644 index 00000000..8b4ea01b --- /dev/null +++ b/service/src/main/resources/db/changesets/20241212_add_quota_consumed_update_pipeline_display_name.yaml @@ -0,0 +1,22 @@ +# add quota_consumed to pipeline_runs table and update display_name for array_imputation in pipelines table + +databaseChangeLog: + - changeSet: + id: add quota_consumed to pipeline_runs table and update display_name for array_imputation in pipelines table + author: mma + changes: + - addColumn: + columns: + - column: + name: quota_consumed + type: int + constraints: + nullable: true + tableName: pipeline_runs + - update: + tableName: pipelines + columns: + - column: + name: display_name + value: "All of Us/AnVIL Array Imputation" + where: name='array_imputation' AND version=0 diff --git a/service/src/test/java/bio/terra/pipelines/common/utils/FlightBeanBagTest.java b/service/src/test/java/bio/terra/pipelines/common/utils/FlightBeanBagTest.java index 8348d68e..49bb5bf9 100644 --- a/service/src/test/java/bio/terra/pipelines/common/utils/FlightBeanBagTest.java +++ b/service/src/test/java/bio/terra/pipelines/common/utils/FlightBeanBagTest.java @@ -11,12 +11,14 @@ import bio.terra.pipelines.dependencies.sam.SamService; import bio.terra.pipelines.dependencies.wds.WdsService; import bio.terra.pipelines.dependencies.workspacemanager.WorkspaceManagerService; +import bio.terra.pipelines.notifications.NotificationService; import bio.terra.pipelines.service.PipelineInputsOutputsService; import bio.terra.pipelines.service.PipelineRunsService; import bio.terra.pipelines.service.PipelinesService; import bio.terra.pipelines.service.QuotasService; import bio.terra.pipelines.testutils.BaseEmbeddedDbTest; import org.junit.jupiter.api.Test; +import org.mockito.Mock; import org.springframework.beans.factory.annotation.Autowired; class FlightBeanBagTest extends BaseEmbeddedDbTest { @@ -31,6 +33,11 @@ class FlightBeanBagTest extends BaseEmbeddedDbTest { @Autowired private WorkspaceManagerService workspaceManagerService; @Autowired private RawlsService rawlsService; @Autowired private QuotasService quotasService; + + @Mock + private NotificationService + notificationService; // mock because at startup tries to auto-create a topic + @Autowired private ImputationConfiguration imputationConfiguration; @Autowired private CbasConfiguration cbasConfiguration; @Autowired private WdlPipelineConfiguration wdlPipelineConfiguration; @@ -48,6 +55,7 @@ void testFlightBeanBag() { cbasService, rawlsService, quotasService, + notificationService, workspaceManagerService, imputationConfiguration, cbasConfiguration, @@ -62,6 +70,7 @@ void testFlightBeanBag() { assertEquals(workspaceManagerService, flightBeanBag.getWorkspaceManagerService()); assertEquals(rawlsService, flightBeanBag.getRawlsService()); assertEquals(quotasService, flightBeanBag.getQuotasService()); + assertEquals(notificationService, flightBeanBag.getNotificationService()); assertEquals(imputationConfiguration, flightBeanBag.getImputationConfiguration()); assertEquals(cbasConfiguration, flightBeanBag.getCbasConfiguration()); assertEquals(wdlPipelineConfiguration, flightBeanBag.getWdlPipelineConfiguration()); diff --git a/service/src/test/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHookTest.java b/service/src/test/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHookTest.java new file mode 100644 index 00000000..4302acb0 --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/common/utils/StairwaySendFailedJobNotificationHookTest.java @@ -0,0 +1,120 @@ +package bio.terra.pipelines.common.utils; + +import static org.junit.jupiter.params.provider.Arguments.arguments; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +import bio.terra.pipelines.dependencies.stairway.JobMapKeys; +import bio.terra.pipelines.notifications.NotificationService; +import bio.terra.pipelines.testutils.BaseEmbeddedDbTest; +import bio.terra.pipelines.testutils.StairwayTestUtils; +import bio.terra.pipelines.testutils.TestFlightContext; +import bio.terra.pipelines.testutils.TestUtils; +import bio.terra.stairway.FlightContext; +import bio.terra.stairway.FlightStatus; +import java.util.UUID; +import java.util.stream.Stream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.mock.mockito.MockBean; + +class StairwaySendFailedJobNotificationHookTest extends BaseEmbeddedDbTest { + @Autowired @InjectMocks + StairwaySendFailedJobNotificationHook stairwaySendFailedJobNotificationHook; + + @MockBean NotificationService notificationService; + + @Mock private FlightContext flightContext; + + private final UUID testJobId = TestUtils.TEST_NEW_UUID; + + @BeforeEach + void setup() { + doNothing() + .when(notificationService) + .configureAndSendPipelineRunFailedNotification( + testJobId, TestUtils.TEST_USER_ID_1, flightContext); + stairwaySendFailedJobNotificationHook = + new StairwaySendFailedJobNotificationHook(notificationService); + } + + private static Stream flightContexts() { + + return Stream.of( + // arguments: whether to include hook key in input Params, + // hook key value, flight status at endFlight time, whether failed job notification should + // be sent + + arguments( + true, + true, + FlightStatus.SUCCESS, + false), // flight was successful, so the pipelineRun status should not be set to FAILED + arguments( + true, + true, + FlightStatus.ERROR, + true), // flight failed, so the pipelineRun status should be set to FAILED + arguments( + true, + true, + FlightStatus.FATAL, + true), // flight failed dismally, so the pipelineRun status should be set to FAILED + arguments( + true, + true, + FlightStatus.QUEUED, + true), // flight in an unexpected status for end of flight, so the pipelineRun status + // should be set to FAILED + arguments( + false, + false, // doesn't matter + FlightStatus.ERROR, + false), // flight failed, but the hook key was not included in the input parameters + arguments( + true, + false, + FlightStatus.ERROR, + false)); // flight failed, but the hook key value is false + } + + @ParameterizedTest + @MethodSource("flightContexts") + void endFlight( + boolean includeHookKey, + boolean hookKeyValue, + FlightStatus endFlightStatus, + boolean shouldSendFailedJobNotification) + throws InterruptedException { + var context = new TestFlightContext(); + + if (includeHookKey) { + // this includes setting the DO_SET_PIPELINE_RUN_STATUS_FAILED_HOOK key to true + StairwayTestUtils.constructCreateJobInputs(context.getInputParameters()); + context + .getInputParameters() + .put(JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK, hookKeyValue); + } + + stairwaySendFailedJobNotificationHook.startFlight(context); + + // set the end flight status + context.flightStatus(endFlightStatus); + + stairwaySendFailedJobNotificationHook.endFlight(context); + + if (shouldSendFailedJobNotification) { + verify(notificationService) + .configureAndSendPipelineRunFailedNotification( + testJobId, TestUtils.TEST_USER_ID_1, context); + } else { + verifyNoInteractions(notificationService); + } + } +} diff --git a/service/src/test/java/bio/terra/pipelines/controller/PipelineRunsApiControllerTest.java b/service/src/test/java/bio/terra/pipelines/controller/PipelineRunsApiControllerTest.java index d952eed6..b3c8f87a 100644 --- a/service/src/test/java/bio/terra/pipelines/controller/PipelineRunsApiControllerTest.java +++ b/service/src/test/java/bio/terra/pipelines/controller/PipelineRunsApiControllerTest.java @@ -87,6 +87,8 @@ class PipelineRunsApiControllerTest { private final Instant updatedTime = Instant.now(); private final Map testOutputs = TestUtils.TEST_PIPELINE_OUTPUTS; + private final Integer testQuotaConsumed = 10; + @BeforeEach void beforeEach() { when(samConfiguration.baseUri()).thenReturn("baseSamUri"); @@ -441,7 +443,9 @@ void startImputationRunStairwayError() throws Exception { void getPipelineRunResultDoneSuccess() throws Exception { String pipelineName = PipelinesEnum.ARRAY_IMPUTATION.getValue(); String jobIdString = newJobId.toString(); - PipelineRun pipelineRun = getPipelineRunWithStatus(CommonPipelineRunStatusEnum.SUCCEEDED); + PipelineRun pipelineRun = + getPipelineRunWithStatusAndQuotaConsumed( + CommonPipelineRunStatusEnum.SUCCEEDED, testQuotaConsumed); ApiPipelineRunOutputs apiPipelineRunOutputs = new ApiPipelineRunOutputs(); apiPipelineRunOutputs.putAll(testOutputs); @@ -483,7 +487,8 @@ void getPipelineRunResultDoneFailed() throws Exception { String jobIdString = newJobId.toString(); String errorMessage = "test exception message"; Integer statusCode = 500; - PipelineRun pipelineRun = getPipelineRunWithStatus(CommonPipelineRunStatusEnum.FAILED); + PipelineRun pipelineRun = + getPipelineRunWithStatusAndQuotaConsumed(CommonPipelineRunStatusEnum.FAILED, null); ApiErrorReport errorReport = new ApiErrorReport().message(errorMessage).statusCode(statusCode); @@ -612,8 +617,10 @@ void getAllPipelineRunsWithNoPageToken() throws Exception { PipelineRun pipelineRunPreparing = getPipelineRunPreparing(preparingDescription); PipelineRun pipelineRunPreparingNoDescription = getPipelineRunPreparing(null); PipelineRun pipelineRunSucceeded = - getPipelineRunWithStatus(CommonPipelineRunStatusEnum.SUCCEEDED); - PipelineRun pipelineRunFailed = getPipelineRunWithStatus(CommonPipelineRunStatusEnum.FAILED); + getPipelineRunWithStatusAndQuotaConsumed( + CommonPipelineRunStatusEnum.SUCCEEDED, testQuotaConsumed); + PipelineRun pipelineRunFailed = + getPipelineRunWithStatusAndQuotaConsumed(CommonPipelineRunStatusEnum.FAILED, null); PageResponse> pageResponse = new PageResponse<>( List.of( @@ -757,7 +764,8 @@ private PipelineRun getPipelineRunPreparing(String description) { createdTime, updatedTime, CommonPipelineRunStatusEnum.PREPARING, - description); + description, + null); } /** helper method to create a PipelineRun object for a running job */ @@ -775,11 +783,16 @@ private PipelineRun getPipelineRunRunning() { createdTime, updatedTime, CommonPipelineRunStatusEnum.RUNNING, - TestUtils.TEST_PIPELINE_DESCRIPTION_1); + TestUtils.TEST_PIPELINE_DESCRIPTION_1, + null); } - /** helper method to create a PipelineRun object for a completed job. */ - private PipelineRun getPipelineRunWithStatus(CommonPipelineRunStatusEnum status) { + /** + * helper method to create a PipelineRun object for a completed job, specifying the status and + * quotaConsumed. + */ + private PipelineRun getPipelineRunWithStatusAndQuotaConsumed( + CommonPipelineRunStatusEnum status, Integer quotaConsumed) { return new PipelineRun( newJobId, testUser.getSubjectId(), @@ -793,6 +806,7 @@ private PipelineRun getPipelineRunWithStatus(CommonPipelineRunStatusEnum status) createdTime, updatedTime, status, - TestUtils.TEST_PIPELINE_DESCRIPTION_1); + TestUtils.TEST_PIPELINE_DESCRIPTION_1, + quotaConsumed); } } diff --git a/service/src/test/java/bio/terra/pipelines/notifications/NotificationServiceTest.java b/service/src/test/java/bio/terra/pipelines/notifications/NotificationServiceTest.java new file mode 100644 index 00000000..5bd3d6ad --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/notifications/NotificationServiceTest.java @@ -0,0 +1,308 @@ +package bio.terra.pipelines.notifications; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import bio.terra.pipelines.app.configuration.internal.NotificationConfiguration; +import bio.terra.pipelines.common.utils.CommonPipelineRunStatusEnum; +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.db.repositories.PipelineRunsRepository; +import bio.terra.pipelines.dependencies.rawls.RawlsServiceApiException; +import bio.terra.pipelines.service.PipelineRunsService; +import bio.terra.pipelines.service.PipelinesService; +import bio.terra.pipelines.service.QuotasService; +import bio.terra.pipelines.testutils.BaseEmbeddedDbTest; +import bio.terra.pipelines.testutils.TestUtils; +import bio.terra.stairway.FlightContext; +import bio.terra.stairway.StepResult; +import bio.terra.stairway.StepStatus; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.UUID; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.mock.mockito.MockBean; + +class NotificationServiceTest extends BaseEmbeddedDbTest { + @InjectMocks @Autowired NotificationService notificationService; + @Autowired PipelineRunsService pipelineRunsService; + @Autowired PipelineRunsRepository pipelineRunsRepository; + @Autowired PipelinesService pipelinesService; + @Autowired QuotasService quotasService; + @Autowired NotificationConfiguration notificationConfiguration; + @Autowired ObjectMapper objectMapper; + @MockBean PubsubService pubsubService; + @Mock private FlightContext flightContext; + + UUID testJobId = TestUtils.TEST_NEW_UUID; + String testUserId = TestUtils.TEST_USER_ID_1; + Integer testQuotaConsumedByJob = 1000; + String testUserDescription = TestUtils.TEST_USER_PROVIDED_DESCRIPTION; + String testErrorMessage = "test error message"; + + @Test + void formatDateTime() { + Instant instant = Instant.parse("2021-08-25T12:34:56.789Z"); + String formattedDateTime = notificationService.formatInstantToReadableString(instant); + assertEquals("Wed, 25 Aug 2021 12:34:56 GMT", formattedDateTime); + } + + @Test + void configureAndSendPipelineRunSucceededNotification() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + PipelineRun writtenPipelineRun = + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.SUCCEEDED); + + // initialize and set user quota + UserQuota userQuota = + quotasService.getOrCreateQuotaForUserAndPipeline(testUserId, pipeline.getName()); + UserQuota updatedUserQuota = + quotasService.updateQuotaConsumed(userQuota, testQuotaConsumedByJob); + int expectedQuotaRemaining = updatedUserQuota.getQuota() - updatedUserQuota.getQuotaConsumed(); + + String stringifiedJobSucceededNotification = + objectMapper.writeValueAsString( + new TeaspoonsJobSucceededNotification( + testUserId, + pipeline.getDisplayName(), + testJobId.toString(), + notificationService.formatInstantToReadableString(writtenPipelineRun.getCreated()), + notificationService.formatInstantToReadableString(writtenPipelineRun.getUpdated()), + testQuotaConsumedByJob.toString(), + String.valueOf(expectedQuotaRemaining), + testUserDescription)); + // success is a void method + doNothing() + .when(pubsubService) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobSucceededNotification); + + notificationService.configureAndSendPipelineRunSucceededNotification(testJobId, testUserId); + + // verify that the pubsub method was called + verify(pubsubService, times(1)) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobSucceededNotification); + } + + @Test + void configureAndSendPipelineRunSucceededNotificationIOException() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.SUCCEEDED); + + doThrow(new IOException()).when(pubsubService).publishMessage(any(), any(), any()); + + // exception should be caught + assertDoesNotThrow( + () -> + notificationService.configureAndSendPipelineRunSucceededNotification( + testJobId, testUserId)); + } + + @Test + void configureAndSendPipelineRunFailedNotification() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + PipelineRun writtenPipelineRun = + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.FAILED); + + // initialize and set a custom user quota value. this is not quota consumed by the job. + int customUserQuota = 2000; + UserQuota userQuota = + quotasService.getOrCreateQuotaForUserAndPipeline(testUserId, pipeline.getName()); + UserQuota updatedUserQuota = quotasService.updateQuotaConsumed(userQuota, customUserQuota); + int expectedQuotaRemaining = updatedUserQuota.getQuota() - updatedUserQuota.getQuotaConsumed(); + + when(flightContext.getFlightId()).thenReturn(testJobId.toString()); + RawlsServiceApiException rawlsServiceApiException = + new RawlsServiceApiException(testErrorMessage); + StepResult stepResultFailedWithException = + new StepResult(StepStatus.STEP_RESULT_FAILURE_FATAL, rawlsServiceApiException); + when(flightContext.getResult()).thenReturn(stepResultFailedWithException); + + String stringifiedJobFailedNotification = + objectMapper.writeValueAsString( + new TeaspoonsJobFailedNotification( + testUserId, + pipeline.getDisplayName(), + testJobId.toString(), + testErrorMessage, + notificationService.formatInstantToReadableString(writtenPipelineRun.getCreated()), + notificationService.formatInstantToReadableString(writtenPipelineRun.getUpdated()), + String.valueOf(expectedQuotaRemaining), + testUserDescription)); + // success is a void method + doNothing() + .when(pubsubService) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + + notificationService.configureAndSendPipelineRunFailedNotification( + testJobId, testUserId, flightContext); + + // verify that the pubsub method was called + verify(pubsubService, times(1)) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + } + + @Test + void configureAndSendPipelineRunFailedNotificationNoUserQuota() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + PipelineRun writtenPipelineRun = + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.FAILED); + + // don't initialize user in user_quota table + int expectedQuotaRemaining = + quotasService.getPipelineQuota(pipeline.getName()).getDefaultQuota(); + + when(flightContext.getFlightId()).thenReturn(testJobId.toString()); + RawlsServiceApiException rawlsServiceApiException = + new RawlsServiceApiException(testErrorMessage); + StepResult stepResultFailedWithException = + new StepResult(StepStatus.STEP_RESULT_FAILURE_FATAL, rawlsServiceApiException); + when(flightContext.getResult()).thenReturn(stepResultFailedWithException); + + String stringifiedJobFailedNotification = + objectMapper.writeValueAsString( + new TeaspoonsJobFailedNotification( + testUserId, + pipeline.getDisplayName(), + testJobId.toString(), + testErrorMessage, + notificationService.formatInstantToReadableString(writtenPipelineRun.getCreated()), + notificationService.formatInstantToReadableString(writtenPipelineRun.getUpdated()), + String.valueOf(expectedQuotaRemaining), + testUserDescription)); + // success is a void method + doNothing() + .when(pubsubService) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + + notificationService.configureAndSendPipelineRunFailedNotification( + testJobId, testUserId, flightContext); + + // verify that the pubsub method was called + verify(pubsubService, times(1)) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + } + + @Test + void configureAndSendPipelineRunFailedNotificationWithoutException() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + PipelineRun writtenPipelineRun = + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.FAILED); + + // don't initialize user in user_quota table + int expectedQuotaRemaining = + quotasService.getPipelineQuota(pipeline.getName()).getDefaultQuota(); + + when(flightContext.getFlightId()).thenReturn(testJobId.toString()); + StepResult stepResultFailedWithoutException = + new StepResult(StepStatus.STEP_RESULT_FAILURE_FATAL); + when(flightContext.getResult()).thenReturn(stepResultFailedWithoutException); + + String stringifiedJobFailedNotification = + objectMapper.writeValueAsString( + new TeaspoonsJobFailedNotification( + testUserId, + pipeline.getDisplayName(), + testJobId.toString(), + "Unknown error", + notificationService.formatInstantToReadableString(writtenPipelineRun.getCreated()), + notificationService.formatInstantToReadableString(writtenPipelineRun.getUpdated()), + String.valueOf(expectedQuotaRemaining), + testUserDescription)); + // success is a void method + doNothing() + .when(pubsubService) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + + notificationService.configureAndSendPipelineRunFailedNotification( + testJobId, testUserId, flightContext); + + // verify that the pubsub method was called + verify(pubsubService, times(1)) + .publishMessage( + notificationConfiguration.projectId(), + notificationConfiguration.topicId(), + stringifiedJobFailedNotification); + } + + @Test + void configureAndSendPipelineRunFailedNotificationIOException() throws IOException { + Pipeline pipeline = pipelinesService.getPipelineById(1L); + createCompletedPipelineRunInDb(pipeline, CommonPipelineRunStatusEnum.FAILED); + + when(flightContext.getFlightId()).thenReturn(testJobId.toString()); + RawlsServiceApiException rawlsServiceApiException = + new RawlsServiceApiException(testErrorMessage); + StepResult stepResultFailedWithException = + new StepResult(StepStatus.STEP_RESULT_FAILURE_FATAL, rawlsServiceApiException); + when(flightContext.getResult()).thenReturn(stepResultFailedWithException); + + doThrow(new IOException()).when(pubsubService).publishMessage(any(), any(), any()); + + // exception should be caught + assertDoesNotThrow( + () -> + notificationService.configureAndSendPipelineRunFailedNotification( + testJobId, testUserId, flightContext)); + } + + /** + * Helper method for tests to create a completed pipeline run in the database. + * + * @param pipeline the pipeline to create the run for + * @param statusEnum the status of the pipeline run + * @return the completed pipeline run + */ + private PipelineRun createCompletedPipelineRunInDb( + Pipeline pipeline, CommonPipelineRunStatusEnum statusEnum) { + PipelineRun completedPipelineRun = + new PipelineRun( + testJobId, + testUserId, + pipeline.getId(), + pipeline.getWdlMethodVersion(), + pipeline.getWorkspaceId(), + pipeline.getWorkspaceBillingProject(), + pipeline.getWorkspaceName(), + pipeline.getWorkspaceStorageContainerName(), + pipeline.getWorkspaceGoogleProject(), + null, // timestamps auto generated by db + null, + statusEnum, + testUserDescription, + testQuotaConsumedByJob); + + return pipelineRunsRepository.save(completedPipelineRun); + } +} diff --git a/service/src/test/java/bio/terra/pipelines/notifications/PubsubServiceTest.java b/service/src/test/java/bio/terra/pipelines/notifications/PubsubServiceTest.java new file mode 100644 index 00000000..42e05eb7 --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/notifications/PubsubServiceTest.java @@ -0,0 +1,44 @@ +package bio.terra.pipelines.notifications; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import bio.terra.pipelines.testutils.BaseEmbeddedDbTest; +import com.google.cloud.pubsub.v1.Publisher; +import com.google.pubsub.v1.TopicName; +import java.io.IOException; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.springframework.beans.factory.annotation.Autowired; + +class PubsubServiceTest extends BaseEmbeddedDbTest { + @Autowired PubsubService pubsubService; + + @Test + void initPublisher() throws IOException { + TopicName topicName = TopicName.of("projectId", "topicId"); + try (MockedStatic mockedStaticPublisher = mockStatic(Publisher.class)) { + Publisher.Builder mockBuilder = mock(Publisher.Builder.class); + mockedStaticPublisher.when(() -> Publisher.newBuilder(topicName)).thenReturn(mockBuilder); + Publisher mockPublisher = mock(Publisher.class); + when(mockBuilder.build()).thenReturn(mockPublisher); + + // to start, publisher should be null + assertNull(PubsubService.publisher); + + // calling init once should set the publisher + PubsubService.initPublisher(topicName); + assertEquals(mockPublisher, PubsubService.publisher); + + // calling init again should not call build again + PubsubService.initPublisher(topicName); + assertEquals(mockPublisher, PubsubService.publisher); + verify(mockBuilder, times(1)).build(); + } + } +} diff --git a/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotificationTest.java b/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotificationTest.java new file mode 100644 index 00000000..836b8f92 --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobFailedNotificationTest.java @@ -0,0 +1,42 @@ +package bio.terra.pipelines.notifications; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import bio.terra.pipelines.testutils.BaseTest; +import org.junit.jupiter.api.Test; + +class TeaspoonsJobFailedNotificationTest extends BaseTest { + String recipientUserId = "test-recipient-user-id"; + String pipelineDisplayName = "test-pipeline-display-name"; + String jobId = "test-job-id"; + String errorMessage = "test-error-message"; + String timeSubmitted = "test-time-submitted"; + String timeCompleted = "test-time-completed"; + String quotaRemaining = "test-quota-remaining"; + String userDescription = "test-user-description"; + + @Test + void teaspoonsJobFailedNotificationFullConstructor() { + + TeaspoonsJobFailedNotification notification = + new TeaspoonsJobFailedNotification( + recipientUserId, + pipelineDisplayName, + jobId, + errorMessage, + timeSubmitted, + timeCompleted, + quotaRemaining, + userDescription); + assertEquals("TeaspoonsJobFailedNotification", notification.getNotificationType()); + assertEquals(recipientUserId, notification.getRecipientUserId()); + assertEquals(pipelineDisplayName, notification.getPipelineDisplayName()); + assertEquals(jobId, notification.getJobId()); + assertEquals(errorMessage, notification.getErrorMessage()); + assertEquals(timeSubmitted, notification.getTimeSubmitted()); + assertEquals(timeCompleted, notification.getTimeCompleted()); + assertEquals("0", notification.getQuotaConsumedByJob()); + assertEquals(quotaRemaining, notification.getQuotaRemaining()); + assertEquals(userDescription, notification.getUserDescription()); + } +} diff --git a/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotificationTest.java b/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotificationTest.java new file mode 100644 index 00000000..fd956614 --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/notifications/TeaspoonsJobSucceededNotificationTest.java @@ -0,0 +1,41 @@ +package bio.terra.pipelines.notifications; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import bio.terra.pipelines.testutils.BaseTest; +import org.junit.jupiter.api.Test; + +class TeaspoonsJobSucceededNotificationTest extends BaseTest { + + String recipientUserId = "test-recipient-user-id"; + String pipelineDisplayName = "test-pipeline-display-name"; + String jobId = "test-job-id"; + String timeSubmitted = "test-time-submitted"; + String timeCompleted = "test-time-completed"; + String quotaConsumedByJob = "test-quota-consumed-by-job"; + String quotaRemaining = "test-quota-remaining"; + String userDescription = "test-user-description"; + + @Test + void teaspoonsJobSucceededNotificationFullConstructor() { + TeaspoonsJobSucceededNotification notification = + new TeaspoonsJobSucceededNotification( + recipientUserId, + pipelineDisplayName, + jobId, + timeSubmitted, + timeCompleted, + quotaConsumedByJob, + quotaRemaining, + userDescription); + assertEquals("TeaspoonsJobSucceededNotification", notification.getNotificationType()); + assertEquals(recipientUserId, notification.getRecipientUserId()); + assertEquals(pipelineDisplayName, notification.getPipelineDisplayName()); + assertEquals(jobId, notification.getJobId()); + assertEquals(timeSubmitted, notification.getTimeSubmitted()); + assertEquals(timeCompleted, notification.getTimeCompleted()); + assertEquals(quotaConsumedByJob, notification.getQuotaConsumedByJob()); + assertEquals(quotaRemaining, notification.getQuotaRemaining()); + assertEquals(userDescription, notification.getUserDescription()); + } +} diff --git a/service/src/test/java/bio/terra/pipelines/service/PipelineRunsServiceTest.java b/service/src/test/java/bio/terra/pipelines/service/PipelineRunsServiceTest.java index 648eaa66..ffd60213 100644 --- a/service/src/test/java/bio/terra/pipelines/service/PipelineRunsServiceTest.java +++ b/service/src/test/java/bio/terra/pipelines/service/PipelineRunsServiceTest.java @@ -71,6 +71,7 @@ class PipelineRunsServiceTest extends BaseEmbeddedDbTest { private final String testControlWorkspaceGoogleProject = TestUtils.CONTROL_WORKSPACE_GOOGLE_PROJECT; private final String testUserDescription = TestUtils.TEST_USER_PROVIDED_DESCRIPTION; + private final Integer testQuotaConsumed = 10; private SimpleMeterRegistry meterRegistry; @@ -538,9 +539,10 @@ void markPipelineRunSuccess() { PipelineRun updatedPipelineRun = pipelineRunsService.markPipelineRunSuccessAndWriteOutputs( - testJobId, testUserId, TestUtils.TEST_PIPELINE_OUTPUTS); + testJobId, testUserId, TestUtils.TEST_PIPELINE_OUTPUTS, testQuotaConsumed); assertTrue(updatedPipelineRun.getStatus().isSuccess()); assertEquals(CommonPipelineRunStatusEnum.SUCCEEDED, updatedPipelineRun.getStatus()); + assertEquals(testQuotaConsumed, updatedPipelineRun.getQuotaConsumed()); PipelineOutput pipelineOutput = pipelineOutputsRepository.findPipelineOutputsByJobId(pipelineRun.getId()); diff --git a/service/src/test/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpFlightTest.java b/service/src/test/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpFlightTest.java index 75b13e72..bc96913c 100644 --- a/service/src/test/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpFlightTest.java +++ b/service/src/test/java/bio/terra/pipelines/stairway/flights/imputation/RunImputationGcpFlightTest.java @@ -39,7 +39,8 @@ class RunImputationGcpFlightTest extends BaseEmbeddedDbTest { "SubmitCromwellSubmissionStep", "PollCromwellSubmissionStatusStep", "CompletePipelineRunStep", - "FetchOutputsFromDataTableStep"); + "FetchOutputsFromDataTableStep", + "SendJobSucceededNotificationStep"); @Autowired FlightBeanBag flightBeanBag; private SimpleMeterRegistry meterRegistry; diff --git a/service/src/test/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStepTest.java b/service/src/test/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStepTest.java index 41fa464a..1ea43727 100644 --- a/service/src/test/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStepTest.java +++ b/service/src/test/java/bio/terra/pipelines/stairway/steps/common/CompletePipelineRunStepTest.java @@ -34,6 +34,7 @@ class CompletePipelineRunStepTest extends BaseEmbeddedDbTest { @Mock private FlightContext flightContext; private final UUID testJobId = TestUtils.TEST_NEW_UUID; + private final Integer effectiveQuotaConsumed = 500; @BeforeEach void setup() { @@ -41,6 +42,7 @@ void setup() { var workingMap = new FlightMap(); workingMap.put(ImputationJobMapKeys.PIPELINE_RUN_OUTPUTS, TestUtils.TEST_PIPELINE_OUTPUTS); + workingMap.put(ImputationJobMapKeys.EFFECTIVE_QUOTA_CONSUMED, effectiveQuotaConsumed); when(flightContext.getInputParameters()).thenReturn(inputParameters); when(flightContext.getWorkingMap()).thenReturn(workingMap); @@ -68,7 +70,8 @@ void doStepSuccess() throws JsonProcessingException { null, null, CommonPipelineRunStatusEnum.SUCCEEDED, - TestUtils.TEST_PIPELINE_DESCRIPTION_1)); + TestUtils.TEST_PIPELINE_DESCRIPTION_1, + null)); // do the step var writeJobStep = new CompletePipelineRunStep(pipelineRunsService); @@ -79,12 +82,13 @@ void doStepSuccess() throws JsonProcessingException { assertEquals(StepStatus.STEP_RESULT_SUCCESS, result.getStepStatus()); - // make sure the run was updated with isSuccess + // make sure the run was updated with isSuccess and quotaConsumed PipelineRun writtenJob = pipelineRunsRepository .findByJobIdAndUserId(testJobId, inputParams.get(JobMapKeys.USER_ID, String.class)) .orElseThrow(); assertEquals(CommonPipelineRunStatusEnum.SUCCEEDED, writtenJob.getStatus()); + assertEquals(effectiveQuotaConsumed, writtenJob.getQuotaConsumed()); assertTrue(writtenJob.getStatus().isSuccess()); // make sure outputs were written to db diff --git a/service/src/test/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStepTest.java b/service/src/test/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStepTest.java new file mode 100644 index 00000000..f34bdfb4 --- /dev/null +++ b/service/src/test/java/bio/terra/pipelines/stairway/steps/common/SendJobSucceededNotificationStepTest.java @@ -0,0 +1,82 @@ +package bio.terra.pipelines.stairway.steps.common; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import bio.terra.pipelines.dependencies.stairway.JobMapKeys; +import bio.terra.pipelines.notifications.NotificationService; +import bio.terra.pipelines.testutils.BaseEmbeddedDbTest; +import bio.terra.pipelines.testutils.TestUtils; +import bio.terra.stairway.FlightContext; +import bio.terra.stairway.FlightMap; +import bio.terra.stairway.StepStatus; +import java.util.UUID; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; + +class SendJobSucceededNotificationStepTest extends BaseEmbeddedDbTest { + @Mock private NotificationService notificationService; + @Mock private FlightContext flightContext; + + private final UUID testJobId = TestUtils.TEST_NEW_UUID; + private final String testUserId = TestUtils.TEST_USER_ID_1; + + @BeforeEach + void setup() { + var inputParameters = new FlightMap(); + var workingMap = new FlightMap(); + + inputParameters.put(JobMapKeys.USER_ID, testUserId); + + when(flightContext.getFlightId()).thenReturn(testJobId.toString()); + when(flightContext.getInputParameters()).thenReturn(inputParameters); + when(flightContext.getWorkingMap()).thenReturn(workingMap); + } + + @Test + void doStepSuccess() { + // assume success + doNothing() + .when(notificationService) + .configureAndSendPipelineRunSucceededNotification(testJobId, testUserId); + + var sendJobSucceededNotificationStep = + new SendJobSucceededNotificationStep(notificationService); + var result = sendJobSucceededNotificationStep.doStep(flightContext); + + assertEquals(StepStatus.STEP_RESULT_SUCCESS, result.getStepStatus()); + verify(notificationService, times(1)) + .configureAndSendPipelineRunSucceededNotification(testJobId, testUserId); + } + + @Test + void doStepFailureStillSuccess() { + // throw exception + doThrow(new RuntimeException()) + .when(notificationService) + .configureAndSendPipelineRunSucceededNotification(testJobId, testUserId); + + var sendJobSucceededNotificationStep = + new SendJobSucceededNotificationStep(notificationService); + var result = sendJobSucceededNotificationStep.doStep(flightContext); + + // step should catch exception and just log the error rather than throwing + assertEquals(StepStatus.STEP_RESULT_SUCCESS, result.getStepStatus()); + verify(notificationService, times(1)) + .configureAndSendPipelineRunSucceededNotification(testJobId, testUserId); + } + + @Test + void undoStepSuccess() { + var sendJobSucceededNotificationStep = + new SendJobSucceededNotificationStep(notificationService); + var result = sendJobSucceededNotificationStep.undoStep(flightContext); + + assertEquals(StepStatus.STEP_RESULT_SUCCESS, result.getStepStatus()); + } +} diff --git a/service/src/test/java/bio/terra/pipelines/testutils/StairwayTestUtils.java b/service/src/test/java/bio/terra/pipelines/testutils/StairwayTestUtils.java index 6820f361..95543ebf 100644 --- a/service/src/test/java/bio/terra/pipelines/testutils/StairwayTestUtils.java +++ b/service/src/test/java/bio/terra/pipelines/testutils/StairwayTestUtils.java @@ -189,6 +189,7 @@ public static FlightMap constructCreateJobInputs( inputParameters.put(JobMapKeys.DOMAIN_NAME, domainName); inputParameters.put(JobMapKeys.PIPELINE_ID, pipelineId); inputParameters.put(JobMapKeys.DO_INCREMENT_METRICS_FAILED_COUNTER_HOOK, true); + inputParameters.put(JobMapKeys.DO_SEND_JOB_FAILURE_NOTIFICATION_HOOK, true); inputParameters.put(JobMapKeys.DO_SET_PIPELINE_RUN_STATUS_FAILED_HOOK, true); inputParameters.put(ImputationJobMapKeys.PIPELINE_INPUT_DEFINITIONS, pipelineInputDefinitions); inputParameters.put(