feature-distributed-locking/src/main/java/org/onap/policy/distributed/locking/DistributedLockingFeature.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * feature-distributed-locking
   4  * ================================================================================
   5  * Copyright (C) 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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20 package org.onap.policy.distributed.locking;
  21
  22 import java.sql.Connection;
  23 import java.sql.PreparedStatement;
  24 import java.sql.SQLException;
  25 import java.util.Properties;
  26 import java.util.UUID;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.Future;
  29 import java.util.concurrent.ScheduledExecutorService;
  30 import java.util.concurrent.TimeUnit;
  31 import org.apache.commons.dbcp2.BasicDataSource;
  32 import org.apache.commons.dbcp2.BasicDataSourceFactory;
  33 import org.onap.policy.common.utils.properties.exception.PropertyException;
  34 import org.onap.policy.drools.core.lock.LockRequestFuture;
  35 import org.onap.policy.drools.core.lock.PolicyResourceLockFeatureAPI;
  36 import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
  37 import org.onap.policy.drools.persistence.SystemPersistence;
  38 import org.onap.policy.drools.system.PolicyEngine;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 public class DistributedLockingFeature implements PolicyEngineFeatureAPI, PolicyResourceLockFeatureAPI {
  43
  44         /**
  45          * Logger instance
  46          */
  47         private static final Logger logger = LoggerFactory.getLogger(DistributedLockingFeature.class);
  48
  49         /**
  50          * Properties Configuration Name
  51          */
  52         public static final String CONFIGURATION_PROPERTIES_NAME = "feature-distributed-locking";
  53
  54         /**
  55          * Properties for locking feature
  56          */
  57         private DistributedLockingProperties lockProps;
  58
  59         /**
  60          *ScheduledExecutorService for LockHeartbeat
  61          */
  62         private ScheduledExecutorService scheduledExecutorService;
  63
  64         /**
  65          * Data source used to connect to the DB containing locks.
  66          */
  67         private BasicDataSource dataSource;
  68
  69         /**
  70          * UUID
  71          */
  72         private static final UUID uuid = UUID.randomUUID();
  73
  74
  75         /**
  76          * Reference to Heartbeat
  77          */
  78         private static Heartbeat heartbeat = null;
  79
  80         @Override
  81         public int getSequenceNumber() {
  82         return 1000;
  83         }
  84
  85         @Override
  86         public Future<Boolean> beforeLock(String resourceId, String owner, Callback callback) {
  87
  88                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps, dataSource);
  89
  90                 return new LockRequestFuture(resourceId, owner, tLock.lock());
  91
  92         }
  93
  94         @Override
  95         public OperResult beforeUnlock(String resourceId, String owner) {
  96                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps, dataSource);
  97
  98                 return(tLock.unlock() ? OperResult.OPER_ACCEPTED : OperResult.OPER_DENIED);
  99         }
 100
 101         @Override
 102         public OperResult beforeIsLockedBy(String resourceId, String owner) {
 103                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps, dataSource);
 104
 105         return(tLock.isActive() ? OperResult.OPER_ACCEPTED : OperResult.OPER_DENIED);
 106         }
 107
 108         @Override
 109         public OperResult beforeIsLocked(String resourceId) {
 110                 TargetLock tLock = new TargetLock(resourceId, uuid, "dummyOwner", lockProps, dataSource);
 111
 112         return(tLock.isLocked() ? OperResult.OPER_ACCEPTED : OperResult.OPER_DENIED);
 113         }
 114
 115         @Override
 116         public boolean afterStart(PolicyEngine engine) {
 117
 118                 try {
 119                         this.lockProps = new DistributedLockingProperties(SystemPersistence.manager.getProperties(DistributedLockingFeature.CONFIGURATION_PROPERTIES_NAME));
 120                         this.dataSource = makeDataSource();
 121                 } catch (PropertyException e) {
 122                         logger.error("DistributedLockingFeature feature properies have not been loaded", e);
 123                         throw new DistributedLockingFeatureException(e);
 124                 } catch(InterruptedException e) {
 125             logger.error("DistributedLockingFeature failed to create data source", e);
 126                     Thread.currentThread().interrupt();
 127             throw new DistributedLockingFeatureException(e);
 128         } catch(Exception e) {
 129             logger.error("DistributedLockingFeature failed to create data source", e);
 130             throw new DistributedLockingFeatureException(e);
 131                 }
 132
 133                 long heartbeatInterval = this.lockProps.getHeartBeatIntervalProperty();
 134
 135                 cleanLockTable();
 136                 initHeartbeat();
 137
 138                 this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
 139                 this.scheduledExecutorService.scheduleAtFixedRate(heartbeat, heartbeatInterval, heartbeatInterval, TimeUnit.MILLISECONDS);
 140                 return false;
 141         }
 142
 143         /**
 144          * @return a new, pooled data source
 145          * @throws Exception
 146          */
 147         private BasicDataSource makeDataSource() throws Exception {
 148         Properties props = new Properties();
 149         props.put("driverClassName", lockProps.getDbDriver());
 150         props.put("url", lockProps.getDbUrl());
 151         props.put("username", lockProps.getDbUser());
 152         props.put("password", lockProps.getDbPwd());
 153         props.put("testOnBorrow", "true");
 154         props.put("poolPreparedStatements", "true");
 155
 156         // additional properties are listed in the GenericObjectPool API
 157
 158         return BasicDataSourceFactory.createDataSource(props);
 159     }
 160
 161     /**
 162          * This method kills the heartbeat thread and calls refreshLockTable which removes
 163          * any records from the db where the current host is the owner.
 164          */
 165         @Override
 166         public boolean beforeShutdown(PolicyEngine engine) {
 167                 scheduledExecutorService.shutdown();
 168                 cleanLockTable();
 169                 return false;
 170         }
 171
 172         /**
 173          * This method removes all records owned by the current host from the db.
 174          */
 175         private void cleanLockTable() {
 176
 177             try (Connection conn = dataSource.getConnection();
 178                 PreparedStatement statement = conn.prepareStatement("DELETE FROM pooling.locks WHERE host = ? OR expirationTime < ?");
 179                 ){
 180
 181                                 statement.setString(1, uuid.toString());
 182                                 statement.setLong(2, System.currentTimeMillis());
 183                                 statement.executeUpdate();
 184
 185                 } catch (SQLException e) {
 186                         logger.error("error in refreshLockTable()", e);
 187                 }
 188
 189         }
 190
 191         /**
 192          * Initialize the static heartbeat object
 193          */
 194         private void initHeartbeat() {
 195                 heartbeat = new Heartbeat(uuid, lockProps, dataSource);
 196
 197         }
 198
 199         public static Heartbeat getHeartbeat() {
 200                 return heartbeat;
 201         }
 202
 203 }