so-etsi-nfvo/so-etsi-nfvo-ns-lcm/so-etsi-nfvo-ns-lcm-bpmn-flows/src/main/java/org/onap/so/etsi/nfvo/ns/lcm/bpmn/flows/service/JobExecutorService.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2020 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.service;
  21
  22 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.CREATE_NS_REQUEST_PARAM_NAME;
  23 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.GLOBAL_CUSTOMER_ID_PARAM_NAME;
  24 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.INSTANTIATE_NS_REQUEST_PARAM_NAME;
  25 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.JOB_ID_PARAM_NAME;
  26 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.NS_INSTANCE_ID_PARAM_NAME;
  27 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.OCC_ID_PARAM_NAME;
  28 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.CamundaVariableNameConstants.SERVICE_TYPE_PARAM_NAME;
  29 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.Constants.CREATE_NS_WORKFLOW_NAME;
  30 import static org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.Constants.INSTANTIATE_NS_WORKFLOW_NAME;
  31 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobAction.INSTANTIATE;
  32 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum.ERROR;
  33 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum.FINISHED;
  34 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum.FINISHED_WITH_ERROR;
  35 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum.IN_PROGRESS;
  36 import static org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum.STARTING;
  37 import static org.slf4j.LoggerFactory.getLogger;
  38 import java.time.Instant;
  39 import java.time.LocalDateTime;
  40 import java.util.HashMap;
  41 import java.util.Map;
  42 import java.util.Optional;
  43 import java.util.concurrent.TimeUnit;
  44 import org.apache.commons.lang3.tuple.ImmutablePair;
  45 import org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.GsonProvider;
  46 import org.onap.so.etsi.nfvo.ns.lcm.bpmn.flows.exceptions.NsRequestProcessingException;
  47 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobAction;
  48 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.JobStatusEnum;
  49 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.NfvoJob;
  50 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.NsLcmOpOcc;
  51 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.NsLcmOpType;
  52 import org.onap.so.etsi.nfvo.ns.lcm.database.beans.OperationStateEnum;
  53 import org.onap.so.etsi.nfvo.ns.lcm.database.service.DatabaseServiceProvider;
  54 import org.onap.so.etsi.nfvo.ns.lcm.model.CreateNsRequest;
  55 import org.onap.so.etsi.nfvo.ns.lcm.model.InlineResponse400;
  56 import org.onap.so.etsi.nfvo.ns.lcm.model.InstantiateNsRequest;
  57 import org.onap.so.etsi.nfvo.ns.lcm.model.NsInstancesNsInstance;
  58 import org.slf4j.Logger;
  59 import org.springframework.beans.factory.annotation.Autowired;
  60 import org.springframework.beans.factory.annotation.Value;
  61 import org.springframework.stereotype.Service;
  62 import com.google.common.collect.ImmutableSet;
  63 import com.google.gson.Gson;
  64
  65 /**
  66  * @author Waqas Ikram (waqas.ikram@est.tech)
  67  *
  68  */
  69 @Service
  70 public class JobExecutorService {
  71
  72     private static final Logger logger = getLogger(JobExecutorService.class);
  73
  74     private static final ImmutableSet<JobStatusEnum> JOB_FINISHED_STATES =
  75             ImmutableSet.of(FINISHED, ERROR, FINISHED_WITH_ERROR);
  76
  77     private static final int SLEEP_TIME_IN_SECONDS = 5;
  78
  79     @Value("${so-etsi-ns-lcm-workflow-engine.requesttimeout.timeoutInSeconds:300}")
  80     private int timeOutInSeconds;
  81
  82     private final DatabaseServiceProvider databaseServiceProvider;
  83     private final WorkflowExecutorService workflowExecutorService;
  84     private final WorkflowQueryService workflowQueryService;
  85     private Gson gson;
  86
  87     @Autowired
  88     public JobExecutorService(final DatabaseServiceProvider databaseServiceProvider,
  89             final WorkflowExecutorService workflowExecutorService, final WorkflowQueryService workflowQueryService,
  90             final GsonProvider gsonProvider) {
  91         this.databaseServiceProvider = databaseServiceProvider;
  92         this.workflowExecutorService = workflowExecutorService;
  93         this.workflowQueryService = workflowQueryService;
  94         gson = gsonProvider.getGson();
  95     }
  96
  97     public NsInstancesNsInstance runCreateNsJob(final CreateNsRequest createNsRequest, final String globalCustomerId,
  98             final String serviceType) {
  99         logger.info("Starting 'Create NS' workflow job for request:\n{}", createNsRequest);
 100         final NfvoJob newJob = new NfvoJob().startTime(LocalDateTime.now()).jobType("NS").jobAction(JobAction.CREATE)
 101                 .resourceId(createNsRequest.getNsdId()).resourceName(createNsRequest.getNsName())
 102                 .status(JobStatusEnum.STARTING).progress(5);
 103         databaseServiceProvider.addJob(newJob);
 104
 105         logger.info("New job created in database :\n{}", newJob);
 106
 107         workflowExecutorService.executeWorkflow(newJob.getJobId(), CREATE_NS_WORKFLOW_NAME,
 108                 getVariables(newJob.getJobId(), createNsRequest, globalCustomerId, serviceType));
 109
 110         final ImmutablePair<String, JobStatusEnum> immutablePair =
 111                 waitForJobToFinish(newJob.getJobId(), JOB_FINISHED_STATES);
 112
 113         if (immutablePair.getRight() == null) {
 114             final String message = "Failed to create NS for request: \n" + createNsRequest;
 115             logger.error(message);
 116             throw new NsRequestProcessingException(message);
 117         }
 118         final JobStatusEnum finalJobStatus = immutablePair.getRight();
 119         final String processInstanceId = immutablePair.getLeft();
 120
 121         if (!FINISHED.equals(finalJobStatus)) {
 122
 123             final Optional<InlineResponse400> optional = workflowQueryService.getProblemDetails(processInstanceId);
 124             if (optional.isPresent()) {
 125                 final InlineResponse400 problemDetails = optional.get();
 126                 final String message =
 127                         "Failed to create NS for request: \n" + createNsRequest + " due to \n" + problemDetails;
 128                 logger.error(message);
 129                 throw new NsRequestProcessingException(message, problemDetails);
 130             }
 131
 132             final String message = "Received unexpected Job Status: " + finalJobStatus
 133                     + " Failed to Create NS for request: \n" + createNsRequest;
 134             logger.error(message);
 135             throw new NsRequestProcessingException(message);
 136         }
 137
 138         logger.debug("Will query for CreateNsResponse using processInstanceId:{}", processInstanceId);
 139         final Optional<NsInstancesNsInstance> optional = workflowQueryService.getCreateNsResponse(processInstanceId);
 140         if (optional.isEmpty()) {
 141             final String message =
 142                     "Unable to find CreateNsReponse in Camunda History for process instance: " + processInstanceId;
 143             logger.error(message);
 144             throw new NsRequestProcessingException(message);
 145         }
 146         return optional.get();
 147     }
 148
 149     public String runInstantiateNsJob(final String nsInstanceId, final InstantiateNsRequest instantiateNsRequest) {
 150
 151         final NfvoJob newJob = new NfvoJob().startTime(LocalDateTime.now()).jobType("NS").jobAction(INSTANTIATE)
 152                 .resourceId(nsInstanceId).status(STARTING).progress(0);
 153         databaseServiceProvider.addJob(newJob);
 154         logger.info("New job created in database :\n{}", newJob);
 155
 156         final LocalDateTime currentDateTime = LocalDateTime.now();
 157         final NsLcmOpOcc newNsLcmOpOcc = new NsLcmOpOcc().id(nsInstanceId).operation(NsLcmOpType.INSTANTIATE)
 158                 .operationState(OperationStateEnum.PROCESSING).stateEnteredTime(currentDateTime)
 159                 .startTime(currentDateTime).isAutoInnovation(false).isCancelPending(false)
 160                 .operationParams(gson.toJson(instantiateNsRequest));
 161         databaseServiceProvider.addNSLcmOpOcc(newNsLcmOpOcc);
 162         logger.info("New NSLcmOpOcc created in database :\n{}", newNsLcmOpOcc);
 163
 164         workflowExecutorService.executeWorkflow(newJob.getJobId(), INSTANTIATE_NS_WORKFLOW_NAME,
 165                 getVariables(nsInstanceId, newJob.getJobId(), newNsLcmOpOcc.getId(), instantiateNsRequest));
 166
 167         final ImmutableSet<JobStatusEnum> jobFinishedStates =
 168                 ImmutableSet.of(FINISHED, ERROR, FINISHED_WITH_ERROR, IN_PROGRESS);
 169         final ImmutablePair<String, JobStatusEnum> immutablePair =
 170                 waitForJobToFinish(newJob.getJobId(), jobFinishedStates);
 171
 172         if (immutablePair.getRight() == null) {
 173             final String message = "Failed to Instantiate NS for request: \n" + instantiateNsRequest;
 174             logger.error(message);
 175             throw new NsRequestProcessingException(message);
 176         }
 177
 178         final JobStatusEnum finalJobStatus = immutablePair.getRight();
 179
 180         if (IN_PROGRESS.equals(finalJobStatus) || FINISHED.equals(finalJobStatus)) {
 181             logger.info("Instantiation Job status: {}", finalJobStatus);
 182
 183
 184             return newNsLcmOpOcc.getId();
 185         }
 186
 187         final String message = "Received unexpected Job Status: " + finalJobStatus
 188                 + " Failed to instantiate NS for request: \n" + instantiateNsRequest;
 189         logger.error(message);
 190         throw new NsRequestProcessingException(message);
 191     }
 192
 193     private ImmutablePair<String, JobStatusEnum> waitForJobToFinish(final String jobId,
 194             final ImmutableSet<JobStatusEnum> jobFinishedStates) {
 195         try {
 196             final long startTimeInMillis = System.currentTimeMillis();
 197             final long timeOutTime = startTimeInMillis + TimeUnit.SECONDS.toMillis(timeOutInSeconds);
 198
 199             logger.info("Will wait till {} for {} job to finish", Instant.ofEpochMilli(timeOutTime).toString(), jobId);
 200             JobStatusEnum currentJobStatus = null;
 201             while (timeOutTime > System.currentTimeMillis()) {
 202
 203                 final Optional<NfvoJob> optional = databaseServiceProvider.getJob(jobId);
 204
 205                 if (optional.isEmpty()) {
 206                     logger.error("Unable to find Job using jobId: {}", jobId);
 207                     return ImmutablePair.nullPair();
 208                 }
 209
 210                 final NfvoJob nfvoJob = optional.get();
 211                 currentJobStatus = nfvoJob.getStatus();
 212                 logger.debug("Received job status response: \n ", nfvoJob);
 213                 if (jobFinishedStates.contains(nfvoJob.getStatus())) {
 214                     logger.info("Job finished \n {}", currentJobStatus);
 215                     return ImmutablePair.of(nfvoJob.getProcessInstanceId(), currentJobStatus);
 216                 }
 217
 218                 logger.debug("Haven't received one of finish state {} yet, will try again in {} seconds",
 219                         jobFinishedStates, SLEEP_TIME_IN_SECONDS);
 220                 TimeUnit.SECONDS.sleep(SLEEP_TIME_IN_SECONDS);
 221
 222             }
 223             logger.warn("Timeout current job status: {}", currentJobStatus);
 224             return ImmutablePair.nullPair();
 225         } catch (final InterruptedException interruptedException) {
 226             Thread.currentThread().interrupt();
 227             logger.error("Sleep was interrupted", interruptedException);
 228             return ImmutablePair.nullPair();
 229         }
 230     }
 231
 232     private Map<String, Object> getVariables(final String jobId, final CreateNsRequest createNsRequest,
 233             final String globalCustomerId, final String serviceType) {
 234         final Map<String, Object> variables = new HashMap<>();
 235         variables.put(JOB_ID_PARAM_NAME, jobId);
 236         variables.put(CREATE_NS_REQUEST_PARAM_NAME, createNsRequest);
 237         variables.put(GLOBAL_CUSTOMER_ID_PARAM_NAME, globalCustomerId);
 238         variables.put(SERVICE_TYPE_PARAM_NAME, serviceType);
 239         return variables;
 240     }
 241
 242     private Map<String, Object> getVariables(final String nsInstanceId, final String jobId, final String occId,
 243             final InstantiateNsRequest instantiateNsRequest) {
 244         final Map<String, Object> variables = new HashMap<>();
 245         variables.put(NS_INSTANCE_ID_PARAM_NAME, nsInstanceId);
 246         variables.put(JOB_ID_PARAM_NAME, jobId);
 247         variables.put(OCC_ID_PARAM_NAME, occId);
 248         variables.put(INSTANTIATE_NS_REQUEST_PARAM_NAME, instantiateNsRequest);
 249         return variables;
 250     }
 251
 252 }