otf-camunda/src/main/java/org/oran/otf/camunda/delegate/otf/common/runnable/AsynchronousTestInstanceCallable.java

   1 /*  Copyright (c) 2019 AT&T Intellectual Property.                             #\r
   2 #                                                                              #\r
   3 #   Licensed under the Apache License, Version 2.0 (the "License");            #\r
   4 #   you may not use this file except in compliance with the License.           #\r
   5 #   You may obtain a copy of the License at                                    #\r
   6 #                                                                              #\r
   7 #       http://www.apache.org/licenses/LICENSE-2.0                             #\r
   8 #                                                                              #\r
   9 #   Unless required by applicable law or agreed to in writing, software        #\r
  10 #   distributed under the License is distributed on an "AS IS" BASIS,          #\r
  11 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   #\r
  12 #   See the License for the specific language governing permissions and        #\r
  13 #   limitations under the License.                                             #\r
  14 ##############################################################################*/\r
  15 \r
  16 \r
  17 package org.oran.otf.camunda.delegate.otf.common.runnable;\r
  18 \r
  19 import org.oran.otf.camunda.configuration.OtfCamundaConfiguration;\r
  20 import org.oran.otf.camunda.exception.TestExecutionException;\r
  21 import org.oran.otf.camunda.exception.WorkflowProcessorException;\r
  22 import org.oran.otf.camunda.service.ProcessEngineAwareService;\r
  23 import org.oran.otf.camunda.workflow.WorkflowProcessor;\r
  24 import org.oran.otf.camunda.workflow.WorkflowRequest;\r
  25 import org.oran.otf.common.model.TestExecution;\r
  26 import org.oran.otf.common.repository.TestExecutionRepository;\r
  27 import org.oran.otf.common.utility.database.TestExecutionUtility;\r
  28 import com.mongodb.client.result.UpdateResult;\r
  29 import java.util.Timer;\r
  30 import java.util.TimerTask;\r
  31 import java.util.concurrent.Callable;\r
  32 \r
  33 import org.camunda.bpm.BpmPlatform;\r
  34 import org.camunda.bpm.engine.RuntimeService;\r
  35 import org.camunda.bpm.engine.runtime.ProcessInstance;\r
  36 import org.oran.otf.camunda.model.ExecutionConstants;\r
  37 import org.slf4j.Logger;\r
  38 import org.slf4j.LoggerFactory;\r
  39 import org.springframework.data.mongodb.core.MongoTemplate;\r
  40 import org.springframework.data.mongodb.core.query.Criteria;\r
  41 import org.springframework.data.mongodb.core.query.Query;\r
  42 import org.springframework.data.mongodb.core.query.Update;\r
  43 \r
  44 public class AsynchronousTestInstanceCallable extends ProcessEngineAwareService\r
  45     implements Callable<TestExecution> {\r
  46 \r
  47   private static final Logger logger =\r
  48       LoggerFactory.getLogger(AsynchronousTestInstanceCallable.class);\r
  49   private final TestExecution parentTestExecution;\r
  50   private final TestExecutionRepository testExecutionRepository;\r
  51   private final WorkflowProcessor processor;\r
  52   private final MongoTemplate mongoOperation;\r
  53 \r
  54   private final WorkflowRequest request;\r
  55   private String processInstanceId;\r
  56 \r
  57   public AsynchronousTestInstanceCallable(\r
  58       WorkflowRequest request,\r
  59       TestExecution parentTestExecution,\r
  60       TestExecutionRepository testExecutionRepository,\r
  61       WorkflowProcessor processor,\r
  62       MongoTemplate mongoOperation) {\r
  63     this.request = request;\r
  64     this.parentTestExecution = parentTestExecution;\r
  65 \r
  66     this.processInstanceId = "";\r
  67 \r
  68     this.testExecutionRepository = testExecutionRepository;\r
  69     this.processor = processor;\r
  70     this.mongoOperation = mongoOperation;\r
  71   }\r
  72 \r
  73   public AsynchronousTestInstanceCallable(\r
  74       WorkflowRequest request,\r
  75       TestExecutionRepository testExecutionRepository,\r
  76       WorkflowProcessor processor,\r
  77       MongoTemplate mongoOperation) {\r
  78     this.request = request;\r
  79     this.parentTestExecution = null;\r
  80 \r
  81     this.processInstanceId = "";\r
  82 \r
  83     this.testExecutionRepository = testExecutionRepository;\r
  84     this.processor = processor;\r
  85     this.mongoOperation = mongoOperation;\r
  86   }\r
  87 \r
  88   @Override\r
  89   public TestExecution call() throws WorkflowProcessorException {\r
  90     try {\r
  91       TestExecution initialTestExecution = processor.processWorkflowRequest(request);\r
  92       this.processInstanceId = initialTestExecution.getProcessInstanceId();\r
  93 \r
  94       // Create a timer task that will call the cancellation after the specified time.\r
  95       TimerTask abortTestInstanceTask =\r
  96           new TimerTask() {\r
  97             @Override\r
  98             public void run() {\r
  99               cancelProcessInstance(processInstanceId);\r
 100 \r
 101               // Find the result after the process instance after it has finished.\r
 102               TestExecution testExecution =\r
 103                   testExecutionRepository\r
 104                       .findFirstByProcessInstanceId(processInstanceId)\r
 105                       .orElse(null);\r
 106               if (testExecution == null) {\r
 107                 logger.error(\r
 108                     String.format(\r
 109                         "Process instance with id %s completed, however, a corresponding test execution was not found in the database.",\r
 110                         processInstanceId));\r
 111               } else {\r
 112                 testExecution.setTestResult(ExecutionConstants.TestResult.TERMINATED);\r
 113                 TestExecutionUtility.saveTestResult(\r
 114                     mongoOperation, testExecution, testExecution.getTestResult());\r
 115 \r
 116                 // Saves the testExecution to the parent test execution if this belongs to a "sub"\r
 117                 // test\r
 118                 // instance call.\r
 119                 // updated terminated\r
 120                 saveToParentTestExecution(testExecution);\r
 121               }\r
 122             }\r
 123           };\r
 124 \r
 125       // Start the daemon that waits the max time for a running test instance.\r
 126       long maxExecutionTimeInMillis = request.getMaxExecutionTimeInMillis();\r
 127       if (maxExecutionTimeInMillis > 0) {\r
 128         new Timer(true).schedule(abortTestInstanceTask, maxExecutionTimeInMillis);\r
 129       }\r
 130 \r
 131       return initialTestExecution;\r
 132     } catch (WorkflowProcessorException e) {\r
 133       throw e;\r
 134     } catch (Exception e) {\r
 135       e.printStackTrace();\r
 136       return null;\r
 137     }\r
 138   }\r
 139 \r
 140   private void saveToParentTestExecution(TestExecution testExecution) {\r
 141     if (parentTestExecution == null) {\r
 142       return;\r
 143     }\r
 144 \r
 145     synchronized (parentTestExecution) {\r
 146       // Add the testExecution to the parentTestExecution\r
 147       parentTestExecution.getTestInstanceResults().add(testExecution);\r
 148       Query query = new Query();\r
 149       query.addCriteria(Criteria.where("_id").is(parentTestExecution.get_id()));\r
 150       // Also add businessKey as a criteria because the object won't be found if the business key\r
 151       // was somehow modified in the workflow.\r
 152       query.addCriteria(Criteria.where("businessKey").is(parentTestExecution.getBusinessKey()));\r
 153       Update update = new Update();\r
 154       update.set("testInstanceResults", parentTestExecution.getTestInstanceResults());\r
 155       UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
 156       // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
 157       if (result.getMatchedCount() == 0) {\r
 158         throw new TestExecutionException(\r
 159             String.format(\r
 160                 "Unable to log the test result because a testExecution associated with _id, %s and businessKey %s, was not found.",\r
 161                 parentTestExecution.get_id(), parentTestExecution.getBusinessKey()));\r
 162       } else if (result.getModifiedCount() == 0) {\r
 163         throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
 164       }\r
 165     }\r
 166   }\r
 167 \r
 168   private boolean isProcessInstanceEnded(String processInstanceId) {\r
 169     try {\r
 170       RuntimeService runtimeService = BpmPlatform.getProcessEngineService().getProcessEngine(OtfCamundaConfiguration.processEngineName).getRuntimeService();\r
 171       ProcessInstance processInstance =\r
 172           runtimeService\r
 173               .createProcessInstanceQuery()\r
 174               .processInstanceId(processInstanceId)\r
 175               .singleResult();\r
 176       return processInstance == null || processInstance.isEnded();\r
 177     } catch (Exception e) {\r
 178       logger.error("Exception :", e);\r
 179       return true;\r
 180     }\r
 181   }\r
 182 \r
 183   private boolean cancelProcessInstance(String processInstanceId) {\r
 184     try {\r
 185       RuntimeService runtimeService = BpmPlatform.getProcessEngineService().getProcessEngine(OtfCamundaConfiguration.processEngineName).getRuntimeService();\r
 186       runtimeService.deleteProcessInstance(\r
 187           processInstanceId, "Triggered by user defined max execution time timeout.");\r
 188       ProcessInstance processInstance =\r
 189           runtimeService\r
 190               .createProcessInstanceQuery()\r
 191               .processInstanceId(processInstanceId)\r
 192               .singleResult();\r
 193       return processInstance == null || processInstance.isEnded();\r
 194     } catch (Exception e) {\r
 195       logger.error("Exception :", e);\r
 196       return true;\r
 197     }\r
 198   }\r
 199 }\r