diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/AwsConfig.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/AwsConfig.java index a540fb39..05a91cb4 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/AwsConfig.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/AwsConfig.java @@ -1,6 +1,8 @@ package au.org.aodn.ogcapi.server.core.configuration; import au.org.aodn.ogcapi.server.processes.RestServices; +import au.org.aodn.ogcapi.server.processes.BatchJobProperties; +import au.org.aodn.ogcapi.server.processes.DownloadJobStatusAggregator; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; @@ -15,12 +17,6 @@ public class AwsConfig { @Value("${aws.region}") private String awsRegion; - @Value("${aws.batch.job.definition}") - private String batchJobDefinition; - - @Value("${aws.batch.job.queue}") - private String batchJobQueue; - @Bean public BatchClient batchClient() { return BatchClient @@ -31,7 +27,15 @@ public BatchClient batchClient() { } @Bean - public RestServices awsBatchService(BatchClient batchClient, ObjectMapper objectMapper) { - return new RestServices(batchClient, objectMapper, batchJobDefinition, batchJobQueue); + public RestServices awsBatchService( + BatchClient batchClient, + ObjectMapper objectMapper, + BatchJobProperties properties) { + return new RestServices(batchClient, objectMapper, properties.definition(), properties.queue()); + } + + @Bean + public DownloadJobStatusAggregator downloadJobStatusAggregator() { + return new DownloadJobStatusAggregator(); } } diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java index 7fa8bd1b..0101b041 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/configuration/Config.java @@ -8,6 +8,7 @@ import au.org.aodn.ogcapi.server.core.util.ConstructUtils; import au.org.aodn.ogcapi.server.core.util.GeometryUtils; import au.org.aodn.ogcapi.server.core.util.RestTemplateUtils; +import au.org.aodn.ogcapi.server.processes.BatchJobProperties; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.PostConstruct; @@ -32,7 +33,8 @@ DdaProperties.class, IndexerProperties.class, GNProperties.class, - DasProperties.class + DasProperties.class, + BatchJobProperties.class }) public class Config { diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobNotFoundException.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobNotFoundException.java new file mode 100644 index 00000000..8b5e7879 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobNotFoundException.java @@ -0,0 +1,7 @@ +package au.org.aodn.ogcapi.server.core.exception; + +public class DownloadJobNotFoundException extends RuntimeException { + public DownloadJobNotFoundException() { + super("Download job not found"); + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobStatusException.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobStatusException.java new file mode 100644 index 00000000..a4065ed9 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/DownloadJobStatusException.java @@ -0,0 +1,11 @@ +package au.org.aodn.ogcapi.server.core.exception; + +public class DownloadJobStatusException extends RuntimeException { + public DownloadJobStatusException() { + super("Unable to retrieve download job status"); + } + + public DownloadJobStatusException(Throwable cause) { + super("Unable to retrieve download job status", cause); + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/GlobalExceptionHandler.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/GlobalExceptionHandler.java index be119fc6..9aca8dca 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/GlobalExceptionHandler.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/exception/GlobalExceptionHandler.java @@ -88,6 +88,34 @@ public ResponseEntity handleResourceNotFoundException(ResourceNot return new ResponseEntity<>(errorResponse, HttpStatus.NOT_FOUND); } + @ExceptionHandler(DownloadJobNotFoundException.class) + public ResponseEntity handleDownloadJobNotFoundException( + DownloadJobNotFoundException ex, + WebRequest request) { + ErrorResponse errorResponse = ErrorResponse + .builder() + .timestamp(LocalDateTime.now()) + .message(ex.getMessage()) + .details(request.getDescription(false)) + .build(); + + return new ResponseEntity<>(errorResponse, HttpStatus.NOT_FOUND); + } + + @ExceptionHandler(DownloadJobStatusException.class) + public ResponseEntity handleDownloadJobStatusException( + DownloadJobStatusException ex, + WebRequest request) { + ErrorResponse errorResponse = ErrorResponse + .builder() + .timestamp(LocalDateTime.now()) + .message(ex.getMessage()) + .details(request.getDescription(false)) + .build(); + + return new ResponseEntity<>(errorResponse, HttpStatus.INTERNAL_SERVER_ERROR); + } + @ExceptionHandler(DasUpstreamException.class) public ResponseEntity handleDasUpstreamException(DasUpstreamException ex, WebRequest request) { ErrorResponse errorResponse = ErrorResponse diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadExecutionResponse.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadExecutionResponse.java new file mode 100644 index 00000000..d2c65996 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadExecutionResponse.java @@ -0,0 +1,13 @@ +package au.org.aodn.ogcapi.server.core.model; + +import au.org.aodn.ogcapi.processes.model.InlineResponse200; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.v3.oas.annotations.media.Schema; + +@Schema(description = "Compatible download execution response with the submitted AWS Batch job ID.") +public record DownloadExecutionResponse( + @JsonProperty("message") InlineValue message, + @JsonProperty("status") InlineValue status, + @JsonProperty("jobID") String jobId +) implements InlineResponse200 { +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadJobStatusInfo.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadJobStatusInfo.java new file mode 100644 index 00000000..e05ea248 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/DownloadJobStatusInfo.java @@ -0,0 +1,65 @@ +package au.org.aodn.ogcapi.server.core.model; + +import au.org.aodn.ogcapi.processes.model.StatusInfo; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.v3.oas.annotations.media.Schema; + +/** + * AODN display metadata added to the standard OGC job status response. + */ +@Schema(description = "AODN download job status, extending the standard OGC StatusInfo model.") +public class DownloadJobStatusInfo extends StatusInfo { + + @JsonProperty("collection") + @JsonInclude(JsonInclude.Include.NON_ABSENT) + @Schema(description = "Display name of the requested collection.") + private String collection; + + @JsonProperty("dataSelection") + @JsonInclude(JsonInclude.Include.NON_ABSENT) + @Schema(description = "Dataset key or data selection requested for the download.") + private String dataSelection; + + @JsonProperty("format") + @JsonInclude(JsonInclude.Include.NON_ABSENT) + @Schema(description = "Requested output document format.") + private String format; + + @JsonProperty("metadataUrl") + @JsonInclude(JsonInclude.Include.NON_ABSENT) + @Schema(description = "Link to the metadata page supplied when the download was submitted.") + private String metadataUrl; + + public String getCollection() { + return collection; + } + + public void setCollection(String collection) { + this.collection = collection; + } + + public String getDataSelection() { + return dataSelection; + } + + public void setDataSelection(String dataSelection) { + this.dataSelection = dataSelection; + } + + public String getFormat() { + return format; + } + + public void setFormat(String format) { + this.format = format; + } + + public String getMetadataUrl() { + return metadataUrl; + } + + public void setMetadataUrl(String metadataUrl) { + this.metadataUrl = metadataUrl; + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/core/model/enumeration/InlineResponseKeyEnum.java b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/enumeration/InlineResponseKeyEnum.java index 76830051..9d3e39b6 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/core/model/enumeration/InlineResponseKeyEnum.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/core/model/enumeration/InlineResponseKeyEnum.java @@ -6,6 +6,7 @@ public enum InlineResponseKeyEnum { MESSAGE("message"), STATUS("status"), + JOB_ID("jobID"), ; private final String value; diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/BatchJobProperties.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/BatchJobProperties.java new file mode 100644 index 00000000..4812aa5d --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/BatchJobProperties.java @@ -0,0 +1,16 @@ +package au.org.aodn.ogcapi.server.processes; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "aws.batch.job") +public record BatchJobProperties( + String queue, + String definition, + String childQueue +) { + public BatchJobProperties { + if (childQueue == null || childQueue.isBlank()) { + childQueue = queue; + } + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregator.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregator.java new file mode 100644 index 00000000..6b1fc779 --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregator.java @@ -0,0 +1,76 @@ +package au.org.aodn.ogcapi.server.processes; + +import au.org.aodn.ogcapi.processes.model.StatusCode; +import software.amazon.awssdk.services.batch.model.JobStatus; + +import java.util.Objects; +import java.util.stream.Stream; + +/** + * Maps an already-fetched AWS Batch workflow snapshot to its public OGC status. + * This class deliberately makes no AWS calls so the workflow rules remain deterministic. + */ +public class DownloadJobStatusAggregator { + + public enum WorkflowMode { + EXPLICIT_ZARR, + CHILD_DISCOVERY_REQUIRED + } + + public record Snapshot( + JobStatus initial, + JobStatus prepare, + JobStatus collect, + WorkflowMode workflowMode, + boolean discoveryWindowExpired + ) { + public Snapshot { + Objects.requireNonNull(initial, "initial status is required"); + Objects.requireNonNull(workflowMode, "workflow mode is required"); + } + } + + public StatusCode aggregate(Snapshot snapshot) { + if (Stream.of(snapshot.initial(), snapshot.prepare(), snapshot.collect()) + .anyMatch(status -> status == JobStatus.FAILED)) { + return StatusCode.FAILED; + } + + if (Stream.of(snapshot.initial(), snapshot.prepare(), snapshot.collect()) + .filter(Objects::nonNull) + .anyMatch(status -> status == JobStatus.UNKNOWN_TO_SDK_VERSION)) { + throw new IllegalStateException("Unsupported AWS Batch job status"); + } + + boolean hasPrepare = snapshot.prepare() != null; + boolean hasCollect = snapshot.collect() != null; + if (hasPrepare != hasCollect) { + if (snapshot.discoveryWindowExpired()) { + throw new IllegalStateException("Only one child workflow job was found after the discovery window"); + } + return StatusCode.RUNNING; + } + + if (snapshot.collect() == JobStatus.SUCCEEDED) { + return StatusCode.SUCCESSFUL; + } + + if (hasPrepare) { + return StatusCode.RUNNING; + } + + return switch (snapshot.initial()) { + case SUBMITTED, PENDING, RUNNABLE -> StatusCode.ACCEPTED; + case STARTING, RUNNING -> StatusCode.RUNNING; + case SUCCEEDED -> { + if (snapshot.workflowMode() == WorkflowMode.EXPLICIT_ZARR + || snapshot.discoveryWindowExpired()) { + yield StatusCode.SUCCESSFUL; + } + yield StatusCode.RUNNING; + } + case FAILED -> StatusCode.FAILED; + default -> throw new IllegalStateException("Unsupported AWS Batch job status"); + }; + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusService.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusService.java new file mode 100644 index 00000000..69123f4c --- /dev/null +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusService.java @@ -0,0 +1,323 @@ +package au.org.aodn.ogcapi.server.processes; + +import au.org.aodn.ogcapi.processes.model.StatusCode; +import au.org.aodn.ogcapi.processes.model.StatusInfo; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobNotFoundException; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobStatusException; +import au.org.aodn.ogcapi.server.core.model.DownloadJobStatusInfo; +import au.org.aodn.ogcapi.server.core.model.enumeration.DatasetDownloadEnums; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import software.amazon.awssdk.services.batch.BatchClient; +import software.amazon.awssdk.services.batch.model.DescribeJobsRequest; +import software.amazon.awssdk.services.batch.model.JobDetail; +import software.amazon.awssdk.services.batch.model.JobStatus; +import software.amazon.awssdk.services.batch.model.KeyValuesPair; +import software.amazon.awssdk.services.batch.model.ListJobsRequest; +import software.amazon.awssdk.services.batch.model.ListJobsResponse; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; + +@Slf4j +@Service +public class DownloadJobStatusService { + + static final String PROCESS_ID = "download-dataset"; + static final Duration CHILD_DISCOVERY_WINDOW = Duration.ofSeconds(30); + + private static final String INITIAL_TYPE = "sub-setting"; + private static final String PREPARE_TYPE = "sub-setting-data-preparation"; + private static final String COLLECT_TYPE = "sub-setting-data-collection"; + private static final String MASTER_JOB_ID = "master_job_id"; + private static final int LIST_PAGE_SIZE = 100; + private static final int DESCRIBE_PAGE_SIZE = 100; + + private final BatchClient batchClient; + private final BatchJobProperties properties; + private final DownloadJobStatusAggregator aggregator; + private final Clock clock; + + @Autowired + public DownloadJobStatusService( + BatchClient batchClient, + BatchJobProperties properties, + DownloadJobStatusAggregator aggregator) { + this(batchClient, properties, aggregator, Clock.systemUTC()); + } + + DownloadJobStatusService( + BatchClient batchClient, + BatchJobProperties properties, + DownloadJobStatusAggregator aggregator, + Clock clock) { + this.batchClient = batchClient; + this.properties = properties; + this.aggregator = aggregator; + this.clock = clock; + } + + public DownloadJobStatusInfo getStatus(String jobId) { + validateJobId(jobId); + + try { + JobDetail initial = describeInitialJob(jobId); + + if (initial.status() == JobStatus.FAILED) { + StatusCode status = aggregator.aggregate(new DownloadJobStatusAggregator.Snapshot( + initial.status(), null, null, + DownloadJobStatusAggregator.WorkflowMode.CHILD_DISCOVERY_REQUIRED, + false)); + return toStatusInfo(jobId, status, initial, null, null); + } + + // These exact names are an internal contract with data-access-service. Any DAS + // naming change must be applied here at the same time. + JobDetail prepare = findChildJob( + "prepare-data-for-job-" + jobId, jobId, PREPARE_TYPE); + JobDetail collect = findChildJob( + "collect-data-for-job-" + jobId, jobId, COLLECT_TYPE); + + boolean discoveryWindowExpired = discoveryWindowExpired(initial); + DownloadJobStatusAggregator.WorkflowMode workflowMode = isExplicitZarr(initial.parameters()) + ? DownloadJobStatusAggregator.WorkflowMode.EXPLICIT_ZARR + : DownloadJobStatusAggregator.WorkflowMode.CHILD_DISCOVERY_REQUIRED; + + StatusCode status = aggregator.aggregate(new DownloadJobStatusAggregator.Snapshot( + initial.status(), + statusOf(prepare), + statusOf(collect), + workflowMode, + discoveryWindowExpired)); + + return toStatusInfo(jobId, status, initial, prepare, collect); + } catch (DownloadJobNotFoundException e) { + throw e; + } catch (Exception e) { + log.error("Failed to reconstruct AWS Batch workflow for download job {}", jobId, e); + throw new DownloadJobStatusException(e); + } + } + + private void validateJobId(String jobId) { + try { + UUID parsed = UUID.fromString(jobId); + if (!parsed.toString().equalsIgnoreCase(jobId)) { + throw new IllegalArgumentException("Non-canonical UUID"); + } + } catch (Exception e) { + throw new DownloadJobNotFoundException(); + } + } + + private JobDetail describeInitialJob(String jobId) { + List jobs = batchClient.describeJobs(DescribeJobsRequest.builder().jobs(jobId).build()).jobs(); + if (jobs.size() != 1) { + throw new DownloadJobNotFoundException(); + } + + JobDetail job = jobs.get(0); + if (!jobId.equalsIgnoreCase(job.jobId()) + || !matchesQueue(properties.queue(), job.jobQueue()) + || !matchesJobDefinition(properties.definition(), job.jobDefinition()) + || !INITIAL_TYPE.equals(job.parameters().get(DatasetDownloadEnums.Parameter.TYPE.getValue()))) { + throw new DownloadJobNotFoundException(); + } + return job; + } + + private JobDetail findChildJob(String exactJobName, String masterJobId, String expectedType) { + Set candidateIds = listCandidateIds(exactJobName); + if (candidateIds.isEmpty()) { + return null; + } + + List valid = new ArrayList<>(describeJobs(candidateIds).stream() + .filter(job -> exactJobName.equals(job.jobName())) + .filter(job -> matchesQueue(properties.childQueue(), job.jobQueue())) + .filter(job -> masterJobId.equals(job.parameters().get(MASTER_JOB_ID))) + .filter(job -> expectedType.equals(job.parameters().get(DatasetDownloadEnums.Parameter.TYPE.getValue()))) + .collect(Collectors.toMap( + JobDetail::jobId, + Function.identity(), + (first, duplicate) -> first, + LinkedHashMap::new)) + .values()); + + if (valid.size() > 1) { + log.error("Ambiguous child workflow job {}: valid AWS job ids {}", exactJobName, + valid.stream().map(JobDetail::jobId).toList()); + throw new DownloadJobStatusException(); + } + return valid.isEmpty() ? null : valid.get(0); + } + + private Set listCandidateIds(String exactJobName) { + Set result = new LinkedHashSet<>(); + String nextToken = null; + do { + ListJobsRequest request = ListJobsRequest.builder() + .jobQueue(properties.childQueue()) + .filters(KeyValuesPair.builder().name("JOB_NAME").values(exactJobName).build()) + .maxResults(LIST_PAGE_SIZE) + .nextToken(nextToken) + .build(); + ListJobsResponse response = batchClient.listJobs(request); + response.jobSummaryList().stream() + .filter(summary -> exactJobName.equals(summary.jobName())) + .map(summary -> summary.jobId()) + .filter(id -> id != null && !id.isBlank()) + .forEach(result::add); + nextToken = response.nextToken(); + } while (nextToken != null); + return result; + } + + private List describeJobs(Set candidateIds) { + List ids = new ArrayList<>(candidateIds); + List result = new ArrayList<>(); + for (int start = 0; start < ids.size(); start += DESCRIBE_PAGE_SIZE) { + int end = Math.min(start + DESCRIBE_PAGE_SIZE, ids.size()); + result.addAll(batchClient.describeJobs(DescribeJobsRequest.builder() + .jobs(ids.subList(start, end)) + .build()).jobs()); + } + return result; + } + + private boolean discoveryWindowExpired(JobDetail initial) { + Long stoppedAt = initial.stoppedAt(); + return stoppedAt != null && stoppedAt > 0 + && !clock.instant().isBefore(Instant.ofEpochMilli(stoppedAt).plus(CHILD_DISCOVERY_WINDOW)); + } + + private boolean isExplicitZarr(Map parameters) { + String keys = parameters.get(DatasetDownloadEnums.Parameter.KEY.getValue()); + if (keys == null || keys.isBlank()) { + return false; + } + String[] splitKeys = keys.split(",", -1); + for (String key : splitKeys) { + String trimmed = key.trim(); + if (trimmed.isEmpty() || !trimmed.endsWith(".zarr")) { + return false; + } + } + return true; + } + + private DownloadJobStatusInfo toStatusInfo( + String jobId, + StatusCode status, + JobDetail initial, + JobDetail prepare, + JobDetail collect) { + DownloadJobStatusInfo result = new DownloadJobStatusInfo(); + result.setProcessID(PROCESS_ID); + result.setType(StatusInfo.TypeEnum.PROCESS); + result.setJobID(jobId); + result.setStatus(status); + result.setMessage(messageFor(status)); + + Map parameters = initial.parameters(); + result.setCollection(nonBlank(parameters.get(DatasetDownloadEnums.Parameter.COLLECTION_TITLE.getValue()))); + result.setDataSelection(nonBlank(parameters.get(DatasetDownloadEnums.Parameter.KEY.getValue()))); + result.setFormat(nonBlank(parameters.get(DatasetDownloadEnums.Parameter.OUTPUT_FORMAT.getValue()))); + result.setMetadataUrl(nonBlank(parameters.get(DatasetDownloadEnums.Parameter.FULL_METADATA_LINK.getValue()))); + result.setCreated(toDate(initial.createdAt())); + result.setStarted(firstStartedAt(initial, prepare, collect)); + result.setFinished(finishedAt(status, initial, prepare, collect)); + return result; + } + + private String nonBlank(String value) { + return value == null || value.isBlank() ? null : value; + } + + private Date firstStartedAt(JobDetail... jobs) { + Long first = null; + for (JobDetail job : jobs) { + if (job != null && job.startedAt() != null && job.startedAt() > 0 + && (first == null || job.startedAt() < first)) { + first = job.startedAt(); + } + } + return toDate(first); + } + + private Date finishedAt(StatusCode status, JobDetail initial, JobDetail prepare, JobDetail collect) { + if (status == StatusCode.SUCCESSFUL) { + return toDate(collect != null && collect.status() == JobStatus.SUCCEEDED + ? collect.stoppedAt() + : initial.stoppedAt()); + } + if (status == StatusCode.FAILED) { + for (JobDetail job : new JobDetail[]{initial, prepare, collect}) { + if (job != null && job.status() == JobStatus.FAILED) { + return toDate(job.stoppedAt()); + } + } + } + return null; + } + + private Date toDate(Long epochMillis) { + return epochMillis == null || epochMillis <= 0 ? null : new Date(epochMillis); + } + + private JobStatus statusOf(JobDetail job) { + return job == null ? null : job.status(); + } + + private String messageFor(StatusCode status) { + return switch (status) { + case ACCEPTED -> "Download job accepted"; + case RUNNING -> "Download job is running"; + case SUCCESSFUL -> "Download job completed successfully"; + case FAILED -> "Download job failed"; + case DISMISSED -> "Download job dismissed"; + }; + } + + static boolean matchesQueue(String configured, String actual) { + if (configured == null || actual == null) { + return false; + } + if (configured.startsWith("arn:")) { + return configured.equals(actual); + } + return configured.equals(resourceName(actual, "job-queue/")); + } + + static boolean matchesJobDefinition(String configured, String actual) { + if (configured == null || actual == null) { + return false; + } + if (configured.startsWith("arn:")) { + return configured.equals(actual); + } + return withoutRevision(configured).equals(withoutRevision(resourceName(actual, "job-definition/"))); + } + + private static String resourceName(String value, String marker) { + int markerIndex = value.indexOf(marker); + return markerIndex >= 0 ? value.substring(markerIndex + marker.length()) : value; + } + + private static String withoutRevision(String value) { + int revision = value.lastIndexOf(':'); + return revision >= 0 ? value.substring(0, revision) : value; + } +} diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestApi.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestApi.java index 354a0aa2..15258db6 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestApi.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestApi.java @@ -1,17 +1,25 @@ package au.org.aodn.ogcapi.server.processes; import au.org.aodn.ogcapi.processes.api.ProcessesApi; +import au.org.aodn.ogcapi.processes.api.JobsApi; import au.org.aodn.ogcapi.processes.model.Execute; import au.org.aodn.ogcapi.processes.model.InlineResponse200; import au.org.aodn.ogcapi.processes.model.ProcessList; import au.org.aodn.ogcapi.processes.model.Results; +import au.org.aodn.ogcapi.processes.model.JobList; +import au.org.aodn.ogcapi.processes.model.StatusInfo; +import au.org.aodn.ogcapi.server.core.model.DownloadExecutionResponse; +import au.org.aodn.ogcapi.server.core.model.DownloadJobStatusInfo; import au.org.aodn.ogcapi.server.core.model.InlineValue; import au.org.aodn.ogcapi.server.core.model.enumeration.DatasetDownloadEnums; import au.org.aodn.ogcapi.server.core.model.enumeration.InlineResponseKeyEnum; import au.org.aodn.ogcapi.server.core.model.enumeration.ProcessIdEnum; import io.swagger.v3.oas.annotations.Parameter; import io.swagger.v3.oas.annotations.enums.ParameterIn; +import io.swagger.v3.oas.annotations.media.Content; +import io.swagger.v3.oas.annotations.media.ExampleObject; import io.swagger.v3.oas.annotations.media.Schema; +import io.swagger.v3.oas.annotations.responses.ApiResponse; import jakarta.validation.Valid; import lombok.extern.slf4j.Slf4j; import org.apache.coyote.BadRequestException; @@ -27,11 +35,14 @@ @Slf4j @RestController("ProcessesRestApi") @RequestMapping(value = "/api/v1/ogc") -public class RestApi implements ProcessesApi { +public class RestApi implements ProcessesApi, JobsApi { @Autowired private RestServices restServices; + @Autowired + private DownloadJobStatusService downloadJobStatusService; + @Override // because the produces value in the interface declaration includes "/_" which may // cause exception thrown sometimes. So i re-declared the produces value here @@ -41,6 +52,19 @@ public class RestApi implements ProcessesApi { consumes = {MediaType.APPLICATION_JSON_VALUE}, method = RequestMethod.POST ) + @ApiResponse( + responseCode = "200", + description = "Download job accepted.", + content = @Content( + mediaType = MediaType.APPLICATION_JSON_VALUE, + schema = @Schema(implementation = DownloadExecutionResponse.class), + examples = @ExampleObject(value = """ + { + "message": {"message": "Job submitted with ID: 123e4567-e89b-12d3-a456-426614174000"}, + "status": {"message": "200"}, + "jobID": "123e4567-e89b-12d3-a456-426614174000" + } + """))) public ResponseEntity execute( @Parameter(in = ParameterIn.PATH, required = true, schema = @Schema()) @PathVariable("processID") @@ -65,7 +89,8 @@ public ResponseEntity execute( String outputFormat = DatasetDownloadEnums.Parameter.OUTPUT_FORMAT.getStringInput(body); Object multiPolygon = DatasetDownloadEnums.Parameter.MULTI_POLYGON.getObjectInput(body); - var response = restServices.downloadData(uuid, key, startDate, endDate, multiPolygon, recipient, collectionTitle, fullMetadataLink, suggestedCitation, outputFormat); + String jobId = restServices.downloadData(uuid, key, startDate, endDate, multiPolygon, recipient, + collectionTitle, fullMetadataLink, suggestedCitation, outputFormat); // The notify user email lives here rather than in data-access-service to make the first // email faster @@ -73,11 +98,9 @@ public ResponseEntity execute( // a job id, otherwise we promise the user a file that will never be produced. restServices.notifyUser(recipient, uuid, key, startDate, endDate, multiPolygon, collectionTitle, fullMetadataLink, suggestedCitation, outputFormat); - var value = new InlineValue(response.getBody()); + var value = new InlineValue("Job submitted with ID: " + jobId); var status = new InlineValue(Integer.toString(HttpStatus.OK.value())); - var results = new Results(); - results.put(InlineResponseKeyEnum.MESSAGE.getValue(), value); - results.put(InlineResponseKeyEnum.STATUS.getValue(), status); + var results = new DownloadExecutionResponse(value, status, jobId); return ResponseEntity.ok(results); @@ -115,6 +138,35 @@ public ResponseEntity getProcesses() { return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED).build(); } + @Override + @ApiResponse(responseCode = "501", description = "Listing jobs is not implemented.") + public ResponseEntity getJobs() { + return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED).build(); + } + + @Override + @ApiResponse(responseCode = "501", description = "Retrieving job results is not implemented.") + public ResponseEntity getResult(String jobId) { + return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED).build(); + } + + @Override + @ApiResponse( + responseCode = "200", + description = "Download job status.", + content = @Content( + mediaType = MediaType.APPLICATION_JSON_VALUE, + schema = @Schema(implementation = DownloadJobStatusInfo.class))) + public ResponseEntity getStatus(String jobId) { + return ResponseEntity.ok(downloadJobStatusService.getStatus(jobId)); + } + + @Override + @ApiResponse(responseCode = "501", description = "Dismissing jobs is not implemented.") + public ResponseEntity dismiss(String jobId) { + return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED).build(); + } + /** * WFS download endpoint with SSE support to handle long-running operations and prevent timeouts */ diff --git a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java index c94e0ed3..a481fbf0 100644 --- a/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java +++ b/server/src/main/java/au/org/aodn/ogcapi/server/processes/RestServices.java @@ -13,7 +13,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.http.ResponseEntity; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import software.amazon.awssdk.services.batch.BatchClient; import software.amazon.awssdk.services.batch.model.SubmitJobRequest; @@ -92,7 +91,7 @@ public void notifyUser(String recipient, String uuid, String key, String startDa } } - public ResponseEntity downloadData( + public String downloadData( String id, String key, String startDate, @@ -124,7 +123,7 @@ public ResponseEntity downloadData( this.batchJobDefinition, parameters); log.info("Job submitted with ID: {}", jobId); - return ResponseEntity.ok("Job submitted with ID: " + jobId); + return jobId; } private String submitJob(String jobName, String jobQueue, String jobDefinition, Map parameters) { diff --git a/server/src/main/resources/application-dev.yaml b/server/src/main/resources/application-dev.yaml index f0891264..099ce22c 100644 --- a/server/src/main/resources/application-dev.yaml +++ b/server/src/main/resources/application-dev.yaml @@ -20,3 +20,9 @@ geonetwork4: elasticsearch: semantic: enabled: false + +aws: + batch: + job: + # DAS dev submits its preparation/collection parents to this separate queue. + child-queue: generate-csv-data-file diff --git a/server/src/main/resources/application.yaml b/server/src/main/resources/application.yaml index 0d7ceb04..b2be0494 100644 --- a/server/src/main/resources/application.yaml +++ b/server/src/main/resources/application.yaml @@ -43,6 +43,9 @@ aws: job: queue: data-access-service-batch-job-queue definition: data-access-service-batch-job-definition + # Defaults to the initial queue in BatchJobProperties. Dev deployments whose DAS + # uses generate-csv-data-file must override this value explicitly. + child-queue: data-access-service-batch-job-queue wfs-default-param: fields: diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregatorTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregatorTest.java new file mode 100644 index 00000000..d3cf5dd4 --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusAggregatorTest.java @@ -0,0 +1,115 @@ +package au.org.aodn.ogcapi.server.processes; + +import au.org.aodn.ogcapi.processes.model.StatusCode; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.batch.model.JobStatus; + +import static au.org.aodn.ogcapi.server.processes.DownloadJobStatusAggregator.WorkflowMode.CHILD_DISCOVERY_REQUIRED; +import static au.org.aodn.ogcapi.server.processes.DownloadJobStatusAggregator.WorkflowMode.EXPLICIT_ZARR; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class DownloadJobStatusAggregatorTest { + + private final DownloadJobStatusAggregator aggregator = new DownloadJobStatusAggregator(); + + @Test + void mapsInitialPreExecutionStatesToAccepted() { + for (JobStatus status : new JobStatus[]{JobStatus.SUBMITTED, JobStatus.PENDING, JobStatus.RUNNABLE}) { + assertEquals(StatusCode.ACCEPTED, aggregate(status, null, null, CHILD_DISCOVERY_REQUIRED, false)); + } + } + + @Test + void mapsInitialExecutionStatesToRunning() { + for (JobStatus status : new JobStatus[]{JobStatus.STARTING, JobStatus.RUNNING}) { + assertEquals(StatusCode.RUNNING, aggregate(status, null, null, CHILD_DISCOVERY_REQUIRED, false)); + } + } + + @Test + void mapsInitialFailureToFailed() { + assertEquals(StatusCode.FAILED, + aggregate(JobStatus.FAILED, null, null, CHILD_DISCOVERY_REQUIRED, false)); + } + + @Test + void mapsDependentPendingRunnableAndRunningStatesToRunning() { + for (JobStatus prepare : new JobStatus[]{JobStatus.PENDING, JobStatus.RUNNABLE, JobStatus.RUNNING}) { + assertEquals(StatusCode.RUNNING, + aggregate(JobStatus.SUCCEEDED, prepare, JobStatus.PENDING, CHILD_DISCOVERY_REQUIRED, false)); + } + for (JobStatus collect : new JobStatus[]{JobStatus.PENDING, JobStatus.RUNNABLE, JobStatus.RUNNING}) { + assertEquals(StatusCode.RUNNING, + aggregate(JobStatus.SUCCEEDED, JobStatus.SUCCEEDED, collect, CHILD_DISCOVERY_REQUIRED, false)); + } + } + + @Test + void mapsCollectSuccessToSuccessful() { + assertEquals(StatusCode.SUCCESSFUL, + aggregate(JobStatus.SUCCEEDED, JobStatus.SUCCEEDED, JobStatus.SUCCEEDED, + CHILD_DISCOVERY_REQUIRED, true)); + } + + @Test + void failureTakesPrecedenceOverCollectSuccess() { + assertEquals(StatusCode.FAILED, + aggregate(JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.SUCCEEDED, + CHILD_DISCOVERY_REQUIRED, true)); + } + + @Test + void explicitZarrWithoutChildrenSucceedsImmediately() { + assertEquals(StatusCode.SUCCESSFUL, + aggregate(JobStatus.SUCCEEDED, null, null, EXPLICIT_ZARR, false)); + } + + @Test + void ambiguousChildlessWorkflowRunsInsideWindowAndSucceedsAfterward() { + assertEquals(StatusCode.RUNNING, + aggregate(JobStatus.SUCCEEDED, null, null, CHILD_DISCOVERY_REQUIRED, false)); + assertEquals(StatusCode.SUCCESSFUL, + aggregate(JobStatus.SUCCEEDED, null, null, CHILD_DISCOVERY_REQUIRED, true)); + } + + @Test + void singleChildRunsInsideWindowAndIsInconsistentAfterward() { + assertEquals(StatusCode.RUNNING, + aggregate(JobStatus.SUCCEEDED, JobStatus.SUCCEEDED, null, + CHILD_DISCOVERY_REQUIRED, false)); + assertEquals(StatusCode.RUNNING, + aggregate(JobStatus.SUCCEEDED, null, JobStatus.PENDING, + CHILD_DISCOVERY_REQUIRED, false)); + + assertThrows(IllegalStateException.class, + () -> aggregate(JobStatus.SUCCEEDED, JobStatus.SUCCEEDED, null, + CHILD_DISCOVERY_REQUIRED, true)); + assertThrows(IllegalStateException.class, + () -> aggregate(JobStatus.SUCCEEDED, null, JobStatus.PENDING, + CHILD_DISCOVERY_REQUIRED, true)); + assertThrows(IllegalStateException.class, + () -> aggregate(JobStatus.SUCCEEDED, null, JobStatus.SUCCEEDED, + CHILD_DISCOVERY_REQUIRED, true)); + } + + @Test + void childFailureIsTerminalEvenWhenOnlyOneChildIsVisible() { + assertEquals(StatusCode.FAILED, + aggregate(JobStatus.SUCCEEDED, JobStatus.FAILED, null, + CHILD_DISCOVERY_REQUIRED, false)); + assertEquals(StatusCode.FAILED, + aggregate(JobStatus.SUCCEEDED, null, JobStatus.FAILED, + CHILD_DISCOVERY_REQUIRED, true)); + } + + private StatusCode aggregate( + JobStatus initial, + JobStatus prepare, + JobStatus collect, + DownloadJobStatusAggregator.WorkflowMode mode, + boolean expired) { + return aggregator.aggregate(new DownloadJobStatusAggregator.Snapshot( + initial, prepare, collect, mode, expired)); + } +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusServiceTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusServiceTest.java new file mode 100644 index 00000000..4d6c073c --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/DownloadJobStatusServiceTest.java @@ -0,0 +1,338 @@ +package au.org.aodn.ogcapi.server.processes; + +import au.org.aodn.ogcapi.processes.model.StatusCode; +import au.org.aodn.ogcapi.processes.model.StatusInfo; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobNotFoundException; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobStatusException; +import au.org.aodn.ogcapi.server.core.model.DownloadJobStatusInfo; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import software.amazon.awssdk.services.batch.BatchClient; +import software.amazon.awssdk.services.batch.model.DescribeJobsRequest; +import software.amazon.awssdk.services.batch.model.DescribeJobsResponse; +import software.amazon.awssdk.services.batch.model.JobDetail; +import software.amazon.awssdk.services.batch.model.JobStatus; +import software.amazon.awssdk.services.batch.model.JobSummary; +import software.amazon.awssdk.services.batch.model.ListJobsRequest; +import software.amazon.awssdk.services.batch.model.ListJobsResponse; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class DownloadJobStatusServiceTest { + + private static final String JOB_ID = "123e4567-e89b-12d3-a456-426614174000"; + private static final String QUEUE_NAME = "initial-queue"; + private static final String CHILD_QUEUE_NAME = "child-queue"; + private static final String QUEUE_ARN = "arn:aws:batch:ap-southeast-2:123456789012:job-queue/" + QUEUE_NAME; + private static final String CHILD_QUEUE_ARN = "arn:aws:batch:ap-southeast-2:123456789012:job-queue/" + CHILD_QUEUE_NAME; + private static final String DEFINITION_NAME = "download-definition"; + private static final String DEFINITION_ARN = "arn:aws:batch:ap-southeast-2:123456789012:job-definition/" + DEFINITION_NAME + ":7"; + private static final Instant NOW = Instant.parse("2026-08-24T02:00:00Z"); + + @Mock + private BatchClient batchClient; + + private final Map describedJobs = new HashMap<>(); + private final Map listedJobs = new HashMap<>(); + private DownloadJobStatusService service; + + @BeforeEach + void setUp() { + lenient().when(batchClient.describeJobs(any(DescribeJobsRequest.class))).thenAnswer(invocation -> { + DescribeJobsRequest request = invocation.getArgument(0); + List jobs = request.jobs().stream() + .map(describedJobs::get) + .filter(job -> job != null) + .toList(); + return DescribeJobsResponse.builder().jobs(jobs).build(); + }); + lenient().when(batchClient.listJobs(any(ListJobsRequest.class))).thenAnswer(invocation -> { + ListJobsRequest request = invocation.getArgument(0); + String name = request.filters().get(0).values().get(0); + String key = name + "|" + request.nextToken(); + return listedJobs.getOrDefault(key, ListJobsResponse.builder().build()); + }); + + service = new DownloadJobStatusService( + batchClient, + new BatchJobProperties(QUEUE_NAME, DEFINITION_NAME, CHILD_QUEUE_NAME), + new DownloadJobStatusAggregator(), + Clock.fixed(NOW, ZoneOffset.UTC)); + } + + @Test + void invalidJobIdIsTheSameGenericNotFoundWithoutCallingAws() { + DownloadJobNotFoundException exception = assertThrows( + DownloadJobNotFoundException.class, () -> service.getStatus("not-a-uuid")); + + assertEquals("Download job not found", exception.getMessage()); + verify(batchClient, never()).describeJobs(any(DescribeJobsRequest.class)); + verify(batchClient, never()).listJobs(any(ListJobsRequest.class)); + } + + @Test + void missingOrExpiredInitialJobIsGenericNotFound() { + DownloadJobNotFoundException exception = assertThrows( + DownloadJobNotFoundException.class, () -> service.getStatus(JOB_ID)); + assertEquals("Download job not found", exception.getMessage()); + } + + @Test + void acceptsQueueAndDefinitionNamesAgainstAwsArnsAndIgnoresDefinitionRevision() { + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "one.zarr", NOW.minusSeconds(5))); + + DownloadJobStatusInfo result = service.getStatus(JOB_ID); + + assertEquals(StatusCode.SUCCESSFUL, result.getStatus()); + assertEquals(DownloadJobStatusService.PROCESS_ID, result.getProcessID()); + assertEquals(StatusInfo.TypeEnum.PROCESS, result.getType()); + assertEquals(JOB_ID, result.getJobID()); + assertEquals("Test Ocean Data Collection", result.getCollection()); + assertEquals("one.zarr", result.getDataSelection()); + assertEquals("netcdf", result.getFormat()); + assertEquals("https://portal.example.test/details/collection-id", result.getMetadataUrl()); + } + + @Test + void optionalDisplayMetadataIsOmittedWhenItWasNotStored() { + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "one.zarr", NOW.minusSeconds(5)) + .toBuilder() + .parameters(Map.of("type", "sub-setting", "key", "one.zarr")) + .build()); + + DownloadJobStatusInfo result = service.getStatus(JOB_ID); + + assertEquals("one.zarr", result.getDataSelection()); + assertNull(result.getCollection()); + assertNull(result.getFormat()); + assertNull(result.getMetadataUrl()); + } + + @Test + void configuredVersionedArnsRequireExactMatches() { + service = new DownloadJobStatusService( + batchClient, + new BatchJobProperties(QUEUE_ARN, DEFINITION_ARN, CHILD_QUEUE_NAME), + new DownloadJobStatusAggregator(), + Clock.fixed(NOW, ZoneOffset.UTC)); + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "one.zarr", NOW.minusSeconds(5))); + assertEquals(StatusCode.SUCCESSFUL, service.getStatus(JOB_ID).getStatus()); + + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "one.zarr", NOW.minusSeconds(5)) + .toBuilder().jobDefinition(DEFINITION_ARN.replace(":7", ":8")).build()); + assertThrows(DownloadJobNotFoundException.class, () -> service.getStatus(JOB_ID)); + } + + @Test + void rejectsJobsFromAnotherQueueDefinitionOrProcessWithTheSameNotFound() { + List invalidJobs = List.of( + initial(JobStatus.RUNNING, "*", null).toBuilder().jobQueue("other-queue").build(), + initial(JobStatus.RUNNING, "*", null).toBuilder().jobDefinition("other-definition:1").build(), + initial(JobStatus.RUNNING, "*", null).toBuilder() + .parameters(Map.of("type", "another-process", "key", "*")).build()); + + for (JobDetail invalid : invalidJobs) { + describedJobs.put(JOB_ID, invalid); + assertThrows(DownloadJobNotFoundException.class, () -> service.getStatus(JOB_ID)); + } + } + + @Test + void awsDescribeFailureBecomesGenericStatusError() { + doThrow(new RuntimeException("credentials and internal endpoint")) + .when(batchClient).describeJobs(any(DescribeJobsRequest.class)); + DownloadJobStatusException describeError = assertThrows( + DownloadJobStatusException.class, () -> service.getStatus(JOB_ID)); + assertEquals("Unable to retrieve download job status", describeError.getMessage()); + } + + @Test + void awsListFailureBecomesGenericStatusError() { + describedJobs.put(JOB_ID, initial(JobStatus.RUNNING, "*", null)); + doThrow(new RuntimeException("secret list failure")) + .when(batchClient).listJobs(any(ListJobsRequest.class)); + DownloadJobStatusException listError = assertThrows( + DownloadJobStatusException.class, () -> service.getStatus(JOB_ID)); + assertEquals("Unable to retrieve download job status", listError.getMessage()); + } + + @Test + void zeroChildrenUsesZarrAndDiscoveryWindowRules() { + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "a.zarr, b.zarr", NOW.minusSeconds(1))); + assertEquals(StatusCode.SUCCESSFUL, service.getStatus(JOB_ID).getStatus()); + + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "*", NOW.minusSeconds(29))); + assertEquals(StatusCode.RUNNING, service.getStatus(JOB_ID).getStatus()); + + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "dataset.parquet", NOW.minusSeconds(30))); + assertEquals(StatusCode.SUCCESSFUL, service.getStatus(JOB_ID).getStatus()); + } + + @Test + void listJobsPaginatesAndAppliesExactCaseSensitiveNameCheck() { + JobDetail initial = initial(JobStatus.SUCCEEDED, "dataset.parquet", NOW.minusSeconds(5)); + describedJobs.put(JOB_ID, initial); + String prepareName = prepareName(); + String prepareId = "223e4567-e89b-12d3-a456-426614174000"; + describedJobs.put(prepareId, child(prepareId, prepareName, "sub-setting-data-preparation", JobStatus.PENDING)); + + listedJobs.put(prepareName + "|null", listResponse("page-2", + summary("323e4567-e89b-12d3-a456-426614174000", prepareName.toUpperCase()))); + listedJobs.put(prepareName + "|page-2", listResponse(null, summary(prepareId, prepareName))); + + StatusInfo result = service.getStatus(JOB_ID); + + assertEquals(StatusCode.RUNNING, result.getStatus()); + ArgumentCaptor captor = ArgumentCaptor.forClass(ListJobsRequest.class); + verify(batchClient, org.mockito.Mockito.atLeast(3)).listJobs(captor.capture()); + assertTrue(captor.getAllValues().stream().anyMatch(request -> "page-2".equals(request.nextToken()))); + } + + @Test + void duplicateSummariesAreDeduplicatedAndInvalidCandidatesAreDiscarded() { + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "*", NOW.minusSeconds(5))); + String prepareName = prepareName(); + String validId = "223e4567-e89b-12d3-a456-426614174000"; + String wrongMasterId = "323e4567-e89b-12d3-a456-426614174000"; + String wrongTypeId = "423e4567-e89b-12d3-a456-426614174000"; + describedJobs.put(validId, child(validId, prepareName, "sub-setting-data-preparation", JobStatus.RUNNING)); + describedJobs.put(wrongMasterId, child(wrongMasterId, prepareName, "sub-setting-data-preparation", JobStatus.RUNNING) + .toBuilder().parameters(Map.of("master_job_id", "another-job", "type", "sub-setting-data-preparation")).build()); + describedJobs.put(wrongTypeId, child(wrongTypeId, prepareName, "wrong-type", JobStatus.RUNNING)); + listedJobs.put(prepareName + "|null", listResponse(null, + summary(validId, prepareName), + summary(validId, prepareName), + summary(wrongMasterId, prepareName), + summary(wrongTypeId, prepareName))); + + assertEquals(StatusCode.RUNNING, service.getStatus(JOB_ID).getStatus()); + } + + @Test + void multipleDistinctValidCandidatesAreAmbiguous() { + describedJobs.put(JOB_ID, initial(JobStatus.SUCCEEDED, "*", NOW.minusSeconds(5))); + String prepareName = prepareName(); + String first = "223e4567-e89b-12d3-a456-426614174000"; + String second = "323e4567-e89b-12d3-a456-426614174000"; + describedJobs.put(first, child(first, prepareName, "sub-setting-data-preparation", JobStatus.RUNNING)); + describedJobs.put(second, child(second, prepareName, "sub-setting-data-preparation", JobStatus.RUNNING)); + listedJobs.put(prepareName + "|null", listResponse(null, + summary(first, prepareName), summary(second, prepareName))); + + DownloadJobStatusException exception = assertThrows( + DownloadJobStatusException.class, () -> service.getStatus(JOB_ID)); + assertEquals("Unable to retrieve download job status", exception.getMessage()); + } + + @Test + void usesConfiguredChildQueueAndBuildsTerminalDatesWithoutSensitiveFields() { + long createdAt = NOW.minusSeconds(100).toEpochMilli(); + JobDetail initial = initial(JobStatus.SUCCEEDED, "dataset.parquet", NOW.minusSeconds(60)) + .toBuilder().createdAt(createdAt).startedAt(NOW.minusSeconds(90).toEpochMilli()).build(); + describedJobs.put(JOB_ID, initial); + + String prepareId = "223e4567-e89b-12d3-a456-426614174000"; + String collectId = "323e4567-e89b-12d3-a456-426614174000"; + JobDetail prepare = child(prepareId, prepareName(), "sub-setting-data-preparation", JobStatus.SUCCEEDED) + .toBuilder().startedAt(NOW.minusSeconds(50).toEpochMilli()).stoppedAt(NOW.minusSeconds(40).toEpochMilli()).build(); + JobDetail collect = child(collectId, collectName(), "sub-setting-data-collection", JobStatus.SUCCEEDED) + .toBuilder().startedAt(NOW.minusSeconds(30).toEpochMilli()).stoppedAt(NOW.minusSeconds(10).toEpochMilli()) + .statusReason("s3://private-bucket/internal-key").build(); + describedJobs.put(prepareId, prepare); + describedJobs.put(collectId, collect); + listedJobs.put(prepareName() + "|null", listResponse(null, summary(prepareId, prepareName()))); + listedJobs.put(collectName() + "|null", listResponse(null, summary(collectId, collectName()))); + + StatusInfo result = service.getStatus(JOB_ID); + + assertEquals(StatusCode.SUCCESSFUL, result.getStatus()); + assertEquals(new Date(createdAt), result.getCreated()); + assertEquals(new Date(NOW.minusSeconds(90).toEpochMilli()), result.getStarted()); + assertEquals(new Date(NOW.minusSeconds(10).toEpochMilli()), result.getFinished()); + assertNull(result.getProgress()); + assertNull(result.getUpdated()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListJobsRequest.class); + verify(batchClient, org.mockito.Mockito.atLeast(2)).listJobs(captor.capture()); + assertTrue(captor.getAllValues().stream().allMatch(request -> CHILD_QUEUE_NAME.equals(request.jobQueue()))); + } + + @Test + void queueAndDefinitionNormalizersHandleNamesAndArns() { + assertTrue(DownloadJobStatusService.matchesQueue(QUEUE_NAME, QUEUE_ARN)); + assertTrue(DownloadJobStatusService.matchesQueue(QUEUE_ARN, QUEUE_ARN)); + assertTrue(DownloadJobStatusService.matchesJobDefinition(DEFINITION_NAME, DEFINITION_ARN)); + assertTrue(DownloadJobStatusService.matchesJobDefinition(DEFINITION_NAME + ":3", DEFINITION_ARN)); + assertTrue(DownloadJobStatusService.matchesJobDefinition(DEFINITION_ARN, DEFINITION_ARN)); + } + + private JobDetail initial(JobStatus status, String key, Instant stoppedAt) { + Map parameters = new HashMap<>(); + parameters.put("type", "sub-setting"); + parameters.put("collection_title", "Test Ocean Data Collection"); + parameters.put("output_format", "netcdf"); + parameters.put("full_metadata_link", "https://portal.example.test/details/collection-id"); + if (key != null) { + parameters.put("key", key); + } + return JobDetail.builder() + .jobId(JOB_ID) + .jobName("initial") + .jobQueue(QUEUE_ARN) + .jobDefinition(DEFINITION_ARN) + .status(status) + .parameters(parameters) + .createdAt(NOW.minusSeconds(120).toEpochMilli()) + .stoppedAt(stoppedAt == null ? null : stoppedAt.toEpochMilli()) + .build(); + } + + private JobDetail child(String id, String name, String type, JobStatus status) { + return JobDetail.builder() + .jobId(id) + .jobName(name) + .jobQueue(CHILD_QUEUE_ARN) + .status(status) + .parameters(Map.of("master_job_id", JOB_ID, "type", type)) + .build(); + } + + private JobSummary summary(String id, String name) { + return JobSummary.builder().jobId(id).jobName(name).build(); + } + + private ListJobsResponse listResponse(String nextToken, JobSummary... jobs) { + return ListJobsResponse.builder().jobSummaryList(jobs).nextToken(nextToken).build(); + } + + private String prepareName() { + return "prepare-data-for-job-" + JOB_ID; + } + + private String collectName() { + return "collect-data-for-job-" + JOB_ID; + } +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiJobsTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiJobsTest.java new file mode 100644 index 00000000..76be1f4f --- /dev/null +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiJobsTest.java @@ -0,0 +1,125 @@ +package au.org.aodn.ogcapi.server.processes; + +import au.org.aodn.ogcapi.processes.model.StatusCode; +import au.org.aodn.ogcapi.processes.model.StatusInfo; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobNotFoundException; +import au.org.aodn.ogcapi.server.core.exception.DownloadJobStatusException; +import au.org.aodn.ogcapi.server.core.exception.GlobalExceptionHandler; +import au.org.aodn.ogcapi.server.core.model.DownloadJobStatusInfo; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.http.MediaType; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.util.Map; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +@ExtendWith(MockitoExtension.class) +class RestApiJobsTest { + + private static final String JOB_ID = "123e4567-e89b-12d3-a456-426614174000"; + + @Mock + private RestServices restServices; + + @Mock + private DownloadJobStatusService downloadJobStatusService; + + private final ObjectMapper objectMapper = new ObjectMapper(); + private MockMvc mockMvc; + + @BeforeEach + void setUp() { + RestApi restApi = new RestApi(); + ReflectionTestUtils.setField(restApi, "restServices", restServices); + ReflectionTestUtils.setField(restApi, "downloadJobStatusService", downloadJobStatusService); + mockMvc = MockMvcBuilders.standaloneSetup(restApi) + .setControllerAdvice(new GlobalExceptionHandler()) + .build(); + } + + @Test + void postKeepsExistingFieldsAndAddsPureJobId() throws Exception { + when(restServices.downloadData(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) + .thenReturn(JOB_ID); + String body = objectMapper.writeValueAsString(Map.of("inputs", Map.of( + "uuid", "collection-id", + "recipient", "person@example.com"))); + + mockMvc.perform(post("/api/v1/ogc/processes/download/execution") + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .content(body)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.message.message").value("Job submitted with ID: " + JOB_ID)) + .andExpect(jsonPath("$.status.message").value("200")) + .andExpect(jsonPath("$.jobID").value(JOB_ID)); + } + + @Test + void getStatusSerializesExtendedStatusInfo() throws Exception { + DownloadJobStatusInfo statusInfo = new DownloadJobStatusInfo(); + statusInfo.setProcessID("download-dataset"); + statusInfo.setType(StatusInfo.TypeEnum.PROCESS); + statusInfo.setJobID(JOB_ID); + statusInfo.setStatus(StatusCode.RUNNING); + statusInfo.setMessage("Download job is running"); + statusInfo.setCollection("Test Ocean Data Collection"); + statusInfo.setDataSelection("satellite_wind_altimeter_delayed_qc.zarr"); + statusInfo.setFormat("netcdf"); + statusInfo.setMetadataUrl("https://portal.example.test/details/collection-id"); + when(downloadJobStatusService.getStatus(JOB_ID)).thenReturn(statusInfo); + + mockMvc.perform(get("/api/v1/ogc/jobs/{jobId}", JOB_ID).accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.processID").value("download-dataset")) + .andExpect(jsonPath("$.type").value("process")) + .andExpect(jsonPath("$.jobID").value(JOB_ID)) + .andExpect(jsonPath("$.status").value("running")) + .andExpect(jsonPath("$.collection").value("Test Ocean Data Collection")) + .andExpect(jsonPath("$.dataSelection").value("satellite_wind_altimeter_delayed_qc.zarr")) + .andExpect(jsonPath("$.format").value("netcdf")) + .andExpect(jsonPath("$.metadataUrl").value("https://portal.example.test/details/collection-id")) + .andExpect(jsonPath("$.progress").doesNotExist()); + } + + @Test + void getStatusReturnsGenericNotFoundAndServerErrors() throws Exception { + when(downloadJobStatusService.getStatus("missing")).thenThrow(new DownloadJobNotFoundException()); + mockMvc.perform(get("/api/v1/ogc/jobs/missing").accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isNotFound()) + .andExpect(jsonPath("$.message").value("Download job not found")) + .andExpect(jsonPath("$.message").value(org.hamcrest.Matchers.not( + org.hamcrest.Matchers.containsString("AWS")))); + + when(downloadJobStatusService.getStatus("broken")).thenThrow(new DownloadJobStatusException()); + mockMvc.perform(get("/api/v1/ogc/jobs/broken").accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isInternalServerError()) + .andExpect(jsonPath("$.message").value("Unable to retrieve download job status")) + .andExpect(jsonPath("$.message").value(org.hamcrest.Matchers.not( + org.hamcrest.Matchers.containsString("secret")))); + } + + @Test + void unsupportedJobsOperationsReturnNotImplemented() throws Exception { + mockMvc.perform(get("/api/v1/ogc/jobs").accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isNotImplemented()); + mockMvc.perform(get("/api/v1/ogc/jobs/{jobId}/results", JOB_ID).accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isNotImplemented()); + mockMvc.perform(delete("/api/v1/ogc/jobs/{jobId}", JOB_ID).accept(MediaType.APPLICATION_JSON)) + .andExpect(status().isNotImplemented()); + } +} diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiTest.java index 12a67788..d2de112d 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestApiTest.java @@ -3,6 +3,7 @@ import au.org.aodn.ogcapi.processes.model.Execute; import au.org.aodn.ogcapi.processes.model.InlineResponse200; import au.org.aodn.ogcapi.processes.model.Results; +import au.org.aodn.ogcapi.server.core.model.DownloadExecutionResponse; import au.org.aodn.ogcapi.server.core.model.InlineValue; import au.org.aodn.ogcapi.server.core.model.enumeration.DatasetDownloadEnums; import au.org.aodn.ogcapi.server.core.model.enumeration.InlineResponseKeyEnum; @@ -55,16 +56,17 @@ public void setUp() { @Test public void testExecuteDownloadDatasetSuccess() throws JsonProcessingException { when(restServices.downloadData(any(), any(), any(), any(), any(), any(), any(), any(), any(), any())) - .thenReturn(ResponseEntity.ok("Job submitted with ID: test-job-id")); + .thenReturn("test-job-id"); ResponseEntity response = restApi.execute(ProcessIdEnum.DOWNLOAD_DATASET.getValue(), executeRequest); assertEquals(200, response.getStatusCode().value()); - assertInstanceOf(Results.class, response.getBody()); - Results results = (Results) response.getBody(); + assertInstanceOf(DownloadExecutionResponse.class, response.getBody()); + DownloadExecutionResponse results = (DownloadExecutionResponse) response.getBody(); assert results != null; - InlineValue message = (InlineValue) results.get("message"); - assertEquals("Job submitted with ID: test-job-id", message.message()); + assertEquals("Job submitted with ID: test-job-id", results.message().message()); + assertEquals("200", results.status().message()); + assertEquals("test-job-id", results.jobId()); // The "processing started" email must go out only after AWS Batch returned a job id InOrder inOrder = inOrder(restServices); diff --git a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestServicesTest.java b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestServicesTest.java index 39acacf0..37c458be 100644 --- a/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestServicesTest.java +++ b/server/src/test/java/au/org/aodn/ogcapi/server/processes/RestServicesTest.java @@ -10,7 +10,6 @@ import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.MockitoAnnotations; -import org.springframework.http.ResponseEntity; import software.amazon.awssdk.services.batch.BatchClient; import software.amazon.awssdk.services.batch.model.SubmitJobRequest; import software.amazon.awssdk.services.batch.model.SubmitJobResponse; @@ -51,11 +50,11 @@ public void testDownloadDataSuccess() throws JsonProcessingException { when(objectMapper.writeValueAsString(any())).thenReturn("test-multipolygon"); // Act - ResponseEntity response = restServices.downloadData( + String response = restServices.downloadData( "test-uuid", "test-dname", "2023-01-01", "2023-01-31", "test-multipolygon", "test@example.com", "Test Ocean Data Collection", "https://metadata.imas.utas.edu.au/.../test-uuid-123", "Cite data as: Mazor, T., Watermeyer, K., Hobley, T., Grinter, V., Holden, R., MacDonald, K. and Ferns, L. (2023).", "geotiff"); // Assert - assertEquals(ResponseEntity.ok("Job submitted with ID: " + jobId), response); + assertEquals(jobId, response); verify(batchClient, times(1)).submitJob(any(SubmitJobRequest.class)); } @@ -81,7 +80,7 @@ public void testDownloadDataCapturesSubmitJobRequest() throws JsonProcessingExce when(objectMapper.writeValueAsString(any())).thenReturn("test-multipolygon"); // Act - ResponseEntity response = restServices.downloadData( + String response = restServices.downloadData( "test-uuid", "test-dname","2023-01-01", "2023-01-31", "non-specified", "test@example.com", "Test Ocean Data Collection", "https://metadata.imas.utas.edu.au/.../test-uuid-123", "Cite data as: Mazor, T., Watermeyer, K., Hobley, T., Grinter, V., Holden, R., MacDonald, K. and Ferns, L. (2023).", "geotiff"); // Capture the submitted request @@ -92,7 +91,12 @@ public void testDownloadDataCapturesSubmitJobRequest() throws JsonProcessingExce // Assert relevant parameters assertEquals("non-specified", captured.parameters().get("multi_polygon")); - assertEquals(ResponseEntity.ok("Job submitted with ID: " + jobId), response); + assertEquals("test-dname", captured.parameters().get("key")); + assertEquals("Test Ocean Data Collection", captured.parameters().get("collection_title")); + assertEquals("geotiff", captured.parameters().get("output_format")); + assertEquals("https://metadata.imas.utas.edu.au/.../test-uuid-123", + captured.parameters().get("full_metadata_link")); + assertEquals(jobId, response); } @Test @@ -104,7 +108,7 @@ public void submitJobReplacesEmptySuggestedCitationWithUnavailable() throws Json // polygons set to 'non-specified' to avoid objectMapper serialization // Act: pass empty suggestedCitation - ResponseEntity response = restServices.downloadData( + String response = restServices.downloadData( "test-uuid", "test-dname","2023-01-01", "2023-01-31", "non-specified", "test@example.com", "Test Ocean Data Collection", "https://metadata.imas.utas.edu.au/.../test-uuid-123", "", "geotiff"); @@ -115,7 +119,7 @@ public void submitJobReplacesEmptySuggestedCitationWithUnavailable() throws Json String suggestedKey = DatasetDownloadEnums.Parameter.SUGGESTED_CITATION.getValue(); assertEquals("unavailable", captured.parameters().get(suggestedKey)); - assertEquals(ResponseEntity.ok("Job submitted with ID: " + jobId), response); + assertEquals(jobId, response); } @Test