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