2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017-2020 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.drools.system.internal;
23 import com.fasterxml.jackson.annotation.JsonIgnore;
24 import java.util.ArrayList;
25 import java.util.Arrays;
26 import java.util.HashMap;
27 import java.util.List;
28 import java.util.Properties;
29 import java.util.stream.Collectors;
30 import org.onap.policy.common.endpoints.event.comm.Topic;
31 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
32 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
33 import org.onap.policy.common.endpoints.event.comm.TopicListener;
34 import org.onap.policy.common.endpoints.event.comm.TopicSink;
35 import org.onap.policy.common.endpoints.event.comm.TopicSource;
36 import org.onap.policy.common.gson.annotation.GsonJsonIgnore;
37 import org.onap.policy.common.utils.services.FeatureApiUtils;
38 import org.onap.policy.drools.controller.DroolsController;
39 import org.onap.policy.drools.controller.DroolsControllerConstants;
40 import org.onap.policy.drools.controller.DroolsControllerFactory;
41 import org.onap.policy.drools.features.PolicyControllerFeatureApi;
42 import org.onap.policy.drools.features.PolicyControllerFeatureApiConstants;
43 import org.onap.policy.drools.persistence.SystemPersistence;
44 import org.onap.policy.drools.persistence.SystemPersistenceConstants;
45 import org.onap.policy.drools.properties.DroolsPropertyConstants;
46 import org.onap.policy.drools.protocol.configuration.DroolsConfiguration;
47 import org.onap.policy.drools.system.PolicyController;
48 import org.onap.policy.drools.utils.PropertyUtil;
49 import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyTypeIdentifier;
50 import org.slf4j.Logger;
51 import org.slf4j.LoggerFactory;
54 * This implementation of the Policy Controller merely aggregates and tracks for management purposes
55 * all underlying resources that this controller depends upon.
57 public class AggregatedPolicyController implements PolicyController, TopicListener {
59 private static final String BEFORE_OFFER_FAILURE = "{}: feature {} before-offer failure because of {}";
60 private static final String AFTER_OFFER_FAILURE = "{}: feature {} after-offer failure because of {}";
65 private static final Logger logger = LoggerFactory.getLogger(AggregatedPolicyController.class);
68 * identifier for this policy controller.
70 private final String name;
73 * Abstracted Event Sources List regardless communication technology.
75 private final List<TopicSource> sources;
78 * Abstracted Event Sinks List regardless communication technology.
80 private final List<TopicSink> sinks;
83 * Mapping topics to sinks.
87 private final HashMap<String, TopicSink> topic2Sinks = new HashMap<>();
90 * Is this Policy Controller running (alive) ? reflects invocation of start()/stop() only.
92 private volatile boolean alive;
95 * Is this Policy Controller locked ? reflects if i/o controller related operations and start
96 * are permitted, more specifically: start(), deliver() and onTopicEvent(). It does not affect
97 * the ability to stop the underlying drools infrastructure
99 private volatile boolean locked;
102 * Policy Drools Controller.
104 private volatile DroolsController droolsController;
107 * Properties used to initialize controller.
109 private final Properties properties;
114 private List<ToscaPolicyTypeIdentifier> policyTypes;
117 * Constructor version mainly used for bootstrapping at initialization time a policy engine
120 * @param name controller name
123 * @throws IllegalArgumentException when invalid arguments are provided
125 public AggregatedPolicyController(String name, Properties properties) {
130 * 1. Register read topics with network infrastructure (ueb, dmaap, rest) 2. Register write
131 * topics with network infrastructure (ueb, dmaap, rest) 3. Register with drools
135 // Create/Reuse Readers/Writers for all event sources endpoints
137 this.sources = getEndpointManager().addTopicSources(properties);
138 this.sinks = getEndpointManager().addTopicSinks(properties);
140 initDrools(properties);
143 /* persist new properties */
144 getPersistenceManager().storeController(name, properties);
145 this.properties = PropertyUtil.getInterpolatedProperties(properties);
147 this.policyTypes = getPolicyTypesFromProperties();
151 public List<ToscaPolicyTypeIdentifier> getPolicyTypes() {
152 if (!policyTypes.isEmpty()) {
156 return droolsController
157 .getBaseDomainNames()
159 .map(d -> new ToscaPolicyTypeIdentifier(d,
160 DroolsPropertyConstants.DEFAULT_CONTROLLER_POLICY_TYPE_VERSION))
161 .collect(Collectors.toList());
164 protected List<ToscaPolicyTypeIdentifier> getPolicyTypesFromProperties() {
165 List<ToscaPolicyTypeIdentifier> policyTypeIds = new ArrayList<>();
167 String ptiPropValue = properties.getProperty(DroolsPropertyConstants.PROPERTY_CONTROLLER_POLICY_TYPES);
168 if (ptiPropValue == null) {
169 return policyTypeIds;
172 List<String> ptiPropList = new ArrayList<>(Arrays.asList(ptiPropValue.split("\\s*,\\s*")));
173 for (String pti : ptiPropList) {
174 String[] ptv = pti.split(":");
175 if (ptv.length == 1) {
176 policyTypeIds.add(new ToscaPolicyTypeIdentifier(ptv[0],
177 DroolsPropertyConstants.DEFAULT_CONTROLLER_POLICY_TYPE_VERSION));
178 } else if (ptv.length == 2) {
179 policyTypeIds.add(new ToscaPolicyTypeIdentifier(ptv[0], ptv[1]));
183 return policyTypeIds;
187 * initialize drools layer.
189 * @throws IllegalArgumentException if invalid parameters are passed in
191 private void initDrools(Properties properties) {
193 // Register with drools infrastructure
194 this.droolsController = getDroolsFactory().build(properties, sources, sinks);
195 } catch (Exception | LinkageError e) {
196 logger.error("{}: cannot init-drools because of {}", this, e.getMessage(), e);
197 throw new IllegalArgumentException(e);
204 * @throws IllegalArgumentException if invalid parameters are passed in
206 private void initSinks() {
207 this.topic2Sinks.clear();
208 for (TopicSink sink : sinks) {
209 this.topic2Sinks.put(sink.getTopic(), sink);
217 public boolean updateDrools(DroolsConfiguration newDroolsConfiguration) {
218 DroolsConfiguration oldDroolsConfiguration = new DroolsConfiguration(this.droolsController.getArtifactId(),
219 this.droolsController.getGroupId(), this.droolsController.getVersion());
221 if (oldDroolsConfiguration.getGroupId().equalsIgnoreCase(newDroolsConfiguration.getGroupId())
222 && oldDroolsConfiguration.getArtifactId().equalsIgnoreCase(newDroolsConfiguration.getArtifactId())
223 && oldDroolsConfiguration.getVersion().equalsIgnoreCase(newDroolsConfiguration.getVersion())) {
224 logger.warn("{}: cannot update-drools: identical configuration {} vs {}", this, oldDroolsConfiguration,
225 newDroolsConfiguration);
229 if (FeatureApiUtils.apply(getProviders(),
230 feature -> feature.beforePatch(this, oldDroolsConfiguration, newDroolsConfiguration),
231 (feature, ex) -> logger.error("{}: feature {} before-patch failure because of {}", this,
232 feature.getClass().getName(), ex.getMessage(), ex))) {
236 if (droolsController.isBrained()
237 && (newDroolsConfiguration.getArtifactId() == null
238 || DroolsControllerConstants.NO_ARTIFACT_ID.equals(newDroolsConfiguration.getArtifactId()))) {
239 // detach maven artifact
240 DroolsControllerConstants.getFactory().destroy(this.droolsController);
243 boolean success = true;
245 this.properties.setProperty(DroolsPropertyConstants.RULES_GROUPID, newDroolsConfiguration.getGroupId());
246 this.properties.setProperty(DroolsPropertyConstants.RULES_ARTIFACTID,
247 newDroolsConfiguration.getArtifactId());
248 this.properties.setProperty(DroolsPropertyConstants.RULES_VERSION, newDroolsConfiguration.getVersion());
249 getPersistenceManager().storeController(name, this.properties);
251 this.initDrools(this.properties);
254 droolsController.lock();
256 droolsController.unlock();
260 droolsController.start();
262 droolsController.stop();
264 } catch (RuntimeException e) {
265 logger.error("{}: cannot update-drools because of {}", this, e.getMessage(), e);
269 boolean finalSuccess = success;
270 FeatureApiUtils.apply(getProviders(),
271 feature -> feature.afterPatch(this, oldDroolsConfiguration, newDroolsConfiguration, finalSuccess),
272 (feature, ex) -> logger.error("{}: feature {} after-patch failure because of {}", this,
273 feature.getClass().getName(), ex.getMessage(), ex));
282 public String getName() {
290 public boolean start() {
291 logger.info("{}: start", this);
293 if (FeatureApiUtils.apply(getProviders(),
294 feature -> feature.beforeStart(this),
295 (feature, ex) -> logger.error("{}: feature {} before-start failure because of {}", this,
296 feature.getClass().getName(), ex.getMessage(), ex))) {
300 if (this.isLocked()) {
301 throw new IllegalStateException("Policy Controller " + name + " is locked");
304 synchronized (this) {
312 final boolean success = this.droolsController.start();
314 // register for events
316 for (TopicSource source : sources) {
317 source.register(this);
320 for (TopicSink sink : sinks) {
323 } catch (Exception e) {
324 logger.error("{}: cannot start {} because of {}", this, sink, e.getMessage(), e);
328 FeatureApiUtils.apply(getProviders(),
329 feature -> feature.afterStart(this),
330 (feature, ex) -> logger.error("{}: feature {} after-start failure because of {}", this,
331 feature.getClass().getName(), ex.getMessage(), ex));
340 public boolean stop() {
341 logger.info("{}: stop", this);
343 if (FeatureApiUtils.apply(getProviders(),
344 feature -> feature.beforeStop(this),
345 (feature, ex) -> logger.error("{}: feature {} before-stop failure because of {}", this,
346 feature.getClass().getName(), ex.getMessage(), ex))) {
350 /* stop regardless locked state */
352 synchronized (this) {
360 // 1. Stop registration
362 for (TopicSource source : sources) {
363 source.unregister(this);
366 boolean success = this.droolsController.stop();
368 FeatureApiUtils.apply(getProviders(),
369 feature -> feature.afterStop(this),
370 (feature, ex) -> logger.error("{}: feature {} after-stop failure because of {}", this,
371 feature.getClass().getName(), ex.getMessage(), ex));
380 public void shutdown() {
381 logger.info("{}: shutdown", this);
383 if (FeatureApiUtils.apply(getProviders(),
384 feature -> feature.beforeShutdown(this),
385 (feature, ex) -> logger.error("{}: feature {} before-shutdown failure because of {}", this,
386 feature.getClass().getName(), ex.getMessage(), ex))) {
392 getDroolsFactory().shutdown(this.droolsController);
394 FeatureApiUtils.apply(getProviders(),
395 feature -> feature.afterShutdown(this),
396 (feature, ex) -> logger.error("{}: feature {} after-shutdown failure because of {}", this,
397 feature.getClass().getName(), ex.getMessage(), ex));
405 logger.info("{}: halt", this);
407 if (FeatureApiUtils.apply(getProviders(),
408 feature -> feature.beforeHalt(this),
409 (feature, ex) -> logger.error("{}: feature {} before-halt failure because of {}", this,
410 feature.getClass().getName(), ex.getMessage(), ex))) {
415 getDroolsFactory().destroy(this.droolsController);
416 getPersistenceManager().deleteController(this.name);
418 FeatureApiUtils.apply(getProviders(),
419 feature -> feature.afterHalt(this),
420 (feature, ex) -> logger.error("{}: feature {} after-halt failure because of {}", this,
421 feature.getClass().getName(), ex.getMessage(), ex));
428 public void onTopicEvent(Topic.CommInfrastructure commType, String topic, String event) {
429 logger.debug("{}: raw event offered from {}:{}: {}", this, commType, topic, event);
435 if (FeatureApiUtils.apply(getProviders(),
436 feature -> feature.beforeOffer(this, commType, topic, event),
437 (feature, ex) -> logger.error(BEFORE_OFFER_FAILURE, this,
438 feature.getClass().getName(), ex.getMessage(), ex))) {
442 boolean success = this.droolsController.offer(topic, event);
444 FeatureApiUtils.apply(getProviders(),
445 feature -> feature.afterOffer(this, commType, topic, event, success),
446 (feature, ex) -> logger.error(AFTER_OFFER_FAILURE, this,
447 feature.getClass().getName(), ex.getMessage(), ex));
451 public <T> boolean offer(T event) {
452 logger.debug("{}: event offered: {}", this, event);
458 if (FeatureApiUtils.apply(getProviders(),
459 feature -> feature.beforeOffer(this, event),
460 (feature, ex) -> logger.error(BEFORE_OFFER_FAILURE, this,
461 feature.getClass().getName(), ex.getMessage(), ex))) {
465 boolean success = this.droolsController.offer(event);
467 FeatureApiUtils.apply(getProviders(),
468 feature -> feature.afterOffer(this, event, success),
469 (feature, ex) -> logger.error(AFTER_OFFER_FAILURE, this,
470 feature.getClass().getName(), ex.getMessage(), ex));
475 private boolean skipOffer() {
476 return isLocked() || !isAlive();
483 public boolean deliver(Topic.CommInfrastructure commType, String topic, Object event) {
485 logger.debug("{}: deliver event to {}:{}: {}", this, commType, topic, event);
487 if (FeatureApiUtils.apply(getProviders(),
488 feature -> feature.beforeDeliver(this, commType, topic, event),
489 (feature, ex) -> logger.error("{}: feature {} before-deliver failure because of {}", this,
490 feature.getClass().getName(), ex.getMessage(), ex))) {
494 if (topic == null || topic.isEmpty()) {
495 throw new IllegalArgumentException("Invalid Topic");
499 throw new IllegalArgumentException("Invalid Event");
502 if (!this.isAlive()) {
503 throw new IllegalStateException("Policy Engine is stopped");
506 if (this.isLocked()) {
507 throw new IllegalStateException("Policy Engine is locked");
510 if (!this.topic2Sinks.containsKey(topic)) {
511 logger.warn("{}: cannot deliver event because the sink {}:{} is not registered: {}", this, commType, topic,
513 throw new IllegalArgumentException("Unsupported topic " + topic + " for delivery");
516 boolean success = this.droolsController.deliver(this.topic2Sinks.get(topic), event);
518 FeatureApiUtils.apply(getProviders(),
519 feature -> feature.afterDeliver(this, commType, topic, event, success),
520 (feature, ex) -> logger.error("{}: feature {} after-deliver failure because of {}", this,
521 feature.getClass().getName(), ex.getMessage(), ex));
530 public boolean isAlive() {
538 public boolean lock() {
539 logger.info("{}: lock", this);
541 if (FeatureApiUtils.apply(getProviders(),
542 feature -> feature.beforeLock(this),
543 (feature, ex) -> logger.error("{}: feature {} before-lock failure because of {}", this,
544 feature.getClass().getName(), ex.getMessage(), ex))) {
548 synchronized (this) {
556 // it does not affect associated sources/sinks, they are
557 // autonomous entities
559 boolean success = this.droolsController.lock();
561 FeatureApiUtils.apply(getProviders(),
562 feature -> feature.afterLock(this),
563 (feature, ex) -> logger.error("{}: feature {} after-lock failure because of {}", this,
564 feature.getClass().getName(), ex.getMessage(), ex));
573 public boolean unlock() {
575 logger.info("{}: unlock", this);
577 if (FeatureApiUtils.apply(getProviders(),
578 feature -> feature.beforeUnlock(this),
579 (feature, ex) -> logger.error("{}: feature {} before-unlock failure because of {}", this,
580 feature.getClass().getName(), ex.getMessage(), ex))) {
584 synchronized (this) {
592 boolean success = this.droolsController.unlock();
594 FeatureApiUtils.apply(getProviders(),
595 feature -> feature.afterUnlock(this),
596 (feature, ex) -> logger.error("{}: feature {} after-unlock failure because of {}", this,
597 feature.getClass().getName(), ex.getMessage(), ex));
606 public boolean isLocked() {
614 public List<TopicSource> getTopicSources() {
622 public List<TopicSink> getTopicSinks() {
630 public DroolsController getDrools() {
631 return this.droolsController;
641 public Properties getProperties() {
642 return this.properties;
646 public String toString() {
647 return "AggregatedPolicyController [name=" + name + ", alive=" + alive
648 + ", locked=" + locked + ", droolsController=" + droolsController + "]";
651 // the following methods may be overridden by junit tests
653 protected SystemPersistence getPersistenceManager() {
654 return SystemPersistenceConstants.getManager();
657 protected TopicEndpoint getEndpointManager() {
658 return TopicEndpointManager.getManager();
661 protected DroolsControllerFactory getDroolsFactory() {
662 return DroolsControllerConstants.getFactory();
665 protected List<PolicyControllerFeatureApi> getProviders() {
666 return PolicyControllerFeatureApiConstants.getProviders().getList();