otf-camunda/src/main/java/org/oran/otf/camunda/delegate/otf/common/runnable/SynchronousTestInstanceCallable.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.Collections;\r
  30 import java.util.HashMap;\r
  31 import java.util.Map;\r
  32 import java.util.Timer;\r
  33 import java.util.TimerTask;\r
  34 import java.util.concurrent.Callable;\r
  35 import org.camunda.bpm.BpmPlatform;\r
  36 import org.camunda.bpm.engine.RuntimeService;\r
  37 import org.camunda.bpm.engine.runtime.ProcessInstance;\r
  38 import org.oran.otf.camunda.model.ExecutionConstants;\r
  39 import org.slf4j.Logger;\r
  40 import org.slf4j.LoggerFactory;\r
  41 import org.springframework.data.mongodb.core.MongoTemplate;\r
  42 import org.springframework.data.mongodb.core.query.Criteria;\r
  43 import org.springframework.data.mongodb.core.query.Query;\r
  44 import org.springframework.data.mongodb.core.query.Update;\r
  45 \r
  46 public class SynchronousTestInstanceCallable extends ProcessEngineAwareService\r
  47     implements Callable<TestExecution> {\r
  48 \r
  49   private static final Logger logger =\r
  50       LoggerFactory.getLogger(SynchronousTestInstanceCallable.class);\r
  51   private final TestExecution parentTestExecution;\r
  52   private final TestExecutionRepository testExecutionRepository;\r
  53   private final WorkflowProcessor processor;\r
  54   private final MongoTemplate mongoOperation;\r
  55 \r
  56   private final WorkflowRequest request;\r
  57   private String processInstanceId;\r
  58 \r
  59   public SynchronousTestInstanceCallable(\r
  60       WorkflowRequest request,\r
  61       TestExecution parentTestExecution,\r
  62       TestExecutionRepository testExecutionRepository,\r
  63       WorkflowProcessor processor,\r
  64       MongoTemplate mongoOperation) {\r
  65     this.request = request;\r
  66     this.parentTestExecution = parentTestExecution;\r
  67 \r
  68     this.processInstanceId = "";\r
  69 \r
  70     this.testExecutionRepository = testExecutionRepository;\r
  71     this.processor = processor;\r
  72     this.mongoOperation = mongoOperation;\r
  73   }\r
  74 \r
  75   public SynchronousTestInstanceCallable(\r
  76       WorkflowRequest request,\r
  77       TestExecutionRepository testExecutionRepository,\r
  78       WorkflowProcessor processor,\r
  79       MongoTemplate mongoOperation) {\r
  80     this.request = request;\r
  81     this.parentTestExecution = null;\r
  82 \r
  83     this.processInstanceId = "";\r
  84 \r
  85     this.testExecutionRepository = testExecutionRepository;\r
  86     this.processor = processor;\r
  87     this.mongoOperation = mongoOperation;\r
  88   }\r
  89 \r
  90   @Override\r
  91   public TestExecution call() throws WorkflowProcessorException {\r
  92     try {\r
  93       TestExecution initialTestExecution = processor.processWorkflowRequest(request);\r
  94       this.processInstanceId = initialTestExecution.getProcessInstanceId();\r
  95       final Map<String, Boolean> abortionStatus = Collections.synchronizedMap(new HashMap<>());\r
  96       abortionStatus.put("isAborted", false);\r
  97 \r
  98       // Create a timer task that will call the cancellation after the specified time.\r
  99       TimerTask abortTestInstanceTask =\r
 100           new TimerTask() {\r
 101             @Override\r
 102             public void run() {\r
 103               cancelProcessInstance(processInstanceId);\r
 104               abortionStatus.put("isAborted", true);\r
 105             }\r
 106           };\r
 107 \r
 108       // Start the daemon that waits the max time for a running test instance.\r
 109       long maxExecutionTimeInMillis = request.getMaxExecutionTimeInMillis();\r
 110       if (maxExecutionTimeInMillis > 0) {\r
 111         new Timer(true).schedule(abortTestInstanceTask, maxExecutionTimeInMillis);\r
 112       }\r
 113 \r
 114       while (!isProcessInstanceEnded(processInstanceId)) {\r
 115         Thread.sleep(1000);\r
 116       }\r
 117 \r
 118       // Find the result after the process instance after it has finished.\r
 119       TestExecution testExecution =\r
 120           testExecutionRepository.findFirstByProcessInstanceId(processInstanceId).orElse(null);\r
 121       if (testExecution == null) {\r
 122         logger.error(\r
 123             String.format(\r
 124                 "Process instance with id %s completed, however, a corresponding test execution was not found in the database.",\r
 125                 processInstanceId));\r
 126       } else {\r
 127         // If the test result was not set in the workflow, set it to completed now that we know the\r
 128         // process instance has finished executing.\r
 129         String testResult = testExecution.getTestResult();\r
 130         if (testResult.equalsIgnoreCase("UNKNOWN") || testResult.equalsIgnoreCase("STARTED")) {\r
 131           if (abortionStatus.get("isAborted")) {\r
 132             testExecution.setTestResult(ExecutionConstants.TestResult.TERMINATED);\r
 133           } else {\r
 134             testExecution.setTestResult(ExecutionConstants.TestResult.COMPLETED);\r
 135           }\r
 136 \r
 137           //TODO: RG remove prints\r
 138           System.out.println(testExecution.getTestHeadResults());\r
 139           System.out.println(request);\r
 140           TestExecutionUtility.saveTestResult(\r
 141               mongoOperation, testExecution, testExecution.getTestResult());\r
 142         }\r
 143 \r
 144         // Saves the testExecution to the parent test execution if this belongs to a "sub" test\r
 145         // instance call.\r
 146         saveToParentTestExecution(testExecution);\r
 147       }\r
 148 \r
 149       return testExecution;\r
 150     } catch (WorkflowProcessorException e) {\r
 151       throw e;\r
 152     } catch (Exception e) {\r
 153       e.printStackTrace();\r
 154       return null;\r
 155     }\r
 156   }\r
 157 \r
 158   private void saveToParentTestExecution(TestExecution testExecution) {\r
 159     if (parentTestExecution == null) {\r
 160       return;\r
 161     }\r
 162 \r
 163     synchronized (parentTestExecution) {\r
 164       // Add the testExecution to the parentTestExecution\r
 165       parentTestExecution.getTestInstanceResults().add(testExecution);\r
 166       Query query = new Query();\r
 167       query.addCriteria(Criteria.where("_id").is(parentTestExecution.get_id()));\r
 168       // Also add businessKey as a criteria because the object won't be found if the business key\r
 169       // was somehow modified in the workflow.\r
 170       query.addCriteria(Criteria.where("businessKey").is(parentTestExecution.getBusinessKey()));\r
 171       Update update = new Update();\r
 172       update.set("testInstanceResults", parentTestExecution.getTestInstanceResults());\r
 173       UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
 174       // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
 175       if (result.getMatchedCount() == 0) {\r
 176         throw new TestExecutionException(\r
 177             String.format(\r
 178                 "Unable to log the test result because a testExecution associated with _id, %s and businessKey %s, was not found.",\r
 179                 parentTestExecution.get_id(), parentTestExecution.getBusinessKey()));\r
 180       } else if (result.getModifiedCount() == 0) {\r
 181         throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
 182       }\r
 183     }\r
 184     logger.info(\r
 185         String.format(\r
 186             "\t[Child-%s] finished saving result to parentTestExecution with result %s.",\r
 187             processInstanceId, testExecution.getTestResult()));\r
 188   }\r
 189 \r
 190   private boolean isProcessInstanceEnded(String processInstanceId) {\r
 191     try {\r
 192       RuntimeService runtimeService =\r
 193           BpmPlatform.getProcessEngineService()\r
 194               .getProcessEngine(OtfCamundaConfiguration.processEngineName)\r
 195               .getRuntimeService();\r
 196       ProcessInstance processInstance =\r
 197           runtimeService\r
 198               .createProcessInstanceQuery()\r
 199               .processInstanceId(processInstanceId)\r
 200               .singleResult();\r
 201       return processInstance == null || processInstance.isEnded();\r
 202     } catch (Exception e) {\r
 203       logger.error("Exception :", e);\r
 204       return true;\r
 205     }\r
 206   }\r
 207 \r
 208   private boolean cancelProcessInstance(String processInstanceId) {\r
 209     try {\r
 210       RuntimeService runtimeService =\r
 211           BpmPlatform.getProcessEngineService()\r
 212               .getProcessEngine(OtfCamundaConfiguration.processEngineName)\r
 213               .getRuntimeService();\r
 214       runtimeService.deleteProcessInstance(\r
 215           processInstanceId, "Triggered by user defined max execution time timeout.");\r
 216       ProcessInstance processInstance =\r
 217           runtimeService\r
 218               .createProcessInstanceQuery()\r
 219               .processInstanceId(processInstanceId)\r
 220               .singleResult();\r
 221       return processInstance == null || processInstance.isEnded();\r
 222     } catch (Exception e) {\r
 223       logger.error("Exception :", e);\r
 224       return true;\r
 225     }\r
 226   }\r
 227 }\r