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 // TODO: delete DroolsSessionEntity where sessionId not in
549 sessInfoCleaned = true;
554 * Determine whether persistence is enabled for a specific container.
556 * @param container container to be checked
557 * @param sessionName name of the session to be checked
558 * @return {@code true} if persistence is enabled for this container, and {@code false} if not
560 private boolean isPersistenceEnabled(PolicyContainer container, String sessionName) {
561 Properties properties = getProperties(container);
562 boolean rval = false;
564 if (properties != null) {
565 // fetch the 'type' property
566 String type = getProperty(properties, sessionName, "type");
567 rval = "auto".equals(type) || "native".equals(type);
574 * Determine the controller properties associated with the policy container.
576 * @param container container whose properties are to be retrieved
577 * @return the container's properties, or {@code null} if not found
579 private Properties getProperties(PolicyContainer container) {
581 return fact.getPolicyController(container).getProperties();
582 } catch (IllegalArgumentException e) {
583 logger.error("getProperties exception: ", e);
589 * Fetch the persistence property associated with a session. The name may have the form:
592 * <li>persistence.SESSION-NAME.PROPERTY
593 * <li>persistence.PROPERTY
596 * @param properties properties from which the value is to be retrieved
597 * @param sessionName session name of interest
598 * @param property property name of interest
599 * @return the property value, or {@code null} if not found
601 private String getProperty(Properties properties, String sessionName, String property) {
602 String value = properties.getProperty("persistence." + sessionName + "." + property);
604 value = properties.getProperty("persistence." + property);
610 /* ============================================================ */
613 * This 'ThreadModel' variant periodically calls 'KieSession.fireAllRules()', because the
614 * 'fireUntilHalt' method isn't compatible with persistence.
616 public class PersistentThreadModel implements Runnable, PolicySession.ThreadModel {
618 /** Session associated with this persistent thread. */
619 private final PolicySession session;
621 /** The session thread. */
622 private final Thread thread;
624 /** Used to indicate that processing should stop. */
625 private final CountDownLatch stopped = new CountDownLatch(1);
627 /** Minimum time, in milli-seconds, that the thread should sleep before firing rules again. */
628 long minSleepTime = 100;
631 * Maximum time, in milli-seconds, that the thread should sleep before firing rules again. This
632 * is a "half" time, so that we can multiply it by two without overflowing the word size.
634 long halfMaxSleepTime = 5000L / 2L;
637 * Constructor - initialize variables and create thread.
639 * @param session the 'PolicySession' instance
640 * @param properties may contain additional session properties
642 public PersistentThreadModel(PolicySession session, Properties properties) {
643 this.session = session;
644 this.thread = new Thread(this, getThreadName());
646 if (properties == null) {
650 // extract 'minSleepTime' and/or 'maxSleepTime'
651 String name = session.getName();
653 // fetch 'minSleepTime' value, and update if defined
654 String sleepTimeString = getProperty(properties, name, "minSleepTime");
655 if (sleepTimeString != null) {
657 minSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
658 } catch (Exception e) {
659 logger.error(sleepTimeString + ": Illegal value for 'minSleepTime'", e);
663 // fetch 'maxSleepTime' value, and update if defined
664 long maxSleepTime = 2 * halfMaxSleepTime;
665 sleepTimeString = getProperty(properties, name, "maxSleepTime");
666 if (sleepTimeString != null) {
668 maxSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
669 } catch (Exception e) {
670 logger.error(sleepTimeString + ": Illegal value for 'maxSleepTime'", e);
674 // swap values if needed
675 if (minSleepTime > maxSleepTime) {
679 + ") is greater than maxSleepTime("
682 long tmp = minSleepTime;
683 minSleepTime = maxSleepTime;
687 halfMaxSleepTime = Math.max(1, maxSleepTime / 2);
693 * @return the String to use as the thread name */
694 private String getThreadName() {
695 return "Session " + session.getFullName() + " (persistent)";
698 /*=========================*/
699 /* 'ThreadModel' interface */
700 /*=========================*/
706 public void start() {
715 // tell the thread to stop
718 // wait up to 10 seconds for the thread to stop
722 } catch (InterruptedException e) {
723 logger.error("stopThread exception: ", e);
724 Thread.currentThread().interrupt();
727 // verify that it's done
728 if (thread.isAlive()) {
729 logger.error("stopThread: still running");
737 public void updated() {
738 // the container artifact has been updated -- adjust the thread name
739 thread.setName(getThreadName());
742 /*======================*/
743 /* 'Runnable' interface */
744 /*======================*/
751 logger.info("PersistentThreadModel running");
753 // set thread local variable
754 session.setPolicySession();
756 KieSession kieSession = session.getKieSession();
757 long sleepTime = 2 * halfMaxSleepTime;
759 // We want to continue, despite any exceptions that occur
760 // while rules are fired.
766 if (kieSession.fireAllRules() > 0) {
767 // some rules fired -- reduce poll delay
768 sleepTime = Math.max(minSleepTime, sleepTime / 2);
770 // no rules fired -- increase poll delay
771 sleepTime = 2 * Math.min(halfMaxSleepTime, sleepTime);
774 } catch (Exception | LinkageError e) {
775 logger.error("Exception during kieSession.fireAllRules", e);
779 if (stopped.await(sleepTime, TimeUnit.MILLISECONDS)) {
783 } catch (InterruptedException e) {
784 logger.error("startThread exception: ", e);
785 Thread.currentThread().interrupt();
790 logger.info("PersistentThreadModel completed");
794 /* ============================================================ */
796 /** DataSource-EntityManagerFactory pair. */
797 private class DsEmf {
798 private BasicDataSource bds;
799 private EntityManagerFactory emf;
802 * Makes an entity manager factory for the given data source.
804 * @param bds pooled data source
806 public DsEmf(BasicDataSource bds) {
808 Map<String, Object> props = new HashMap<>();
809 props.put(org.hibernate.cfg.Environment.JPA_JTA_DATASOURCE, bds);
812 this.emf = fact.makeEntMgrFact(props);
814 } catch (RuntimeException e) {
820 /** Closes the entity manager factory and the data source. */
821 public void close() {
825 } catch (RuntimeException e) {
833 /** Closes the data source only. */
834 private void closeDataSource() {
838 } catch (SQLException e) {
839 throw new PersistenceFeatureException(e);
844 private static class SingletonRegistry {
845 private static final TransactionSynchronizationRegistry transreg =
846 new com.arjuna.ats.internal.jta.transaction.arjunacore
847 .TransactionSynchronizationRegistryImple();
849 private SingletonRegistry() {
854 /** Factory for various items. Methods can be overridden for junit testing. */
855 protected static class Factory {
858 * Gets the transaction manager.
860 * @return the transaction manager
862 public TransactionManager getTransMgr() {
863 return com.arjuna.ats.jta.TransactionManager.transactionManager();
867 * Gets the user transaction.
869 * @return the user transaction
871 public UserTransaction getUserTrans() {
872 return com.arjuna.ats.jta.UserTransaction.userTransaction();
876 * Gets the transaction synchronization registry.
878 * @return the transaction synchronization registry
880 public TransactionSynchronizationRegistry getTransSyncReg() {
881 return SingletonRegistry.transreg;
885 * Gets the KIE services.
887 * @return the KIE services
889 public KieServices getKieServices() {
890 return KieServices.Factory.get();
894 * Loads properties from a file.
896 * @param filenm name of the file to load
897 * @return properties, as loaded from the file
898 * @throws IOException if an error occurs reading from the file
900 public Properties loadProperties(String filenm) throws IOException {
901 return PropertyUtil.getProperties(filenm);
905 * Makes a Data Source.
907 * @param dsProps data source properties
908 * @return a new data source
910 public BasicDataSource makeDataSource(Properties dsProps) {
912 return BasicDataSourceFactory.createDataSource(dsProps);
914 } catch (Exception e) {
915 throw new PersistenceFeatureException(e);
920 * Makes a new JPA connector for drools sessions.
922 * @param emf entity manager factory
923 * @return a new JPA connector for drools sessions
925 public DroolsSessionConnector makeJpaConnector(EntityManagerFactory emf) {
926 return new JpaDroolsSessionConnector(emf);
930 * Makes a new entity manager factory.
932 * @param props properties with which the factory should be configured
933 * @return a new entity manager factory
935 public EntityManagerFactory makeEntMgrFact(Map<String, Object> props) {
936 return Persistence.createEntityManagerFactory("onapsessionsPU", props);
940 * Gets the policy controller associated with a given policy container.
942 * @param container container whose controller is to be retrieved
943 * @return the container's controller
945 public PolicyController getPolicyController(PolicyContainer container) {
946 return PolicyController.factory.get(container.getGroupId(), container.getArtifactId());
951 * Runtime exceptions generated by this class. Wraps exceptions generated by delegated operations,
952 * particularly when they are not, themselves, Runtime exceptions.
954 public static class PersistenceFeatureException extends RuntimeException {
955 private static final long serialVersionUID = 1L;
960 public PersistenceFeatureException(Exception ex) {