otf-camunda/src/main/java/org/oran/otf/camunda/workflow/handler/FailedJobIncidentHandler.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.workflow.handler;\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.model.ExecutionConstants.TestResult;\r
  22 import org.oran.otf.camunda.workflow.utility.WorkflowTask;\r
  23 import org.oran.otf.common.model.TestExecution;\r
  24 import org.oran.otf.common.repository.TestExecutionRepository;\r
  25 import org.oran.otf.common.utility.Utility;\r
  26 import org.oran.otf.service.impl.DeleteProcessInstanceServiceImpl;\r
  27 import com.google.common.base.Strings;\r
  28 import com.mongodb.client.result.UpdateResult;\r
  29 \r
  30 import java.util.Date;\r
  31 import java.util.List;\r
  32 \r
  33 import org.camunda.bpm.BpmPlatform;\r
  34 import org.camunda.bpm.engine.RuntimeService;\r
  35 import org.camunda.bpm.engine.impl.incident.IncidentContext;\r
  36 import org.camunda.bpm.engine.impl.incident.IncidentHandler;\r
  37 import org.camunda.bpm.engine.runtime.Execution;\r
  38 import org.camunda.bpm.engine.runtime.Incident;\r
  39 import org.slf4j.Logger;\r
  40 import org.slf4j.LoggerFactory;\r
  41 import org.springframework.beans.factory.annotation.Autowired;\r
  42 import org.springframework.boot.context.event.ApplicationReadyEvent;\r
  43 import org.springframework.context.event.EventListener;\r
  44 import org.springframework.data.mongodb.core.MongoTemplate;\r
  45 import org.springframework.data.mongodb.core.query.Criteria;\r
  46 import org.springframework.data.mongodb.core.query.Query;\r
  47 import org.springframework.data.mongodb.core.query.Update;\r
  48 import org.springframework.stereotype.Service;\r
  49 \r
  50 @Service\r
  51 public class FailedJobIncidentHandler implements IncidentHandler {\r
  52 \r
  53   private static final Logger logger = LoggerFactory.getLogger(FailedJobIncidentHandler.class);\r
  54   private static final String logPrefix = Utility.getLoggerPrefix();\r
  55 \r
  56   @Autowired\r
  57   private TestExecutionRepository testExecutionRepository;\r
  58   @Autowired\r
  59   private MongoTemplate mongoOperation;\r
  60   @Autowired\r
  61   private DeleteProcessInstanceServiceImpl deleteProcessInstanceService;\r
  62 \r
  63   @Override\r
  64   public String getIncidentHandlerType() {\r
  65     return Incident.FAILED_JOB_HANDLER_TYPE;\r
  66   }\r
  67 \r
  68   @Override\r
  69   public Incident handleIncident(IncidentContext context, String message) {\r
  70     String executionId = context.getExecutionId();\r
  71     if (Strings.isNullOrEmpty(executionId)) {\r
  72       return null;\r
  73     }\r
  74     RuntimeService runtimeService = BpmPlatform.getProcessEngineService().getProcessEngine(OtfCamundaConfiguration.processEngineName).getRuntimeService();\r
  75 \r
  76     Execution execution = runtimeService.createExecutionQuery().executionId(executionId).singleResult();\r
  77     String processInstanceId = execution.getProcessInstanceId();\r
  78     TestExecution testExecution =\r
  79         testExecutionRepository.findFirstByProcessInstanceId(processInstanceId).orElse(null);\r
  80 \r
  81     if (testExecution == null) {\r
  82       String error = String.format(\r
  83           "%sUnable to find testExecution with processInstanceId %s. This process instance will forcefully be terminated to avoid a rogue process.",\r
  84           logPrefix, processInstanceId);\r
  85       logger.error(error);\r
  86       deleteProcessInstanceService.deleteProcessInstanceInternal(processInstanceId, error);\r
  87 \r
  88     } else {\r
  89       if(!testExecution.getTestResult().equals(TestResult.TERMINATED)){\r
  90         updateTestResult(testExecution, TestResult.WORKFLOW_ERROR, message);\r
  91       }\r
  92       deleteProcessInstanceService.deleteProcessInstanceInternal(processInstanceId, message);\r
  93 \r
  94     }\r
  95 \r
  96     List<WorkflowTask> workflowTasks =\r
  97         WorkflowTask.workflowTasksByExecutionId.getOrDefault(processInstanceId, null);\r
  98 \r
  99     if (workflowTasks != null) {\r
 100       logger.debug("Forcefully terminating workflow tasks for processInstanceId: " + processInstanceId);\r
 101       for (WorkflowTask workflowTask : workflowTasks) {\r
 102         workflowTask.shutdown(true);\r
 103       }\r
 104     }\r
 105 \r
 106     return null;\r
 107   }\r
 108 \r
 109   private void updateTestResult(TestExecution testExecution, String testResult, String testResultMessage) {\r
 110     // Set the test result\r
 111     testExecution.setTestResult(testResult);\r
 112     testExecution.setTestResultMessage(testResultMessage);\r
 113     Query query = new Query();\r
 114     query.addCriteria(Criteria.where("_id").is(testExecution.get_id()));\r
 115     // Also add businessKey as a criteria because the object won't be found if the business key\r
 116     // was somehow modified in the workflow.\r
 117     query.addCriteria(Criteria.where("businessKey").is(testExecution.getBusinessKey()));\r
 118     Update update = new Update();\r
 119     update.set("testResult", testExecution.getTestResult());\r
 120     update.set("testResultMessage", testExecution.getTestResultMessage());\r
 121     update.set("endTime", new Date(System.currentTimeMillis()));\r
 122     UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
 123     // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
 124     if (result.getMatchedCount() == 0) {\r
 125       throw new TestExecutionException(\r
 126           String.format(\r
 127               "Unable to log the test result because a testExecution associated with _id, %s and businessKey %s, was not found.",\r
 128               testExecution.get_id(), testExecution.getBusinessKey()));\r
 129     } else if (result.getModifiedCount() == 0) {\r
 130       throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
 131     }\r
 132   }\r
 133 \r
 134   @Override\r
 135   public void resolveIncident(IncidentContext context) {\r
 136     //    logger.info("incident resolved");\r
 137 \r
 138   }\r
 139 \r
 140   @Override\r
 141   public void deleteIncident(IncidentContext context) {\r
 142     //    logger.info("incident deleted");\r
 143 \r
 144   }\r
 145 }\r