Improve handling of multiple policy in APEX PDP
Change-Id: Ic4adf5bd8876dc31fc93993298e90389baaa2c39
Issue-ID: POLICY-2883
Signed-off-by: a.sreekumar <ajith.sreekumar@bell.ca>
diff --git a/examples/examples-servlet/src/main/java/org/onap/policy/apex/examples/servlet/ApexServletListener.java b/examples/examples-servlet/src/main/java/org/onap/policy/apex/examples/servlet/ApexServletListener.java
index 80ef735..3375ff8 100644
--- a/examples/examples-servlet/src/main/java/org/onap/policy/apex/examples/servlet/ApexServletListener.java
+++ b/examples/examples-servlet/src/main/java/org/onap/policy/apex/examples/servlet/ApexServletListener.java
@@ -1,6 +1,7 @@
/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2018 Ericsson. All rights reserved.
+ * Modifications Copyright (C) 2021 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.
@@ -75,7 +76,11 @@
}
// Initialize apex
- apexMain = new ApexMain(argsList.toArray(new String[argsList.size()]));
+ try {
+ apexMain = new ApexMain(argsList.toArray(new String[argsList.size()]));
+ } catch (ApexException e) {
+ LOGGER.error("Apex Engine initialization failed", e);
+ }
}
diff --git a/model/policy-model/src/main/java/org/onap/policy/apex/model/policymodel/handling/PolicyModelMerger.java b/model/policy-model/src/main/java/org/onap/policy/apex/model/policymodel/handling/PolicyModelMerger.java
index 90d5b7f..ae0dda8 100644
--- a/model/policy-model/src/main/java/org/onap/policy/apex/model/policymodel/handling/PolicyModelMerger.java
+++ b/model/policy-model/src/main/java/org/onap/policy/apex/model/policymodel/handling/PolicyModelMerger.java
@@ -2,7 +2,7 @@
* ============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.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -145,7 +145,16 @@
return mergedPolicyModel;
}
- private static <V> void checkForDuplicateItem(Map<AxArtifactKey, V> mergedItemsMap,
+ /**
+ * Method to check for duplicate items.
+ *
+ * @param <V> the concept type
+ * @param mergedItemsMap the map to which items are copied
+ * @param copyOverItemsMap the map from where items are copied
+ * @param errorMessage error message in case of any duplicate concepts
+ * @param itemType the type of concept to specify distinguished error messages
+ */
+ public static <V> void checkForDuplicateItem(Map<AxArtifactKey, V> mergedItemsMap,
Map<AxArtifactKey, V> copyOverItemsMap, StringBuilder errorMessage, String itemType) {
for (Entry<AxArtifactKey, V> entry : copyOverItemsMap.entrySet()) {
V item = mergedItemsMap.get(entry.getKey());
diff --git a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/test/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RestRequestorTest.java b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/test/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RestRequestorTest.java
index 2a36006..0130464 100644
--- a/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/test/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RestRequestorTest.java
+++ b/plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/test/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/RestRequestorTest.java
@@ -2,6 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2020 Nordix Foundation.
+ * Modifications Copyright (C) 2021 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.
@@ -21,6 +22,7 @@
package org.onap.policy.apex.plugins.event.carrier.restrequestor;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -267,28 +269,14 @@
/**
* Test REST requestor producer alone.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
*/
@Test
- public void testRestRequestorProducerAlone() throws MessagingException, ApexException, IOException {
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
+ public void testRestRequestorProducerAlone() {
final String[] args = {"src/test/resources/prodcons/File2RESTRequest2FileGetProducerAlone.json"};
+ assertThatThrownBy(() -> new ApexMain(args))
+ .hasRootCauseMessage("REST Requestor producer (RestRequestorProducer) "
+ + "must run in peered requestor mode with a REST Requestor consumer");
- ApexMain apexMain = new ApexMain(args);
- apexMain.shutdown();
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- assertTrue(outString.contains("REST Requestor producer (RestRequestorProducer) "
- + "must run in peered requestor mode with a REST Requestor consumer"));
}
/**
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 cadcc7e..3451c12 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
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2021 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -22,35 +22,41 @@
package org.onap.policy.apex.service.engine.main;
-import java.util.AbstractMap;
-import java.util.ArrayList;
import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Collectors;
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.concepts.AxKeyInfo;
+import org.onap.policy.apex.model.basicmodel.concepts.AxKeyInformation;
import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
import org.onap.policy.apex.model.basicmodel.service.ModelService;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbum;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbums;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchema;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchemas;
import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
+import org.onap.policy.apex.model.eventmodel.concepts.AxEvent;
+import org.onap.policy.apex.model.eventmodel.concepts.AxEvents;
+import org.onap.policy.apex.model.policymodel.concepts.AxPolicies;
+import org.onap.policy.apex.model.policymodel.concepts.AxPolicy;
import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
+import org.onap.policy.apex.model.policymodel.concepts.AxTask;
+import org.onap.policy.apex.model.policymodel.concepts.AxTasks;
import org.onap.policy.apex.model.policymodel.handling.PolicyModelMerger;
import org.onap.policy.apex.service.engine.event.ApexEventException;
import org.onap.policy.apex.service.engine.runtime.EngineService;
import org.onap.policy.apex.service.engine.runtime.impl.EngineServiceImpl;
+import org.onap.policy.apex.service.parameters.ApexParameterConstants;
import org.onap.policy.apex.service.parameters.ApexParameters;
import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
import org.onap.policy.common.parameters.ParameterService;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.slf4j.ext.XLogger;
import org.slf4j.ext.XLoggerFactory;
@@ -66,13 +72,12 @@
// The logger for this class
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexActivator.class);
- // The parameters of the Apex activator when running with multiple policies
@Getter
@Setter
- private Map<ToscaConceptIdentifier, ApexParameters> apexParametersMap;
+ private ApexParameters apexParameters;
@Getter
- Map<ToscaConceptIdentifier, AxPolicyModel> policyModelsMap;
+ private AxPolicyModel policyModel;
// Event unmarshalers are used to receive events asynchronously into Apex
private final Map<String, ApexEventUnmarshaller> unmarshallerMap = new LinkedHashMap<>();
@@ -93,10 +98,10 @@
/**
* Instantiate the activator for the Apex engine as a complete service.
*
- * @param parametersMap the apex parameters map for the Apex service
+ * @param parameters the apex parameters for the Apex service
*/
- public ApexActivator(Map<ToscaConceptIdentifier, ApexParameters> parametersMap) {
- apexParametersMap = parametersMap;
+ public ApexActivator(ApexParameters parameters) {
+ apexParameters = parameters;
}
/**
@@ -108,96 +113,131 @@
LOGGER.debug("Apex engine starting as a service . . .");
try {
- ApexParameters apexParameters = apexParametersMap.values().iterator().next();
- // totalInstanceCount is the sum of instance counts required as per each policy
- int totalInstanceCount = apexParametersMap.values().stream()
- .mapToInt(p -> p.getEngineServiceParameters().getInstanceCount()).sum();
- apexParameters.getEngineServiceParameters().setInstanceCount(totalInstanceCount);
- engineKey = apexParameters.getEngineServiceParameters().getEngineKey();
instantiateEngine(apexParameters);
setUpModelMarshallerAndUnmarshaller(apexParameters);
} catch (final Exception e) {
- LOGGER.debug(APEX_ENGINE_FAILED_MSG, e);
+ try {
+ terminate();
+ } catch (ApexException e1) {
+ LOGGER.error("Terminating the ApexActivator encountered error", e1);
+ }
throw new ApexActivatorException(APEX_ENGINE_FAILED_MSG, e);
}
LOGGER.debug("Apex engine started as a service");
}
- private void setUpModelMarshallerAndUnmarshaller(ApexParameters apexParameters) throws ApexException {
- policyModelsMap = new LinkedHashMap<>();
- Map<String, EventHandlerParameters> inputParametersMap = new LinkedHashMap<>();
- Map<String, EventHandlerParameters> outputParametersMap = new LinkedHashMap<>();
- Set<Entry<ToscaConceptIdentifier, ApexParameters>> apexParamsEntrySet = new LinkedHashSet<>(
- apexParametersMap.entrySet());
- apexParamsEntrySet.stream().forEach(apexParamsEntry -> {
- ApexParameters apexParams = apexParamsEntry.getValue();
- List<String> duplicateInputParameters = new ArrayList<>(apexParams.getEventInputParameters().keySet());
- duplicateInputParameters.retainAll(inputParametersMap.keySet());
- List<String> duplicateOutputParameters = new ArrayList<>(apexParams.getEventOutputParameters().keySet());
- duplicateOutputParameters.retainAll(outputParametersMap.keySet());
- if (!(duplicateInputParameters.isEmpty() && duplicateOutputParameters.isEmpty())) {
- LOGGER.error("I/O Parameters {}/{} for {}:{} are duplicates. So this policy is not executed.",
- duplicateInputParameters, duplicateOutputParameters, apexParamsEntry.getKey().getName(),
- apexParamsEntry.getKey().getVersion());
- apexParametersMap.remove(apexParamsEntry.getKey());
- return;
- }
+ private void instantiateEngine(ApexParameters apexParameters) throws ApexException {
+ if (null != apexEngineService && apexEngineService.getKey().equals(engineKey)) {
+ throw new ApexException("Apex Engine already initialized.");
+ }
+ // Create engine with specified thread count
+ LOGGER.debug("starting apex engine service . . .");
+ apexEngineService = EngineServiceImpl.create(apexParameters.getEngineServiceParameters());
- inputParametersMap.putAll(apexParams.getEventInputParameters());
- outputParametersMap.putAll(apexParams.getEventOutputParameters());
- // Check if a policy model file has been specified
- if (apexParams.getEngineServiceParameters().getPolicyModel() != null) {
- LOGGER.debug("deploying policy model to the apex engines . . .");
- try {
- final String policyModelString = apexParams.getEngineServiceParameters().getPolicyModel();
- AxPolicyModel policyModel = EngineServiceImpl.createModel(engineKey, policyModelString);
- policyModelsMap.put(apexParamsEntry.getKey(), policyModel);
- } catch (ApexException e) {
- throw new ApexRuntimeException("Failed to create the apex model.", e);
- }
- }
- });
- AxPolicyModel finalPolicyModel = aggregatePolicyModels(policyModelsMap);
-
- // Set the policy model in the engine
- apexEngineService.updateModel(engineKey, finalPolicyModel, true);
-
- handleExistingMarshallerAndUnmarshaller(inputParametersMap, outputParametersMap);
- setUpNewMarshallerAndUnmarshaller(apexParameters.getEngineServiceParameters(), inputParametersMap,
- outputParametersMap);
-
- // Wire up pairings between marhsallers and unmarshallers
- setUpMarshalerPairings(inputParametersMap);
-
- // Start event processing
- startUnmarshallers(inputParametersMap);
+ // Create the engine holder to hold the engine's references and act as an event
+ // receiver
+ engineServiceHandler = new ApexEngineServiceHandler(apexEngineService);
}
- private AxPolicyModel aggregatePolicyModels(Map<ToscaConceptIdentifier, AxPolicyModel> policyModelsMap) {
- // Doing a deep copy so that original values in policyModelsMap is retained
- // after reduction operation
- Set<Entry<ToscaConceptIdentifier, AxPolicyModel>> policyModelsEntries = policyModelsMap.entrySet().stream()
- .map(entry -> new AbstractMap.SimpleEntry<>(entry.getKey(), entry.getValue())).collect(Collectors.toSet());
- Optional<Entry<ToscaConceptIdentifier, AxPolicyModel>> finalPolicyModelEntry = policyModelsEntries.stream()
- .reduce((entry1, entry2) -> {
- try {
- entry1.setValue(
- PolicyModelMerger.getMergedPolicyModel(entry1.getValue(), entry2.getValue(), true, true));
- } catch (ApexModelException exc) {
- LOGGER.error("Policy model for {} : {} has duplicates. So this policy is not executed.",
- entry2.getKey().getName(), entry2.getKey().getVersion(), exc);
- apexParametersMap.remove(entry2.getKey());
- policyModelsMap.remove(entry2.getKey());
- }
- return entry1;
- });
- AxPolicyModel finalPolicyModel = null;
- if (finalPolicyModelEntry.isPresent()) {
- finalPolicyModel = new AxPolicyModel(finalPolicyModelEntry.get().getValue());
+ private void setUpModelMarshallerAndUnmarshaller(ApexParameters apexParameters) throws ApexException {
+ AxPolicyModel model;
+ try {
+ final String policyModelString = apexParameters.getEngineServiceParameters().getPolicyModel();
+ model = EngineServiceImpl.createModel(apexParameters.getEngineServiceParameters().getEngineKey(),
+ policyModelString);
+ } catch (ApexException e) {
+ throw new ApexRuntimeException("Failed to create the apex model.", e);
}
- return finalPolicyModel;
+
+ AxKeyInformation existingKeyInformation = null;
+ AxContextSchemas existingSchemas = null;
+ AxEvents existingEvents = null;
+ AxContextAlbums existingAlbums = null;
+ AxTasks existingTasks = null;
+ AxPolicies existingPolicies = null;
+ if (ModelService.existsModel(AxPolicyModel.class)) {
+ existingKeyInformation = new AxKeyInformation(ModelService.getModel(AxKeyInformation.class));
+ existingSchemas = new AxContextSchemas(ModelService.getModel(AxContextSchemas.class));
+ existingEvents = new AxEvents(ModelService.getModel(AxEvents.class));
+ existingAlbums = new AxContextAlbums(ModelService.getModel(AxContextAlbums.class));
+ existingTasks = new AxTasks(ModelService.getModel(AxTasks.class));
+ existingPolicies = new AxPolicies(ModelService.getModel(AxPolicies.class));
+ }
+ // Set the policy model in the engine
+ try {
+ apexEngineService.updateModel(apexParameters.getEngineServiceParameters().getEngineKey(), model, true);
+ policyModel = new AxPolicyModel(model);
+ } catch (ApexException exp) {
+ // Discard concepts from the current policy
+ if (null != existingKeyInformation) {
+ // Update model service with other policies' concepts.
+ ModelService.registerModel(AxKeyInformation.class, existingKeyInformation);
+ ModelService.registerModel(AxContextSchemas.class, existingSchemas);
+ ModelService.registerModel(AxEvents.class, existingEvents);
+ ModelService.registerModel(AxContextAlbums.class, existingAlbums);
+ ModelService.registerModel(AxTasks.class, existingTasks);
+ ModelService.registerModel(AxPolicies.class, existingPolicies);
+ } else {
+ ModelService.clear();
+ }
+ throw exp;
+ }
+ if (null != existingKeyInformation) {
+ // Make sure all concepts in previously deployed policies are retained in ModelService
+ // during multi policy deployment.
+ updateModelService(existingKeyInformation, existingSchemas, existingEvents, existingAlbums, existingTasks,
+ existingPolicies);
+ }
+
+ setUpNewMarshallerAndUnmarshaller(apexParameters.getEngineServiceParameters(),
+ apexParameters.getEventInputParameters(), apexParameters.getEventOutputParameters());
+
+ // Wire up pairings between marhsallers and unmarshallers
+ setUpMarshalerPairings(apexParameters.getEventInputParameters());
+
+ // Start event processing
+ startUnmarshallers(apexParameters.getEventInputParameters());
+ }
+
+ private void updateModelService(AxKeyInformation existingKeyInformation,
+ AxContextSchemas existingSchemas, AxEvents existingEvents,
+ AxContextAlbums existingAlbums, AxTasks existingTasks,
+ AxPolicies existingPolicies) throws ApexModelException {
+
+ AxContextSchemas axContextSchemas = ModelService.getModel(AxContextSchemas.class);
+ AxEvents axEvents = ModelService.getModel(AxEvents.class);
+ AxContextAlbums axContextAlbums = ModelService.getModel(AxContextAlbums.class);
+ AxTasks axTasks = ModelService.getModel(AxTasks.class);
+ AxPolicies axPolicies = ModelService.getModel(AxPolicies.class);
+
+ Map<AxArtifactKey, AxContextSchema> newSchemasMap = axContextSchemas.getSchemasMap();
+ Map<AxArtifactKey, AxEvent> newEventsMap = axEvents.getEventMap();
+ Map<AxArtifactKey, AxContextAlbum> newAlbumsMap = axContextAlbums.getAlbumsMap();
+ Map<AxArtifactKey, AxTask> newTasksMap = axTasks.getTaskMap();
+ Map<AxArtifactKey, AxPolicy> newPoliciesMap = axPolicies.getPolicyMap();
+
+ StringBuilder errorMessage = new StringBuilder();
+ PolicyModelMerger.checkForDuplicateItem(existingSchemas.getSchemasMap(), newSchemasMap, errorMessage, "schema");
+ PolicyModelMerger.checkForDuplicateItem(existingEvents.getEventMap(), newEventsMap, errorMessage, "event");
+ PolicyModelMerger.checkForDuplicateItem(existingAlbums.getAlbumsMap(), newAlbumsMap, errorMessage, "album");
+ PolicyModelMerger.checkForDuplicateItem(existingTasks.getTaskMap(), newTasksMap, errorMessage, "task");
+ PolicyModelMerger.checkForDuplicateItem(existingPolicies.getPolicyMap(), newPoliciesMap, errorMessage,
+ "policy");
+ if (errorMessage.length() > 0) {
+ throw new ApexModelException(errorMessage.toString());
+ }
+
+ AxKeyInformation axKeyInformation = ModelService.getModel(AxKeyInformation.class);
+ Map<AxArtifactKey, AxKeyInfo> newKeyInfoMap = axKeyInformation.getKeyInfoMap();
+ // Now add all the concepts that must be copied over
+ newKeyInfoMap.putAll(existingKeyInformation.getKeyInfoMap());
+ newSchemasMap.putAll(existingSchemas.getSchemasMap());
+ newEventsMap.putAll(existingEvents.getEventMap());
+ newAlbumsMap.putAll(existingAlbums.getAlbumsMap());
+ newTasksMap.putAll(existingTasks.getTaskMap());
+ newPoliciesMap.putAll(existingPolicies.getPolicyMap());
}
private void setUpNewMarshallerAndUnmarshaller(EngineServiceParameters engineServiceParameters,
@@ -226,38 +266,6 @@
}
}
- 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(engineKey)) {
- throw new ApexException("Apex Engine already initialized.");
- }
- // Create engine with specified thread count
- LOGGER.debug("starting apex engine service . . .");
- apexEngineService = EngineServiceImpl.create(apexParameters.getEngineServiceParameters());
-
- // Create the engine holder to hold the engine's references and act as an event
- // receiver
- engineServiceHandler = new ApexEngineServiceHandler(apexEngineService);
- }
-
/**
* Set up unmarshaler/marshaler pairing for synchronized event handling. We only
* need to traverse the unmarshalers because the unmarshalers and marshalers are
@@ -275,8 +283,8 @@
// Check if the unmarshaler is synchronized with a marshaler
if (inputParameters.getValue().isPeeredMode(peeredMode)) {
// Find the unmarshaler and marshaler
- final ApexEventMarshaller peeredMarshaler = marshallerMap
- .get(inputParameters.getValue().getPeer(peeredMode));
+ final ApexEventMarshaller peeredMarshaler =
+ marshallerMap.get(inputParameters.getValue().getPeer(peeredMode));
// Connect the unmarshaler and marshaler
unmarshaller.connectMarshaler(peeredMode, peeredMarshaler);
@@ -298,22 +306,6 @@
}
/**
- * Updates the APEX Engine with the model created from new Policies.
- *
- * @param apexParamsMap the apex parameters map for the Apex service
- * @throws ApexException on errors
- */
- public void updateModel(Map<ToscaConceptIdentifier, ApexParameters> apexParamsMap) throws ApexException {
- try {
- ApexParameters apexParameters = apexParamsMap.values().iterator().next();
- setUpModelMarshallerAndUnmarshaller(apexParameters);
- } catch (final Exception e) {
- LOGGER.debug(APEX_ENGINE_FAILED_MSG, e);
- throw new ApexActivatorException(APEX_ENGINE_FAILED_MSG, e);
- }
- }
-
- /**
* Get the Apex engine worker stats.
*/
public List<AxEngineModel> getEngineStats() {
@@ -338,19 +330,25 @@
engineServiceHandler.terminate();
engineServiceHandler = null;
}
-
- // Clear the services
- ModelService.clear();
- ParameterService.clear();
+ // Clear the services in case if this was the only policy running in the engine
+ ApexParameters apexParams = null;
+ if (ParameterService.contains(ApexParameterConstants.MAIN_GROUP_NAME)) {
+ apexParams = ParameterService.get(ApexParameterConstants.MAIN_GROUP_NAME);
+ }
+ if (null != apexParams
+ && apexParameters.getEventInputParameters().size() == apexParams.getEventInputParameters().size()) {
+ ModelService.clear();
+ ParameterService.clear();
+ }
}
/**
* Shuts down all marshallers and unmarshallers.
*/
private void shutdownMarshallerAndUnmarshaller() {
- marshallerMap.values().forEach(ApexEventMarshaller::stop);
- marshallerMap.clear();
unmarshallerMap.values().forEach(ApexEventUnmarshaller::stop);
unmarshallerMap.clear();
+ marshallerMap.values().forEach(ApexEventMarshaller::stop);
+ marshallerMap.clear();
}
}
diff --git a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexEventUnmarshaller.java b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexEventUnmarshaller.java
index fd6ac44..fdc404c 100644
--- a/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexEventUnmarshaller.java
+++ b/services/services-engine/src/main/java/org/onap/policy/apex/service/engine/main/ApexEventUnmarshaller.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2020 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -347,17 +347,16 @@
// Order the stop
stopOrderedFlag = true;
- // Order a stop on the synchronous cache if one exists
- if (consumerParameters != null && consumerParameters.isPeeredMode(EventHandlerPeeredMode.SYNCHRONOUS)
- && consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS) != null) {
- ((SynchronousEventCache) consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS)).stop();
- }
-
// Wait for thread shutdown
while (unmarshallerThread != null && unmarshallerThread.isAlive()) {
ThreadUtilities.sleep(UNMARSHALLER_SHUTDOWN_WAIT_INTERVAL);
}
+ // Order a stop on the synchronous cache if one exists
+ if (consumerParameters != null && consumerParameters.isPeeredMode(EventHandlerPeeredMode.SYNCHRONOUS)
+ && consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS) != null) {
+ ((SynchronousEventCache) consumer.getPeeredReference(EventHandlerPeeredMode.SYNCHRONOUS)).stop();
+ }
LOGGER.exit("shut down Apex event unmarshaller");
}
}
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 5707c95..65c2acf 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,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modification Copyright (C) 2019-2021 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -24,26 +24,20 @@
import java.util.Arrays;
import java.util.Base64;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import java.util.Map.Entry;
-import java.util.TreeMap;
+import java.util.stream.Collectors;
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;
-import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbum;
-import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbums;
import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
-import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
+import org.onap.policy.apex.service.parameters.ApexParameterConstants;
import org.onap.policy.apex.service.parameters.ApexParameterHandler;
import org.onap.policy.apex.service.parameters.ApexParameters;
+import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
import org.onap.policy.common.parameters.ParameterService;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.slf4j.ext.XLogger;
import org.slf4j.ext.XLoggerFactory;
@@ -58,14 +52,12 @@
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ApexMain.class);
// The Apex Activator that activates the Apex engine
+ @Getter
private ApexActivator activator;
// The parameters read in from JSON for each policy
@Getter
- private Map<ToscaConceptIdentifier, ApexParameters> apexParametersMap;
-
- //engineParameters are aggregated in case of multiple policies
- private EngineParameters aggregatedEngineParameters;
+ private ApexParameters apexParameters;
private ApexParameterHandler apexParameterHandler = new ApexParameterHandler();
@@ -77,28 +69,27 @@
* Instantiates the Apex service.
*
* @param args the command line arguments
+ * @throws ApexException the apex exception.
*/
- public ApexMain(final String[] args) {
+ public ApexMain(final String[] args) throws ApexException {
LOGGER.entry("Starting Apex service with parameters " + Arrays.toString(args) + " . . .");
- apexParametersMap = new LinkedHashMap<>();
- aggregatedEngineParameters = new EngineParameters();
try {
- apexParametersMap.put(new ToscaConceptIdentifier(), populateApexParameters(args));
+ apexParameters = populateApexParameters(args);
} catch (ApexException e) {
LOGGER.error(APEX_SERVICE_FAILED_MSG, e);
return;
}
- apexParameterHandler.registerParameters(apexParametersMap.values().iterator().next());
+ aggregateParametersAndRegister();
+
// Now, create the activator for the Apex service
- activator = new ApexActivator(apexParametersMap);
+ activator = new ApexActivator(apexParameters);
// Start the activator
try {
activator.initialize();
setAlive(true);
} catch (final ApexActivatorException e) {
- LOGGER.error("start of Apex service failed, used parameters are " + Arrays.toString(args), e);
- return;
+ throw new ApexException("start of Apex service failed, used parameters are " + Arrays.toString(args), e);
}
// Add a shutdown hook to shut everything down in an orderly manner
@@ -106,120 +97,6 @@
LOGGER.exit("Started Apex");
}
- /**
- * Instantiates the Apex service for multiple policies.
- *
- * @param policyArgumentsMap the map with command line arguments as value and policy-id as key
- * @throws ApexException on errors
- */
- public ApexMain(Map<ToscaConceptIdentifier, String[]> policyArgumentsMap) throws ApexException {
- apexParametersMap = new LinkedHashMap<>();
- aggregatedEngineParameters = new EngineParameters();
- for (Entry<ToscaConceptIdentifier, String[]> policyArgsEntry: policyArgumentsMap.entrySet()) {
- try {
- apexParametersMap.put(policyArgsEntry.getKey(), populateApexParameters(policyArgsEntry.getValue()));
- } catch (ApexException e) {
- LOGGER.error("Invalid arguments specified for policy - " + policyArgsEntry.getKey().getName() + ":"
- + policyArgsEntry.getKey().getVersion(), e);
- }
- }
- if (apexParametersMap.isEmpty()) {
- 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);
-
- // Start the activator
- try {
- activator.initialize();
- apexParametersMap = activator.getApexParametersMap();
- setAlive(true);
- } catch (final ApexActivatorException e) {
- LOGGER.error(APEX_SERVICE_FAILED_MSG, e);
- activator.terminate();
- return;
- }
-
- // Add a shutdown hook to shut everything down in an orderly manner
- Runtime.getRuntime().addShutdownHook(new ApexMainShutdownHookClass());
- LOGGER.exit("Started Apex");
- }
-
- /**
- * Updates the APEX Engine with the model created from new Policies.
- *
- * @param policyArgsMap the map with command line arguments as value and policy-id as key
- * @throws ApexException on errors
- */
- public void updateModel(Map<ToscaConceptIdentifier, 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<ToscaConceptIdentifier, String[]> policyArgsEntry : policyArgsMap.entrySet()) {
- findAlbumsToHold(albumsMap, policyArgsEntry.getKey());
- try {
- apexParametersMap.put(policyArgsEntry.getKey(), populateApexParameters(policyArgsEntry.getValue()));
- } catch (ApexException e) {
- LOGGER.error("Invalid arguments specified for policy - {}:{}", policyArgsEntry.getKey().getName(),
- 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() && !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 parameters again in this case
- apexParameterHandler.registerParameters(apexParameters);
- activator = new ApexActivator(apexParametersMap);
- activator.initialize();
- setAlive(true);
- } else {
- albums.setAlbumsMap(albumsMap);
- activator.setApexParametersMap(apexParametersMap);
- activator.updateModel(apexParametersMap);
- }
- } catch (final ApexException e) {
- LOGGER.error(APEX_SERVICE_FAILED_MSG, e);
- activator.terminate();
- throw new ApexException(e.getMessage());
- }
- }
-
- /**
- * Find the context albums which should be retained when updating the policies.
- *
- * @param albumsMap the albums which should be kept during model update
- * @param policyId the policy id of current policy
- */
- private void findAlbumsToHold(Map<AxArtifactKey, AxContextAlbum> albumsMap, ToscaConceptIdentifier policyId) {
- for (Entry<ToscaConceptIdentifier, AxPolicyModel> policyModelsEntry : activator.getPolicyModelsMap()
- .entrySet()) {
- // If a policy with the same major version is received in PDP_UPDATE,
- // context for that policy has to be retained. For this, take such policies' albums
- if (policyModelsEntry.getKey().getName().equals(policyId.getName())
- && policyModelsEntry.getKey().getVersion().split("\\.")[0]
- .equals(policyId.getVersion().split("\\.")[0])) {
- albumsMap.putAll(policyModelsEntry.getValue().getAlbums().getAlbumsMap());
- }
- }
- }
-
private ApexParameters populateApexParameters(String[] args) throws ApexException {
// Check the arguments
final ApexCommandLineArguments arguments = new ApexCommandLineArguments();
@@ -263,11 +140,32 @@
ehParameterEntry.getValue().setName(ehParameterEntry.getKey());
}
}
- aggregateEngineParameters(axParameters.getEngineServiceParameters().getEngineParameters());
return axParameters;
}
- private void aggregateEngineParameters(EngineParameters engineParameters) {
+ private void aggregateParametersAndRegister() throws ApexException {
+ ApexParameters aggregatedParameters = null;
+ if (ParameterService.contains(ApexParameterConstants.MAIN_GROUP_NAME)) {
+ aggregatedParameters = ParameterService.get(ApexParameterConstants.MAIN_GROUP_NAME);
+ } else {
+ aggregatedParameters = new ApexParameters();
+ aggregatedParameters.setEngineServiceParameters(new EngineServiceParameters());
+ apexParameterHandler.registerParameters(aggregatedParameters);
+ }
+ List<String> duplicateInputParameters = aggregatedParameters.getEventInputParameters().keySet().stream()
+ .filter(apexParameters.getEventInputParameters()::containsKey).collect(Collectors.toList());
+ List<String> duplicateOutputParameters = aggregatedParameters.getEventOutputParameters().keySet().stream()
+ .filter(apexParameters.getEventOutputParameters()::containsKey).collect(Collectors.toList());
+ if (!(duplicateInputParameters.isEmpty() && duplicateOutputParameters.isEmpty())) {
+ throw new ApexException(
+ "start of Apex service failed because this policy has the following duplicate I/O parameters: "
+ + duplicateInputParameters + "/" + duplicateOutputParameters);
+ }
+ aggregatedParameters.getEventInputParameters().putAll(apexParameters.getEventInputParameters());
+ aggregatedParameters.getEventOutputParameters().putAll(apexParameters.getEventOutputParameters());
+ EngineParameters aggregatedEngineParameters =
+ aggregatedParameters.getEngineServiceParameters().getEngineParameters();
+ EngineParameters engineParameters = apexParameters.getEngineServiceParameters().getEngineParameters();
aggregatedEngineParameters.getTaskParameters().addAll(engineParameters.getTaskParameters());
aggregatedEngineParameters.getExecutorParameterMap().putAll(engineParameters.getExecutorParameterMap());
aggregatedEngineParameters.getContextParameters().getSchemaParameters().getSchemaHelperParameterMap()
@@ -346,8 +244,9 @@
* The main method.
*
* @param args the arguments
+ * @throws ApexException the apex exception.
*/
- public static void main(final String[] args) {
+ public static void main(final String[] args) throws ApexException {
new ApexMain(args);
}
}
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 9a07b4a..673d9ca 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,7 +2,7 @@
* ============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.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -73,8 +73,6 @@
*/
public ApexParameters getParameters(final ApexCommandLineArguments arguments) throws ParameterException {
- ParameterService.clear();
-
ApexParameters parameters = null;
String toscaPolicyFilePath = arguments.getToscaPolicyFilePath();
// Read the parameters
diff --git a/services/services-engine/src/test/java/org/onap/policy/apex/service/engine/main/ApexMainTest.java b/services/services-engine/src/test/java/org/onap/policy/apex/service/engine/main/ApexMainTest.java
index 68472d4..2cb12c3 100644
--- a/services/services-engine/src/test/java/org/onap/policy/apex/service/engine/main/ApexMainTest.java
+++ b/services/services-engine/src/test/java/org/onap/policy/apex/service/engine/main/ApexMainTest.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2020-2021 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -23,21 +23,22 @@
package org.onap.policy.apex.service.engine.main;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import java.io.ByteArrayOutputStream;
import java.io.OutputStream;
import java.io.PrintStream;
-import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Test;
import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
-import org.onap.policy.apex.service.parameters.ApexParameters;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
+import org.onap.policy.apex.model.basicmodel.service.ModelService;
+import org.onap.policy.common.parameters.ParameterService;
/**
* Test the ApexMain class.
@@ -114,13 +115,13 @@
OutputStream outContent = new ByteArrayOutputStream();
System.setOut(new PrintStream(outContent));
- String[] args = { "-p", "src/test/resources/parameters/correctParams.json" };
+ String[] args = {"-p", "src/test/resources/parameters/correctParams.json"};
final ApexMain apexMain = new ApexMain(args);
- assertEquals("MyApexEngine",
- apexMain.getApexParametersMap().values().iterator().next().getEngineServiceParameters().getName());
- await().atMost(200, TimeUnit.MILLISECONDS).until(() -> outContent.toString()
- .contains("Added the action listener to the engine"));
+ assertEquals("MyApexEngine", apexMain.getApexParameters().getEngineServiceParameters().getName());
+ await().atMost(200, TimeUnit.MILLISECONDS)
+ .until(() -> outContent.toString().contains("Added the action listener to the engine"));
+ assertTrue(apexMain.isAlive());
apexMain.shutdown();
}
@@ -129,16 +130,15 @@
OutputStream outContent = new ByteArrayOutputStream();
System.setOut(new PrintStream(outContent));
- String[] args = { "-p", "src/test/resources/parameters/correctParamsJavaProperties.json" };
+ String[] args = {"-p", "src/test/resources/parameters/correctParamsJavaProperties.json"};
final ApexMain apexMain = new ApexMain(args);
- assertEquals("MyApexEngine",
- apexMain.getApexParametersMap().values().iterator().next().getEngineServiceParameters().getName());
+ assertEquals("MyApexEngine", apexMain.getApexParameters().getEngineServiceParameters().getName());
assertEquals("trust-store-file", System.getProperty("javax.net.ssl.trustStore"));
assertEquals("Pol1cy_0nap", System.getProperty("javax.net.ssl.trustStorePassword"));
- await().atMost(10000, TimeUnit.MILLISECONDS).until(() -> outContent.toString()
- .contains("Added the action listener to the engine"));
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .until(() -> outContent.toString().contains("Added the action listener to the engine"));
apexMain.shutdown();
}
@@ -146,46 +146,41 @@
public void testCorrectParametersWithMultiplePolicies() throws ApexException {
OutputStream outContent = new ByteArrayOutputStream();
System.setOut(new PrintStream(outContent));
- Map<ToscaConceptIdentifier, String[]> argsMap = new HashMap<ToscaConceptIdentifier, String[]>();
- String[] args = {"-p", "src/test/resources/parameters/correctParams.json"};
- argsMap.put(new ToscaConceptIdentifier("id1", "v1"), args);
- final ApexMain apexMain = new ApexMain(argsMap);
- ApexParameters apexParam = (ApexParameters) apexMain.getApexParametersMap().values().toArray()[0];
- assertEquals("MyApexEngine", apexParam.getEngineServiceParameters().getName());
- apexMain.shutdown();
+ String[] args1 = {"-p", "src/test/resources/parameters/correctParams.json"};
+ String[] args2 = {"-p", "src/test/resources/parameters/correctParams2.json"};
+ final ApexMain apexMain1 = new ApexMain(args1);
+ final ApexMain apexMain2 = new ApexMain(args2);
+ assertEquals("MyApexEngine", apexMain1.getApexParameters().getEngineServiceParameters().getName());
+ assertEquals("MyApexEngine2", apexMain2.getApexParameters().getEngineServiceParameters().getName());
final String outString = outContent.toString();
- assertThat(outString).contains("Added the action listener to the engine");
+ assertThat(outString).contains("Added the action listener to the engine")
+ .contains("Created apex engine MyApexEngine").contains("Created apex engine MyApexEngine2");
+ assertTrue(apexMain1.isAlive());
+ assertTrue(apexMain2.isAlive());
+ apexMain1.shutdown();
+ apexMain2.shutdown();
+ ModelService.clear();
+ ParameterService.clear();
}
@Test
public void testInCorrectParametersWithMultiplePolicies() throws ApexException {
- OutputStream outContent = new ByteArrayOutputStream();
- System.setOut(new PrintStream(outContent));
- Map<ToscaConceptIdentifier, String[]> argsMap = new HashMap<ToscaConceptIdentifier, String[]>();
String[] args = {"-p", "src/test/resources/parameters/correctParams.json"};
- argsMap.put(new ToscaConceptIdentifier("id1", "v1"), args);
- argsMap.put(new ToscaConceptIdentifier("id2", "v2"), args);
- final ApexMain apexMain = new ApexMain(argsMap);
- ApexParameters apexParam = (ApexParameters) apexMain.getApexParametersMap().values().toArray()[0];
- assertEquals("MyApexEngine", apexParam.getEngineServiceParameters().getName());
- apexMain.shutdown();
- final String outString = outContent.toString();
- assertThat(outString).contains("I/O Parameters [TheFileConsumer1]/[FirstProducer] for id2:v2 are duplicates. "
- + "So this policy is not executed");
- assertEquals(1, apexMain.getApexParametersMap().size()); // only id1:v1 is kept in the map, id2:v2 failed
+ final ApexMain apexMain1 = new ApexMain(args);
+ assertThatThrownBy(() -> new ApexMain(args)).hasMessage("start of Apex service failed because this"
+ + " policy has the following duplicate I/O parameters: [TheFileConsumer1]/[FirstProducer]");
+ apexMain1.shutdown();
}
@Test
public void testInvalidArgsWithMultiplePolicies() throws ApexException {
OutputStream outContent = new ByteArrayOutputStream();
System.setOut(new PrintStream(outContent));
- Map<ToscaConceptIdentifier, String[]> argsMap = new HashMap<ToscaConceptIdentifier, String[]>();
String[] args = {"-c", "file1", "-m", "file2"};
- argsMap.put(new ToscaConceptIdentifier("id1", "v1"), args);
- final ApexMain apexMain = new ApexMain(argsMap);
+ final ApexMain apexMain = new ApexMain(args);
final String outString = outContent.toString();
apexMain.shutdown();
assertThat(outString).contains("Arguments validation failed", "start of Apex service failed");
- assertThat(apexMain.getApexParametersMap()).isEmpty(); // No policy is running in the engine
+ assertFalse(apexMain.isAlive()); // No policy is running in the engine
}
}
diff --git a/services/services-engine/src/test/resources/parameters/correctParams.json b/services/services-engine/src/test/resources/parameters/correctParams.json
index 43fef1c..7ea06d7 100644
--- a/services/services-engine/src/test/resources/parameters/correctParams.json
+++ b/services/services-engine/src/test/resources/parameters/correctParams.json
@@ -501,7 +501,8 @@
},
"eventProtocolParameters": {
"eventProtocol": "JSON"
- }
+ },
+ "eventName": "BasicEvent"
}
}
}
diff --git a/services/services-engine/src/test/resources/parameters/correctParams2.json b/services/services-engine/src/test/resources/parameters/correctParams2.json
new file mode 100644
index 0000000..2d2ebff
--- /dev/null
+++ b/services/services-engine/src/test/resources/parameters/correctParams2.json
@@ -0,0 +1,513 @@
+{
+ "tosca_definitions_version": "tosca_simple_yaml_1_1_0",
+ "topology_template": {
+ "policies": [
+ {
+ "onap.policies.apex.Simplecontrolloop": {
+ "type": "onap.policies.native.Apex",
+ "type_version": "1.0.0",
+ "name": "onap.policies.apex.Simplecontrolloop",
+ "version": "1.0.0",
+ "properties": {
+ "engineServiceParameters": {
+ "name": "MyApexEngine2",
+ "version": "0.0.1",
+ "id": 45,
+ "instanceCount": 2,
+ "deploymentPort": 65522,
+ "policy_type_impl": {
+ "apexPolicyModel": {
+ "key": {
+ "name": "SmallModel",
+ "version": "0.0.1"
+ },
+ "keyInformation": {
+ "key": {
+ "name": "SmallModel_KeyInfo",
+ "version": "0.0.1"
+ },
+ "keyInfoMap": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ },
+ "UUID": "fec1b353-b35f-4384-b7d9-69622059c248",
+ "description": "Generated description for a concept called \"BasicContextAlbum\" with version \"0.0.1\" and UUID \"fec1b353-b35f-4384-b7d9-69622059c248\""
+ }
+ },
+ {
+ "key": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "UUID": "cc8d3c1a-e975-459a-bcd2-69f423eaa1f3",
+ "description": "Generated description for a concept called \"BasicEvent\" with version \"0.0.1\" and UUID \"cc8d3c1a-e975-459a-bcd2-69f423eaa1f3\""
+ }
+ },
+ {
+ "key": {
+ "name": "BasicPolicy",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicPolicy",
+ "version": "0.0.1"
+ },
+ "UUID": "d0c5d8ee-5fe7-4978-89ce-4a3e69cad043",
+ "description": "Generated description for a concept called \"BasicPolicy\" with version \"0.0.1\" and UUID \"d0c5d8ee-5fe7-4978-89ce-4a3e69cad043\""
+ }
+ },
+ {
+ "key": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "UUID": "c5651414-fc1c-493b-878d-75f0ce685c36",
+ "description": "Generated description for a concept called \"BasicTask\" with version \"0.0.1\" and UUID \"c5651414-fc1c-493b-878d-75f0ce685c36\""
+ }
+ },
+ {
+ "key": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "UUID": "790ff718-8dc0-44e0-89d8-1b3bbe238310",
+ "description": "Generated description for a concept called \"IntType\" with version \"0.0.1\" and UUID \"790ff718-8dc0-44e0-89d8-1b3bbe238310\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel",
+ "version": "0.0.1"
+ },
+ "UUID": "a1bd1f4e-713b-456b-b1a8-bb48beee28e8",
+ "description": "Generated description for a concept called \"SmallModel\" with version \"0.0.1\" and UUID \"a1bd1f4e-713b-456b-b1a8-bb48beee28e8\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_Albums",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_Albums",
+ "version": "0.0.1"
+ },
+ "UUID": "72bed9af-ab7d-3379-b9f7-b5eca5c9ef22",
+ "description": "Generated description for concept referred to by key \"SmallModel_Albums:0.0.1\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_Events",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_Events",
+ "version": "0.0.1"
+ },
+ "UUID": "796dc6b0-627d-34ae-a5e2-1bc4b4b486b8",
+ "description": "Generated description for concept referred to by key \"SmallModel_Events:0.0.1\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_KeyInfo",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_KeyInfo",
+ "version": "0.0.1"
+ },
+ "UUID": "b4876774-6907-3d27-a2b8-f05737c5ee4a",
+ "description": "Generated description for concept referred to by key \"SmallModel_KeyInfo:0.0.1\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_Policies",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_Policies",
+ "version": "0.0.1"
+ },
+ "UUID": "5bcf946b-67be-3190-a906-f954896f999f",
+ "description": "Generated description for concept referred to by key \"SmallModel_Policies:0.0.1\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_Schemas",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_Schemas",
+ "version": "0.0.1"
+ },
+ "UUID": "c25bf5c3-7f1e-3667-b8a9-971ba21517bc",
+ "description": "Generated description for concept referred to by key \"SmallModel_Schemas:0.0.1\""
+ }
+ },
+ {
+ "key": {
+ "name": "SmallModel_Tasks",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "SmallModel_Tasks",
+ "version": "0.0.1"
+ },
+ "UUID": "43b015ca-2ed1-3a35-b103-e8a5aa68f1ef",
+ "description": "Generated description for concept referred to by key \"SmallModel_Tasks:0.0.1\""
+ }
+ }
+ ]
+ }
+ },
+ "policies": {
+ "key": {
+ "name": "SmallModel_Policies",
+ "version": "0.0.1"
+ },
+ "policyMap": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicPolicy",
+ "version": "0.0.1"
+ },
+ "value": {
+ "policyKey": {
+ "name": "BasicPolicy",
+ "version": "0.0.1"
+ },
+ "template": "FREEFORM",
+ "state": {
+ "entry": [
+ {
+ "key": "OnlyState",
+ "value": {
+ "stateKey": {
+ "parentKeyName": "BasicPolicy",
+ "parentKeyVersion": "0.0.1",
+ "parentLocalName": "NULL",
+ "localName": "OnlyState"
+ },
+ "trigger": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "stateOutputs": {
+ "entry": [
+ {
+ "key": "OnlyOutput",
+ "value": {
+ "key": {
+ "parentKeyName": "BasicPolicy",
+ "parentKeyVersion": "0.0.1",
+ "parentLocalName": "OnlyState",
+ "localName": "OnlyOutput"
+ },
+ "outgoingEvent": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "nextState": {
+ "parentKeyName": "NULL",
+ "parentKeyVersion": "0.0.0",
+ "parentLocalName": "NULL",
+ "localName": "NULL"
+ }
+ }
+ }
+ ]
+ },
+ "contextAlbumReference": [
+ {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ }
+ ],
+ "taskSelectionLogic": {
+ "key": "NULL",
+ "logicFlavour": "UNDEFINED",
+ "logic": ""
+ },
+ "stateFinalizerLogicMap": {
+ "entry": []
+ },
+ "defaultTask": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "taskReferences": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "parentKeyName": "BasicPolicy",
+ "parentKeyVersion": "0.0.1",
+ "parentLocalName": "OnlyState",
+ "localName": "BasicTask"
+ },
+ "outputType": "DIRECT",
+ "output": {
+ "parentKeyName": "BasicPolicy",
+ "parentKeyVersion": "0.0.1",
+ "parentLocalName": "OnlyState",
+ "localName": "OnlyOutput"
+ }
+ }
+ }
+ ]
+ }
+ }
+ }
+ ]
+ },
+ "firstState": "OnlyState"
+ }
+ }
+ ]
+ }
+ },
+ "tasks": {
+ "key": {
+ "name": "SmallModel_Tasks",
+ "version": "0.0.1"
+ },
+ "taskMap": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicTask",
+ "version": "0.0.1"
+ },
+ "inputFields": {
+ "entry": [
+ {
+ "key": "intPar",
+ "value": {
+ "key": "intPar",
+ "fieldSchemaKey": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "optional": false
+ }
+ }
+ ]
+ },
+ "outputFields": {
+ "entry": [
+ {
+ "key": "intPar",
+ "value": {
+ "key": "intPar",
+ "fieldSchemaKey": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "optional": false
+ }
+ }
+ ]
+ },
+ "taskParameters": {
+ "entry": []
+ },
+ "contextAlbumReference": [
+ {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ }
+ ],
+ "taskLogic": {
+ "key": "TaskLogic",
+ "logicFlavour": "JAVASCRIPT",
+ "logic": "executor.logger.debug(executor.subject.id);\nvar gc = executor.getContextAlbum(\"BasicContextAlbum\");\nexecutor.logger.debug(gc.name);\nexecutor.logger.debug(executor.inFields);\n\nexecutor.logger.debug(executor.eo);\n\nvar returnValue = executor.isTrue;"
+ }
+ }
+ }
+ ]
+ }
+ },
+ "events": {
+ "key": {
+ "name": "SmallModel_Events",
+ "version": "0.0.1"
+ },
+ "eventMap": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicEvent",
+ "version": "0.0.1"
+ },
+ "nameSpace": "org.onap.policy.apex.events",
+ "source": "source",
+ "target": "target",
+ "parameter": {
+ "entry": [
+ {
+ "key": "intPar",
+ "value": {
+ "key": "intPar",
+ "fieldSchemaKey": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "optional": false
+ }
+ }
+ ]
+ }
+ }
+ }
+ ]
+ }
+ },
+ "albums": {
+ "key": {
+ "name": "SmallModel_Albums",
+ "version": "0.0.1"
+ },
+ "albums": {
+ "entry": [
+ {
+ "key": {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "BasicContextAlbum",
+ "version": "0.0.1"
+ },
+ "scope": "GLOBAL",
+ "isWritable": true,
+ "itemSchema": {
+ "name": "IntType",
+ "version": "0.0.1"
+ }
+ }
+ }
+ ]
+ }
+ },
+ "schemas": {
+ "key": {
+ "name": "SmallModel_Schemas",
+ "version": "0.0.1"
+ },
+ "schemas": {
+ "entry": [
+ {
+ "key": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "value": {
+ "key": {
+ "name": "IntType",
+ "version": "0.0.1"
+ },
+ "schemaFlavour": "Java",
+ "schemaDefinition": "java.lang.Integer"
+ }
+ }
+ ]
+ }
+ }
+ }
+ },
+ "engineParameters": {
+ "executorParameters": {
+ "JAVASCRIPT": {
+ "parameterClassName": "org.onap.policy.apex.service.engine.parameters.dummyclasses.SuperDooperExecutorParameters"
+ }
+ }
+ }
+ },
+ "eventOutputParameters": {
+ "FirstProducer2": {
+ "carrierTechnologyParameters": {
+ "carrierTechnology": "FILE",
+ "parameters": {
+ "standardIo": true
+ }
+ },
+ "eventProtocolParameters": {
+ "eventProtocol": "JSON"
+ }
+ }
+ },
+ "eventInputParameters": {
+ "TheFileConsumer2": {
+ "carrierTechnologyParameters": {
+ "carrierTechnology": "FILE",
+ "parameters": {
+ "fileName": "src/test/resources/events/TestPojoEvent.json"
+ }
+ },
+ "eventProtocolParameters": {
+ "eventProtocol": "JSON"
+ },
+ "eventName": "BasicEvent"
+ }
+ }
+ }
+ }
+ }
+ ]
+ }
+}
\ No newline at end of file
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/services/onappf/handler/ApexEngineHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/services/onappf/handler/ApexEngineHandler.java
index 4d560a3..03e9721 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/services/onappf/handler/ApexEngineHandler.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/services/onappf/handler/ApexEngineHandler.java
@@ -1,7 +1,7 @@
/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2019-2021 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -24,13 +24,36 @@
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.onap.policy.apex.core.engine.EngineParameters;
+import org.onap.policy.apex.core.engine.TaskParameters;
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.concepts.AxKeyInfo;
+import org.onap.policy.apex.model.basicmodel.concepts.AxKeyInformation;
+import org.onap.policy.apex.model.basicmodel.service.ModelService;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbum;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextAlbums;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchema;
+import org.onap.policy.apex.model.contextmodel.concepts.AxContextSchemas;
import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
+import org.onap.policy.apex.model.eventmodel.concepts.AxEvent;
+import org.onap.policy.apex.model.eventmodel.concepts.AxEvents;
+import org.onap.policy.apex.model.policymodel.concepts.AxPolicies;
+import org.onap.policy.apex.model.policymodel.concepts.AxPolicy;
+import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
+import org.onap.policy.apex.model.policymodel.concepts.AxTask;
+import org.onap.policy.apex.model.policymodel.concepts.AxTasks;
import org.onap.policy.apex.service.engine.main.ApexMain;
+import org.onap.policy.apex.service.parameters.ApexParameterConstants;
+import org.onap.policy.apex.service.parameters.ApexParameters;
import org.onap.policy.apex.services.onappf.exception.ApexStarterException;
+import org.onap.policy.common.parameters.ParameterService;
import org.onap.policy.common.utils.coder.CoderException;
import org.onap.policy.common.utils.coder.StandardCoder;
import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
@@ -49,7 +72,7 @@
private static final Logger LOGGER = LoggerFactory.getLogger(ApexEngineHandler.class);
- private ApexMain apexMain;
+ private Map<ToscaConceptIdentifier, ApexMain> apexMainMap;
/**
* Constructs the object. Extracts the config and model files from each policy and instantiates the apex engine.
@@ -57,13 +80,13 @@
* @param policies the list of policies
* @throws ApexStarterException if the apex engine instantiation failed using the policies passed
*/
- public ApexEngineHandler(List<ToscaPolicy> policies) throws ApexStarterException {
- Map<ToscaConceptIdentifier, String[]> policyArgsMap = createPolicyArgsMap(policies);
+ public ApexEngineHandler(List<ToscaPolicy> policies) throws ApexStarterException {
LOGGER.debug("Starting apex engine.");
- try {
- apexMain = new ApexMain(policyArgsMap);
- } catch (ApexException e) {
- throw new ApexStarterException(e);
+ apexMainMap = initiateApexEngineForPolicies(policies);
+ if (apexMainMap.isEmpty()) {
+ ModelService.clear();
+ ParameterService.clear();
+ throw new ApexStarterException("Apex Engine failed to start.");
}
}
@@ -74,20 +97,136 @@
* @throws ApexStarterException if the apex engine instantiation failed using the policies passed
*/
public void updateApexEngine(List<ToscaPolicy> policies) throws ApexStarterException {
- if (null == apexMain || !apexMain.isAlive()) {
- throw new ApexStarterException("Apex Engine not initialized.");
+ List<ToscaConceptIdentifier> runningPolicies = getRunningPolicies();
+ List<ToscaPolicy> policiesToDeploy = policies.stream()
+ .filter(policy -> !runningPolicies.contains(policy.getIdentifier())).collect(Collectors.toList());
+ List<ToscaConceptIdentifier> policiesToUnDeploy = runningPolicies.stream()
+ .filter(polId -> policies.stream().noneMatch(policy -> policy.getIdentifier().equals(polId)))
+ .collect(Collectors.toList());
+ Map<ToscaConceptIdentifier, ApexMain> undeployedPoliciesMainMap = new LinkedHashMap<>();
+ policiesToUnDeploy.forEach(policyId -> {
+ ApexMain apexMain = apexMainMap.get(policyId);
+ try {
+ apexMain.shutdown();
+ undeployedPoliciesMainMap.put(policyId, apexMain);
+ apexMainMap.remove(policyId);
+ } catch (ApexException e) {
+ LOGGER.error("Shutting down policy {} failed", policyId, e);
+ }
+ });
+ if (!undeployedPoliciesMainMap.isEmpty()) {
+ updateModelAndParameterServices(undeployedPoliciesMainMap);
}
- Map<ToscaConceptIdentifier, String[]> policyArgsMap = createPolicyArgsMap(policies);
- try {
- apexMain.updateModel(policyArgsMap);
- } catch (ApexException e) {
- throw new ApexStarterException(e);
+ if (!policiesToDeploy.isEmpty()) {
+ Map<ToscaConceptIdentifier, ApexMain> mainMap = initiateApexEngineForPolicies(policiesToDeploy);
+ if (mainMap.isEmpty()) {
+ throw new ApexStarterException("Updating the APEX engine with new policies failed.");
+ }
+ apexMainMap.putAll(mainMap);
+ }
+ if (apexMainMap.isEmpty()) {
+ ModelService.clear();
+ ParameterService.clear();
}
}
- private Map<ToscaConceptIdentifier, String[]> createPolicyArgsMap(List<ToscaPolicy> policies)
+ /**
+ * Clear the corresponding items from ModelService and ParameterService.
+ *
+ * @param undeployedPoliciesMainMap the policies that are undeployed
+ */
+ private void updateModelAndParameterServices(Map<ToscaConceptIdentifier, ApexMain> undeployedPoliciesMainMap) {
+ Set<String> inputParamKeysToRetain = new HashSet<>();
+ Set<String> outputParamKeysToRetain = new HashSet<>();
+ List<TaskParameters> taskParametersToRetain = new ArrayList<>();
+ List<String> executorParamKeysToRetain = new ArrayList<>();
+ List<String> schemaParamKeysToRetain = new ArrayList<>();
+
+ List<AxArtifactKey> keyInfoKeystoRetain = new ArrayList<>();
+ List<AxArtifactKey> schemaKeystoRetain = new ArrayList<>();
+ List<AxArtifactKey> eventKeystoRetain = new ArrayList<>();
+ List<AxArtifactKey> albumKeystoRetain = new ArrayList<>();
+ List<AxArtifactKey> taskKeystoRetain = new ArrayList<>();
+ List<AxArtifactKey> policyKeystoRetain = new ArrayList<>();
+
+ apexMainMap.values().forEach(main -> {
+ inputParamKeysToRetain.addAll(main.getApexParameters().getEventInputParameters().keySet());
+ outputParamKeysToRetain.addAll(main.getApexParameters().getEventOutputParameters().keySet());
+ taskParametersToRetain.addAll(
+ main.getApexParameters().getEngineServiceParameters().getEngineParameters().getTaskParameters());
+ executorParamKeysToRetain.addAll(main.getApexParameters().getEngineServiceParameters().getEngineParameters()
+ .getExecutorParameterMap().keySet());
+ schemaParamKeysToRetain.addAll(main.getApexParameters().getEngineServiceParameters().getEngineParameters()
+ .getContextParameters().getSchemaParameters().getSchemaHelperParameterMap().keySet());
+
+ keyInfoKeystoRetain
+ .addAll(main.getActivator().getPolicyModel().getKeyInformation().getKeyInfoMap().keySet());
+ schemaKeystoRetain.addAll(main.getActivator().getPolicyModel().getSchemas().getSchemasMap().keySet());
+ eventKeystoRetain.addAll(main.getActivator().getPolicyModel().getEvents().getEventMap().keySet());
+ albumKeystoRetain.addAll(main.getActivator().getPolicyModel().getAlbums().getAlbumsMap().keySet());
+ taskKeystoRetain.addAll(main.getActivator().getPolicyModel().getTasks().getTaskMap().keySet());
+ policyKeystoRetain.addAll(main.getActivator().getPolicyModel().getPolicies().getPolicyMap().keySet());
+ });
+ for (ApexMain main : undeployedPoliciesMainMap.values()) {
+ ApexParameters existingParameters = ParameterService.get(ApexParameterConstants.MAIN_GROUP_NAME);
+ List<String> eventInputParamKeysToRemove = main.getApexParameters().getEventInputParameters().keySet()
+ .stream().filter(key -> !inputParamKeysToRetain.contains(key)).collect(Collectors.toList());
+ List<String> eventOutputParamKeysToRemove = main.getApexParameters().getEventOutputParameters().keySet()
+ .stream().filter(key -> !outputParamKeysToRetain.contains(key)).collect(Collectors.toList());
+ eventInputParamKeysToRemove.forEach(existingParameters.getEventInputParameters()::remove);
+ eventOutputParamKeysToRemove.forEach(existingParameters.getEventOutputParameters()::remove);
+ EngineParameters engineParameters =
+ main.getApexParameters().getEngineServiceParameters().getEngineParameters();
+ final List<TaskParameters> taskParametersToRemove = engineParameters.getTaskParameters().stream()
+ .filter(taskParameter -> !taskParametersToRetain.contains(taskParameter)).collect(Collectors.toList());
+ final List<String> executorParamKeysToRemove = engineParameters.getExecutorParameterMap().keySet().stream()
+ .filter(key -> !executorParamKeysToRetain.contains(key)).collect(Collectors.toList());
+ final List<String> schemaParamKeysToRemove =
+ engineParameters.getContextParameters().getSchemaParameters().getSchemaHelperParameterMap().keySet()
+ .stream().filter(key -> !schemaParamKeysToRetain.contains(key)).collect(Collectors.toList());
+ EngineParameters aggregatedEngineParameters =
+ existingParameters.getEngineServiceParameters().getEngineParameters();
+ aggregatedEngineParameters.getTaskParameters().removeAll(taskParametersToRemove);
+ executorParamKeysToRemove.forEach(aggregatedEngineParameters.getExecutorParameterMap()::remove);
+ schemaParamKeysToRemove.forEach(aggregatedEngineParameters.getContextParameters().getSchemaParameters()
+ .getSchemaHelperParameterMap()::remove);
+
+ final AxPolicyModel policyModel = main.getActivator().getPolicyModel();
+ final List<AxArtifactKey> keyInfoKeystoRemove = policyModel.getKeyInformation().getKeyInfoMap().keySet()
+ .stream().filter(key -> !keyInfoKeystoRetain.contains(key)).collect(Collectors.toList());
+ final List<AxArtifactKey> schemaKeystoRemove = policyModel.getSchemas().getSchemasMap().keySet().stream()
+ .filter(key -> !schemaKeystoRetain.contains(key)).collect(Collectors.toList());
+ final List<AxArtifactKey> eventKeystoRemove = policyModel.getEvents().getEventMap().keySet().stream()
+ .filter(key -> !eventKeystoRetain.contains(key)).collect(Collectors.toList());
+ final List<AxArtifactKey> albumKeystoRemove = policyModel.getAlbums().getAlbumsMap().keySet().stream()
+ .filter(key -> !albumKeystoRetain.contains(key)).collect(Collectors.toList());
+ final List<AxArtifactKey> taskKeystoRemove = policyModel.getTasks().getTaskMap().keySet().stream()
+ .filter(key -> !taskKeystoRetain.contains(key)).collect(Collectors.toList());
+ final List<AxArtifactKey> policyKeystoRemove = policyModel.getPolicies().getPolicyMap().keySet().stream()
+ .filter(key -> !policyKeystoRetain.contains(key)).collect(Collectors.toList());
+
+ final Map<AxArtifactKey, AxKeyInfo> keyInfoMap =
+ ModelService.getModel(AxKeyInformation.class).getKeyInfoMap();
+ final Map<AxArtifactKey, AxContextSchema> schemasMap =
+ ModelService.getModel(AxContextSchemas.class).getSchemasMap();
+ final Map<AxArtifactKey, AxEvent> eventMap = ModelService.getModel(AxEvents.class).getEventMap();
+ final Map<AxArtifactKey, AxContextAlbum> albumsMap =
+ ModelService.getModel(AxContextAlbums.class).getAlbumsMap();
+ final Map<AxArtifactKey, AxTask> taskMap = ModelService.getModel(AxTasks.class).getTaskMap();
+ final Map<AxArtifactKey, AxPolicy> policyMap = ModelService.getModel(AxPolicies.class).getPolicyMap();
+
+ keyInfoKeystoRemove.forEach(keyInfoMap::remove);
+ schemaKeystoRemove.forEach(schemasMap::remove);
+ eventKeystoRemove.forEach(eventMap::remove);
+ albumKeystoRemove.forEach(albumsMap::remove);
+ taskKeystoRemove.forEach(taskMap::remove);
+ policyKeystoRemove.forEach(policyMap::remove);
+ }
+ }
+
+ private Map<ToscaConceptIdentifier, ApexMain> initiateApexEngineForPolicies(List<ToscaPolicy> policies)
throws ApexStarterException {
- Map<ToscaConceptIdentifier, String[]> policyArgsMap = new LinkedHashMap<>();
+ Map<ToscaConceptIdentifier, ApexMain> mainMap = new LinkedHashMap<>();
for (ToscaPolicy policy : policies) {
String policyName = policy.getIdentifier().getName();
final StandardCoder standardCoder = new StandardCoder();
@@ -103,34 +242,38 @@
throw new ApexStarterException(e);
}
final String[] apexArgs = {"-p", file.getAbsolutePath()};
- policyArgsMap.put(policy.getIdentifier(), apexArgs);
+ LOGGER.info("Starting apex engine for policy {}", policy.getIdentifier());
+ try {
+ ApexMain apexMain = new ApexMain(apexArgs);
+ mainMap.put(policy.getIdentifier(), apexMain);
+ } catch (Exception e) {
+ LOGGER.error("Execution of policy {} failed", policy.getIdentifier(), e);
+ }
}
- return policyArgsMap;
+ return mainMap;
}
/**
* Method to get the APEX engine statistics.
*/
public List<AxEngineModel> getEngineStats() {
- List<AxEngineModel> engineStats = null;
- if (null != apexMain && apexMain.isAlive()) {
- engineStats = apexMain.getEngineStats();
- }
- return engineStats;
+ // engineStats from all the apexMain instances running individual tosca policies are combined here.
+ return apexMainMap.values().stream().filter(apexMain -> (null != apexMain && apexMain.isAlive()))
+ .flatMap(m -> m.getEngineStats().stream()).collect(Collectors.toList());
}
/**
* Method to check whether the apex engine is running or not.
*/
public boolean isApexEngineRunning() {
- return null != apexMain && apexMain.isAlive();
+ return apexMainMap.values().stream().anyMatch(apexMain -> (null != apexMain && apexMain.isAlive()));
}
/**
* Method that return the list of running policies in the apex engine.
*/
public List<ToscaConceptIdentifier> getRunningPolicies() {
- return new ArrayList<>(apexMain.getApexParametersMap().keySet());
+ return new ArrayList<>(apexMainMap.keySet());
}
/**
@@ -139,8 +282,12 @@
public void shutdown() throws ApexStarterException {
try {
LOGGER.debug("Shutting down apex engine.");
- apexMain.shutdown();
- apexMain = null;
+ for (ApexMain apexMain : apexMainMap.values()) {
+ apexMain.shutdown();
+ }
+ apexMainMap.clear();
+ ModelService.clear();
+ ParameterService.clear();
} catch (final ApexException e) {
throw new ApexStarterException(e);
}
diff --git a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestFile2Rest.java b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestFile2Rest.java
index e3d64d2..568e9f3 100644
--- a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestFile2Rest.java
+++ b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestFile2Rest.java
@@ -2,6 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2020 Nordix Foundation.
+ * Modifications Copyright (C) 2021 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.
@@ -21,6 +22,7 @@
package org.onap.policy.apex.testsuites.integration.uservice.adapt.restclient;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -260,24 +262,10 @@
*/
@Test
public void testFileEventsBadHttpMethod() throws MessagingException, ApexException, IOException {
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
final String[] args = {"src/test/resources/prodcons/File2RESTJsonEventBadHTTPMethod.json"};
- final ApexMain apexMain = new ApexMain(args);
-
- ThreadUtilities.sleep(200);
- apexMain.shutdown();
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- LOGGER.info("BadHttpMethod-OUTSTRING=\n" + outString + "\nEnd-BadHttpMethod");
- assertTrue(outString
- .contains("specified HTTP method of \"DELETE\" is invalid, only HTTP methods \"POST\" and \"PUT\" "
- + "are supported for event sending on REST client producer"));
+ assertThatThrownBy(() -> new ApexMain(args)).hasRootCauseMessage(
+ "specified HTTP method of \"DELETE\" is invalid, only HTTP methods \"POST\" and \"PUT\" "
+ + "are supported for event sending on REST client producer (FirstProducer)");
}
/**
diff --git a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestRest2File.java b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestRest2File.java
index 6711728..25ff122 100644
--- a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestRest2File.java
+++ b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restclient/TestRest2File.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2020 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -22,6 +22,7 @@
package org.onap.policy.apex.testsuites.integration.uservice.adapt.restclient;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.fail;
import java.io.ByteArrayOutputStream;
@@ -202,29 +203,14 @@
/**
* Test file events bad http method.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
*/
@Test
- public void testFileEventsBadHttpMethod() throws MessagingException, ApexException, IOException {
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
+ public void testFileEventsBadHttpMethod() {
final String[] args = {"src/test/resources/prodcons/REST2FileJsonEventBadHTTPMethod.json"};
- final ApexMain apexMain = new ApexMain(args);
+ assertThatThrownBy(() -> new ApexMain(args))
+ .hasRootCauseMessage("specified HTTP method of \"POST\" is invalid, "
+ + "only HTTP method \"GET\" is supported for event reception on REST client consumer (FirstConsumer)");
- ThreadUtilities.sleep(5000);
- apexMain.shutdown();
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- checkRequiredString(outString, "specified HTTP method of \"POST\" is invalid, "
- + "only HTTP method \"GET\" is supported for event reception on REST client consumer");
}
/**
diff --git a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restserver/TestRestServer.java b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restserver/TestRestServer.java
index e3057ad..39173bc 100644
--- a/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restserver/TestRestServer.java
+++ b/testsuites/integration/integration-uservice-test/src/test/java/org/onap/policy/apex/testsuites/integration/uservice/adapt/restserver/TestRestServer.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* Modifications Copyright (C) 2019-2020 Nordix Foundation.
- * Modifications Copyright (C) 2020 Bell Canada. All rights reserved.
+ * Modifications Copyright (C) 2020-2021 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.
@@ -22,6 +22,7 @@
package org.onap.policy.apex.testsuites.integration.uservice.adapt.restserver;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -285,35 +286,14 @@
/**
* Test rest server producer standalone.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
- * @throws InterruptedException interrupted exception
*/
@Test
- public void testRestServerProducerStandalone()
- throws MessagingException, ApexException, IOException, InterruptedException {
+ public void testRestServerProducerStandalone() {
LOGGER.debug("testRestServerProducerStandalone start");
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
final String[] args = {"src/test/resources/prodcons/RESTServerJsonEventProducerStandalone.json"};
-
- final ApexMain apexMain = new ApexMain(args);
- ThreadUtilities.sleep(200);
- apexMain.shutdown();
-
- await().atMost(10L, TimeUnit.SECONDS).until(() -> !apexMain.isAlive());
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- assertTrue(outString
- .contains("the parameters \"host\", \"port\", and \"standalone\" are illegal on REST Server producer"));
- LOGGER.debug("testRestServerProducerStandalone end");
+ assertThatThrownBy(() -> new ApexMain(args))
+ .hasRootCauseMessage("the parameters \"host\", \"port\", and \"standalone\" are illegal"
+ + " on REST Server producer (FirstProducer)");
}
/**
@@ -382,94 +362,40 @@
/**
* Test rest server consumer standalone no host.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
*/
@Test
- public void testRestServerConsumerStandaloneNoHost() throws MessagingException, ApexException, IOException {
+ public void testRestServerConsumerStandaloneNoHost() {
LOGGER.debug("testRestServerConsumerStandaloneNoHost start");
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
final String[] args = {"src/test/resources/prodcons/RESTServerJsonEventConsumerStandaloneNoHost.json"};
+ assertThatThrownBy(() -> new ApexMain(args))
+ .hasRootCauseMessage("the parameters \"host\" and \"port\" must be defined for REST Server consumer "
+ + "(FirstConsumer) in standalone mode");
- final ApexMain apexMain = new ApexMain(args);
- ThreadUtilities.sleep(200);
- apexMain.shutdown();
-
- await().atMost(10L, TimeUnit.SECONDS).until(() -> !apexMain.isAlive());
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- assertTrue(outString.contains("the parameters \"host\" and \"port\" must be defined for REST Server consumer "
- + "(FirstConsumer) in standalone mode"));
LOGGER.debug("testRestServerConsumerStandaloneNoHost end");
}
/**
* Test rest server consumer standalone no port.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
*/
@Test
- public void testRestServerConsumerStandaloneNoPort() throws MessagingException, ApexException, IOException {
+ public void testRestServerConsumerStandaloneNoPort() {
LOGGER.debug("testRestServerConsumerStandaloneNoPort start");
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
final String[] args = {"src/test/resources/prodcons/RESTServerJsonEventConsumerStandaloneNoPort.json"};
-
- final ApexMain apexMain = new ApexMain(args);
- ThreadUtilities.sleep(200);
- apexMain.shutdown();
-
- await().atMost(10L, TimeUnit.SECONDS).until(() -> !apexMain.isAlive());
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- assertTrue(outString.contains("the parameters \"host\" and \"port\" must be defined for REST Server consumer "
- + "(FirstConsumer) in standalone mode"));
+ assertThatThrownBy(() -> new ApexMain(args))
+ .hasRootCauseMessage("the parameters \"host\" and \"port\" must be defined for REST Server consumer "
+ + "(FirstConsumer) in standalone mode");
LOGGER.debug("testRestServerConsumerStandaloneNoPort end");
}
/**
* Test rest server producer not sync.
- *
- * @throws MessagingException the messaging exception
- * @throws ApexException the apex exception
- * @throws IOException Signals that an I/O exception has occurred.
*/
@Test
- public void testRestServerProducerNotSync() throws MessagingException, ApexException, IOException {
+ public void testRestServerProducerNotSync() {
LOGGER.debug("testRestServerProducerNotSync start");
- System.setOut(new PrintStream(outContent));
- System.setErr(new PrintStream(errContent));
-
final String[] args = {"src/test/resources/prodcons/RESTServerJsonEventProducerNotSync.json"};
-
- final ApexMain apexMain = new ApexMain(args);
- ThreadUtilities.sleep(200);
- apexMain.shutdown();
-
- await().atMost(10L, TimeUnit.SECONDS).until(() -> !apexMain.isAlive());
-
- final String outString = outContent.toString();
-
- System.setOut(stdout);
- System.setErr(stderr);
-
- assertTrue(outString.contains(
- "REST Server producer (FirstProducer) must run in synchronous mode " + "with a REST Server consumer"));
+ assertThatThrownBy(() -> new ApexMain(args)).hasRootCauseMessage(
+ "REST Server producer (FirstProducer) must run in synchronous mode " + "with a REST Server consumer");
LOGGER.debug("testRestServerProducerNotSync end");
}