gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2019-2021 Nordix Foundation.
   4  *  Modifications Copyright (C) 2020-2021 Bell Canada. All rights reserved.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.apex.services.onappf.handler;
  23
  24 import java.io.File;
  25 import java.io.IOException;
  26 import java.util.ArrayList;
  27 import java.util.HashSet;
  28 import java.util.LinkedHashMap;
  29 import java.util.List;
  30 import java.util.Map;
  31 import java.util.Set;
  32 import java.util.stream.Collectors;
  33 import org.onap.policy.apex.core.engine.EngineParameters;
  34 import org.onap.policy.apex.core.engine.TaskParameters;
  35 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
  36 import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
  37 import org.onap.policy.apex.model.basicmodel.concepts.AxKeyInfo;
  38 import org.onap.policy.apex.model.basicmodel.concepts.AxKeyInformation;
  39 import org.onap.policy.apex.model.basicmodel.service.ModelService;
  40 import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbum;
  41 import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbums;
  42 import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchema;
  43 import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchemas;
  44 import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
  45 import org.onap.policy.apex.model.eventmodel.concepts.AxEvent;
  46 import org.onap.policy.apex.model.eventmodel.concepts.AxEvents;
  47 import org.onap.policy.apex.model.policymodel.concepts.AxPolicies;
  48 import org.onap.policy.apex.model.policymodel.concepts.AxPolicy;
  49 import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
  50 import org.onap.policy.apex.model.policymodel.concepts.AxTask;
  51 import org.onap.policy.apex.model.policymodel.concepts.AxTasks;
  52 import org.onap.policy.apex.service.engine.main.ApexMain;
  53 import org.onap.policy.apex.service.parameters.ApexParameterConstants;
  54 import org.onap.policy.apex.service.parameters.ApexParameters;
  55 import org.onap.policy.apex.services.onappf.exception.ApexStarterException;
  56 import org.onap.policy.common.parameters.ParameterService;
  57 import org.onap.policy.common.utils.coder.CoderException;
  58 import org.onap.policy.common.utils.coder.StandardCoder;
  59 import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
  60 import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
  61 import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate;
  62 import org.onap.policy.models.tosca.authorative.concepts.ToscaTopologyTemplate;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 /**
  67  * This class instantiates the Apex Engine based on instruction from PAP.
  68  *
  69  * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
  70  */
  71 public class ApexEngineHandler {
  72
  73     private static final Logger LOGGER = LoggerFactory.getLogger(ApexEngineHandler.class);
  74
  75     private Map<ToscaConceptIdentifier, ApexMain> apexMainMap;
  76
  77     /**
  78      * Constructs the object. Extracts the config and model files from each policy and instantiates the apex engine.
  79      *
  80      * @param policies the list of policies
  81      * @throws ApexStarterException if the apex engine instantiation failed using the policies passed
  82      */
  83     public ApexEngineHandler(List<ToscaPolicy> policies) throws ApexStarterException {
  84         LOGGER.debug("Starting apex engine.");
  85         apexMainMap = initiateApexEngineForPolicies(policies);
  86         if (apexMainMap.isEmpty()) {
  87             ModelService.clear();
  88             ParameterService.clear();
  89             throw new ApexStarterException("Apex Engine failed to start.");
  90         }
  91     }
  92
  93     /**
  94      * Updates the Apex Engine with the policy model created from new list of policies.
  95      *
  96      * @param policies the list of policies
  97      * @throws ApexStarterException if the apex engine instantiation failed using the policies passed
  98      */
  99     public void updateApexEngine(List<ToscaPolicy> policies) throws ApexStarterException {
 100         List<ToscaConceptIdentifier> runningPolicies = getRunningPolicies();
 101         List<ToscaPolicy> policiesToDeploy = policies.stream()
 102             .filter(policy -> !runningPolicies.contains(policy.getIdentifier())).collect(Collectors.toList());
 103         List<ToscaConceptIdentifier> policiesToUnDeploy = runningPolicies.stream()
 104             .filter(polId -> policies.stream().noneMatch(policy -> policy.getIdentifier().equals(polId)))
 105             .collect(Collectors.toList());
 106         Map<ToscaConceptIdentifier, ApexMain> undeployedPoliciesMainMap = new LinkedHashMap<>();
 107         policiesToUnDeploy.forEach(policyId -> {
 108             ApexMain apexMain = apexMainMap.get(policyId);
 109             try {
 110                 apexMain.shutdown();
 111                 undeployedPoliciesMainMap.put(policyId, apexMain);
 112                 apexMainMap.remove(policyId);
 113             } catch (ApexException e) {
 114                 LOGGER.error("Shutting down policy {} failed", policyId, e);
 115             }
 116         });
 117         if (!undeployedPoliciesMainMap.isEmpty()) {
 118             updateModelAndParameterServices(undeployedPoliciesMainMap);
 119         }
 120         if (!policiesToDeploy.isEmpty()) {
 121             Map<ToscaConceptIdentifier, ApexMain> mainMap = initiateApexEngineForPolicies(policiesToDeploy);
 122             if (mainMap.isEmpty()) {
 123                 throw new ApexStarterException("Updating the APEX engine with new policies failed.");
 124             }
 125             apexMainMap.putAll(mainMap);
 126         }
 127         if (apexMainMap.isEmpty()) {
 128             ModelService.clear();
 129             ParameterService.clear();
 130         }
 131     }
 132
 133     /**
 134      * Clear the corresponding items from ModelService and ParameterService.
 135      *
 136      * @param undeployedPoliciesMainMap the policies that are undeployed
 137      */
 138     private void updateModelAndParameterServices(Map<ToscaConceptIdentifier, ApexMain> undeployedPoliciesMainMap) {
 139         Set<String> inputParamKeysToRetain = new HashSet<>();
 140         Set<String> outputParamKeysToRetain = new HashSet<>();
 141         List<TaskParameters> taskParametersToRetain = new ArrayList<>();
 142         List<String> executorParamKeysToRetain = new ArrayList<>();
 143         List<String> schemaParamKeysToRetain = new ArrayList<>();
 144
 145         List<AxArtifactKey> keyInfoKeystoRetain = new ArrayList<>();
 146         List<AxArtifactKey> schemaKeystoRetain = new ArrayList<>();
 147         List<AxArtifactKey> eventKeystoRetain = new ArrayList<>();
 148         List<AxArtifactKey> albumKeystoRetain = new ArrayList<>();
 149         List<AxArtifactKey> taskKeystoRetain = new ArrayList<>();
 150         List<AxArtifactKey> policyKeystoRetain = new ArrayList<>();
 151
 152         apexMainMap.values().forEach(main -> {
 153             inputParamKeysToRetain.addAll(main.getApexParameters().getEventInputParameters().keySet());
 154             outputParamKeysToRetain.addAll(main.getApexParameters().getEventOutputParameters().keySet());
 155             taskParametersToRetain.addAll(
 156                 main.getApexParameters().getEngineServiceParameters().getEngineParameters().getTaskParameters());
 157             executorParamKeysToRetain.addAll(main.getApexParameters().getEngineServiceParameters().getEngineParameters()
 158                 .getExecutorParameterMap().keySet());
 159             schemaParamKeysToRetain.addAll(main.getApexParameters().getEngineServiceParameters().getEngineParameters()
 160                 .getContextParameters().getSchemaParameters().getSchemaHelperParameterMap().keySet());
 161
 162             keyInfoKeystoRetain
 163                 .addAll(main.getActivator().getPolicyModel().getKeyInformation().getKeyInfoMap().keySet());
 164             schemaKeystoRetain.addAll(main.getActivator().getPolicyModel().getSchemas().getSchemasMap().keySet());
 165             eventKeystoRetain.addAll(main.getActivator().getPolicyModel().getEvents().getEventMap().keySet());
 166             albumKeystoRetain.addAll(main.getActivator().getPolicyModel().getAlbums().getAlbumsMap().keySet());
 167             taskKeystoRetain.addAll(main.getActivator().getPolicyModel().getTasks().getTaskMap().keySet());
 168             policyKeystoRetain.addAll(main.getActivator().getPolicyModel().getPolicies().getPolicyMap().keySet());
 169         });
 170         for (ApexMain main : undeployedPoliciesMainMap.values()) {
 171             ApexParameters existingParameters = ParameterService.get(ApexParameterConstants.MAIN_GROUP_NAME);
 172             List<String> eventInputParamKeysToRemove = main.getApexParameters().getEventInputParameters().keySet()
 173                 .stream().filter(key -> !inputParamKeysToRetain.contains(key)).collect(Collectors.toList());
 174             List<String> eventOutputParamKeysToRemove = main.getApexParameters().getEventOutputParameters().keySet()
 175                 .stream().filter(key -> !outputParamKeysToRetain.contains(key)).collect(Collectors.toList());
 176             eventInputParamKeysToRemove.forEach(existingParameters.getEventInputParameters()::remove);
 177             eventOutputParamKeysToRemove.forEach(existingParameters.getEventOutputParameters()::remove);
 178             EngineParameters engineParameters =
 179                 main.getApexParameters().getEngineServiceParameters().getEngineParameters();
 180             final List<TaskParameters> taskParametersToRemove = engineParameters.getTaskParameters().stream()
 181                 .filter(taskParameter -> !taskParametersToRetain.contains(taskParameter)).collect(Collectors.toList());
 182             final List<String> executorParamKeysToRemove = engineParameters.getExecutorParameterMap().keySet().stream()
 183                 .filter(key -> !executorParamKeysToRetain.contains(key)).collect(Collectors.toList());
 184             final List<String> schemaParamKeysToRemove =
 185                 engineParameters.getContextParameters().getSchemaParameters().getSchemaHelperParameterMap().keySet()
 186                     .stream().filter(key -> !schemaParamKeysToRetain.contains(key)).collect(Collectors.toList());
 187             EngineParameters aggregatedEngineParameters =
 188                 existingParameters.getEngineServiceParameters().getEngineParameters();
 189             aggregatedEngineParameters.getTaskParameters().removeAll(taskParametersToRemove);
 190             executorParamKeysToRemove.forEach(aggregatedEngineParameters.getExecutorParameterMap()::remove);
 191             schemaParamKeysToRemove.forEach(aggregatedEngineParameters.getContextParameters().getSchemaParameters()
 192                 .getSchemaHelperParameterMap()::remove);
 193
 194             final AxPolicyModel policyModel = main.getActivator().getPolicyModel();
 195             final List<AxArtifactKey> keyInfoKeystoRemove = policyModel.getKeyInformation().getKeyInfoMap().keySet()
 196                 .stream().filter(key -> !keyInfoKeystoRetain.contains(key)).collect(Collectors.toList());
 197             final List<AxArtifactKey> schemaKeystoRemove = policyModel.getSchemas().getSchemasMap().keySet().stream()
 198                 .filter(key -> !schemaKeystoRetain.contains(key)).collect(Collectors.toList());
 199             final List<AxArtifactKey> eventKeystoRemove = policyModel.getEvents().getEventMap().keySet().stream()
 200                 .filter(key -> !eventKeystoRetain.contains(key)).collect(Collectors.toList());
 201             final List<AxArtifactKey> albumKeystoRemove = policyModel.getAlbums().getAlbumsMap().keySet().stream()
 202                 .filter(key -> !albumKeystoRetain.contains(key)).collect(Collectors.toList());
 203             final List<AxArtifactKey> taskKeystoRemove = policyModel.getTasks().getTaskMap().keySet().stream()
 204                 .filter(key -> !taskKeystoRetain.contains(key)).collect(Collectors.toList());
 205             final List<AxArtifactKey> policyKeystoRemove = policyModel.getPolicies().getPolicyMap().keySet().stream()
 206                 .filter(key -> !policyKeystoRetain.contains(key)).collect(Collectors.toList());
 207
 208             final Map<AxArtifactKey, AxKeyInfo> keyInfoMap =
 209                 ModelService.getModel(AxKeyInformation.class).getKeyInfoMap();
 210             final Map<AxArtifactKey, AxContextSchema> schemasMap =
 211                 ModelService.getModel(AxContextSchemas.class).getSchemasMap();
 212             final Map<AxArtifactKey, AxEvent> eventMap = ModelService.getModel(AxEvents.class).getEventMap();
 213             final Map<AxArtifactKey, AxContextAlbum> albumsMap =
 214                 ModelService.getModel(AxContextAlbums.class).getAlbumsMap();
 215             final Map<AxArtifactKey, AxTask> taskMap = ModelService.getModel(AxTasks.class).getTaskMap();
 216             final Map<AxArtifactKey, AxPolicy> policyMap = ModelService.getModel(AxPolicies.class).getPolicyMap();
 217
 218             keyInfoKeystoRemove.forEach(keyInfoMap::remove);
 219             schemaKeystoRemove.forEach(schemasMap::remove);
 220             eventKeystoRemove.forEach(eventMap::remove);
 221             albumKeystoRemove.forEach(albumsMap::remove);
 222             taskKeystoRemove.forEach(taskMap::remove);
 223             policyKeystoRemove.forEach(policyMap::remove);
 224         }
 225     }
 226
 227     private Map<ToscaConceptIdentifier, ApexMain> initiateApexEngineForPolicies(List<ToscaPolicy> policies)
 228         throws ApexStarterException {
 229         Map<ToscaConceptIdentifier, ApexMain> mainMap = new LinkedHashMap<>();
 230         for (ToscaPolicy policy : policies) {
 231             String policyName = policy.getIdentifier().getName();
 232             final StandardCoder standardCoder = new StandardCoder();
 233             ToscaServiceTemplate toscaServiceTemplate = new ToscaServiceTemplate();
 234             ToscaTopologyTemplate toscaTopologyTemplate = new ToscaTopologyTemplate();
 235             toscaTopologyTemplate.setPolicies(List.of(Map.of(policyName, policy)));
 236             toscaServiceTemplate.setToscaTopologyTemplate(toscaTopologyTemplate);
 237             File file;
 238             try {
 239                 file = File.createTempFile(policyName, ".json");
 240                 standardCoder.encode(file, toscaServiceTemplate);
 241             } catch (CoderException | IOException e) {
 242                 throw new ApexStarterException(e);
 243             }
 244             final String[] apexArgs = {"-p", file.getAbsolutePath()};
 245             LOGGER.info("Starting apex engine for policy {}", policy.getIdentifier());
 246             try {
 247                 ApexMain apexMain = new ApexMain(apexArgs);
 248                 mainMap.put(policy.getIdentifier(), apexMain);
 249             } catch (Exception e) {
 250                 LOGGER.error("Execution of policy {} failed", policy.getIdentifier(), e);
 251             }
 252         }
 253         return mainMap;
 254     }
 255
 256     /**
 257      * Method to get the APEX engine statistics.
 258      */
 259     public List<AxEngineModel> getEngineStats() {
 260         // engineStats from all the apexMain instances running individual tosca policies are combined here.
 261         return apexMainMap.values().stream().filter(apexMain -> (null != apexMain && apexMain.isAlive()))
 262             .flatMap(m -> m.getEngineStats().stream()).collect(Collectors.toList());
 263     }
 264
 265     /**
 266      * Method to check whether the apex engine is running or not.
 267      */
 268     public boolean isApexEngineRunning() {
 269         return apexMainMap.values().stream().anyMatch(apexMain -> (null != apexMain && apexMain.isAlive()));
 270     }
 271
 272     /**
 273      * Method that return the list of running policies in the apex engine.
 274      */
 275     public List<ToscaConceptIdentifier> getRunningPolicies() {
 276         return new ArrayList<>(apexMainMap.keySet());
 277     }
 278
 279     /**
 280      * Method to shut down the apex engine.
 281      */
 282     public void shutdown() throws ApexStarterException {
 283         try {
 284             LOGGER.debug("Shutting down apex engine.");
 285             for (ApexMain apexMain : apexMainMap.values()) {
 286                 apexMain.shutdown();
 287             }
 288             apexMainMap.clear();
 289             ModelService.clear();
 290             ParameterService.clear();
 291         } catch (final ApexException e) {
 292             throw new ApexStarterException(e);
 293         }
 294     }
 295 }