Bug 570871 added multi-threaded service execution Change-Id: I41cd1ed746cdab40feb60d8b374ad008d046ff32 Signed-off-by: Mahmoud Bazzal <mahmood1994ha@gmail.com>
diff --git a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/CloudServiceExecutionRunnable.java b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/CloudServiceExecutionRunnable.java index 4c25288..f5574c2 100644 --- a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/CloudServiceExecutionRunnable.java +++ b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/CloudServiceExecutionRunnable.java
@@ -9,27 +9,18 @@ * * Contributors: * Robert Bosch GmbH - initial API and implementation + * Dortmund University of Applied Sciences and Arts - Bug 570871 ******************************************************************************** */ package org.eclipse.app4mc.cloud.manager; -import java.io.File; -import java.nio.file.Files; import java.nio.file.Path; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import org.apache.http.HttpStatus; import org.eclipse.app4mc.cloud.manager.ProcessLog.Action; -import org.eclipse.app4mc.cloud.manager.administration.CloudServiceDefinition; import org.eclipse.app4mc.cloud.manager.storage.StorageService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.messaging.simp.SimpMessageSendingOperations; -import org.springframework.util.StringUtils; - -import kong.unirest.HttpResponse; -import kong.unirest.MultipartBody; -import kong.unirest.Unirest; public class CloudServiceExecutionRunnable implements Runnable { @@ -39,6 +30,8 @@ private String uuid; private String originalFilename; private WorkflowStatus workflowStatus; + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); public CloudServiceExecutionRunnable( StorageService storageService, @@ -60,7 +53,15 @@ try { Path inputFile = this.storageService.load(this.uuid, this.originalFilename); - processServiceNode(this.workflowStatus.getServiceRootNode(), inputFile); + ServiceNodeProcessingTask rootTask = new ServiceNodeProcessingTask(this.workflowStatus.getServiceRootNode(), + inputFile, storageService, messagingTemplate, uuid, workflowStatus); + + logger.info("started root task"); + rootTask.fork(); + logger.info("waiting for root task to finish"); + // join the task to indicate the entire workflow is done + rootTask.join(); + logger.info("finished root task"); } finally { this.workflowStatus.done(); WorkflowStatusHelper.saveWorkflowStatus( @@ -96,310 +97,4 @@ } } - private Path processServiceNode(ServiceNode serviceNode, Path inputFile) { - Path nodeResult = inputFile; - - if (!serviceNode.isStructuralNode()) { - nodeResult = executeCloudService(serviceNode, inputFile); - } - - Path childInput = nodeResult; - for (ServiceNode node : serviceNode.getChildren()) { - // if an error occurred stop the workflow - if (!this.workflowStatus.getErrors().isEmpty()) { - break; - } - - try { - if (serviceNode.isStructuralNode()) { - // inside a structural node the services are executed in sequential order, - // passing the result of the previous service to the next one - - // TODO if the structural node is not the root node, open a new thread for the execution - - childInput = processServiceNode(node, childInput); - } else { - // if the node is a service node, all child services should get the result of - // the parent service - - // TODO if serviceNode.getChildren().size() > 1 open new thread for each child - - processServiceNode(node, childInput); - } - } catch (ProcessingFailedException e) { - this.workflowStatus.addError(e.getMessage()); - } - - if (this.workflowStatus.isCancelled()) { - addMessage(this.workflowStatus, "Workflow cancelled by user"); - break; - } - } - - return nodeResult; - } - - private Path executeCloudService(ServiceNode node, Path inputFile) { - try { - CloudServiceDefinition csd = node.getService(); - ServiceConfiguration config = node.getServiceConfiguration(); - String serviceName = csd.getName(); - String baseUrl = csd.getBaseUrl(); - - // upload to service - MultipartBody multipartBody = Unirest.post(baseUrl) - .field("file", Files.newInputStream(inputFile), inputFile.getFileName().toString()); - - if (config != null) { - config.getParameterList().forEach(param -> { - if (!StringUtils.isEmpty(param.getValue()) && !param.isManagerParameter()) { - if ("multiple".equals(param.getCardinality())) { - multipartBody.field(param.getKey(), Arrays.asList(param.getValue().split(","))); - } else { - multipartBody.field(param.getKey(), param.getValue()); - } - } - }); - } - - HttpResponse<?> uploadResponse = multipartBody.asEmpty(); - - // extract status link from result - String statusUrl = null; - if (uploadResponse.getStatus() == 201) { - statusUrl = uploadResponse.getHeaders().getFirst("Location"); - if (StringUtils.isEmpty(statusUrl)) { - // fallback check for Link header if Location header is not set - statusUrl = HeaderHelper.getUrlFromLink(uploadResponse.getHeaders().get("Link"), "status", baseUrl); - } else if (!HeaderHelper.isValid(statusUrl, baseUrl)) { - throw new ProcessingFailedException( - "The status URL from the Location header does not match with the service base URL '" + baseUrl + "': " + statusUrl); - } - } else if (uploadResponse.getStatus() == 200) { - // fallback if return code is 200, then follow up needs to be placed in Link header - statusUrl = HeaderHelper.getUrlFromLink(uploadResponse.getHeaders().get("Link"), "status", baseUrl); - } else { - // error - Object body = uploadResponse.getBody(); - if (body != null && !body.toString().isEmpty()) { - workflowStatus.addError("Upload to " + serviceName + " failed! Error code: " + uploadResponse.getStatus() + " - " + body +" - Workflow stopped!"); - } else { - workflowStatus.addError("Upload to " + serviceName + " failed! Error code: " + uploadResponse.getStatus() + " - Workflow stopped!"); - } - return null; - } - - addMessage(workflowStatus, "Upload to " + serviceName + " service succeeded"); - - Path result = inputFile; - - if (workflowStatus.isCancelled()) { - return null; - } - - // trigger status URL until process is finished or error occured - if (statusUrl != null) { - HttpResponse<?> statusResponse = Unirest.get(statusUrl).asEmpty(); - List<String> linkHeaders = HeaderHelper.getLinkHeaders(statusResponse); - - long start = System.currentTimeMillis(); - long end = System.currentTimeMillis(); - - long timeout = 60_000l; - if (config != null) { - ServiceConfigurationParameter timeoutParam = config.getParameter("timeout"); - if (timeoutParam != null) { - try { - timeout = Long.valueOf(timeoutParam.getValue()); - } catch (Exception e) { - addMessage(workflowStatus, serviceName + " timeout value " + timeoutParam.getValue() + " invalid. Using default of 60_000"); - timeout = 60_000l; - } - } - } - - while (linkHeaders.size() <= 1) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - // Restore interrupted state... - Thread.currentThread().interrupt(); - } - - statusResponse = Unirest.get(statusUrl).asEmpty(); - linkHeaders = HeaderHelper.getLinkHeaders(statusResponse); - - end = System.currentTimeMillis(); - - // don't request for more than configured timeout - if (timeout > 0 && (end - start) > timeout) { - addMessage(workflowStatus, serviceName + " status requested for " + (timeout/1000) + " seconds"); - break; - } - - if (workflowStatus.isCancelled()) { - // TODO trigger cancel at service? - return null; - } - - // send a processing message every 5 seconds to the client to indicate that the - // service is still progressing - if ((((end - start) / 1000) % 5) == 0) { - this.messagingTemplate.convertAndSend( - "/topic/process-updates/" + this.uuid, - new ProcessLog(Action.PROCESSING, serviceName + " is processing ...")); - - } - } - - // create the service result directory relative to the input file - Path serviceSubDir = getServiceDir(node); - Files.createDirectories(serviceSubDir); - - String servicePath = storageService.load(workflowStatus.getUuid()).relativize(serviceSubDir).toString(); - // replace backslashes with forward slashes in case the service is executed on Windows - servicePath = servicePath.replaceAll("\\\\", "/"); - if (!servicePath.isEmpty()) { - servicePath += "/"; - } - - // first check if there is a result link - List<String> resultUrls = HeaderHelper.getUrlsFromLink(linkHeaders, "result", baseUrl); - - String deleteUrl = null; - if (!resultUrls.isEmpty()) { - - addMessage(workflowStatus, serviceName + " processing finished"); - - boolean error = false; - for (String resultUrl : resultUrls) { - - // download file - HttpResponse<File> resultResponse = Unirest.get(resultUrl) - .asFile(serviceSubDir.resolve("result").toString()); - - Path migrationResult = resultResponse.getBody().toPath(); - String filename = HeaderHelper.getFilenameFromHeader(resultResponse.getHeaders().getFirst("Content-Disposition")); - if (filename != null) { - migrationResult = Files.move(migrationResult, serviceSubDir.resolve(filename)); - } - - String resultName = resultResponse.getHeaders().getFirst("x-app4mc-result-name"); - if (StringUtils.isEmpty(resultName)) { - resultName = serviceName; - } - - addMessage(workflowStatus, resultName + " result downloaded"); - - workflowStatus.addResult( - resultName + " Result", - servicePath + migrationResult.getFileName().toString()); - - // extract header information if result should be used in workflow - String useResult = resultResponse.getHeaders().getFirst("x-app4mc-use-result"); - if (useResult == null || !useResult.toLowerCase().equals("false")) { - // the result should be used in the workflow - result = migrationResult; - } - - if (resultResponse.getStatus() != HttpStatus.SC_OK) { - error = true; - } - - if (deleteUrl == null) { - // extract delete - deleteUrl = HeaderHelper.getUrlFromLink(resultResponse.getHeaders().get("Link"), "delete", baseUrl); - } - } - - if (error) { - workflowStatus.addError(serviceName + " failed with errors"); - } else { - addMessage(workflowStatus, serviceName + " successfull"); - } - } else { - String errorUrl = HeaderHelper.getUrlFromLink(linkHeaders, "error", baseUrl); - if (errorUrl != null) { - workflowStatus.addError(serviceName + " processing finished with error"); - - // download error file - HttpResponse<File> errorResponse = Unirest.get(errorUrl) - .asFile(serviceSubDir.resolve("error").toString()); - - Path migrationError = errorResponse.getBody().toPath(); - String filename = HeaderHelper.getFilenameFromHeader(errorResponse.getHeaders().getFirst("Content-Disposition")); - if (filename != null) { - migrationError = Files.move(migrationError, serviceSubDir.resolve(filename)); - } - - addMessage(workflowStatus, serviceName + " error result downloaded"); - - workflowStatus.addResult( - serviceName + " Error", - servicePath + migrationError.getFileName().toString()); - - // extract delete - deleteUrl = HeaderHelper.getUrlFromLink(errorResponse.getHeaders().get("Link"), "delete", baseUrl); - } else { - workflowStatus.addError(serviceName + " has no result and no error"); - } - } - - boolean deleteResult = true; - if (config != null) { - ServiceConfigurationParameter deleteParam = config.getParameter("deleteResult"); - if (deleteParam != null) { - deleteResult = Boolean.valueOf(deleteParam.getValue()); - } - } - - if (deleteUrl != null && deleteResult) { - // delete upload again - Unirest.delete(deleteUrl).asEmpty(); - addMessage(workflowStatus, serviceName + " cleaned up"); - } - - } else { - // no status URL in the upload response, stop further processing - workflowStatus.addMessage("No status URL found for service " + serviceName); - return inputFile; - } - - return result; - - } catch (Exception e) { - throw new ProcessingFailedException("Error in " + node.getService().getName() + " workflow: " + e.getMessage(), e); - } - } - - private Path getServiceDir(ServiceNode node) { - Path path = storageService.load(workflowStatus.getUuid()); - ServiceNode sn = node; - ArrayList<String> segments = new ArrayList<>(); - segments.add(getServiceNodeDirName(sn)); - while (sn.getParentNode() != null && !sn.getParentNode().getId().equals(WorkflowStatus.ROOT_NODE_ID)) { - sn = sn.getParentNode(); - segments.add(0, getServiceNodeDirName(sn)); - } - - for (String segment : segments) { - path = path.resolve(segment); - } - - return path; - } - - private String getServiceNodeDirName(ServiceNode node) { - if (node.getService() != null) { - return "_" + node.getService().getKey().toLowerCase(); - } - return "_" + node.getId().toLowerCase(); - } - - private void addMessage(WorkflowStatus workflowStatus, String message) { - workflowStatus.addMessage(message); - this.messagingTemplate.convertAndSend( - "/topic/process-updates/" + this.uuid, - new ProcessLog(Action.UPDATE, message)); - } }
diff --git a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNode.java b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNode.java index 3f25fb6..36e6dc8 100644 --- a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNode.java +++ b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNode.java
@@ -30,6 +30,7 @@ private ServiceNode parentNode; + private boolean failed = false; /** * Create a structural node like for example the root node. */ @@ -193,4 +194,12 @@ public String toString() { return this.id; } + + public void markFailed() { + this.failed = true; + } + + public boolean isFailed() { + return this.failed; + } }
diff --git a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNodeProcessingTask.java b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNodeProcessingTask.java new file mode 100644 index 0000000..92b9b4b --- /dev/null +++ b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/ServiceNodeProcessingTask.java
@@ -0,0 +1,429 @@ +/********************************************************************************* + * Copyright (c) 2020, 2021 Robert Bosch GmbH and others. + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + * + * Contributors: + * Robert Bosch GmbH - initial API and implementation + * Dortmund University of Applied Sciences and Arts - Bug 570871 + ******************************************************************************** + */ +package org.eclipse.app4mc.cloud.manager; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ForkJoinTask; +import java.util.concurrent.RecursiveTask; + +import org.apache.http.HttpStatus; +import org.eclipse.app4mc.cloud.manager.ProcessLog.Action; +import org.eclipse.app4mc.cloud.manager.administration.CloudServiceDefinition; +import org.eclipse.app4mc.cloud.manager.storage.StorageService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.messaging.simp.SimpMessageSendingOperations; +import org.springframework.util.StringUtils; + +import kong.unirest.HttpResponse; +import kong.unirest.MultipartBody; +import kong.unirest.Unirest; + +public class ServiceNodeProcessingTask extends RecursiveTask<Path> { + + private static final long serialVersionUID = 8286285942428911612L; + + private final String uuid; + private final ServiceNode serviceNode; + private final Path inputFile; + + private final StorageService storageService; + private final SimpMessageSendingOperations messagingTemplate; + private WorkflowStatus workflowStatus; + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + public ServiceNodeProcessingTask(ServiceNode n, Path inputFile, StorageService storageService, + SimpMessageSendingOperations messagingTemplate, String uuid, WorkflowStatus ws) { + this.storageService = storageService; + this.messagingTemplate = messagingTemplate; + this.workflowStatus = ws; + this.serviceNode = n; + this.inputFile = inputFile; + this.uuid = uuid; + } + + @Override + protected Path compute() { + try { + if (this.serviceNode.isStructuralNode()) { + logger.info("created structural node thread: {}", this.serviceNode.getId()); + Path retval = processStructuralNode(this.serviceNode, this.inputFile); + return retval; + } else { + logger.info("created branch node thread: {}", this.serviceNode.getId()); + Path retval = processNonStructuralNode(this.serviceNode, this.inputFile); + if (this.workflowStatus.isCancelled()) { + addMessage(this.workflowStatus, "Workflow cancelled by user"); + } + return retval; + } + } catch (ProcessingFailedException e) { + this.workflowStatus.addError(e.getMessage()); + return null; + } + + } + + private Path processStructuralNode(ServiceNode node, Path inputFile) { + Path nextInput = inputFile; + for (ServiceNode n : node.getChildren()) { + try { + nextInput = processNonStructuralNode(n, nextInput); + // if an error occurs without throwing an exception, successor tasks do not get executed + if (n.isFailed()) { + break; + } + if (this.workflowStatus.isCancelled()) { + addMessage(this.workflowStatus, "Workflow cancelled by user"); + break; + } + } catch (ProcessingFailedException e) { + this.workflowStatus.addError(e.getMessage()); + break; + } + } + return nextInput; + } + + private Path processNonStructuralNode(ServiceNode node, Path inputFile) { + if (node.getChildren().isEmpty()) { + return processLeafNode(node, inputFile); + } else if (node.getChildren().size() > 1) { + try { + Path currOutput = executeCloudService(node, inputFile); + // if an error occurs without throwing an exception, successor tasks do not get executed + if (node.isFailed()) { + return null; + } + List<ServiceNodeProcessingTask> lActiveProcessingTask = new ArrayList<>(); + for (ServiceNode n : node.getChildren()) { + lActiveProcessingTask.add(new ServiceNodeProcessingTask(n, currOutput, storageService, + messagingTemplate, uuid, workflowStatus)); + } + logger.info("launching new branches"); + ForkJoinTask.invokeAll(lActiveProcessingTask); + return currOutput; + } catch (ProcessingFailedException e) { + this.workflowStatus.addError(e.getMessage()); + return null; + } + } else { + try { + Path currOutput = executeCloudService(node, inputFile); + // if an error occurs without throwing an exception, successor tasks do not get executed + if (node.isFailed()) { + return null; + } + ServiceNode nextNode = node.getChildren().get(0); + if (nextNode.isStructuralNode()) { + ServiceNodeProcessingTask nextNodeTask = new ServiceNodeProcessingTask(nextNode, currOutput, + storageService, messagingTemplate, uuid, workflowStatus); + ForkJoinPool.commonPool().invoke(nextNodeTask); + } else { + processNonStructuralNode(nextNode, currOutput); + } + return currOutput; + } catch (ProcessingFailedException e) { + this.workflowStatus.addError(e.getMessage()); + return null; + } + } + } + + private Path processLeafNode(ServiceNode node, Path inputFile) { + return executeCloudService(node, inputFile); + } + + private Path executeCloudService(ServiceNode node, Path inputFile) { + try { + CloudServiceDefinition csd = node.getService(); + ServiceConfiguration config = node.getServiceConfiguration(); + String serviceName = csd.getName(); + String baseUrl = csd.getBaseUrl(); + + // upload to service + MultipartBody multipartBody = Unirest.post(baseUrl) + .field("file", Files.newInputStream(inputFile), inputFile.getFileName().toString()); + + if (config != null) { + config.getParameterList().forEach(param -> { + if (!StringUtils.isEmpty(param.getValue()) && !param.isManagerParameter()) { + if ("multiple".equals(param.getCardinality())) { + multipartBody.field(param.getKey(), Arrays.asList(param.getValue().split(","))); + } else { + multipartBody.field(param.getKey(), param.getValue()); + } + } + }); + } + + HttpResponse<?> uploadResponse = multipartBody.asEmpty(); + + // extract status link from result + String statusUrl = null; + if (uploadResponse.getStatus() == 201) { + statusUrl = uploadResponse.getHeaders().getFirst("Location"); + if (StringUtils.isEmpty(statusUrl)) { + // fallback check for Link header if Location header is not set + statusUrl = HeaderHelper.getUrlFromLink(uploadResponse.getHeaders().get("Link"), "status", baseUrl); + } else if (!HeaderHelper.isValid(statusUrl, baseUrl)) { + throw new ProcessingFailedException( + "The status URL from the Location header does not match with the service base URL '" + baseUrl + "': " + statusUrl); + } + } else if (uploadResponse.getStatus() == 200) { + // fallback if return code is 200, then follow up needs to be placed in Link header + statusUrl = HeaderHelper.getUrlFromLink(uploadResponse.getHeaders().get("Link"), "status", baseUrl); + } else { + // error + Object body = uploadResponse.getBody(); + if (body != null && !body.toString().isEmpty()) { + workflowStatus.addError("Upload to " + serviceName + " failed! Error code: " + uploadResponse.getStatus() + " - " + body +" - Workflow stopped!"); + node.markFailed(); + } else { + workflowStatus.addError("Upload to " + serviceName + " failed! Error code: " + uploadResponse.getStatus() + " - Workflow stopped!"); + node.markFailed(); + } + return null; + } + + addMessage(workflowStatus, "Upload to " + serviceName + " service succeeded"); + + Path result = inputFile; + + if (workflowStatus.isCancelled()) { + return null; + } + + // trigger status URL until process is finished or error occured + if (statusUrl != null) { + HttpResponse<?> statusResponse = Unirest.get(statusUrl).asEmpty(); + List<String> linkHeaders = HeaderHelper.getLinkHeaders(statusResponse); + + long start = System.currentTimeMillis(); + long end = System.currentTimeMillis(); + + long timeout = 60_000l; + if (config != null) { + ServiceConfigurationParameter timeoutParam = config.getParameter("timeout"); + if (timeoutParam != null) { + try { + timeout = Long.valueOf(timeoutParam.getValue()); + } catch (Exception e) { + addMessage(workflowStatus, serviceName + " timeout value " + timeoutParam.getValue() + " invalid. Using default of 60_000"); + timeout = 60_000l; + } + } + } + + while (linkHeaders.size() <= 1) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // Restore interrupted state... + Thread.currentThread().interrupt(); + } + + statusResponse = Unirest.get(statusUrl).asEmpty(); + linkHeaders = HeaderHelper.getLinkHeaders(statusResponse); + + end = System.currentTimeMillis(); + + // don't request for more than configured timeout + if (timeout > 0 && (end - start) > timeout) { + addMessage(workflowStatus, serviceName + " status requested for " + (timeout/1000) + " seconds"); + break; + } + + if (workflowStatus.isCancelled()) { + // TODO trigger cancel at service? + return null; + } + + // send a processing message every 5 seconds to the client to indicate that the + // service is still progressing + if ((((end - start) / 1000) % 5) == 0) { + synchronized (this.messagingTemplate) { + this.messagingTemplate.convertAndSend( + "/topic/process-updates/" + this.uuid, + new ProcessLog(Action.PROCESSING, serviceName + " is processing ...")); + + } + } + } + + // create the service result directory relative to the input file + Path serviceSubDir = getServiceDir(node); + Files.createDirectories(serviceSubDir); + + String servicePath = storageService.load(workflowStatus.getUuid()).relativize(serviceSubDir).toString(); + // replace backslashes with forward slashes in case the service is executed on Windows + servicePath = servicePath.replaceAll("\\\\", "/"); + if (!servicePath.isEmpty()) { + servicePath += "/"; + } + + // first check if there is a result link + List<String> resultUrls = HeaderHelper.getUrlsFromLink(linkHeaders, "result", baseUrl); + + String deleteUrl = null; + if (!resultUrls.isEmpty()) { + + addMessage(workflowStatus, serviceName + " processing finished"); + + boolean error = false; + for (String resultUrl : resultUrls) { + + // download file + HttpResponse<File> resultResponse = Unirest.get(resultUrl) + .asFile(serviceSubDir.resolve("result").toString()); + + Path migrationResult = resultResponse.getBody().toPath(); + String filename = HeaderHelper.getFilenameFromHeader(resultResponse.getHeaders().getFirst("Content-Disposition")); + if (filename != null) { + migrationResult = Files.move(migrationResult, serviceSubDir.resolve(filename)); + } + + String resultName = resultResponse.getHeaders().getFirst("x-app4mc-result-name"); + if (StringUtils.isEmpty(resultName)) { + resultName = serviceName; + } + + addMessage(workflowStatus, resultName + " result downloaded"); + + workflowStatus.addResult( + resultName + " Result", + servicePath + migrationResult.getFileName().toString()); + + // extract header information if result should be used in workflow + String useResult = resultResponse.getHeaders().getFirst("x-app4mc-use-result"); + if (useResult == null || !useResult.toLowerCase().equals("false")) { + // the result should be used in the workflow + result = migrationResult; + } + + if (resultResponse.getStatus() != HttpStatus.SC_OK) { + error = true; + } + + if (deleteUrl == null) { + // extract delete + deleteUrl = HeaderHelper.getUrlFromLink(resultResponse.getHeaders().get("Link"), "delete", baseUrl); + } + } + + if (error) { + workflowStatus.addError(serviceName + " failed with errors"); + node.markFailed(); + } else { + addMessage(workflowStatus, serviceName + " successfull"); + } + } else { + String errorUrl = HeaderHelper.getUrlFromLink(linkHeaders, "error", baseUrl); + if (errorUrl != null) { + workflowStatus.addError(serviceName + " processing finished with error"); + node.markFailed(); + + // download error file + HttpResponse<File> errorResponse = Unirest.get(errorUrl) + .asFile(serviceSubDir.resolve("error").toString()); + + Path migrationError = errorResponse.getBody().toPath(); + String filename = HeaderHelper.getFilenameFromHeader(errorResponse.getHeaders().getFirst("Content-Disposition")); + if (filename != null) { + migrationError = Files.move(migrationError, serviceSubDir.resolve(filename)); + } + + addMessage(workflowStatus, serviceName + " error result downloaded"); + + workflowStatus.addResult( + serviceName + " Error", + servicePath + migrationError.getFileName().toString()); + + // extract delete + deleteUrl = HeaderHelper.getUrlFromLink(errorResponse.getHeaders().get("Link"), "delete", baseUrl); + } else { + workflowStatus.addError(serviceName + " has no result and no error"); + node.markFailed(); + } + } + + boolean deleteResult = true; + if (config != null) { + ServiceConfigurationParameter deleteParam = config.getParameter("deleteResult"); + if (deleteParam != null) { + deleteResult = Boolean.valueOf(deleteParam.getValue()); + } + } + + if (deleteUrl != null && deleteResult) { + // delete upload again + Unirest.delete(deleteUrl).asEmpty(); + addMessage(workflowStatus, serviceName + " cleaned up"); + } + + } else { + // no status URL in the upload response, stop further processing + workflowStatus.addMessage("No status URL found for service " + serviceName); + return inputFile; + } + + return result; + + } catch (Exception e) { + throw new ProcessingFailedException("Error in " + node.getService().getName() + " workflow: " + e.getMessage(), e); + } + } + + private Path getServiceDir(ServiceNode node) { + Path path = storageService.load(workflowStatus.getUuid()); + ServiceNode sn = node; + ArrayList<String> segments = new ArrayList<>(); + segments.add(getServiceNodeDirName(sn)); + while (sn.getParentNode() != null && !sn.getParentNode().getId().equals(WorkflowStatus.ROOT_NODE_ID)) { + sn = sn.getParentNode(); + segments.add(0, getServiceNodeDirName(sn)); + } + + for (String segment : segments) { + path = path.resolve(segment); + } + + return path; + } + + private String getServiceNodeDirName(ServiceNode node) { + if (node.getService() != null) { + return "_" + node.getService().getKey().toLowerCase(); + } + return "_" + node.getId().toLowerCase(); + } + + private void addMessage(WorkflowStatus workflowStatus, String message) { + synchronized(this.messagingTemplate){ + workflowStatus.addMessage(message); + this.messagingTemplate.convertAndSend( + "/topic/process-updates/" + this.uuid, + new ProcessLog(Action.UPDATE, message)); + } + } + +}
diff --git a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatus.java b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatus.java index d42e003..47d17fb 100644 --- a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatus.java +++ b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatus.java
@@ -9,6 +9,7 @@ * * Contributors: * Robert Bosch GmbH - initial API and implementation + * Dortmund University of Applied Sciences and Arts - Bug 570871 ******************************************************************************** */ package org.eclipse.app4mc.cloud.manager; @@ -248,16 +249,24 @@ return node; } - public ArrayList<String> getMessages() { - return messages; + public List<String> getMessages() { + return new ArrayList<>(this.messages); + } + + public boolean hasMessages() { + return !this.messages.isEmpty(); } public void addMessage(String message) { this.messages.add(message); } - public ArrayList<String> getErrors() { - return errors; + public List<String> getErrors() { + return new ArrayList<>(this.errors); + } + + public boolean hasErrors() { + return !this.errors.isEmpty(); } public void addError(String error) { @@ -265,7 +274,11 @@ } public HashMap<String, String> getResults() { - return this.results; + return new LinkedHashMap<>(this.results); + } + + public boolean hasResults() { + return !this.results.isEmpty(); } public void addResult(String key, String resultFile) {
diff --git a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusSerializer.java b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusSerializer.java index 2f487bc..553cc36 100644 --- a/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusSerializer.java +++ b/manager/src/main/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusSerializer.java
@@ -65,16 +65,15 @@ gen.writeBooleanField("done", value.isDone()); } - if (!value.getMessages().isEmpty()) { + if (value.hasMessages()) { gen.writeObjectField("messages", value.getMessages()); } - if (!value.getErrors().isEmpty()) { + if (value.hasErrors()) { gen.writeObjectField("errors", value.getErrors()); } - if (!value.getResults().isEmpty()) { + if (value.hasResults()) { gen.writeObjectField("results", value.getResults()); } - gen.writeEndObject(); }
diff --git a/manager/src/test/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusHelperTest.java b/manager/src/test/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusHelperTest.java index c9994d4..19dd243 100644 --- a/manager/src/test/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusHelperTest.java +++ b/manager/src/test/java/org/eclipse/app4mc/cloud/manager/WorkflowStatusHelperTest.java
@@ -9,6 +9,7 @@ * * Contributors: * Robert Bosch GmbH - initial API and implementation + * Dortmund University of Applied Sciences and Arts - Bug 570871 ******************************************************************************** */ package org.eclipse.app4mc.cloud.manager; @@ -215,14 +216,14 @@ assertFalse(ws.isCancelled()); assertTrue(ws.isDone()); - ArrayList<String> messages = ws.getMessages(); + List<String> messages = ws.getMessages(); assertNotNull(messages); assertEquals(3, messages.size()); assertEquals("Message One", messages.get(0)); assertEquals("Message Two", messages.get(1)); assertEquals("Message Three", messages.get(2)); - ArrayList<String> errors = ws.getErrors(); + List<String> errors = ws.getErrors(); assertNotNull(errors); assertEquals(0, errors.size());