2 * ============LICENSE_START=======================================================
3 * feature-session-persistence
4 * ================================================================================
5 * Copyright (C) 2017-2018 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.persistence;
23 import java.io.IOException;
24 import java.sql.Connection;
25 import java.sql.PreparedStatement;
26 import java.sql.SQLException;
27 import java.util.HashMap;
29 import java.util.Properties;
30 import java.util.concurrent.CountDownLatch;
31 import java.util.concurrent.TimeUnit;
33 import javax.persistence.EntityManagerFactory;
34 import javax.persistence.Persistence;
35 import javax.transaction.TransactionManager;
36 import javax.transaction.TransactionSynchronizationRegistry;
37 import javax.transaction.UserTransaction;
39 import org.apache.commons.dbcp2.BasicDataSource;
40 import org.apache.commons.dbcp2.BasicDataSourceFactory;
41 import org.kie.api.KieServices;
42 import org.kie.api.runtime.Environment;
43 import org.kie.api.runtime.EnvironmentName;
44 import org.kie.api.runtime.KieSession;
45 import org.kie.api.runtime.KieSessionConfiguration;
46 import org.onap.policy.drools.core.PolicyContainer;
47 import org.onap.policy.drools.core.PolicySession;
48 import org.onap.policy.drools.core.PolicySessionFeatureAPI;
49 import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
50 import org.onap.policy.drools.system.PolicyController;
51 import org.onap.policy.drools.system.PolicyEngine;
52 import org.onap.policy.drools.utils.PropertyUtil;
53 import org.slf4j.Logger;
54 import org.slf4j.LoggerFactory;
57 * If this feature is supported, there is a single instance of it. It adds persistence to Drools
58 * sessions. In addition, if an active-standby feature exists, then that is used to determine the
59 * active and last-active PDP. If it does not exist, then the current host name is used as the PDP
62 * <p>The bulk of the code here was once in other classes, such as 'PolicyContainer' and 'Main'. It
63 * was moved here as part of making this a separate optional feature.
65 public class PersistenceFeature implements PolicySessionFeatureAPI, PolicyEngineFeatureAPI {
67 private static final Logger logger = LoggerFactory.getLogger(PersistenceFeature.class);
69 /** Standard factory used to get various items. */
70 private static Factory stdFactory = new Factory();
72 /** Factory used to get various items. */
73 private Factory fact = stdFactory;
75 /** KieService factory. */
76 private KieServices kieSvcFact;
78 /** Persistence properties. */
79 private Properties persistProps;
81 /** Whether or not the SessionInfo records should be cleaned out. */
82 private boolean sessInfoCleaned;
84 /** SessionInfo timeout, in milli-seconds, as read from
85 * {@link #persistProps}. */
86 private long sessionInfoTimeoutMs;
88 /** Object used to serialize cleanup of sessioninfo table. */
89 private Object cleanupLock = new Object();
92 * Sets the factory to be used during junit testing.
94 * @param fact factory to be used
96 protected void setFactory(Factory fact) {
101 * Lookup the adjunct for this feature that is associated with the specified PolicyContainer. If
102 * not found, create one.
104 * @param policyContainer the container whose adjunct we are looking up, and possibly creating
105 * @return the associated 'ContainerAdjunct' instance, which may be new
107 private ContainerAdjunct getContainerAdjunct(PolicyContainer policyContainer) {
109 Object rval = policyContainer.getAdjunct(this);
111 if (rval == null || !(rval instanceof ContainerAdjunct)) {
112 // adjunct does not exist, or has the wrong type (should never
114 rval = new ContainerAdjunct(policyContainer);
115 policyContainer.setAdjunct(this, rval);
118 return (ContainerAdjunct) rval;
125 public int getSequenceNumber() {
133 public void globalInit(String[] args, String configDir) {
135 kieSvcFact = fact.getKieServices();
138 persistProps = fact.loadProperties(configDir + "/feature-session-persistence.properties");
140 } catch (IOException e1) {
141 logger.error("initializePersistence: ", e1);
144 sessionInfoTimeoutMs = getPersistenceTimeout();
148 * Creates a persistent KieSession, loading it from the persistent store, or creating one, if it
149 * does not exist yet.
152 public KieSession activatePolicySession(
153 PolicyContainer policyContainer, String name, String kieBaseName) {
155 if (isPersistenceEnabled(policyContainer, name)) {
156 cleanUpSessionInfo();
158 return getContainerAdjunct(policyContainer).newPersistentKieSession(name, kieBaseName);
168 public PolicySession.ThreadModel selectThreadModel(PolicySession session) {
170 PolicyContainer policyContainer = session.getPolicyContainer();
171 if (isPersistenceEnabled(policyContainer, session.getName())) {
172 return new PersistentThreadModel(session, getProperties(policyContainer));
181 public void disposeKieSession(PolicySession policySession) {
183 ContainerAdjunct contAdj =
184 (ContainerAdjunct) policySession.getPolicyContainer().getAdjunct(this);
185 if (contAdj != null) {
186 contAdj.disposeKieSession(policySession.getName());
194 public void destroyKieSession(PolicySession policySession) {
196 ContainerAdjunct contAdj =
197 (ContainerAdjunct) policySession.getPolicyContainer().getAdjunct(this);
198 if (contAdj != null) {
199 contAdj.destroyKieSession(policySession.getName());
207 public boolean afterStart(PolicyEngine engine) {
215 public boolean beforeStart(PolicyEngine engine) {
216 synchronized (cleanupLock) {
217 sessInfoCleaned = false;
227 public boolean beforeActivate(PolicyEngine engine) {
228 synchronized (cleanupLock) {
229 sessInfoCleaned = false;
239 public boolean afterActivate(PolicyEngine engine) {
243 /* ============================================================ */
246 * Gets the persistence timeout value for sessioninfo records.
248 * @return the timeout value, in milli-seconds, or {@code -1} if it is unspecified or invalid
250 private long getPersistenceTimeout() {
251 String timeoutString = null;
254 timeoutString = persistProps.getProperty(DroolsPersistenceProperties.DB_SESSIONINFO_TIMEOUT);
256 if (timeoutString != null) {
257 // timeout parameter is specified
258 return Long.valueOf(timeoutString) * 1000;
261 } catch (NumberFormatException e) {
263 "Invalid value for Drools persistence property persistence.sessioninfo.timeout: {}",
271 /* ============================================================ */
274 * Each instance of this class is a logical extension of a 'PolicyContainer' instance. Its
275 * reference is stored in the 'adjuncts' table within the 'PolicyContainer', and will be
276 * garbage-collected with the container.
278 protected class ContainerAdjunct {
279 /** 'PolicyContainer' instance that this adjunct is extending. */
280 private PolicyContainer policyContainer;
282 /** Maps a KIE session name to its data source. */
283 private Map<String, DsEmf> name2ds = new HashMap<>();
286 * Constructor - initialize a new 'ContainerAdjunct'.
288 * @param policyContainer the 'PolicyContainer' instance this adjunct is extending
290 private ContainerAdjunct(PolicyContainer policyContainer) {
291 this.policyContainer = policyContainer;
295 * Create a new persistent KieSession. If there is already a corresponding entry in the
296 * database, it is used to initialize the KieSession. If not, a completely new session is
299 * @param name the name of the KieSession (which is also the name of the associated
301 * @param kieBaseName the name of the 'KieBase' instance containing this session
302 * @return a new KieSession with persistence enabled
304 private KieSession newPersistentKieSession(String name, String kieBaseName) {
308 BasicDataSource ds = fact.makeDataSource(getDataSourceProperties());
309 DsEmf dsemf = new DsEmf(ds);
312 EntityManagerFactory emf = dsemf.emf;
313 DroolsSessionConnector conn = fact.makeJpaConnector(emf);
315 long desiredSessionId = getSessionId(conn, name);
318 "\n\nThis controller is primary... coming up with session {} \n\n", desiredSessionId);
320 // session does not exist -- attempt to create one
322 "getPolicySession:session does not exist -- attempt to create one with name {}", name);
324 Environment env = kieSvcFact.newEnvironment();
326 configureKieEnv(env, emf);
328 KieSessionConfiguration kieConf = kieSvcFact.newKieSessionConfiguration();
330 KieSession kieSession =
331 (desiredSessionId >= 0
332 ? loadKieSession(kieBaseName, desiredSessionId, env, kieConf)
335 if (kieSession == null) {
336 // loadKieSession() returned null or desiredSessionId < 0
338 "LOADING We cannot load session {}. Going to create a new one", desiredSessionId);
340 kieSession = newKieSession(kieBaseName, env);
343 replaceSession(conn, name, kieSession);
345 name2ds.put(name, dsemf);
349 } catch (RuntimeException e) {
356 * Loads an existing KieSession from the persistent store.
358 * @param kieBaseName the name of the 'KieBase' instance containing this session
359 * @param desiredSessionId id of the desired KieSession
360 * @param env Kie Environment for the session
361 * @param kConf Kie Configuration for the session
362 * @return the persistent session, or {@code null} if it could not be loaded
364 private KieSession loadKieSession(
365 String kieBaseName, long desiredSessionId, Environment env, KieSessionConfiguration kieConf) {
367 KieSession kieSession =
372 policyContainer.getKieContainer().getKieBase(kieBaseName),
376 logger.info("LOADING Loaded session {}", desiredSessionId);
380 } catch (Exception e) {
381 logger.error("loadKieSession error: ", e);
387 * Creates a new, persistent KieSession.
389 * @param kieBaseName the name of the 'KieBase' instance containing this session
390 * @param env Kie Environment for the session
391 * @return a new, persistent session
393 private KieSession newKieSession(String kieBaseName, Environment env) {
394 KieSession kieSession =
397 .newKieSession(policyContainer.getKieContainer().getKieBase(kieBaseName), null, env);
399 logger.info("LOADING CREATED {}", kieSession.getIdentifier());
405 * Closes the data source associated with a session.
407 * @param name name of the session being destroyed
409 private void destroyKieSession(String name) {
410 closeDataSource(name);
414 * Closes the data source associated with a session.
416 * @param name name of the session being disposed of
418 private void disposeKieSession(String name) {
419 closeDataSource(name);
423 * Closes the data source associated with a session.
425 * @param name name of the session whose data source is to be closed
427 private void closeDataSource(String name) {
428 DsEmf ds = name2ds.remove(name);
434 /** Configures java system properties for JPA/JTA. */
435 private void configureSysProps() {
436 System.setProperty("com.arjuna.ats.arjuna.coordinator.defaultTimeout", "60");
438 "com.arjuna.ats.arjuna.objectstore.objectStoreDir",
439 persistProps.getProperty(DroolsPersistenceProperties.JTA_OBJECTSTORE_DIR));
441 "ObjectStoreEnvironmentBean.objectStoreDir",
442 persistProps.getProperty(DroolsPersistenceProperties.JTA_OBJECTSTORE_DIR));
446 * Configures a Kie Environment.
448 * @param env environment to be configured
449 * @param emf entity manager factory
451 private void configureKieEnv(Environment env, EntityManagerFactory emf) {
452 env.set(EnvironmentName.ENTITY_MANAGER_FACTORY, emf);
453 env.set(EnvironmentName.TRANSACTION, fact.getUserTrans());
454 env.set(EnvironmentName.TRANSACTION_SYNCHRONIZATION_REGISTRY, fact.getTransSyncReg());
455 env.set(EnvironmentName.TRANSACTION_MANAGER, fact.getTransMgr());
459 * Gets a session's ID from the persistent store.
461 * @param conn persistence connector
462 * @param sessnm name of the session
463 * @return the session's id, or {@code -1} if the session is not found
465 private long getSessionId(DroolsSessionConnector conn, String sessnm) {
466 DroolsSession sess = conn.get(sessnm);
467 return sess != null ? sess.getSessionId() : -1;
471 * Replaces a session within the persistent store, if it exists. Adds it otherwise.
473 * @param conn persistence connector
474 * @param sessnm name of session to be updated
475 * @param kieSession new session information
477 private void replaceSession(DroolsSessionConnector conn, String sessnm, KieSession kieSession) {
479 DroolsSessionEntity sess = new DroolsSessionEntity();
481 sess.setSessionName(sessnm);
482 sess.setSessionId(kieSession.getIdentifier());
488 /* ============================================================ */
491 * Gets the data source properties.
493 * @return the data source properties
495 private Properties getDataSourceProperties() {
496 Properties props = new Properties();
497 props.put("driverClassName", persistProps.getProperty(DroolsPersistenceProperties.DB_DRIVER));
498 props.put("url", persistProps.getProperty(DroolsPersistenceProperties.DB_URL));
499 props.put("username", persistProps.getProperty(DroolsPersistenceProperties.DB_USER));
500 props.put("password", persistProps.getProperty(DroolsPersistenceProperties.DB_PWD));
501 props.put("maxActive", "3");
502 props.put("maxIdle", "1");
503 props.put("maxWait", "120000");
504 props.put("whenExhaustedAction", "2");
505 props.put("testOnBorrow", "false");
506 props.put("poolPreparedStatements", "true");
512 * Removes "old" Drools 'sessioninfo' records, so they aren't used to restore data to Drools
513 * sessions. This also has the useful side-effect of removing abandoned records as well.
515 private void cleanUpSessionInfo() {
517 synchronized (cleanupLock) {
518 if (sessInfoCleaned) {
519 logger.info("Clean up of sessioninfo table: already done");
523 if (sessionInfoTimeoutMs < 0) {
524 logger.info("Clean up of sessioninfo table: no timeout specified");
528 // now do the record deletion
529 try (BasicDataSource ds = fact.makeDataSource(getDataSourceProperties());
530 Connection connection = ds.getConnection();
531 PreparedStatement statement =
532 connection.prepareStatement(
533 "DELETE FROM sessioninfo WHERE timestampdiff(second,lastmodificationdate,now()) > ?")) {
535 connection.setAutoCommit(true);
537 statement.setLong(1, sessionInfoTimeoutMs / 1000);
539 int count = statement.executeUpdate();
540 logger.info("Cleaning up sessioninfo table -- {} records removed", count);
542 } catch (SQLException e) {
543 logger.error("Clean up of sessioninfo table failed", e);
546 // delete DroolsSessionEntity where sessionId not in (sessinfo.xxx)?
548 sessInfoCleaned = true;
553 * Determine whether persistence is enabled for a specific container.
555 * @param container container to be checked
556 * @param sessionName name of the session to be checked
557 * @return {@code true} if persistence is enabled for this container, and {@code false} if not
559 private boolean isPersistenceEnabled(PolicyContainer container, String sessionName) {
560 Properties properties = getProperties(container);
561 boolean rval = false;
563 if (properties != null) {
564 // fetch the 'type' property
565 String type = getProperty(properties, sessionName, "type");
566 rval = "auto".equals(type) || "native".equals(type);
573 * Determine the controller properties associated with the policy container.
575 * @param container container whose properties are to be retrieved
576 * @return the container's properties, or {@code null} if not found
578 private Properties getProperties(PolicyContainer container) {
580 return fact.getPolicyController(container).getProperties();
581 } catch (IllegalArgumentException e) {
582 logger.error("getProperties exception: ", e);
588 * Fetch the persistence property associated with a session. The name may have the form:
591 * <li>persistence.SESSION-NAME.PROPERTY
592 * <li>persistence.PROPERTY
595 * @param properties properties from which the value is to be retrieved
596 * @param sessionName session name of interest
597 * @param property property name of interest
598 * @return the property value, or {@code null} if not found
600 private String getProperty(Properties properties, String sessionName, String property) {
601 String value = properties.getProperty("persistence." + sessionName + "." + property);
603 value = properties.getProperty("persistence." + property);
609 /* ============================================================ */
612 * This 'ThreadModel' variant periodically calls 'KieSession.fireAllRules()', because the
613 * 'fireUntilHalt' method isn't compatible with persistence.
615 public class PersistentThreadModel implements Runnable, PolicySession.ThreadModel {
617 /** Session associated with this persistent thread. */
618 private final PolicySession session;
620 /** The session thread. */
621 private final Thread thread;
623 /** Used to indicate that processing should stop. */
624 private final CountDownLatch stopped = new CountDownLatch(1);
626 /** Minimum time, in milli-seconds, that the thread should sleep before firing rules again. */
627 long minSleepTime = 100;
630 * Maximum time, in milli-seconds, that the thread should sleep before firing rules again. This
631 * is a "half" time, so that we can multiply it by two without overflowing the word size.
633 long halfMaxSleepTime = 5000L / 2L;
636 * Constructor - initialize variables and create thread.
638 * @param session the 'PolicySession' instance
639 * @param properties may contain additional session properties
641 public PersistentThreadModel(PolicySession session, Properties properties) {
642 this.session = session;
643 this.thread = new Thread(this, getThreadName());
645 if (properties == null) {
649 // extract 'minSleepTime' and/or 'maxSleepTime'
650 String name = session.getName();
652 // fetch 'minSleepTime' value, and update if defined
653 String sleepTimeString = getProperty(properties, name, "minSleepTime");
654 if (sleepTimeString != null) {
656 minSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
657 } catch (Exception e) {
658 logger.error(sleepTimeString + ": Illegal value for 'minSleepTime'", e);
662 // fetch 'maxSleepTime' value, and update if defined
663 long maxSleepTime = 2 * halfMaxSleepTime;
664 sleepTimeString = getProperty(properties, name, "maxSleepTime");
665 if (sleepTimeString != null) {
667 maxSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
668 } catch (Exception e) {
669 logger.error(sleepTimeString + ": Illegal value for 'maxSleepTime'", e);
673 // swap values if needed
674 if (minSleepTime > maxSleepTime) {
678 + ") is greater than maxSleepTime("
681 long tmp = minSleepTime;
682 minSleepTime = maxSleepTime;
686 halfMaxSleepTime = Math.max(1, maxSleepTime / 2);
692 * @return the String to use as the thread name */
693 private String getThreadName() {
694 return "Session " + session.getFullName() + " (persistent)";
697 /*=========================*/
698 /* 'ThreadModel' interface */
699 /*=========================*/
705 public void start() {
714 // tell the thread to stop
717 // wait up to 10 seconds for the thread to stop
721 } catch (InterruptedException e) {
722 logger.error("stopThread exception: ", e);
723 Thread.currentThread().interrupt();
726 // verify that it's done
727 if (thread.isAlive()) {
728 logger.error("stopThread: still running");
736 public void updated() {
737 // the container artifact has been updated -- adjust the thread name
738 thread.setName(getThreadName());
741 /*======================*/
742 /* 'Runnable' interface */
743 /*======================*/
750 logger.info("PersistentThreadModel running");
752 // set thread local variable
753 session.setPolicySession();
755 KieSession kieSession = session.getKieSession();
756 long sleepTime = 2 * halfMaxSleepTime;
758 // We want to continue, despite any exceptions that occur
759 // while rules are fired.
765 if (kieSession.fireAllRules() > 0) {
766 // some rules fired -- reduce poll delay
767 sleepTime = Math.max(minSleepTime, sleepTime / 2);
769 // no rules fired -- increase poll delay
770 sleepTime = 2 * Math.min(halfMaxSleepTime, sleepTime);
773 } catch (Exception | LinkageError e) {
774 logger.error("Exception during kieSession.fireAllRules", e);
778 if (stopped.await(sleepTime, TimeUnit.MILLISECONDS)) {
782 } catch (InterruptedException e) {
783 logger.error("startThread exception: ", e);
784 Thread.currentThread().interrupt();
789 logger.info("PersistentThreadModel completed");
793 /* ============================================================ */
795 /** DataSource-EntityManagerFactory pair. */
796 private class DsEmf {
797 private BasicDataSource bds;
798 private EntityManagerFactory emf;
801 * Makes an entity manager factory for the given data source.
803 * @param bds pooled data source
805 public DsEmf(BasicDataSource bds) {
807 Map<String, Object> props = new HashMap<>();
808 props.put(org.hibernate.cfg.Environment.JPA_JTA_DATASOURCE, bds);
811 this.emf = fact.makeEntMgrFact(props);
813 } catch (RuntimeException e) {
819 /** Closes the entity manager factory and the data source. */
820 public void close() {
824 } catch (RuntimeException e) {
832 /** Closes the data source only. */
833 private void closeDataSource() {
837 } catch (SQLException e) {
838 throw new PersistenceFeatureException(e);
843 private static class SingletonRegistry {
844 private static final TransactionSynchronizationRegistry transreg =
845 new com.arjuna.ats.internal.jta.transaction.arjunacore
846 .TransactionSynchronizationRegistryImple();
848 private SingletonRegistry() {
853 /** Factory for various items. Methods can be overridden for junit testing. */
854 protected static class Factory {
857 * Gets the transaction manager.
859 * @return the transaction manager
861 public TransactionManager getTransMgr() {
862 return com.arjuna.ats.jta.TransactionManager.transactionManager();
866 * Gets the user transaction.
868 * @return the user transaction
870 public UserTransaction getUserTrans() {
871 return com.arjuna.ats.jta.UserTransaction.userTransaction();
875 * Gets the transaction synchronization registry.
877 * @return the transaction synchronization registry
879 public TransactionSynchronizationRegistry getTransSyncReg() {
880 return SingletonRegistry.transreg;
884 * Gets the KIE services.
886 * @return the KIE services
888 public KieServices getKieServices() {
889 return KieServices.Factory.get();
893 * Loads properties from a file.
895 * @param filenm name of the file to load
896 * @return properties, as loaded from the file
897 * @throws IOException if an error occurs reading from the file
899 public Properties loadProperties(String filenm) throws IOException {
900 return PropertyUtil.getProperties(filenm);
904 * Makes a Data Source.
906 * @param dsProps data source properties
907 * @return a new data source
909 public BasicDataSource makeDataSource(Properties dsProps) {
911 return BasicDataSourceFactory.createDataSource(dsProps);
913 } catch (Exception e) {
914 throw new PersistenceFeatureException(e);
919 * Makes a new JPA connector for drools sessions.
921 * @param emf entity manager factory
922 * @return a new JPA connector for drools sessions
924 public DroolsSessionConnector makeJpaConnector(EntityManagerFactory emf) {
925 return new JpaDroolsSessionConnector(emf);
929 * Makes a new entity manager factory.
931 * @param props properties with which the factory should be configured
932 * @return a new entity manager factory
934 public EntityManagerFactory makeEntMgrFact(Map<String, Object> props) {
935 return Persistence.createEntityManagerFactory("onapsessionsPU", props);
939 * Gets the policy controller associated with a given policy container.
941 * @param container container whose controller is to be retrieved
942 * @return the container's controller
944 public PolicyController getPolicyController(PolicyContainer container) {
945 return PolicyController.factory.get(container.getGroupId(), container.getArtifactId());
950 * Runtime exceptions generated by this class. Wraps exceptions generated by delegated operations,
951 * particularly when they are not, themselves, Runtime exceptions.
953 public static class PersistenceFeatureException extends RuntimeException {
954 private static final long serialVersionUID = 1L;
959 public PersistenceFeatureException(Exception ex) {