gerrit.onap Code Review - policy/drools-pdp.git/blob

   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.DriverManager;
  24 import java.sql.PreparedStatement;
  25 import java.sql.SQLException;
  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
  32 import org.onap.policy.common.utils.properties.exception.PropertyException;
  33 import org.onap.policy.drools.core.lock.LockRequestFuture;
  34 import org.onap.policy.drools.core.lock.PolicyResourceLockFeatureAPI;
  35 import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
  36 import org.onap.policy.drools.persistence.SystemPersistence;
  37 import org.onap.policy.drools.system.PolicyEngine;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 public class DistributedLockingFeature implements PolicyEngineFeatureAPI, PolicyResourceLockFeatureAPI {
  42
  43         /**
  44          * Logger instance
  45          */
  46         private static final Logger logger = LoggerFactory.getLogger(DistributedLockingFeature.class);
  47
  48         /**
  49          * Properties Configuration Name
  50          */
  51         public static final String CONFIGURATION_PROPERTIES_NAME = "feature-distributed-locking";
  52
  53         /**
  54          * Properties for locking feature
  55          */
  56         private DistributedLockingProperties lockProps;
  57
  58         /**
  59          *ScheduledExecutorService for LockHeartbeat
  60          */
  61         private ScheduledExecutorService scheduledExecutorService;
  62
  63         /**
  64          * UUID
  65          */
  66         private static final UUID uuid = UUID.randomUUID();
  67
  68
  69         /**
  70          * Reference to Heartbeat
  71          */
  72         private static Heartbeat heartbeat = null;
  73
  74         @Override
  75         public int getSequenceNumber() {
  76         return 1000;
  77         }
  78
  79         @Override
  80         public Future<Boolean> beforeLock(String resourceId, String owner, Callback callback) {
  81
  82                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps);
  83
  84                 return new LockRequestFuture(resourceId, owner, tLock.lock());
  85
  86         }
  87
  88         @Override
  89         public Boolean beforeUnlock(String resourceId, String owner) {
  90                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps);
  91
  92                 return tLock.unlock();
  93         }
  94
  95         @Override
  96         public Boolean beforeIsLockedBy(String resourceId, String owner) {
  97                 TargetLock tLock = new TargetLock(resourceId, uuid, owner, lockProps);
  98
  99                 return tLock.isActive();
 100         }
 101
 102         @Override
 103         public Boolean beforeIsLocked(String resourceId) {
 104                 TargetLock tLock = new TargetLock(resourceId, uuid, "dummyOwner", lockProps);
 105
 106                 return tLock.isLocked();
 107         }
 108
 109         @Override
 110         public boolean afterStart(PolicyEngine engine) {
 111
 112                 try {
 113                         this.lockProps = new DistributedLockingProperties(SystemPersistence.manager.getProperties(DistributedLockingFeature.CONFIGURATION_PROPERTIES_NAME));
 114                 } catch (PropertyException e) {
 115                         logger.error("DistributedLockingFeature feature properies have not been loaded", e);
 116                         throw new DistributedLockingFeatureException(e);
 117                 }
 118
 119                 long heartbeatInterval = this.lockProps.getHeartBeatIntervalProperty();
 120
 121                 cleanLockTable();
 122                 heartbeat = new Heartbeat(uuid, lockProps);
 123
 124                 this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
 125                 this.scheduledExecutorService.scheduleAtFixedRate(heartbeat, heartbeatInterval, heartbeatInterval, TimeUnit.MILLISECONDS);
 126                 return false;
 127         }
 128
 129         /**
 130          * This method kills the heartbeat thread and calls refreshLockTable which removes
 131          * any records from the db where the current host is the owner.
 132          */
 133         @Override
 134         public boolean beforeShutdown(PolicyEngine engine) {
 135                 scheduledExecutorService.shutdown();
 136                 cleanLockTable();
 137                 return false;
 138         }
 139
 140         /**
 141          * This method removes all records owned by the current host from the db.
 142          */
 143         private void cleanLockTable() {
 144
 145             try (Connection conn = DriverManager.getConnection(lockProps.getDbUrl(),
 146                         lockProps.getDbUser(),
 147                         lockProps.getDbPwd());
 148                 PreparedStatement statement = conn.prepareStatement("DELETE FROM pooling.locks WHERE host = ? OR expirationTime < ?");
 149                 ){
 150
 151                                 statement.setString(1, uuid.toString());
 152                                 statement.setLong(2, System.currentTimeMillis());
 153                                 statement.executeUpdate();
 154
 155                 } catch (SQLException e) {
 156                         logger.error("error in refreshLockTable()", e);
 157                 }
 158
 159         }
 160
 161         public static Heartbeat getHeartbeat() {
 162                 return heartbeat;
 163         }
 164
 165 }