Store async upload job in the request's thread

IvanBorislavovDimitrov · IvanBorislavovDimitrov · commit 84f38cc80182 · 2023-12-15T17:25:41.000+02:00
diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/Messages.java
@@ -20,6 +20,7 @@ public final class Messages {
     public static final String CANNOT_PARSE_CONTAINER_URI_OF_OBJECT_STORE = "Cannot parse container_uri of object store";
     public static final String UNSUPPORTED_SERVICE_PLAN_FOR_OBJECT_STORE = "Unsupported service plan for object store!";
     public static final String REQUEST_0_1_FAILED_WITH_2 = "Request \"{0} {1}\" failed with \"{2}\"";
+    public static final String ERROR_OCCURRED_WHILE_DELETING_JOB_ENTRY = "Error occurred while deleting job entry";
 
     // Audit log messages
 
@@ -63,7 +64,7 @@ public final class Messages {
     public static final String CREATING_ASYNC_UPLOAD_JOB = "Creating async upload job for URL {} with ID: {}";
     public static final String ASYNC_UPLOAD_JOB_REJECTED = "Async upload job {} rejected. Deleting entry";
     public static final String STARTING_DOWNLOAD_OF_MTAR = "Starting download of MTAR from remote endpoint: {}";
-    public static final String UPLOADED_MTAR_FROM_REMOTE_ENDPOINT = "Uploaded MTAR from remote endpoint {} in {} ms";
+    public static final String UPLOADED_MTAR_FROM_REMOTE_ENDPOINT_AND_JOB_ID = "Uploaded MTAR from remote endpoint {} with job id: {} in {} ms";
     public static final String ASYNC_UPLOAD_JOB_FINISHED = "Async upload job {} finished";
     public static final String UPLOADING_MTAR_STREAM_FROM_REMOTE_ENDPOINT = "Uploading MTAR stream from remote endpoint: {}";
     public static final String CALLING_REMOTE_MTAR_ENDPOINT = "Calling remote MTAR endpoint {}";
diff --git a/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/FilesApiServiceImpl.java b/multiapps-controller-web/src/main/java/org/cloudfoundry/multiapps/controller/web/api/impl/FilesApiServiceImpl.java
@@ -113,13 +113,14 @@ public ResponseEntity<FileMetadata> uploadFile(MultipartHttpServletRequest reque
         var multipartFile = getFileFromRequest(request);
         try (InputStream in = new BufferedInputStream(multipartFile.getInputStream(), INPUT_STREAM_BUFFER_SIZE)) {
             var startTime = LocalDateTime.now();
-            FileEntry fileEntry = fileService.addFile(spaceGuid, namespace, multipartFile.getOriginalFilename(), in, multipartFile.getSize());
+            FileEntry fileEntry = fileService.addFile(spaceGuid, namespace, multipartFile.getOriginalFilename(), in,
+                                                      multipartFile.getSize());
             FileMetadata file = parseFileEntry(fileEntry);
             AuditLoggingProvider.getFacade()
                                 .logConfigCreate(file);
             var endTime = LocalDateTime.now();
-            LOGGER.trace(Messages.UPLOADED_FILE, file.getId(), file.getName(), file.getSize(), file.getDigest(),
-                         file.getDigestAlgorithm(), ChronoUnit.MILLIS.between(startTime, endTime));
+            LOGGER.trace(Messages.UPLOADED_FILE, file.getId(), file.getName(), file.getSize(), file.getDigest(), file.getDigestAlgorithm(),
+                         ChronoUnit.MILLIS.between(startTime, endTime));
             return ResponseEntity.status(HttpStatus.CREATED)
                                  .body(file);
         } catch (Exception e) {
@@ -145,19 +146,32 @@ public ResponseEntity<Void> startUploadFromUrl(String spaceGuid, String namespac
         var entry = createJobEntry(spaceGuid, namespace, urlWithoutUserInfo);
         LOGGER.debug(Messages.CREATING_ASYNC_UPLOAD_JOB, urlWithoutUserInfo, entry.getId());
         try {
+            uploadJobService.add(entry);
             deployFromUrlExecutor.execute(() -> uploadFileFromUrl(entry, spaceGuid, namespace, decodedUrl));
         } catch (RejectedExecutionException ignored) {
             LOGGER.debug(Messages.ASYNC_UPLOAD_JOB_REJECTED, entry.getId());
+            deleteAsyncJobEntry(entry);
             return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                                  .header(HttpHeaders.RETRY_AFTER, RETRY_AFTER_SECONDS)
                                  .build();
         }
         return ResponseEntity.accepted()
-                             .header("x-cf-app-instance", configuration.getApplicationGuid() + ":" + configuration.getApplicationInstanceIndex())
+                             .header("x-cf-app-instance",
+                                     configuration.getApplicationGuid() + ":" + configuration.getApplicationInstanceIndex())
                              .header(HttpHeaders.LOCATION, getLocationHeader(spaceGuid, entry.getId()))
                              .build();
     }
 
+    private void deleteAsyncJobEntry(AsyncUploadJobEntry entry) {
+        try {
+            uploadJobService.createQuery()
+                            .id(entry.getId())
+                            .delete();
+        } catch (Exception e) {
+            LOGGER.error(Messages.ERROR_OCCURRED_WHILE_DELETING_JOB_ENTRY, e);
+        }
+    }
+
     private String getLocationHeader(String spaceGuid, String jobId) {
         return "spaces/" + spaceGuid + "/files/jobs/" + jobId;
     }
@@ -259,8 +273,8 @@ private AsyncUploadJobEntry getJob(String id, String spaceGuid, String namespace
         try {
             return uploadJobService.createQuery()
                                    .id(id)
-                                   //even though the ID fully qualifies the job, we add these filters
-                                   //to prevent accessing a job from another space, namespace or a different user
+                                   // even though the ID fully qualifies the job, we add these filters
+                                   // to prevent accessing a job from another space, namespace or a different user
                                    .spaceGuid(spaceGuid)
                                    .user(SecurityContextUtil.getUsername())
                                    .namespace(namespace)
@@ -280,29 +294,32 @@ private AsyncUploadResult createErrorResult(String error) {
     private void uploadFileFromUrl(AsyncUploadJobEntry jobEntry, String spaceGuid, String namespace, String fileUrl) {
         var counter = new AtomicLong(0);
         jobCounters.put(jobEntry.getId(), counter);
+        LOGGER.debug(Messages.STARTING_DOWNLOAD_OF_MTAR, jobEntry.getUrl());
+        var startTime = LocalDateTime.now();
+        AsyncUploadJobEntry jobEntryWithTimestamp = ImmutableAsyncUploadJobEntry.copyOf(jobEntry)
+                                                                                .withState(State.RUNNING)
+                                                                                .withStartedAt(startTime);
         try {
-            uploadJobService.add(jobEntry);
-            LOGGER.debug(Messages.STARTING_DOWNLOAD_OF_MTAR, jobEntry.getUrl());
-            var startTime = LocalDateTime.now();
-            uploadJobService.update(jobEntry, ImmutableAsyncUploadJobEntry.copyOf(jobEntry)
-                                                                          .withState(State.RUNNING)
-                                                                          .withStartedAt(startTime));
-            FileEntry fileEntry = resilientOperationExecutor.execute((CheckedSupplier<FileEntry>) () -> doUploadFileFromUrl(spaceGuid, namespace, fileUrl, counter));
-            LOGGER.trace(Messages.UPLOADED_MTAR_FROM_REMOTE_ENDPOINT, jobEntry.getUrl(),
+            jobEntryWithTimestamp = uploadJobService.update(jobEntry, jobEntryWithTimestamp);
+            FileEntry fileEntry = resilientOperationExecutor.execute((CheckedSupplier<FileEntry>) () -> doUploadFileFromUrl(spaceGuid,
+                                                                                                                            namespace,
+                                                                                                                            fileUrl,
+                                                                                                                            counter));
+            LOGGER.trace(Messages.UPLOADED_MTAR_FROM_REMOTE_ENDPOINT_AND_JOB_ID, jobEntry.getUrl(), jobEntry.getId(),
                          ChronoUnit.MILLIS.between(startTime, LocalDateTime.now()));
 
             var descriptor = fileService.processFileContent(spaceGuid, fileEntry.getId(), this::extractDeploymentDescriptor);
             LOGGER.debug(Messages.ASYNC_UPLOAD_JOB_FINISHED, jobEntry.getId());
-            uploadJobService.update(jobEntry, ImmutableAsyncUploadJobEntry.copyOf(jobEntry)
-                                                                          .withFileId(fileEntry.getId())
-                                                                          .withMtaId(descriptor.getId())
-                                                                          .withFinishedAt(LocalDateTime.now())
-                                                                          .withState(State.FINISHED));
+            uploadJobService.update(jobEntryWithTimestamp, ImmutableAsyncUploadJobEntry.copyOf(jobEntryWithTimestamp)
+                                                                                       .withFileId(fileEntry.getId())
+                                                                                       .withMtaId(descriptor.getId())
+                                                                                       .withFinishedAt(LocalDateTime.now())
+                                                                                       .withState(State.FINISHED));
         } catch (Exception e) {
             LOGGER.error(MessageFormat.format(Messages.ASYNC_UPLOAD_JOB_FAILED, jobEntry.getId(), e.getMessage()), e);
-            uploadJobService.update(jobEntry, ImmutableAsyncUploadJobEntry.copyOf(jobEntry)
-                                                                          .withError(e.getMessage())
-                                                                          .withState(State.ERROR));
+            uploadJobService.update(jobEntryWithTimestamp, ImmutableAsyncUploadJobEntry.copyOf(jobEntryWithTimestamp)
+                                                                                       .withError(e.getMessage())
+                                                                                       .withState(State.ERROR));
         }
     }
 
@@ -325,11 +342,11 @@ private FileEntry doUploadFileFromUrl(String spaceGuid, String namespace, String
 
         String fileName = extractFileName(fileUrl);
         FileUtils.validateFileHasExtension(fileName);
-        counter.set(0); //reset counter on retry
+        counter.set(0); // reset counter on retry
         // Normal stream returned from the http response always returns 0 when InputStream::available() is executed which seems to break
         // JClods library: https://issues.apache.org/jira/browse/JCLOUDS-1623
         try (CountingInputStream source = new CountingInputStream(response.body(), counter);
-             BufferedInputStream bufferedContent = new BufferedInputStream(source, INPUT_STREAM_BUFFER_SIZE)) {
+            BufferedInputStream bufferedContent = new BufferedInputStream(source, INPUT_STREAM_BUFFER_SIZE)) {
             LOGGER.debug(Messages.UPLOADING_MTAR_STREAM_FROM_REMOTE_ENDPOINT, response.uri());
             return fileService.addFile(spaceGuid, namespace, fileName, bufferedContent, fileSize);
         }
@@ -342,8 +359,8 @@ private HttpResponse<InputStream> callRemoteEndpointWithRetry(HttpClient client,
             var response = client.send(request, BodyHandlers.ofInputStream());
             if (response.statusCode() / 100 != 2) {
                 String error = readErrorBodyFromResponse(response);
-                throw new SLException(MessageFormat.format(Messages.ERROR_FROM_REMOTE_MTAR_ENDPOINT, request.uri(),
-                                                           response.statusCode(), error));
+                throw new SLException(MessageFormat.format(Messages.ERROR_FROM_REMOTE_MTAR_ENDPOINT, request.uri(), response.statusCode(),
+                                                           error));
             }
             return response;
         });
diff --git a/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/FilesApiServiceImplTest.java b/multiapps-controller-web/src/test/java/org/cloudfoundry/multiapps/controller/web/api/impl/FilesApiServiceImplTest.java
@@ -1,6 +1,7 @@
 package org.cloudfoundry.multiapps.controller.web.api.impl;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
 import java.io.InputStream;
@@ -19,6 +20,8 @@
 import java.util.UUID;
 import java.util.concurrent.Executor;
 
+import javax.persistence.NoResultException;
+
 import org.apache.commons.lang3.RandomStringUtils;
 import org.cloudfoundry.multiapps.common.SLException;
 import org.cloudfoundry.multiapps.controller.api.model.AsyncUploadResult;
@@ -59,8 +62,6 @@
 import org.springframework.web.multipart.MultipartFile;
 import org.springframework.web.multipart.MultipartHttpServletRequest;
 
-import javax.persistence.NoResultException;
-
 class FilesApiServiceImplTest {
 
     @Mock
@@ -125,8 +126,9 @@ public void initialize() throws Exception {
             Runnable r = invocationOnMock.getArgument(0);
             r.run();
             return null;
-        }).when(asyncFileUploadExecutor)
-                .execute(Mockito.any());
+        })
+               .when(asyncFileUploadExecutor)
+               .execute(Mockito.any());
     }
 
     @Test
@@ -176,8 +178,8 @@ void testUploadMtaFile() throws Exception {
         Mockito.verify(file)
                .getInputStream();
         Mockito.verify(fileService)
-               .addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName),
-                        Mockito.any(InputStream.class), Mockito.eq(fileSize));
+               .addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName), Mockito.any(InputStream.class),
+                        Mockito.eq(fileSize));
 
         FileMetadata fileMetadata = response.getBody();
         assertMetadataMatches(fileEntry, fileMetadata);
@@ -206,14 +208,17 @@ void testUploadFileFromUrl() throws Exception {
                .thenReturn(jobEntry);
         Mockito.when(uploadJobService.createQuery())
                .thenReturn(query);
+        Mockito.when(uploadJobService.update(any(), any()))
+               .thenReturn(jobEntry);
 
-        Mockito.when(fileService.addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName),
-                                         Mockito.any(), Mockito.eq(20L)))
+        Mockito.when(fileService.addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName), Mockito.any(),
+                                         Mockito.eq(20L)))
                .thenReturn(fileEntry);
         Mockito.when(fileService.getFile(Mockito.eq(SPACE_GUID), Mockito.eq(fileEntry.getId())))
                .thenReturn(fileEntry);
 
-        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID, ImmutableFileUrl.of(FILE_URL));
+        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID,
+                                                                                  ImmutableFileUrl.of(FILE_URL));
 
         assertEquals(startUploadResponse.getStatusCode(), HttpStatus.ACCEPTED);
 
@@ -226,8 +231,7 @@ void testUploadFileFromUrl() throws Exception {
         assertEquals(uploadJobResponse.getStatusCode(), HttpStatus.CREATED);
 
         Mockito.verify(fileService)
-               .addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName), Mockito.any(),
-                        Mockito.eq(20L));
+               .addFile(Mockito.eq(SPACE_GUID), Mockito.eq(NAMESPACE_GUID), Mockito.eq(fileName), Mockito.any(), Mockito.eq(20L));
 
         var responseBody = uploadJobResponse.getBody();
         var fileMetadata = responseBody.getFile();
@@ -247,7 +251,7 @@ void testUploadFileFromUrlWithInvalidJobId() {
                .singleResult();
 
         Mockito.when(uploadJobService.createQuery())
-                .thenReturn(query);
+               .thenReturn(query);
 
         ResponseEntity<AsyncUploadResult> response = testedClass.getUploadFromUrlJob(SPACE_GUID, NAMESPACE_GUID, "invalid");
 
@@ -273,8 +277,11 @@ void testFileUrlDoesntReturnContentLength() throws Exception {
                .thenReturn(jobEntry);
         Mockito.when(uploadJobService.createQuery())
                .thenReturn(query);
+        Mockito.when(uploadJobService.update(any(), any()))
+               .thenReturn(jobEntry);
 
-        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID, ImmutableFileUrl.of(FILE_URL));
+        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID,
+                                                                                  ImmutableFileUrl.of(FILE_URL));
 
         assertEquals(startUploadResponse.getStatusCode(), HttpStatus.ACCEPTED);
 
@@ -312,8 +319,11 @@ void testFileUrlReturnsContentLengthAboveMaxUploadSize() throws Exception {
                .thenReturn(jobEntry);
         Mockito.when(uploadJobService.createQuery())
                .thenReturn(query);
+        Mockito.when(uploadJobService.update(any(), any()))
+               .thenReturn(jobEntry);
 
-        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID, ImmutableFileUrl.of(FILE_URL));
+        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID,
+                                                                                  ImmutableFileUrl.of(FILE_URL));
 
         assertEquals(startUploadResponse.getStatusCode(), HttpStatus.ACCEPTED);
 
@@ -331,7 +341,7 @@ void testFileUrlReturnsContentLengthAboveMaxUploadSize() throws Exception {
     }
 
     @ParameterizedTest
-    @ValueSource(strings = {"https://host.domain/path/file?query=true", "http://host.domain/path/file.mtar?query=true"})
+    @ValueSource(strings = { "https://host.domain/path/file?query=true", "http://host.domain/path/file.mtar?query=true" })
     void testUploadFileWithInvalidUrl(String url) throws Exception {
         AsyncUploadJobsQuery query = Mockito.mock(AsyncUploadJobsQuery.class, Answers.RETURNS_SELF);
         HttpHeaders headers = HttpHeaders.of(Map.of("Content-Length", List.of("20")), (a, b) -> true);
@@ -349,11 +359,14 @@ void testUploadFileWithInvalidUrl(String url) throws Exception {
                .thenReturn(jobEntry);
         Mockito.when(uploadJobService.createQuery())
                .thenReturn(query);
+        Mockito.when(uploadJobService.update(any(), any()))
+               .thenReturn(jobEntry);
 
         String invalidFileUrl = Base64.getUrlEncoder()
                                       .encodeToString(url.getBytes(StandardCharsets.UTF_8));
 
-        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID, ImmutableFileUrl.of(invalidFileUrl));
+        ResponseEntity<Void> startUploadResponse = testedClass.startUploadFromUrl(SPACE_GUID, NAMESPACE_GUID,
+                                                                                  ImmutableFileUrl.of(invalidFileUrl));
 
         assertEquals(startUploadResponse.getStatusCode(), HttpStatus.ACCEPTED);
 
@@ -400,6 +413,7 @@ private AsyncUploadJobEntry mockUploadJobEntry(String fileId, AsyncUploadJobEntr
         when(jobEntry.getStartedAt()).thenReturn(LocalDateTime.MIN);
         when(jobEntry.getFileId()).thenReturn(fileId);
         when(jobEntry.getState()).thenReturn(jobState);
+        when(jobEntry.getUrl()).thenReturn("https://artifactory.sap/mta");
         if (jobState == AsyncUploadJobEntry.State.ERROR) {
             when(jobEntry.getError()).thenReturn(error);
         }
