Merge "APEX changes to support policy disable/enable and some improvements"

3 files changed, 74 insertions, 42 deletions

diff --git a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexActivator.java b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexActivator.java
index 4d84fa394..1e2447b2e 100644
--- a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexActivator.java
+++ b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexActivator.java
@@ -36,6 +36,7 @@ import lombok.Getter;
 import lombok.Setter;
 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
 import org.onap.policy.apex.model.basicmodel.concepts.ApexRuntimeException;
+import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
 import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
 import org.onap.policy.apex.model.basicmodel.service.ModelService;
 import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
@@ -88,6 +89,7 @@ public class ApexActivator {
 
     // The engine service
     private EngineService apexEngineService;
+    private AxArtifactKey engineKey;
 
     /**
      * Instantiate the activator for the Apex engine as a complete service.
@@ -112,8 +114,9 @@ public class ApexActivator {
             int totalInstanceCount = apexParametersMap.values().stream()
                 .mapToInt(p -> p.getEngineServiceParameters().getInstanceCount()).sum();
             apexParameters.getEngineServiceParameters().setInstanceCount(totalInstanceCount);
+            engineKey = apexParameters.getEngineServiceParameters().getEngineKey();
             instantiateEngine(apexParameters);
-            setUpModelMarhsallerAndUnmarshaller(apexParameters);
+            setUpModelMarshallerAndUnmarshaller(apexParameters);
         } catch (final Exception e) {
             LOGGER.debug(APEX_ENGINE_FAILED_MSG, e);
             throw new ApexActivatorException(APEX_ENGINE_FAILED_MSG, e);
@@ -122,7 +125,7 @@ public class ApexActivator {
         LOGGER.debug("Apex engine started as a service");
     }
 
-    private void setUpModelMarhsallerAndUnmarshaller(ApexParameters apexParameters) throws ApexException {
+    private void setUpModelMarshallerAndUnmarshaller(ApexParameters apexParameters) throws ApexException {
         policyModelsMap = new LinkedHashMap<>();
         Map<String, EventHandlerParameters> inputParametersMap = new LinkedHashMap<>();
         Map<String, EventHandlerParameters> outputParametersMap = new LinkedHashMap<>();
@@ -149,8 +152,7 @@ public class ApexActivator {
                 try {
                     final String policyModelString = TextFileUtils
                         .getTextFileAsString(apexParams.getEngineServiceParameters().getPolicyModelFileName());
-                    AxPolicyModel policyModel = EngineServiceImpl
-                        .createModel(apexParams.getEngineServiceParameters().getEngineKey(), policyModelString);
+                    AxPolicyModel policyModel = EngineServiceImpl.createModel(engineKey, policyModelString);
                     policyModelsMap.put(apexParamsEntry.getKey(), policyModel);
                 } catch (ApexException | IOException e) {
                     throw new ApexRuntimeException("Failed to create the apex model.", e);
@@ -160,10 +162,10 @@ public class ApexActivator {
         AxPolicyModel finalPolicyModel = aggregatePolicyModels(policyModelsMap);
 
         // Set the policy model in the engine
-        apexEngineService.updateModel(apexParameters.getEngineServiceParameters().getEngineKey(), finalPolicyModel,
-            true);
+        apexEngineService.updateModel(engineKey, finalPolicyModel, true);
 
-        setUpMarshallerAndUnmarshaller(apexParameters.getEngineServiceParameters(), inputParametersMap,
+        handleExistingMarshallerAndUnmarshaller(inputParametersMap, outputParametersMap);
+        setUpNewMarshallerAndUnmarshaller(apexParameters.getEngineServiceParameters(), inputParametersMap,
             outputParametersMap);
 
         // Wire up pairings between marhsallers and unmarshallers
@@ -197,9 +199,10 @@ public class ApexActivator {
         return finalPolicyModel;
     }
 
-    private void setUpMarshallerAndUnmarshaller(EngineServiceParameters engineServiceParameters,
+    private void setUpNewMarshallerAndUnmarshaller(EngineServiceParameters engineServiceParameters,
         Map<String, EventHandlerParameters> inputParametersMap, Map<String, EventHandlerParameters> outputParametersMap)
         throws ApexEventException {
+
         // Producer parameters specify what event marshalers to handle events leaving Apex are
         // set up and how they are set up
         for (Entry<String, EventHandlerParameters> outputParameters : outputParametersMap.entrySet()) {
@@ -220,9 +223,25 @@ public class ApexActivator {
         }
     }
 
+    private void handleExistingMarshallerAndUnmarshaller(Map<String, EventHandlerParameters> inputParametersMap,
+        Map<String, EventHandlerParameters> outputParametersMap) {
+        // stop and remove any marshaller/unmarshaller that is part of a policy that is undeployed
+        marshallerMap.entrySet().stream()
+            .filter(marshallerEntry -> !outputParametersMap.containsKey(marshallerEntry.getKey()))
+            .forEach(marshallerEntry -> marshallerEntry.getValue().stop());
+        marshallerMap.keySet().removeIf(marshallerKey -> !outputParametersMap.containsKey(marshallerKey));
+        unmarshallerMap.entrySet().stream()
+            .filter(unmarshallerEntry -> !inputParametersMap.containsKey(unmarshallerEntry.getKey()))
+            .forEach(unmarshallerEntry -> unmarshallerEntry.getValue().stop());
+        unmarshallerMap.keySet().removeIf(unmarshallerKey -> !inputParametersMap.containsKey(unmarshallerKey));
+
+        // If a marshaller/unmarshaller is already initialized, they don't need to be reinitialized during model update.
+        outputParametersMap.keySet().removeIf(marshallerMap::containsKey);
+        inputParametersMap.keySet().removeIf(unmarshallerMap::containsKey);
+    }
+
     private void instantiateEngine(ApexParameters apexParameters) throws ApexException {
-        if (null != apexEngineService
-            && apexEngineService.getKey().equals(apexParameters.getEngineServiceParameters().getEngineKey())) {
+        if (null != apexEngineService && apexEngineService.getKey().equals(engineKey)) {
             throw new ApexException("Apex Engine already initialized.");
         }
         // Create engine with specified thread count
@@ -284,9 +303,8 @@ public class ApexActivator {
      */
     public void updateModel(Map<ToscaPolicyIdentifier, ApexParameters> apexParamsMap) throws ApexException {
         try {
-            shutdownMarshallerAndUnmarshaller();
             ApexParameters apexParameters = apexParamsMap.values().iterator().next();
-            setUpModelMarhsallerAndUnmarshaller(apexParameters);
+            setUpModelMarshallerAndUnmarshaller(apexParameters);
         } catch (final Exception e) {
             LOGGER.debug(APEX_ENGINE_FAILED_MSG, e);
             throw new ApexActivatorException(APEX_ENGINE_FAILED_MSG, e);
diff --git a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexMain.java b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexMain.java
index 0e38230be..459cdff24 100644
--- a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexMain.java
+++ b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexMain.java
@@ -2,6 +2,7 @@
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
  *  Modification Copyright (C) 2019-2020 Nordix Foundation.
+ *  Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,6 +31,7 @@ import java.util.Map.Entry;
 import java.util.TreeMap;
 import lombok.Getter;
 import lombok.Setter;
+import org.onap.policy.apex.core.engine.EngineParameters;
 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
 import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
 import org.onap.policy.apex.model.basicmodel.service.ModelService;
@@ -40,6 +42,7 @@ import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
 import org.onap.policy.apex.service.parameters.ApexParameterHandler;
 import org.onap.policy.apex.service.parameters.ApexParameters;
 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
+import org.onap.policy.common.parameters.ParameterService;
 import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
 import org.slf4j.ext.XLogger;
 import org.slf4j.ext.XLoggerFactory;
@@ -61,6 +64,11 @@ public class ApexMain {
     @Getter
     private Map<ToscaPolicyIdentifier, ApexParameters> apexParametersMap;
 
+    //engineParameters are aggregated in case of multiple policies
+    private EngineParameters aggregatedEngineParameters;
+
+    private ApexParameterHandler apexParameterHandler = new ApexParameterHandler();
+
     @Getter
     @Setter(lombok.AccessLevel.PRIVATE)
     private volatile boolean alive = false;
@@ -73,13 +81,14 @@ public class ApexMain {
     public ApexMain(final String[] args) {
         LOGGER.entry("Starting Apex service with parameters " + Arrays.toString(args) + " . . .");
         apexParametersMap = new LinkedHashMap<>();
+        aggregatedEngineParameters = new EngineParameters();
         try {
             apexParametersMap.put(new ToscaPolicyIdentifier(), populateApexParameters(args));
         } catch (ApexException e) {
             LOGGER.error(APEX_SERVICE_FAILED_MSG, e);
             return;
         }
-
+        apexParameterHandler.registerParameters(apexParametersMap.values().iterator().next());
         // Now, create the activator for the Apex service
         activator = new ApexActivator(apexParametersMap);
 
@@ -105,6 +114,7 @@ public class ApexMain {
      */
     public ApexMain(Map<ToscaPolicyIdentifier, String[]> policyArgumentsMap) throws ApexException {
         apexParametersMap = new LinkedHashMap<>();
+        aggregatedEngineParameters = new EngineParameters();
         for (Entry<ToscaPolicyIdentifier, String[]> policyArgsEntry: policyArgumentsMap.entrySet()) {
             try {
                 apexParametersMap.put(policyArgsEntry.getKey(), populateApexParameters(policyArgsEntry.getValue()));
@@ -117,6 +127,11 @@ public class ApexMain {
             LOGGER.error(APEX_SERVICE_FAILED_MSG);
             return;
         }
+
+        // Set the aggregated engineParameters which will be used later while creating the engine
+        ApexParameters apexParameters = apexParametersMap.values().iterator().next();
+        apexParameters.getEngineServiceParameters().setEngineParameters(aggregatedEngineParameters);
+        apexParameterHandler.registerParameters(apexParameters);
         // Now, create the activator for the Apex service
         activator = new ApexActivator(apexParametersMap);
 
@@ -143,7 +158,10 @@ public class ApexMain {
      * @throws ApexException on errors
      */
     public void updateModel(Map<ToscaPolicyIdentifier, String[]> policyArgsMap) throws ApexException {
+        // flag that determines if any policy received in PDP_UPDATE is already deployed in the engine
+        boolean isAnyPolicyDeployed = policyArgsMap.keySet().stream().anyMatch(apexParametersMap::containsKey);
         apexParametersMap.clear();
+        aggregatedEngineParameters = new EngineParameters();
         AxContextAlbums albums = ModelService.getModel(AxContextAlbums.class);
         Map<AxArtifactKey, AxContextAlbum> albumsMap = new TreeMap<>();
         for (Entry<ToscaPolicyIdentifier, String[]> policyArgsEntry : policyArgsMap.entrySet()) {
@@ -155,14 +173,19 @@ public class ApexMain {
                     policyArgsEntry.getKey().getVersion(), e);
             }
         }
+        // Set the aggregated engineParameters
+        ApexParameters apexParameters = apexParametersMap.values().iterator().next();
+        apexParameters.getEngineServiceParameters().setEngineParameters(aggregatedEngineParameters);
+        ParameterService.clear();
+        apexParameterHandler.registerParameters(apexParameters);
         try {
-            if (albumsMap.isEmpty()) {
+            if (albumsMap.isEmpty() && !isAnyPolicyDeployed) {
                 // clear context since none of the policies' context albums has to be retained
                 // this could be because all policies have a major version change,
                 // or a full set of new policies are received in the update message
                 activator.terminate();
-                // ParameterService is cleared when activator is terminated. Register the engine parameters in this case
-                new ApexParameterHandler().registerParameters(apexParametersMap.values().iterator().next());
+                // ParameterService is cleared when activator is terminated. Register the parameters again in this case
+                apexParameterHandler.registerParameters(apexParameters);
                 activator = new ApexActivator(apexParametersMap);
                 activator.initialize();
                 setAlive(true);
@@ -218,10 +241,6 @@ public class ApexMain {
         ApexParameters axParameters;
         // Read the parameters
         try {
-            ApexParameterHandler apexParameterHandler = new ApexParameterHandler();
-            // In case of multiple policies received from PAP, do not clear ParameterService if parameters of one policy
-            // already registered
-            apexParameterHandler.setKeepParameterServiceFlag(null != apexParametersMap && !apexParametersMap.isEmpty());
             axParameters = apexParameterHandler.getParameters(arguments);
         } catch (final Exception e) {
             LOGGER.error("Cannot create APEX Parameters from the arguments provided.", e);
@@ -244,9 +263,17 @@ public class ApexMain {
                 ehParameterEntry.getValue().setName(ehParameterEntry.getKey());
             }
         }
+        aggregateEngineParameters(axParameters.getEngineServiceParameters().getEngineParameters());
         return axParameters;
     }
 
+    private void aggregateEngineParameters(EngineParameters engineParameters) {
+        aggregatedEngineParameters.getTaskParameters().addAll(engineParameters.getTaskParameters());
+        aggregatedEngineParameters.getExecutorParameterMap().putAll(engineParameters.getExecutorParameterMap());
+        aggregatedEngineParameters.getContextParameters().getSchemaParameters().getSchemaHelperParameterMap()
+            .putAll(engineParameters.getContextParameters().getSchemaParameters().getSchemaHelperParameterMap());
+    }
+
     /**
      * Shut down Execution.
      *
diff --git a/services/services-engine/src/main/java/org/onap/policy/apex/service/parameters/ApexParameterHandler.java b/services/services-engine/src/main/java/org/onap/policy/apex/service/parameters/ApexParameterHandler.java
index 02d39733c..f88733f60 100644
--- a/services/services-engine/src/main/java/org/onap/policy/apex/service/parameters/ApexParameterHandler.java
+++ b/services/services-engine/src/main/java/org/onap/policy/apex/service/parameters/ApexParameterHandler.java
@@ -2,19 +2,20 @@
  * ============LICENSE_START=======================================================
  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
  *  Modifications Copyright (C) 2019 Nordix Foundation.
+ *  Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- * 
+ *
  *      http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
- * 
+ *
  * SPDX-License-Identifier: Apache-2.0
  * ============LICENSE_END=========================================================
  */
@@ -24,7 +25,6 @@ package org.onap.policy.apex.service.parameters;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import java.io.FileReader;
-import lombok.Setter;
 import org.onap.policy.apex.core.engine.EngineParameters;
 import org.onap.policy.apex.service.engine.main.ApexCommandLineArguments;
 import org.onap.policy.apex.service.parameters.carriertechnology.CarrierTechnologyParameters;
@@ -46,9 +46,6 @@ import org.slf4j.ext.XLoggerFactory;
 public class ApexParameterHandler {
     private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexParameterHandler.class);
 
-    @Setter
-    private boolean keepParameterServiceFlag;
-
     /**
      * Read the parameters from the parameter file.
      *
@@ -57,11 +54,8 @@ public class ApexParameterHandler {
      * @throws ParameterException on parameter exceptions
      */
     public ApexParameters getParameters(final ApexCommandLineArguments arguments) throws ParameterException {
-        // when populating parameters for multiple policies, do not clear the ParameterService already registered
-        // otherwise clear all existing parameters
-        if (!keepParameterServiceFlag) {
-            ParameterService.clear();
-        }
+
+        ParameterService.clear();
 
         ApexParameters parameters = null;
 
@@ -70,11 +64,11 @@ public class ApexParameterHandler {
             // Register the adapters for our carrier technologies and event protocols with GSON
             // @formatter:off
             final Gson gson = new GsonBuilder()
-                            .registerTypeAdapter(EngineParameters.class, 
+                            .registerTypeAdapter(EngineParameters.class,
                                             new EngineServiceParametersJsonAdapter())
-                            .registerTypeAdapter(CarrierTechnologyParameters.class, 
+                            .registerTypeAdapter(CarrierTechnologyParameters.class,
                                             new CarrierTechnologyParametersJsonAdapter())
-                            .registerTypeAdapter(EventProtocolParameters.class, 
+                            .registerTypeAdapter(EventProtocolParameters.class,
                                             new EventProtocolParametersJsonAdapter())
                             .create();
             // @formatter:on
@@ -118,13 +112,6 @@ public class ApexParameterHandler {
             LOGGER.info(returnMessage);
         }
 
-        // engine parameters in multiple policies are expected to be same.
-        // no need to do registration if already registered
-        if (!keepParameterServiceFlag) {
-            // Register the parameters with the parameter service
-            registerParameters(parameters);
-        }
-
         return parameters;
     }