appc-config/appc-flow-controller/provider/src/main/java/org/onap/appc/flow/controller/node/FlowControlNode.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP : APPC
   4  * ================================================================================
   5  * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
   6  * =============================================================================
   7  * Modifications Copyright (C) 2019 Ericsson
   8  * =============================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  *
  21  * ============LICENSE_END=========================================================
  22  */
  23
  24 package org.onap.appc.flow.controller.node;
  25
  26 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.ACTION_LEVEL;
  27 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.DESINGTIME;
  28 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.GRAPH;
  29 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.INPUT_PARAM_RESPONSE_PREFIX;
  30 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.NODE;
  31 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_PARAM_ERROR_CODE;
  32 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_PARAM_ERROR_MESSAGE;
  33 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_PARAM_STATUS;
  34 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_STATUS_FAILURE;
  35 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_STATUS_MESSAGE;
  36 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.OUTPUT_STATUS_SUCCESS;
  37 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.REQUEST_ACTION;
  38 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.REQUEST_ID;
  39 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.RESPONSE_PREFIX;
  40 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.REST;
  41 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.SEQUENCE_TYPE;
  42 import static org.onap.appc.flow.controller.utils.FlowControllerConstants.VNF_TYPE;
  43 import java.io.IOException;
  44 import java.util.HashMap;
  45 import java.util.List;
  46 import java.util.Map;
  47 import org.apache.commons.lang3.StringUtils;
  48 import org.onap.appc.flow.controller.ResponseHandlerImpl.DefaultResponseHandler;
  49 import org.onap.appc.flow.controller.data.PrecheckOption;
  50 import org.onap.appc.flow.controller.data.ResponseAction;
  51 import org.onap.appc.flow.controller.data.Transaction;
  52 import org.onap.appc.flow.controller.data.Transactions;
  53 import org.onap.appc.flow.controller.dbervices.FlowControlDBService;
  54 import org.onap.appc.flow.controller.executorImpl.GraphExecutor;
  55 import org.onap.appc.flow.controller.executorImpl.NodeExecutor;
  56 import org.onap.appc.flow.controller.executorImpl.RestExecutor;
  57 import org.onap.appc.flow.controller.interfaces.FlowExecutorInterface;
  58 import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
  59 import org.onap.ccsdk.sli.core.sli.SvcLogicException;
  60 import org.onap.ccsdk.sli.core.sli.SvcLogicJavaPlugin;
  61 import com.att.eelf.configuration.EELFLogger;
  62 import com.att.eelf.configuration.EELFManager;
  63 import com.fasterxml.jackson.databind.ObjectMapper;
  64
  65 public class FlowControlNode implements SvcLogicJavaPlugin {
  66
  67     private static final EELFLogger log = EELFManager.getInstance().getLogger(FlowControlNode.class);
  68
  69     private final FlowControlDBService dbService;
  70     private final FlowSequenceGenerator flowSequenceGenerator;
  71
  72     public FlowControlNode() {
  73         this.dbService = FlowControlDBService.initialise();
  74         this.flowSequenceGenerator = new FlowSequenceGenerator();
  75     }
  76
  77     FlowControlNode(FlowControlDBService dbService, FlowSequenceGenerator flowSequenceGenerator) {
  78         this.dbService = dbService;
  79         this.flowSequenceGenerator = flowSequenceGenerator;
  80     }
  81
  82     public void processFlow(Map<String, String> inParams, SvcLogicContext ctx) throws SvcLogicException {
  83         log.debug("Received processParamKeys call with params : " + inParams);
  84         String responsePrefix = inParams.get(INPUT_PARAM_RESPONSE_PREFIX);
  85         try {
  86             responsePrefix = StringUtils.isNotBlank(responsePrefix) ? (responsePrefix + ".") : "";
  87             SvcLogicContext localContext = new SvcLogicContext();
  88
  89             localContext.setAttribute(REQUEST_ID, ctx.getAttribute(REQUEST_ID));
  90             localContext.setAttribute(VNF_TYPE, ctx.getAttribute(VNF_TYPE));
  91             localContext.setAttribute(REQUEST_ACTION, ctx.getAttribute(REQUEST_ACTION));
  92             localContext.setAttribute(ACTION_LEVEL, ctx.getAttribute(ACTION_LEVEL));
  93             localContext.setAttribute(RESPONSE_PREFIX, responsePrefix);
  94             ctx.setAttribute(RESPONSE_PREFIX, responsePrefix);
  95
  96             dbService.getFlowReferenceData(ctx, inParams, localContext);
  97
  98             for (String key : localContext.getAttributeKeySet()) {
  99                 log.debug("processFlow " + key + "=" + ctx.getAttribute(key));
 100             }
 101             processFlowSequence(inParams, ctx, localContext);
 102             if (!ctx.getAttribute(responsePrefix + OUTPUT_PARAM_STATUS).equals(OUTPUT_STATUS_SUCCESS)) {
 103                 throw new SvcLogicException(ctx.getAttribute(responsePrefix + OUTPUT_STATUS_MESSAGE));
 104             }
 105         } catch (Exception e) {
 106             ctx.setAttribute(responsePrefix + OUTPUT_PARAM_STATUS, OUTPUT_STATUS_FAILURE);
 107             ctx.setAttribute(responsePrefix + OUTPUT_PARAM_ERROR_MESSAGE, e.getMessage());
 108             ctx.setAttribute(responsePrefix + OUTPUT_PARAM_ERROR_CODE, ctx.getAttribute(OUTPUT_PARAM_ERROR_CODE));
 109             log.error("Error occurred in processFlow ", e);
 110             throw new SvcLogicException(e.getMessage());
 111         }
 112     }
 113
 114     private void processFlowSequence(Map<String, String> inParams, SvcLogicContext ctx, SvcLogicContext localContext)
 115             throws Exception {
 116
 117         String fn = "FlowExecutorNode.processflowSequence";
 118         log.debug(fn + "Received model for flow : " + localContext.toString());
 119
 120         localContext.getAttributeKeySet().forEach(key -> log.debug(key + "=" + ctx.getAttribute(key)));
 121
 122         String flowSequence = flowSequenceGenerator.getFlowSequence(inParams, ctx, localContext);
 123
 124         log.debug("Received Flow Sequence : " + flowSequence);
 125         HashMap<Integer, Transaction> transactionMap = createTransactionMap(flowSequence, localContext);
 126         executeAllTransaction(transactionMap, ctx);
 127         log.info("Executed all the transaction successfully");
 128     }
 129
 130     private void executeAllTransaction(HashMap<Integer, Transaction> transactionMap, SvcLogicContext ctx)
 131             throws Exception {
 132
 133         String fn = "FlowExecutorNode.executeAllTransaction ";
 134         int retry = 0;
 135         FlowExecutorInterface flowExecutor;
 136         for (int key = 1; key <= transactionMap.size(); key++) {
 137             log.debug(fn + "Starting transactions ID " + key + " :)=" + retry);
 138             Transaction transaction = transactionMap.get(key);
 139             if (!preProcessor(transactionMap, transaction)) {
 140                 log.info("Skipping Transaction ID " + transaction.getTransactionId());
 141                 continue;
 142             }
 143             if (transaction.getExecutionType() != null) {
 144                 switch (transaction.getExecutionType()) {
 145                 case GRAPH:
 146                     flowExecutor = new GraphExecutor();
 147                     break;
 148                 case NODE:
 149                     flowExecutor = new NodeExecutor();
 150                     break;
 151                 case REST:
 152                     flowExecutor = new RestExecutor();
 153                     break;
 154                 default:
 155                     throw new Exception("No Executor found for transaction ID" + transaction.getTransactionId());
 156                 }
 157                 flowExecutor.execute(transaction, ctx);
 158                 ResponseAction responseAction = handleResponse(transaction, ctx);
 159
 160                 if (responseAction.getWait() != null && Integer.parseInt(responseAction.getWait()) > 0) {
 161                     log.debug(fn + "Going to Sleep .... " + responseAction.getWait());
 162                     Thread.sleep(Integer.parseInt(responseAction.getWait()) * 1000L);
 163                 }
 164                 if (responseAction.isIntermediateMessage()) {
 165                     log.debug(fn + "Sending Intermediate Message back  .... ");
 166                     sendIntermediateMessage();
 167                 }
 168                 if (responseAction.getRetry() != null && Integer.parseInt(responseAction.getRetry()) > retry) {
 169                     log.debug(fn + "Ooppss!!! We will retry again ....... ");
 170                     key--;
 171                     retry++;
 172                     log.debug(fn + "key =" + key + "retry =" + retry);
 173                 }
 174                 if (responseAction.isIgnore()) {
 175                     log.debug(fn + "Ignoring this Error and moving ahead  ....... ");
 176                     continue;
 177                 }
 178                 if (responseAction.isStop()) {
 179                     log.debug(fn + "Need to Stop  ....... ");
 180                     break;
 181                 }
 182                 if (responseAction.getJump() != null && Integer.parseInt(responseAction.getJump()) > 0) {
 183                     key = Integer.parseInt(responseAction.getJump());
 184                     key--;
 185                 }
 186                 log.debug(fn + "key =" + key + "retry =" + retry);
 187
 188             } else {
 189                 throw new Exception("Don't know how to execute transaction ID " + transaction.getTransactionId());
 190             }
 191         }
 192     }
 193
 194     protected void sendIntermediateMessage() {
 195         // TODO Auto-generated method stub
 196     }
 197
 198     protected ResponseAction handleResponse(Transaction transaction, SvcLogicContext ctx) {
 199         log.info("Handling Response for transaction Id " + transaction.getTransactionId());
 200         DefaultResponseHandler defaultHandler = new DefaultResponseHandler();
 201         return defaultHandler.handlerResponse(transaction, ctx);
 202     }
 203
 204     private boolean preProcessor(HashMap<Integer, Transaction> transactionMap, Transaction transaction)
 205             throws IOException {
 206
 207         log.debug("Starting Preprocessing Logic ");
 208         boolean runThisStep = false;
 209         try {
 210             if (transaction.getPrecheck() != null && transaction.getPrecheck().getPrecheckOptions() != null
 211                     && !transaction.getPrecheck().getPrecheckOptions().isEmpty()) {
 212
 213                 List<PrecheckOption> precheckOptions = transaction.getPrecheck().getPrecheckOptions();
 214                 for (PrecheckOption precheck : precheckOptions) {
 215                     Transaction trans = transactionMap.get(precheck.getpTransactionID());
 216                     ObjectMapper mapper = new ObjectMapper();
 217                     log.info("Mapper= " + mapper.writeValueAsString(trans));
 218                     HashMap trmap = mapper.readValue(mapper.writeValueAsString(trans), HashMap.class);
 219                     runThisStep = trmap.get(precheck.getParamName()) != null
 220                             && ((String) trmap.get(precheck.getParamName())).equalsIgnoreCase(precheck.getParamValue());
 221
 222                     if (("any").equalsIgnoreCase(transaction.getPrecheck().getPrecheckOperator()) && runThisStep) {
 223                         break;
 224                     }
 225                 }
 226             } else {
 227                 log.debug("No Pre check defined for transaction ID " + transaction.getTransactionId());
 228                 runThisStep = true;
 229             }
 230         } catch (Exception e) {
 231             log.error("Error occured when Preprocessing Logic ", e);
 232             throw e;
 233         }
 234         log.debug("Returing process current Transaction = " + runThisStep);
 235         return runThisStep;
 236     }
 237
 238     private HashMap<Integer, Transaction> createTransactionMap(String flowSequence, SvcLogicContext localContext)
 239             throws Exception {
 240
 241         ObjectMapper mapper = new ObjectMapper();
 242         Transactions transactions = mapper.readValue(flowSequence, Transactions.class);
 243         HashMap<Integer, Transaction> transMap = new HashMap<>();
 244         for (Transaction transaction : transactions.getTransactions()) {
 245             compileFlowDependencies(transaction, localContext);
 246             // parse the Transactions Object and create records in process_flow_status table
 247             // loadTransactionIntoStatus(transactions, ctx);
 248             transMap.put(transaction.getTransactionId(), transaction);
 249         }
 250         return transMap;
 251     }
 252
 253     private void compileFlowDependencies(Transaction transaction, SvcLogicContext localContext) throws Exception {
 254
 255         dbService.populateModuleAndRPC(transaction, localContext.getAttribute(VNF_TYPE));
 256         ObjectMapper mapper = new ObjectMapper();
 257         log.debug("Individual Transaction Details :" + transaction.toString());
 258
 259         if ((localContext.getAttribute(SEQUENCE_TYPE) == null) || (localContext.getAttribute(SEQUENCE_TYPE) != null
 260                 && !localContext.getAttribute(SEQUENCE_TYPE).equalsIgnoreCase(DESINGTIME))) {
 261
 262             localContext.setAttribute("artifact-content", mapper.writeValueAsString(transaction));
 263             dbService.loadSequenceIntoDB(localContext);
 264         }
 265         // get a field in transction class as transactionhandle interface and register
 266         // the Handler here for each transactions
 267     }
 268
 269 }