[#186091347] Create endpoints to clean up a run

src/main/java/com/dnastack/wes/api/RunFile.java

@@ -0,0 +1,26 @@
+package com.dnastack.wes.api;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.*;
+
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+@EqualsAndHashCode
+@ToString
+@Builder
+public class RunFile {
+
+    @JsonProperty(value = "file_type")
+    FileType fileType;
+
+    String path;
+
+    public enum FileType {
+        FINAL,
+        SECONDARY,
+        LOG
+    }
+
+}

src/main/java/com/dnastack/wes/api/RunFileDeletion.java

@@ -0,0 +1,13 @@
+package com.dnastack.wes.api;
+
+import com.fasterxml.jackson.annotation.JsonUnwrapped;
+
+public record RunFileDeletion(@JsonUnwrapped RunFile runFile, DeletionState state, @JsonUnwrapped ErrorResponse errorResponse) {
+
+    public enum DeletionState {
+        DELETED,
+        ASYNC,
+        FAILED
+    }
+
+}

src/main/java/com/dnastack/wes/api/RunFileDeletions.java

@@ -0,0 +1,5 @@
+package com.dnastack.wes.api;
+
+import java.util.List;
+
+public record RunFileDeletions(List<RunFileDeletion> deletions) {}

src/main/java/com/dnastack/wes/api/RunFiles.java

@@ -0,0 +1,5 @@
+package com.dnastack.wes.api;
+
+import java.util.List;
+
+public record RunFiles(List<RunFile> runFiles) {}

src/main/java/com/dnastack/wes/api/WesV1Controller.java

@@ -123,11 +123,28 @@ public RunId cancelRun(@PathVariable("runId") String runId) {
         return adapter.cancel(runId);
     }
 
+    @AuditActionUri("wes:run:files:list")
+    @PreAuthorize("@accessEvaluator.canAccessResource('/ga4gh/wes/v1/runs/' + #runId + '/files', 'wes:runs:read', 'wes')")
+    @GetMapping(value = "/runs/{run_id}/files", produces = { MediaType.APPLICATION_JSON_VALUE })
+    public RunFiles getRunFiles(@PathVariable("run_id") String runId) {
+        return adapter.getRunFiles(runId);
+    }
+
+    @AuditActionUri("wes:run:files:delete")
+    @PreAuthorize("@accessEvaluator.canAccessResource('/ga4gh/wes/v1/runs/' + #runId + '/files', 'wes:runs:write', 'wes')")
+    @DeleteMapping(value = "/runs/{run_id}/files", produces = { MediaType.APPLICATION_JSON_VALUE })
+    public RunFileDeletions deleteRunFiles(
+        @PathVariable("run_id") String runId,
+        @RequestParam(value = "async", required = false) boolean async
+    ) {
+        return adapter.deleteRunFiles(runId, async);
+    }
+
     @AuditActionUri("wes:run:stderr")
     @PreAuthorize("@accessEvaluator.canAccessResource('/ga4gh/wes/v1/runs/' + #runId, 'wes:runs:read', 'wes')")
     @GetMapping(value = "/runs/{runId}/logs/stderr", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
     public void getStderr(HttpServletResponse response, @RequestHeader HttpHeaders headers, @PathVariable String runId) throws IOException {
-        adapter.getLogBytes(response.getOutputStream(), runId,getRangeFromHeaders(response,headers));
+        adapter.getLogBytes(response.getOutputStream(), runId, getRangeFromHeaders(response, headers));
     }
 
     @AuditActionUri("wes:run:stderr")
@@ -139,7 +156,7 @@ public void getTaskStderr(
         @PathVariable String runId,
         @PathVariable String taskId
     ) throws IOException {
-        adapter.getLogBytes(response.getOutputStream(), runId, taskId, "stderr",getRangeFromHeaders(response,headers));
+        adapter.getLogBytes(response.getOutputStream(), runId, taskId, "stderr", getRangeFromHeaders(response, headers));
     }
 
     @AuditActionUri("wes:run:stdout")
@@ -151,7 +168,7 @@ public void getTaskStdout(
         @PathVariable String runId,
         @PathVariable String taskId
     ) throws IOException {
-        adapter.getLogBytes(response.getOutputStream(), runId, taskId, "stdout",getRangeFromHeaders(response,headers));
+        adapter.getLogBytes(response.getOutputStream(), runId, taskId, "stdout", getRangeFromHeaders(response, headers));
     }
 
     @AuditActionUri("wes:run:stderr")
@@ -164,7 +181,7 @@ public void getTaskStderr(
         @PathVariable String taskName,
         @PathVariable int index
     ) throws IOException {
-        adapter.getLogBytes(response.getOutputStream(), runId, taskName, index, "stderr",getRangeFromHeaders(response,headers));
+        adapter.getLogBytes(response.getOutputStream(), runId, taskName, index, "stderr", getRangeFromHeaders(response, headers));
     }
 
     @AuditActionUri("wes:run:stdout")
@@ -180,9 +197,9 @@ public void getTaskStdout(
         adapter.getLogBytes(response.getOutputStream(), runId, taskName, index, "stdout", getRangeFromHeaders(response, headers));
     }
 
-    private HttpRange getRangeFromHeaders(HttpServletResponse response, HttpHeaders headers){
+    private HttpRange getRangeFromHeaders(HttpServletResponse response, HttpHeaders headers) {
         List<HttpRange> ranges = headers.getRange();
-        if (ranges.isEmpty()){
+        if (ranges.isEmpty()) {
             return null;
         } else if (ranges.size() > 1) {
             // only return the first range parsed

src/main/java/com/dnastack/wes/config/AsyncConfig.java

@@ -0,0 +1,31 @@
+package com.dnastack.wes.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Slf4j
+@Configuration
+public class AsyncConfig {
+
+    @Bean
+    public ThreadPoolTaskExecutor defaultAsyncOperationExecutor(
+        @Value("${app.executors.default.core-pool-size:8}") int corePoolSize,
+        @Value("${app.executors.default.max-pool-size:16}") int maxPoolSize,
+        @Value("${app.executors.default.queue-capacity:5000}") int queueCapacity
+    ) {
+        final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(corePoolSize);
+        executor.setMaxPoolSize(maxPoolSize);
+        executor.setQueueCapacity(queueCapacity);
+        executor.setThreadNamePrefix("defaultAsyncOp-");
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.initialize();
+        return executor;
+    }
+
+}

src/main/java/com/dnastack/wes/cromwell/CromwellService.java

@@ -13,13 +13,16 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
 import feign.FeignException;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.http.HttpRange;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 
@@ -31,6 +34,7 @@
 import java.nio.file.Paths;
 import java.util.*;
 import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -51,6 +55,7 @@ public class CromwellService {
     private final PathTranslatorFactory pathTranslatorFactory;
     private final CromwellWesMapper cromwellWesMapper;
     private final CromwellConfig cromwellConfig;
+    private final ThreadPoolTaskExecutor executor;
 
     private final AppConfig appConfig;
 
@@ -61,14 +66,16 @@ public class CromwellService {
         PathTranslatorFactory pathTranslatorFactory,
         CromwellWesMapper cromwellWesMapper,
         AppConfig appConfig,
-        CromwellConfig config
+        CromwellConfig config,
+        @Qualifier("defaultAsyncOperationExecutor") ThreadPoolTaskExecutor executor
     ) {
         this.client = cromwellClient;
         this.pathTranslatorFactory = pathTranslatorFactory;
         this.storageClient = storageClient;
         this.cromwellWesMapper = cromwellWesMapper;
         this.appConfig = appConfig;
         this.cromwellConfig = config;
+        this.executor = executor;
     }
 
 
@@ -211,13 +218,81 @@ private CromwellStatus getStatus(String runId) {
      *
      * @param runId The cromwell id
      *
-     * @return a complete run log
+     * @return the cromwell id
      */
     public RunId cancel(String runId) {
         client.abortWorkflow(runId);
         return RunId.builder().runId(runId).build();
     }
 
+    /**
+     * Get the files for a specific run.
+     *
+     * @param runId The cromwell id
+     *
+     * @return a list of generated files for the run
+     */
+    public RunFiles getRunFiles(String runId) throws NotFoundException {
+        CromwellMetadataResponse metadataResponse = getMetadata(runId);
+        Set<String> finalFileSet = new HashSet<>();
+        Set<String> secondaryFileSet = new HashSet<>();
+        Set<String> logFileSet = new HashSet<>();
+        List<RunFile> files = new ArrayList<>();
+
+        Map<String, Object> outputs = metadataResponse.getOutputs();
+        if (!outputs.isEmpty()) {
+            outputs.values().forEach(output -> extractFilesFromValue(finalFileSet, output));
+        }
+        extractSecondaryLogFiles(secondaryFileSet, logFileSet, metadataResponse);
+
+        finalFileSet.forEach(path -> files.add(new RunFile(RunFile.FileType.FINAL, path)));
+        secondaryFileSet.forEach(path -> {
+            if (!finalFileSet.contains(path) && !logFileSet.contains(path)) {
+                files.add(new RunFile(RunFile.FileType.SECONDARY, path));
+            }
+        });
+        logFileSet.forEach(path -> {
+            if (!finalFileSet.contains(path)) {
+                files.add(new RunFile(RunFile.FileType.LOG, path));
+            }
+        });
+        return new RunFiles(files);
+    }
+
+    /**
+     * Request to delete the files associated with the run.
+     *
+     * @param runId The cromwell id
+     *
+     * @return the cromwell id
+     */
+    public RunFileDeletions deleteRunFiles(String runId, boolean async) {
+        List<RunFile> files = getRunFiles(runId).runFiles();
+        List<RunFileDeletion> outcomes = files.stream().filter(runFile -> RunFile.FileType.SECONDARY.equals(runFile.getFileType()))
+            .map(runFile -> {
+                if (async) {
+                    return deleteRunFileAsync(runFile);
+                } else {
+                    return deleteRunFile(runFile);
+                }
+            }).toList();
+        return new RunFileDeletions(outcomes);
+    }
+
+    public RunFileDeletion deleteRunFileAsync(RunFile runFile) {
+        CompletableFuture.runAsync(() -> deleteRunFile(runFile), executor);
+        return new RunFileDeletion(runFile, RunFileDeletion.DeletionState.ASYNC,null);
+    }
+
+    public RunFileDeletion deleteRunFile(RunFile runFile) {
+        try {
+            storageClient.deleteFile(runFile.getPath());
+            return new RunFileDeletion(runFile, RunFileDeletion.DeletionState.DELETED, null);
+        } catch (IOException e) {
+            return new RunFileDeletion(runFile, RunFileDeletion.DeletionState.FAILED, ErrorResponse.builder().errorCode(400).msg(e.getMessage()).build());
+        }
+    }
+
     public void getLogBytes(OutputStream outputStream, String runId, String taskId, String logKey, HttpRange range) throws IOException {
         String logPath = getLogPath(runId, taskId, logKey);
 
@@ -448,6 +523,63 @@ private JsonNode extractJsonNode(String value) throws IOException {
         }
     }
 
+    private void extractSecondaryLogFiles(Set<String> secondaryFileSet, Set<String> logFileSet, CromwellMetadataResponse metadataResponse){
+        Map<String, Object> outputs = metadataResponse.getOutputs();
+        if (outputs != null && !outputs.isEmpty()) {
+            outputs.values().forEach(output -> extractFilesFromValue(secondaryFileSet, output));
+        }
+        Map<String, List<CromwellTaskCall>> calls = metadataResponse.getCalls();
+        if (calls != null && !calls.isEmpty()) {
+            calls.values().stream().flatMap(List::stream).forEach(call -> extractSecondaryLogFilesFromCall(secondaryFileSet, logFileSet, call));
+        }
+    }
+
+    private void extractSecondaryLogFilesFromCall(Set<String> secondaryFileSet, Set<String> logFileSet, CromwellTaskCall call){
+        Map<String, Object> outputs = call.getOutputs();
+        if (outputs != null && !outputs.isEmpty()) {
+            outputs.values().forEach(output -> extractFilesFromValue(secondaryFileSet, output));
+        }
+        String stderr = call.getStderr();
+        String stdout = call.getStdout();
+        if (stderr != null && storageClient.isFile(stderr)) {
+            logFileSet.add(stderr);
+        }
+        if (stdout != null && storageClient.isFile(stdout)) {
+            logFileSet.add(stdout);
+        }
+        Map<String, String> backendLogs = call.getBackendLogs();
+        if (backendLogs != null && !backendLogs.isEmpty()) {
+            backendLogs.values().forEach(log -> extractFilesFromValue(logFileSet, log));
+        }
+        CromwellMetadataResponse subWorkflowMetadata = call.getSubWorkflowMetadata();
+        if (subWorkflowMetadata != null) {
+            extractSecondaryLogFiles(secondaryFileSet,logFileSet,subWorkflowMetadata);
+        }
+    }
+
+    private void extractFilesFromValue(Set<String> fileSet, Object output) {
+        JsonNode node = mapper.valueToTree(output);
+        if (node.isTextual()) {
+            if (storageClient.isFile(node.asText())) {
+                fileSet.add(node.asText());
+            }
+        } else if (node.isArray()) {
+            ArrayNode arrayOutput = mapper.valueToTree(output);
+            extractFilesFromArrayNode(fileSet, arrayOutput);
+        } else if (node.isObject()) {
+            ObjectNode objectOutput = mapper.valueToTree(output);
+            extractFilesFromObjectNode(fileSet, objectOutput);
+        }
+    }
+
+    private void extractFilesFromArrayNode(Set<String> fileSet, ArrayNode outputs) {
+        outputs.forEach(output -> extractFilesFromValue(fileSet, output));
+    }
+
+    private void extractFilesFromObjectNode(Set<String> fileSet, ObjectNode outputs) {
+        outputs.forEach(output -> extractFilesFromValue(fileSet, output));
+    }
+
     private void setWorkflowSourceAndDependencies(Path tempDirectory, RunRequest runRequest, CromwellExecutionRequest cromwellRequest) throws IOException {
         if (runRequest.getWorkflowAttachments() == null || runRequest.getWorkflowAttachments().length == 0) {
             throw new InvalidRequestException("Url provided is relative however no workflowAttachments are defined");

src/main/java/com/dnastack/wes/storage/AzureBlobStorageClient.java

@@ -20,7 +20,6 @@
 
 public class AzureBlobStorageClient implements BlobStorageClient {
 
-
     private final BlobServiceClient client;
     private final long signedUrlTtl;
     private final String container;
@@ -49,7 +48,6 @@ public AzureBlobStorageClient(AzureBlobStorageClientConfig config) {
         stagingPath = config.getStagingPath();
     }
 
-
     @Override
     public URL getSignedUrl(String blobUri) {
         BlobUrlParts parts = BlobUrlParts.parse(blobUri);
@@ -99,7 +97,6 @@ public void readBytes(OutputStream outputStream, String blobUri, HttpRange httpR
             containerName = container;
         }
 
-
         BlobContainerClient containerClient = client.getBlobContainerClient(containerName);
         BlobClient blobClient = containerClient.getBlobClient(blobName);
 
@@ -122,4 +119,18 @@ public void readBytes(OutputStream outputStream, String blobUri, HttpRange httpR
             .setMaxRetryRequests(3), null, false, null, null);
     }
 
+    @Override
+    public boolean isFile(String filePath) {
+        try {
+            return client.getBlobContainerClient(container).getBlobClient(filePath).exists();
+        } catch (IllegalArgumentException e) {
+            return false;
+        }
+    }
+
+    @Override
+    public void deleteFile(String filePath) {
+        client.getBlobContainerClient(container).getBlobClient(filePath).delete();
+    }
+
 }