diff options
Diffstat (limited to 'src/main/java/org/onap/music/main')
| -rwxr-xr-x | src/main/java/org/onap/music/main/CachingUtil.java | 75 | ||||
| -rw-r--r-- | src/main/java/org/onap/music/main/CronJobManager.java | 94 | ||||
| -rw-r--r-- | src/main/java/org/onap/music/main/MusicCore.java | 776 | ||||
| -rwxr-xr-x | src/main/java/org/onap/music/main/MusicUtil.java | 83 | ||||
| -rwxr-xr-x | src/main/java/org/onap/music/main/PropertiesListener.java | 9 | ||||
| -rw-r--r-- | src/main/java/org/onap/music/main/VotingAppJar.java | 115 |
6 files changed, 507 insertions, 645 deletions
diff --git a/src/main/java/org/onap/music/main/CachingUtil.java b/src/main/java/org/onap/music/main/CachingUtil.java index a81887a0..d3654118 100755 --- a/src/main/java/org/onap/music/main/CachingUtil.java +++ b/src/main/java/org/onap/music/main/CachingUtil.java @@ -21,24 +21,28 @@ */ package org.onap.music.main; +import java.util.Arrays; import java.util.Calendar; import java.util.HashMap; import java.util.Iterator; -import java.util.List; import java.util.Map; import java.util.UUID; +import javax.ws.rs.core.HttpHeaders; import javax.ws.rs.core.MediaType; import org.apache.commons.codec.binary.Base64; import org.apache.commons.jcs.JCS; import org.apache.commons.jcs.access.CacheAccess; +import org.codehaus.jackson.map.ObjectMapper; import org.mindrot.jbcrypt.BCrypt; import org.onap.music.datastore.PreparedQueryObject; +import org.onap.music.datastore.jsonobjects.AAFResponse; import org.onap.music.eelf.logging.EELFLoggerDelegate; import org.onap.music.eelf.logging.format.AppMessages; import org.onap.music.eelf.logging.format.ErrorSeverity; import org.onap.music.eelf.logging.format.ErrorTypes; import org.onap.music.exceptions.MusicServiceException; -import org.onap.music.datastore.jsonobjects.JsonCallback; + +import com.att.eelf.configuration.EELFLogger; import com.datastax.driver.core.DataType; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; @@ -61,8 +65,6 @@ public class CachingUtil implements Runnable { private static CacheAccess<String, Map<String, String>> aafCache = JCS.getInstance("aafCache"); private static CacheAccess<String, String> appNameCache = JCS.getInstance("appNameCache"); private static CacheAccess<String, Map<String, String>> musicValidateCache = JCS.getInstance("musicValidateCache"); - private static CacheAccess<String, JsonCallback> callBackCache = JCS.getInstance("callBackCache"); - private static CacheAccess<String, List<String>> callbackNotifyList = JCS.getInstance("callbackNotifyList"); private static Map<String, Number> userAttempts = new HashMap<>(); private static Map<String, Calendar> lastFailedTime = new HashMap<>(); @@ -71,25 +73,7 @@ public class CachingUtil implements Runnable { return true; return false; } - - public static void updateCallBackCache(String appName, JsonCallback jsonCallBack) { - logger.info("updateCallBackCache: updating cache....."); - callBackCache.put(appName, jsonCallBack); - } - - public static JsonCallback getCallBackCache(String appName) { - return callBackCache.get(appName); - } - public static void updateCallbackNotifyList(List<String> notifyList) { - logger.info("callbackNotifyList: updating cache....."); - callbackNotifyList.put("callbackNotify", notifyList); - } - - public static List<String> getCallbackNotifyList() { - return callbackNotifyList.get("callbackNotify"); - } - public void initializeMusicCache() { logger.info(EELFLoggerDelegate.applicationLogger,"Initializing Music Cache..."); musicCache.put("isInitialized", "true"); @@ -105,6 +89,7 @@ public class CachingUtil implements Runnable { pQuery.addValue(MusicUtil.convertToActualDataType(DataType.cboolean(), false)); } catch (Exception e1) { logger.error(EELFLoggerDelegate.errorLogger, e1.getMessage(),AppMessages.CACHEERROR, ErrorSeverity.CRITICAL, ErrorTypes.GENERALSERVICEERROR); + e1.printStackTrace(); } ResultSet rs = MusicCore.get(pQuery); Iterator<Row> it = rs.iterator(); @@ -117,8 +102,8 @@ public class CachingUtil implements Runnable { String keySpace = row.getString("application_name"); try { userAttempts.put(nameSpace, 0); - boolean responseObj = triggerAAF(nameSpace, userId, password); - if (responseObj) { + AAFResponse responseObj = triggerAAF(nameSpace, userId, password); + if (responseObj.getNs().size() > 0) { map = new HashMap<>(); map.put(userId, password); aafCache.put(nameSpace, map); @@ -129,6 +114,7 @@ public class CachingUtil implements Runnable { } catch (Exception e) { logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.UNKNOWNERROR, ErrorSeverity.INFO, ErrorTypes.GENERALSERVICEERROR); logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),"Something at AAF was changed for ns: " + nameSpace+" So not updating Cache for the namespace. "); + e.printStackTrace(); } } @@ -147,7 +133,7 @@ public class CachingUtil implements Runnable { public static boolean authenticateAAFUser(String nameSpace, String userId, String password, String keySpace) throws Exception { - if (aafCache.get(nameSpace) != null && musicCache.get(keySpace)!=null) { + if (aafCache.get(nameSpace) != null) { if (keySpace != null && !musicCache.get(keySpace).equals(nameSpace)) { logger.info(EELFLoggerDelegate.applicationLogger,"Create new application for the same namespace."); } else if (aafCache.get(nameSpace).get(userId).equals(password)) { @@ -178,13 +164,12 @@ public class CachingUtil implements Runnable { } } - boolean responseObj = triggerAAF(nameSpace, userId, password); - if (responseObj) { + AAFResponse responseObj = triggerAAF(nameSpace, userId, password); + if (responseObj.getNs().size() > 0) { //if (responseObj.getNs().get(0).getAdmin().contains(userId)) { - Map<String, String> map = new HashMap<>(); - map.put(userId, password); - aafCache.put(nameSpace, map); - CachingUtil.updateMusicCache(keySpace, nameSpace); + //Map<String, String> map = new HashMap<>(); + //map.put(userId, password); + //aafCache.put(nameSpace, map); return true; //} } @@ -192,7 +177,7 @@ public class CachingUtil implements Runnable { return false; } - private static boolean triggerAAF(String nameSpace, String userId, String password) + private static AAFResponse triggerAAF(String nameSpace, String userId, String password) throws Exception { if (MusicUtil.getAafEndpointUrl() == null) { logger.error(EELFLoggerDelegate.errorLogger,"",AppMessages.UNKNOWNERROR,ErrorSeverity.WARN, ErrorTypes.GENERALSERVICEERROR); @@ -225,14 +210,14 @@ public class CachingUtil implements Runnable { // TODO Allow for 2-3 times and forbid any attempt to trigger AAF with invalid values // for specific time. } - /*response.getHeaders().put(HttpHeaders.CONTENT_TYPE, + response.getHeaders().put(HttpHeaders.CONTENT_TYPE, Arrays.asList(MediaType.APPLICATION_JSON)); // AAFResponse output = response.getEntity(AAFResponse.class); response.bufferEntity(); String x = response.getEntity(String.class); - AAFResponse responseObj = new ObjectMapper().readValue(x, AAFResponse.class);*/ + AAFResponse responseObj = new ObjectMapper().readValue(x, AAFResponse.class); - return true; + return responseObj; } public static void updateMusicCache(String keyspace, String nameSpace) { @@ -271,6 +256,7 @@ public class CachingUtil implements Runnable { appNameCache.put(namespace, isAAF); } catch (Exception e) { logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.QUERYERROR,ErrorSeverity.ERROR, ErrorTypes.QUERYERROR); + e.printStackTrace(); } } return isAAF; @@ -288,6 +274,7 @@ public class CachingUtil implements Runnable { uuid = rs.getUUID("uuid").toString(); } catch (Exception e) { logger.error(EELFLoggerDelegate.errorLogger,"Exception occured during uuid retrieval from DB."+e.getMessage()); + e.printStackTrace(); } } return uuid; @@ -304,6 +291,7 @@ public class CachingUtil implements Runnable { appName = rs.getString("application_name"); } catch (Exception e) { logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.QUERYERROR, ErrorSeverity.ERROR, ErrorTypes.QUERYERROR); + e.printStackTrace(); } return appName; } @@ -350,6 +338,7 @@ public class CachingUtil implements Runnable { rs = MusicCore.get(queryObject).one(); } catch (MusicServiceException e) { // TODO Auto-generated catch block + e.printStackTrace(); resultMap.put("Exception", "Unable to process operation. Error is "+e.getMessage()); return resultMap; } catch (InvalidQueryException e) { @@ -393,13 +382,14 @@ public class CachingUtil implements Runnable { try { queryObject.addValue(MusicUtil.convertToActualDataType(DataType.text(), keyspace)); } catch (Exception e) { - logger.error(EELFLoggerDelegate.errorLogger,"", AppMessages.AUTHENTICATIONERROR, ErrorSeverity.WARN, ErrorTypes.AUTHENTICATIONERROR); + e.printStackTrace(); } Row rs = null; try { rs = MusicCore.get(queryObject).one(); } catch (MusicServiceException e) { - resultMap.put("Exception", "Unable to process operation. Error is "+e.getMessage()); + e.printStackTrace(); + resultMap.put("Exception", "Unable to process operation. Error is "+e.getMessage()); return resultMap; } if(rs == null) { @@ -427,15 +417,4 @@ public class CachingUtil implements Runnable { CachingUtil.updateMusicValidateCache(nameSpace, userId, pwd); return resultMap; } - - public static void deleteKeysFromDB(String deleteKeys) { - PreparedQueryObject pQuery = new PreparedQueryObject(); - pQuery.appendQueryString( - "DELETE FROM admin.locks WHERE lock_id IN ("+deleteKeys+")"); - try { - MusicCore.nonKeyRelatedPut(pQuery, "eventual"); - } catch (Exception e) { - logger.error(EELFLoggerDelegate.errorLogger,"", AppMessages.AUTHENTICATIONERROR, "Error in deleteKeysFromDB"); - } - } } diff --git a/src/main/java/org/onap/music/main/CronJobManager.java b/src/main/java/org/onap/music/main/CronJobManager.java index 4928044e..5b7a8de4 100644 --- a/src/main/java/org/onap/music/main/CronJobManager.java +++ b/src/main/java/org/onap/music/main/CronJobManager.java @@ -32,21 +32,16 @@ import javax.servlet.ServletContextListener; import javax.servlet.annotation.WebListener; import org.onap.music.datastore.PreparedQueryObject; -import org.onap.music.eelf.logging.EELFLoggerDelegate; -import org.onap.music.eelf.logging.format.AppMessages; -import org.onap.music.eelf.logging.format.ErrorSeverity; -import org.onap.music.eelf.logging.format.ErrorTypes; import org.onap.music.exceptions.MusicLockingException; import org.onap.music.exceptions.MusicServiceException; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; -//@WebListener +@WebListener public class CronJobManager implements ServletContextListener { private ScheduledExecutorService scheduler; - private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(CronJobManager.class); @Override public void contextInitialized(ServletContextEvent event) { @@ -58,25 +53,10 @@ public class CronJobManager implements ServletContextListener { try { ResultType result = MusicCore.nonKeyRelatedPut(pQuery, consistency); } catch (MusicServiceException e1) { - logger.error(EELFLoggerDelegate.errorLogger, e1.getMessage(),ErrorSeverity.ERROR); + e1.printStackTrace(); } - - //Zookeeper cleanup - scheduler.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - deleteLocksFromDB(); - } - } , 0, 24, TimeUnit.HOURS); - } - - @Override - public void contextDestroyed(ServletContextEvent event) { - scheduler.shutdownNow(); - } - - public void deleteLocksFromDB() { - PreparedQueryObject pQuery = new PreparedQueryObject(); + + pQuery = new PreparedQueryObject(); pQuery.appendQueryString( "select * from admin.locks"); try { @@ -91,22 +71,68 @@ public class CronJobManager implements ServletContextListener { if(System.currentTimeMillis() >= ctime + 24 * 60 * 60 * 1000) { expiredKeys = true; String new_id = id.substring(1); - try { - MusicCore.deleteLock(new_id); - } catch (MusicLockingException e) { - logger.info(EELFLoggerDelegate.applicationLogger, - e.getMessage()); - } - deleteKeys.append("'").append(id).append("'").append(","); + MusicCore.deleteLock(new_id); + deleteKeys.append(id).append(","); } - } + else { + MusicUtil.zkNodeMap.put(id, ctime); + } + }; if(expiredKeys) { deleteKeys.deleteCharAt(deleteKeys.length()-1); - CachingUtil.deleteKeysFromDB(deleteKeys.toString()); + deleteKeysFromDB(deleteKeys); } } catch (MusicServiceException e) { - logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),ErrorSeverity.ERROR); + e.printStackTrace(); + } catch (MusicLockingException e) { + e.printStackTrace(); + } + + //Zookeeper cleanup + scheduler.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + Iterator<Entry<String, Long>> it = MusicUtil.zkNodeMap.entrySet().iterator(); + StringBuilder deleteKeys = new StringBuilder(); + Boolean expiredKeys = false; + while (it.hasNext()) { + Map.Entry<String, Long> pair = (Map.Entry<String, Long>)it.next(); + long ctime = pair.getValue(); + if (System.currentTimeMillis() >= ctime + 24 * 60 * 60 * 1000) { + try { + expiredKeys = true; + String id = pair.getKey(); + deleteKeys.append("'").append(id).append("'").append(","); + MusicCore.deleteLock(id.substring(1)); + MusicUtil.zkNodeMap.remove(id); + + } catch (MusicLockingException e) { + e.printStackTrace(); + } + } + } + if(expiredKeys) { + deleteKeys.deleteCharAt(deleteKeys.length()-1); + deleteKeysFromDB(deleteKeys); + } } + } , 0, 24, TimeUnit.HOURS); + } + + @Override + public void contextDestroyed(ServletContextEvent event) { + scheduler.shutdownNow(); + } + + public void deleteKeysFromDB(StringBuilder deleteKeys) { + PreparedQueryObject pQuery = new PreparedQueryObject(); + pQuery.appendQueryString( + "DELETE FROM admin.locks WHERE lock_id IN ("+deleteKeys+")"); + try { + MusicCore.nonKeyRelatedPut(pQuery, "eventual"); + } catch (Exception e) { + e.printStackTrace(); + } } } diff --git a/src/main/java/org/onap/music/main/MusicCore.java b/src/main/java/org/onap/music/main/MusicCore.java index 9f7b060b..6b32fc53 100644 --- a/src/main/java/org/onap/music/main/MusicCore.java +++ b/src/main/java/org/onap/music/main/MusicCore.java @@ -26,12 +26,13 @@ import java.io.StringWriter; import java.util.HashMap; import java.util.Map; import java.util.StringTokenizer; +import java.util.UUID; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.KeeperException.NoNodeException; -import org.onap.music.datastore.MusicDataStore; +import org.onap.music.datastore.CassaDataStore; +import org.onap.music.datastore.CassaLockStore; +import org.onap.music.datastore.CassaLockStore.LockObject; +import org.onap.music.datastore.MusicLockState; import org.onap.music.datastore.PreparedQueryObject; -import org.onap.music.datastore.jsonobjects.JsonKeySpace; import org.onap.music.eelf.logging.EELFLoggerDelegate; import org.onap.music.eelf.logging.format.AppMessages; import org.onap.music.eelf.logging.format.ErrorSeverity; @@ -39,9 +40,6 @@ import org.onap.music.eelf.logging.format.ErrorTypes; import org.onap.music.exceptions.MusicLockingException; import org.onap.music.exceptions.MusicQueryException; import org.onap.music.exceptions.MusicServiceException; -import org.onap.music.lockingservice.MusicLockState; -import org.onap.music.lockingservice.MusicLockState.LockStatus; -import org.onap.music.lockingservice.MusicLockingService; import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.ColumnDefinitions.Definition; @@ -50,6 +48,7 @@ import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; import com.datastax.driver.core.TableMetadata; + /** * This class ..... * @@ -57,10 +56,11 @@ import com.datastax.driver.core.TableMetadata; */ public class MusicCore { - public static MusicLockingService mLockHandle = null; - public static MusicDataStore mDstoreHandle = null; + public static CassaLockStore mLockHandle = null; + public static CassaDataStore mDstoreHandle = null; private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicCore.class); - + private static boolean unitTestRun=true; + public static class Condition { Map<String, Object> conditions; PreparedQueryObject selectQueryForTheRow; @@ -73,22 +73,19 @@ public class MusicCore { public boolean testCondition() throws Exception { // first generate the row ResultSet results = quorumGet(selectQueryForTheRow); - Row row = null; - if(results != null) { - row = results.one(); - } + Row row = results.one(); return getDSHandle().doesRowSatisfyCondition(row, conditions); } } - public static MusicLockingService getLockingServiceHandle() throws MusicLockingException { + public static CassaLockStore getLockingServiceHandle() throws MusicLockingException { logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle"); long start = System.currentTimeMillis(); if (mLockHandle == null) { try { - mLockHandle = new MusicLockingService(); + mLockHandle = new CassaLockStore(getDSHandle()); } catch (Exception e) { logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); throw new MusicLockingException("Failed to aquire Locl store handle " + e); @@ -104,11 +101,11 @@ public class MusicCore { * @param remoteIp * @return */ - public static MusicDataStore getDSHandle(String remoteIp) { + public static CassaDataStore getDSHandle(String remoteIp) { logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring data store handle"); long start = System.currentTimeMillis(); if (mDstoreHandle == null) { - mDstoreHandle = new MusicDataStore(remoteIp); + mDstoreHandle = new CassaDataStore(remoteIp); } long end = System.currentTimeMillis(); logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire data store handle:" + (end - start) + " ms"); @@ -120,15 +117,16 @@ public class MusicCore { * @return * @throws MusicServiceException */ - public static MusicDataStore getDSHandle() throws MusicServiceException { + public static CassaDataStore getDSHandle() throws MusicServiceException { + logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring data store handle"); long start = System.currentTimeMillis(); if (mDstoreHandle == null) { // Quick Fix - Best to put this into every call to getDSHandle? - if (! "localhost".equals(MusicUtil.getMyCassaHost())) { - mDstoreHandle = new MusicDataStore(MusicUtil.getMyCassaHost()); + if (! MusicUtil.getMyCassaHost().equals("localhost") ) { + mDstoreHandle = new CassaDataStore(MusicUtil.getMyCassaHost()); } else { - mDstoreHandle = new MusicDataStore(); + mDstoreHandle = new CassaDataStore(); } } if(mDstoreHandle.getSession() == null) { @@ -142,232 +140,160 @@ public class MusicCore { return mDstoreHandle; } - public static String createLockReference(String lockName) { + public static String createLockReference(String fullyQualifiedKey) { + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + String lockName = splitString[2]; + logger.info(EELFLoggerDelegate.applicationLogger,"Creating lock reference for lock name:" + lockName); long start = System.currentTimeMillis(); - String lockId = null; + String lockReference = null; try { - lockId = getLockingServiceHandle().createLockId("/" + lockName); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.CREATELOCK+lockName,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - - } + lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName); + } catch (MusicLockingException | MusicServiceException | MusicQueryException e) { + e.printStackTrace(); + } long end = System.currentTimeMillis(); logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to create lock reference:" + (end - start) + " ms"); - return lockId; - } - - /** - * - * @param key - * @return - */ - public static boolean isTableOrKeySpaceLock(String key) { - String[] splitString = key.split("\\."); - if (splitString.length > 2) - return false; - else - return true; + return lockReference; } - /** - * - * @param key - * @return - */ - public static MusicLockState getMusicLockState(String key) { - long start = System.currentTimeMillis(); - try { - String[] splitString = key.split("\\."); - String keyspaceName = splitString[0]; - String tableName = splitString[1]; - String primaryKey = splitString[2]; - MusicLockState mls; - String lockName = keyspaceName + "." + tableName + "." + primaryKey; - mls = getLockingServiceHandle().getLockState(lockName); - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to get lock state:" + (end - start) + " ms"); - return mls; - } catch (NullPointerException | MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - return null; - } - public static ReturnType acquireLockWithLease(String key, String lockId, long leasePeriod) { - try { - long start = System.currentTimeMillis(); - /* check if the current lock has exceeded its lease and if yes, release that lock */ - MusicLockState mls = getMusicLockState(key); - if (mls != null) { - if (mls.getLockStatus().equals(LockStatus.LOCKED)) { - logger.info(EELFLoggerDelegate.applicationLogger,"The current lock holder for " + key + " is " + mls.getLockHolder() - + ". Checking if it has exceeded lease"); - long currentLockPeriod = System.currentTimeMillis() - mls.getLeaseStartTime(); - long currentLeasePeriod = mls.getLeasePeriod(); - if (currentLockPeriod > currentLeasePeriod) { - logger.info(EELFLoggerDelegate.applicationLogger,"Lock period " + currentLockPeriod - + " has exceeded lease period " + currentLeasePeriod); - boolean voluntaryRelease = false; - String currentLockHolder = mls.getLockHolder(); - mls = releaseLock(currentLockHolder, voluntaryRelease); - } - } - } else - logger.error(EELFLoggerDelegate.errorLogger,key, AppMessages.INVALIDLOCK,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - - /* - * call the traditional acquire lock now and if the result returned is true, set the - * begin time-stamp and lease period - */ - if (acquireLock(key, lockId).getResult() == ResultType.SUCCESS) { - mls = getMusicLockState(key);// get latest state - if ( mls == null ) { - logger.info(EELFLoggerDelegate.applicationLogger,"Music Lock State is null"); - return new ReturnType(ResultType.FAILURE, "Could not acquire lock, Lock State is null"); - } - if (mls.getLeaseStartTime() == -1) {// set it again only if it is not set already - mls.setLeaseStartTime(System.currentTimeMillis()); - mls.setLeasePeriod(leasePeriod); - getLockingServiceHandle().setLockState(key, mls); - } - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire leased lock:" + (end - start) + " ms"); - return new ReturnType(ResultType.SUCCESS, "Accquired lock"); - } else { - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to fail to acquire leased lock:" + (end - start) + " ms"); - return new ReturnType(ResultType.FAILURE, "Could not acquire lock"); - } - } catch (Exception e) { - StringWriter sw = new StringWriter(); - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - - String exceptionAsString = sw.toString(); - return new ReturnType(ResultType.FAILURE, - "Exception thrown in acquireLockWithLease:\n" + exceptionAsString); - } + public static ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException { + evictExpiredLockHolder(fullyQualifiedKey,leasePeriod); + return acquireLock(fullyQualifiedKey, lockReference); } - public static ReturnType acquireLock(String key, String lockId) throws MusicLockingException { - /* - * first check if I am on top. Since ids are not reusable there is no need to check - * lockStatus If the status is unlocked, then the above call will automatically return - * false. - */ - Boolean result = false; - try { - result = getLockingServiceHandle().isMyTurn(lockId); - } catch (MusicLockingException e2) { - logger.error(EELFLoggerDelegate.errorLogger,AppMessages.INVALIDLOCK + lockId + " " + e2); - throw new MusicLockingException(); - } - if (!result) { - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Not your turn, someone else has the lock"); - try { - if (!getLockingServiceHandle().lockIdExists(lockId)) { - logger.info(EELFLoggerDelegate.applicationLogger, "In acquire lock: this lockId doesn't exist"); - return new ReturnType(ResultType.FAILURE, "Lockid doesn't exist"); - } - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - throw new MusicLockingException(); - } - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: returning failure"); - return new ReturnType(ResultType.FAILURE, "Not your turn, someone else has the lock"); - } - - - // this is for backward compatibility where locks could also be acquired on just - // keyspaces or tables. - if (isTableOrKeySpaceLock(key)) { - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: A table or keyspace lock so no need to perform sync...so returning true"); - return new ReturnType(ResultType.SUCCESS, "A table or keyspace lock so no need to perform sync...so returning true"); - } + private static void evictExpiredLockHolder(String fullyQualifiedKey, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException { - // read the lock name corresponding to the key and if the status is locked or being locked, - // then return false - MusicLockState currentMls = null; - MusicLockState newMls = null; - try { - currentMls = getMusicLockState(key); - String currentLockHolder = null; - if(currentMls != null) { currentLockHolder = currentMls.getLockHolder(); }; - if (lockId.equals(currentLockHolder)) { - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: You already have the lock!"); - return new ReturnType(ResultType.SUCCESS, "You already have the lock!"); - } - } catch (NullPointerException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.INVALIDLOCK+lockId,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - - // change status to "being locked". This state transition is necessary to ensure syncing - // before granting the lock - String lockHolder = null; - boolean needToSyncQuorum = false; - if (currentMls != null) - needToSyncQuorum = currentMls.isNeedToSyncQuorum(); + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + String primaryKeyValue = splitString[2]; + LockObject currentLockHolderObject = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue); + + /* Release the lock of the previous holder if it has expired. if the update to the acquire time has not reached due to network delays, simply use the create time as the + * reference*/ + + long referenceTime = Math.max(Long.parseLong(currentLockHolderObject.acquireTime), Long.parseLong(currentLockHolderObject.createTime)); + if((System.currentTimeMillis() - referenceTime) > leasePeriod) { + forciblyReleaseLock(fullyQualifiedKey, currentLockHolderObject.lockRef+""); + logger.info(EELFLoggerDelegate.applicationLogger, currentLockHolderObject.lockRef+" forcibly released"); + } + } + + private static ReturnType isTopOfLockStore(String keyspace, String table, String primaryKeyValue, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException { + + //return failure to lock holders too early or already evicted from the lock store + String topOfLockStoreS = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue).lockRef; + long topOfLockStoreL = Long.parseLong(topOfLockStoreS); + long lockReferenceL = Long.parseLong(lockReference); - newMls = new MusicLockState(MusicLockState.LockStatus.BEING_LOCKED, lockHolder, - needToSyncQuorum); - try { - getLockingServiceHandle().setLockState(key, newMls); - } catch (MusicLockingException e1) { - logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); + if(lockReferenceL > topOfLockStoreL) { + logger.info(EELFLoggerDelegate.applicationLogger, lockReference+" is not the lock holder yet"); + return new ReturnType(ResultType.FAILURE, lockReference+" is not the lock holder yet"); } - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to being_locked"); + - // do syncing if this was a forced lock release - if (needToSyncQuorum) { - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Since there was a forcible release, need to sync quorum!"); - try { - syncQuorum(key); - } catch (Exception e) { - logger.error(EELFLoggerDelegate.errorLogger,"Failed to set Lock state " + e); - } - } + if(lockReferenceL < topOfLockStoreL) { + logger.info(EELFLoggerDelegate.applicationLogger, lockReference+" is no longer/or was never in the lock store queue"); + return new ReturnType(ResultType.FAILURE, lockReference+" is no longer/or was never in the lock store queue"); + } + + return new ReturnType(ResultType.SUCCESS, lockReference+" is top of lock store"); + } + + public static ReturnType acquireLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException { + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + String primaryKeyValue = splitString[2]; - // change status to locked - lockHolder = lockId; - needToSyncQuorum = false; - newMls = new MusicLockState(MusicLockState.LockStatus.LOCKED, lockHolder, needToSyncQuorum); - try { - getLockingServiceHandle().setLockState(key, newMls); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKSTATE+key,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - logger.info(EELFLoggerDelegate.applicationLogger,"In acquire lock: Set lock state to locked and assigned current lock ref " - + lockId + " as holder"); + ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockReference); - return new ReturnType(result?ResultType.SUCCESS:ResultType.FAILURE, "Set lock state to locked and assigned a lock holder"); + if(result.getResult().equals(ResultType.FAILURE)) + return result;//not top of the lock store q + + //check to see if the value of the key has to be synced in case there was a forceful release + String syncTable = keyspace+".unsyncedKeys_"+table; + String query = "select * from "+syncTable+" where key='"+fullyQualifiedKey+"';"; + PreparedQueryObject readQueryObject = new PreparedQueryObject(); + readQueryObject.appendQueryString(query); + ResultSet results = getDSHandle().executeCriticalGet(readQueryObject); + if (results.all().size() != 0) { + logger.info("In acquire lock: Since there was a forcible release, need to sync quorum!"); + try { + syncQuorum(keyspace, table, primaryKeyValue); + } catch (Exception e) { + StringWriter sw = new StringWriter(); + logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR506E] Failed to aquire lock ",ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); + String exceptionAsString = sw.toString(); + return new ReturnType(ResultType.FAILURE, "Exception thrown while syncing key:\n" + exceptionAsString); + } + String cleanQuery = "delete * from music_internal.unsynced_keys where key='"+fullyQualifiedKey+"';"; + PreparedQueryObject deleteQueryObject = new PreparedQueryObject(); + deleteQueryObject.appendQueryString(cleanQuery); + getDSHandle().executePut(deleteQueryObject, "critical"); + } + + getLockingServiceHandle().updateLockAcquireTime(keyspace, table, primaryKeyValue, lockReference); + + return new ReturnType(ResultType.SUCCESS, lockReference+" is the lock holder for the key"); } /** * - * @param keyspaceName - * @param kspObject - * @return - * @throws Exception + * @param tableQueryObject + * @param consistency + * @return Boolean Indicates success or failure + * @throws MusicServiceException + * + * */ - public boolean createKeyspace(String keyspaceName, JsonKeySpace kspObject) throws Exception { - return true; - } - - - private static void syncQuorum(String key) throws Exception { + public static ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException { + boolean result = false; + + try { + //create shadow locking table + result = getLockingServiceHandle().createLockQueue(keyspace, table); + if(result == false) + return ResultType.FAILURE; + + result = false; + + //create table to track unsynced_keys + table = "unsyncedKeys_"+table; + + String tabQuery = "CREATE TABLE IF NOT EXISTS "+keyspace+"."+table + + " ( key text,PRIMARY KEY (key) );"; + System.out.println(tabQuery); + PreparedQueryObject queryObject = new PreparedQueryObject(); + + queryObject.appendQueryString(tabQuery); + result = false; + result = getDSHandle().executePut(queryObject, "eventual"); + + + //create actual table + result = getDSHandle().executePut(tableQueryObject, consistency); + } catch (MusicQueryException | MusicServiceException | MusicLockingException ex) { + logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR); + throw new MusicServiceException(ex.getMessage()); + } + return result?ResultType.SUCCESS:ResultType.FAILURE; + } + + private static void syncQuorum(String keyspace, String table, String primaryKeyValue) throws Exception { logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---"); - String[] splitString = key.split("\\."); - String keyspaceName = splitString[0]; - String tableName = splitString[1]; - String primaryKeyValue = splitString[2]; PreparedQueryObject selectQuery = new PreparedQueryObject(); PreparedQueryObject updateQuery = new PreparedQueryObject(); // get the primary key d - TableMetadata tableInfo = returnColumnMetadata(keyspaceName, tableName); + TableMetadata tableInfo = returnColumnMetadata(keyspace, table); String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();// we only support single // primary key DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType(); @@ -375,7 +301,7 @@ public class MusicCore { MusicUtil.convertToActualDataType(primaryKeyType, primaryKeyValue); // get the row of data from a quorum - selectQuery.appendQueryString("SELECT * FROM " + keyspaceName + "." + tableName + " WHERE " + selectQuery.appendQueryString("SELECT * FROM " + keyspace + "." + table + " WHERE " + primaryKeyName + "= ?" + ";"); selectQuery.addValue(cqlFormattedPrimaryKeyValue); ResultSet results = null; @@ -400,7 +326,7 @@ public class MusicCore { fieldValueString.append(","); counter = counter + 1; } - updateQuery.appendQueryString("UPDATE " + keyspaceName + "." + tableName + " SET " + updateQuery.appendQueryString("UPDATE " + keyspace + "." + table + " SET " + fieldValueString + " WHERE " + primaryKeyName + "= ? " + ";"); updateQuery.addValue(cqlFormattedPrimaryKeyValue); @@ -442,81 +368,68 @@ public class MusicCore { /** * - * @param lockName + * @param fullyQualifiedKey lockName * @return */ - public static String whoseTurnIsIt(String lockName) { - + public static String whoseTurnIsIt(String fullyQualifiedKey) { + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + String primaryKeyValue = splitString[2]; try { - return getLockingServiceHandle().whoseTurnIsIt("/" + lockName) + ""; - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+lockName ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); + return getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue)+""; + } catch (MusicLockingException | MusicServiceException | MusicQueryException e) { + logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); } return null; - - } /** * - * @param lockId + * @param lockReference * @return */ - public static String getLockNameFromId(String lockId) { - StringTokenizer st = new StringTokenizer(lockId); + public static String getLockNameFromId(String lockReference) { + StringTokenizer st = new StringTokenizer(lockReference); return st.nextToken("$"); } - public static void destroyLockRef(String lockId) { + public static MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) { long start = System.currentTimeMillis(); + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + String primaryKeyValue = splitString[2]; try { - getLockingServiceHandle().unlockAndDeleteId(lockId); - } catch (MusicLockingException | NoNodeException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockId ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); + getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockReference); + } catch (MusicLockingException | MusicServiceException | MusicQueryException e) { + logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockReference ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); } long end = System.currentTimeMillis(); logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms"); + return getMusicLockState(fullyQualifiedKey); } - public static MusicLockState releaseLock(String lockId, boolean voluntaryRelease) { - long start = System.currentTimeMillis(); - try { - getLockingServiceHandle().unlockAndDeleteId(lockId); - } catch (MusicLockingException e1) { - logger.error(EELFLoggerDelegate.errorLogger,e1.getMessage(), AppMessages.RELEASELOCK+lockId ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } catch (KeeperException.NoNodeException nne) { - logger.error(EELFLoggerDelegate.errorLogger,"Failed to release Lock " + lockId + " " + nne); - MusicLockState mls = new MusicLockState("Lock doesn't exists. Release lock operation failed."); - return mls; - } - String lockName = getLockNameFromId(lockId); - MusicLockState mls; - String lockHolder = null; - if (voluntaryRelease) { - mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder); - logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock voluntarily released for " + lockId); - } else { - boolean needToSyncQuorum = true; - mls = new MusicLockState(MusicLockState.LockStatus.UNLOCKED, lockHolder, - needToSyncQuorum); - logger.info(EELFLoggerDelegate.applicationLogger,"In unlock: lock forcibly released for " + lockId); - } - try { - getLockingServiceHandle().setLockState(lockName, mls); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.RELEASELOCK+lockId ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to release lock:" + (end - start) + " ms"); - return mls; - } - - public static void voluntaryReleaseLock(String lockId) throws MusicLockingException{ - try { - getLockingServiceHandle().unlockAndDeleteId(lockId); - } catch (KeeperException.NoNodeException e) { - // ??? No way - } + public static MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException{ + return destroyLockRef(fullyQualifiedKey, lockReference); + } + + public static MusicLockState forciblyReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicServiceException, MusicQueryException{ + String[] splitString = fullyQualifiedKey.split("\\."); + String keyspace = splitString[0]; + String table = splitString[1]; + + //leave a signal that this key could potentially be unsynchronized + String syncTable = keyspace+".unsyncedKeys_"+table; + PreparedQueryObject queryObject = new PreparedQueryObject(); + String values = "(?)"; + queryObject.addValue(fullyQualifiedKey); + String insQuery = "insert into "+syncTable+" (key) values "+values+"';"; + queryObject.appendQueryString(insQuery); + getDSHandle().executePut(queryObject, "critical"); + + //now release the lock + return destroyLockRef(fullyQualifiedKey, lockReference); } /** @@ -525,17 +438,8 @@ public class MusicCore { * @throws MusicLockingException */ public static void deleteLock(String lockName) throws MusicLockingException { - long start = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Deleting lock for " + lockName); - try { - getLockingServiceHandle().deleteLock("/" + lockName); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DELTELOCK+lockName ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - throw new MusicLockingException(e.getMessage()); - } - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to delete lock:" + (end - start) + " ms"); - } + //deprecated + } @@ -551,63 +455,12 @@ public class MusicCore { } - /** - * - * @param nodeName - */ - public static void pureZkCreate(String nodeName) { - try { - getLockingServiceHandle().getzkLockHandle().createNode(nodeName); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle " ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - } - - /** - * - * @param nodeName - * @param data - */ - public static void pureZkWrite(String nodeName, byte[] data) { - long start = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Performing zookeeper write to " + nodeName); - try { - getLockingServiceHandle().getzkLockHandle().setNodeData(nodeName, data); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle " ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - logger.info(EELFLoggerDelegate.applicationLogger,"Performed zookeeper write to " + nodeName); - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms"); - } - - /** - * - * @param nodeName - * @return - */ - public static byte[] pureZkRead(String nodeName) { - long start = System.currentTimeMillis(); - byte[] data = null; - try { - data = getLockingServiceHandle().getzkLockHandle().getNodeData(nodeName); - } catch (MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), "[ERR512E] Failed to get ZK Lock Handle " ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR); - } - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the actual zk put:" + (end - start) + " ms"); - return data; - } - // Prepared Query Additions. /** * - * @param keyspaceName - * @param tableName - * @param primaryKey * @param queryObject * @return ReturnType * @throws MusicServiceException @@ -627,52 +480,51 @@ public class MusicCore { return new ReturnType(ResultType.FAILURE, "Failure"); } } - + /** * - * @param keyspaceName - * @param tableName - * @param primaryKey + * @param keyspace + * @param table + * @param primaryKeyValue * @param queryObject - * @param lockId + * @param lockReference * @return */ - public static ReturnType criticalPut(String keyspaceName, String tableName, String primaryKey, - PreparedQueryObject queryObject, String lockId, Condition conditionInfo) { + public static ReturnType criticalPut(String keyspace, String table, String primaryKeyValue, + PreparedQueryObject queryObject, String lockReference, Condition conditionInfo) { long start = System.currentTimeMillis(); - try { - MusicLockState mls = getLockingServiceHandle() - .getLockState(keyspaceName + "." + tableName + "." + primaryKey); - if (mls.getLockHolder().equals(lockId) == true) { - if (conditionInfo != null) - try { - if (conditionInfo.testCondition() == false) - return new ReturnType(ResultType.FAILURE, - "Lock acquired but the condition is not true"); - } catch (Exception e) { - return new ReturnType(ResultType.FAILURE, - "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n" - + e.getMessage()); - } - getDSHandle().executePut(queryObject, MusicUtil.CRITICAL); - long end = System.currentTimeMillis(); - logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms"); - return new ReturnType(ResultType.SUCCESS, "Update performed"); - } else - return new ReturnType(ResultType.FAILURE, - "Cannot perform operation since you are the not the lock holder"); - } catch (MusicQueryException | MusicServiceException e) { + ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockReference); + if(result.getResult().equals(ResultType.FAILURE)) + return result;//not top of the lock store q + + if (conditionInfo != null) + try { + if (conditionInfo.testCondition() == false) + return new ReturnType(ResultType.FAILURE, + "Lock acquired but the condition is not true"); + } catch (Exception e) { + return new ReturnType(ResultType.FAILURE, + "Exception thrown while checking the condition, check its sanctity:\n" + + e.getMessage()); + } + + String query = queryObject.getQuery(); + query = query.replaceFirst("SET", "using TIMESTAMP "+ v2sTimeStampInMicroseconds(lockReference, System.currentTimeMillis())+ " SET"); + queryObject.replaceQueryString(query); + getDSHandle().executePut(queryObject, MusicUtil.CRITICAL); + long end = System.currentTimeMillis(); + logger.info(EELFLoggerDelegate.applicationLogger,"Time taken for the critical put:" + (end - start) + " ms"); + }catch (MusicQueryException | MusicServiceException | MusicLockingException e) { logger.error(EELFLoggerDelegate.errorLogger,e.getMessage()); return new ReturnType(ResultType.FAILURE, - "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n" + "Exception thrown while doing the critical put\n" + e.getMessage()); - }catch(MusicLockingException ex){ - return new ReturnType(ResultType.FAILURE,ex.getMessage()); } - + return new ReturnType(ResultType.SUCCESS, "Update performed"); } + /** * * @param queryObject @@ -689,10 +541,11 @@ public class MusicCore { try { result = getDSHandle().executePut(queryObject, consistency); } catch (MusicQueryException | MusicServiceException ex) { - logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR); + logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR, + ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR); throw new MusicServiceException(ex.getMessage()); } - return result?ResultType.SUCCESS:ResultType.FAILURE; + return result ? ResultType.SUCCESS : ResultType.FAILURE; } /** @@ -712,46 +565,29 @@ public class MusicCore { } return results; } - - public static String getMyHostId() { - PreparedQueryObject pQuery = new PreparedQueryObject(); - pQuery.appendQueryString("SELECT HOST_ID FROM SYSTEM.LOCAL"); - ResultSet rs = null; - try { - rs = getDSHandle().executeEventualGet(pQuery); - Row row = rs.one(); - return (row == null) ? "UNKNOWN" : row.getUUID("HOST_ID").toString(); - } catch (Exception e) { - e.printStackTrace(); - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage()); - } - logger.error(EELFLoggerDelegate.errorLogger, "Some issue during MusicCore.getMyHostId"); - return "UNKNOW"; - } /** * This method performs DDL operations on cassandra, if the the resource is available. Lock ID * is used to check if the resource is free. * - * @param keyspaceName name of the keyspace - * @param tableName name of the table - * @param primaryKey primary key value + * @param keyspace name of the keyspace + * @param table name of the table + * @param primaryKeyValue primary key value * @param queryObject query object containing prepared query and values - * @param lockId lock ID to check if the resource is free to perform the operation. + * @param lockReference lock ID to check if the resource is free to perform the operation. * @return ResultSet */ - public static ResultSet criticalGet(String keyspaceName, String tableName, String primaryKey, - PreparedQueryObject queryObject, String lockId) throws MusicServiceException { + public static ResultSet criticalGet(String keyspace, String table, String primaryKeyValue, + PreparedQueryObject queryObject, String lockReference) throws MusicServiceException { ResultSet results = null; + try { - MusicLockState mls = getLockingServiceHandle() - .getLockState(keyspaceName + "." + tableName + "." + primaryKey); - if (mls.getLockHolder().equals(lockId)) { + ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockReference); + if(result.getResult().equals(ResultType.FAILURE)) + return null;//not top of the lock store q results = getDSHandle().executeCriticalGet(queryObject); - } else - throw new MusicServiceException("YOU DO NOT HAVE THE LOCK"); } catch (MusicQueryException | MusicServiceException | MusicLockingException e) { - logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR); + logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR); } return results; } @@ -765,22 +601,23 @@ public class MusicCore { * @param queryObject query object containing prepared query and values * @return ReturnType * @throws MusicLockingException + * @throws MusicServiceException + * @throws MusicQueryException */ public static ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey, - PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException { - + PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException { long start = System.currentTimeMillis(); - String key = keyspaceName + "." + tableName + "." + primaryKey; - String lockId = createLockReference(key); + String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey; + String lockReference = createLockReference(fullyQualifiedKey); long lockCreationTime = System.currentTimeMillis(); - ReturnType lockAcqResult = acquireLock(key, lockId); + ReturnType lockAcqResult = acquireLock(fullyQualifiedKey, lockReference); long lockAcqTime = System.currentTimeMillis(); if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) { - logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId); + logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockReference); ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey, - queryObject, lockId, conditionInfo); + queryObject, lockReference, conditionInfo); long criticalPutTime = System.currentTimeMillis(); - voluntaryReleaseLock(lockId); + voluntaryReleaseLock(fullyQualifiedKey,lockReference); long lockDeleteTime = System.currentTimeMillis(); String timingInfo = "|lock creation time:" + (lockCreationTime - start) + "|lock accquire time:" + (lockAcqTime - lockCreationTime) @@ -789,49 +626,8 @@ public class MusicCore { criticalPutResult.setTimingInfo(timingInfo); return criticalPutResult; } else { - logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId); - destroyLockRef(lockId); - return lockAcqResult; - } - } - - /** - * this function is mainly for the benchmarks to see the effect of lock deletion. - * - * @param keyspaceName - * @param tableName - * @param primaryKey - * @param queryObject - * @param conditionInfo - * @return - * @throws MusicLockingException - */ - public static ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName, - String primaryKey, PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException { - - long start = System.currentTimeMillis(); - String key = keyspaceName + "." + tableName + "." + primaryKey; - String lockId = createLockReference(key); - long lockCreationTime = System.currentTimeMillis(); - long leasePeriod = MusicUtil.getDefaultLockLeasePeriod(); - ReturnType lockAcqResult = acquireLock(key, lockId); - long lockAcqTime = System.currentTimeMillis(); - if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) { - logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId); - ReturnType criticalPutResult = criticalPut(keyspaceName, tableName, primaryKey, - queryObject, lockId, conditionInfo); - long criticalPutTime = System.currentTimeMillis(); - deleteLock(key); - long lockDeleteTime = System.currentTimeMillis(); - String timingInfo = "|lock creation time:" + (lockCreationTime - start) - + "|lock accquire time:" + (lockAcqTime - lockCreationTime) - + "|critical put time:" + (criticalPutTime - lockAcqTime) - + "|lock delete time:" + (lockDeleteTime - criticalPutTime) + "|"; - criticalPutResult.setTimingInfo(timingInfo); - return criticalPutResult; - } else { - logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId); - deleteLock(key); + logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockReference); + voluntaryReleaseLock(fullyQualifiedKey,lockReference); return lockAcqResult; } } @@ -849,47 +645,31 @@ public class MusicCore { * @return ResultSet * @throws MusicServiceException * @throws MusicLockingException + * @throws MusicQueryException */ public static ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey, - PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException { - String key = keyspaceName + "." + tableName + "." + primaryKey; - String lockId = createLockReference(key); + PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException { + String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey; + String lockReference = createLockReference(fullyQualifiedKey); long leasePeriod = MusicUtil.getDefaultLockLeasePeriod(); - ReturnType lockAcqResult = acquireLock(key, lockId); + ReturnType lockAcqResult = acquireLock(fullyQualifiedKey, lockReference); if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) { - logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockId); + logger.info(EELFLoggerDelegate.applicationLogger,"acquired lock with id " + lockReference); ResultSet result = - criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId); - voluntaryReleaseLock(lockId); + criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockReference); + voluntaryReleaseLock(fullyQualifiedKey,lockReference); return result; } else { - destroyLockRef(lockId); - logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockId); + voluntaryReleaseLock(fullyQualifiedKey,lockReference); + logger.info(EELFLoggerDelegate.applicationLogger,"unable to acquire lock, id " + lockReference); return null; } } - public static ResultSet atomicGetWithDeleteLock(String keyspaceName, String tableName, String primaryKey, - PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException { - String key = keyspaceName + "." + tableName + "." + primaryKey; - String lockId = createLockReference(key); - long leasePeriod = MusicUtil.getDefaultLockLeasePeriod(); - - ReturnType lockAcqResult = acquireLock(key, lockId); - - if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) { - logger.info(EELFLoggerDelegate.applicationLogger, "acquired lock with id " + lockId); - ResultSet result = criticalGet(keyspaceName, tableName, primaryKey, queryObject, lockId); - deleteLock(key); - return result; - } else { - deleteLock(key); - logger.info(EELFLoggerDelegate.applicationLogger, "unable to acquire lock, id " + lockId); - return null; - } - } - + public static MusicLockState getMusicLockState(String fullyQualifiedKey) { + return null; + } /** * authenticate user logic @@ -903,7 +683,7 @@ public class MusicCore { * @return * @throws Exception */ - public static Map<String, Object> autheticateUser(String nameSpace, String userId, + public static Map<String, Object> authenticate(String nameSpace, String userId, String password, String keyspace, String aid, String operation) throws Exception { Map<String, Object> resultMap = new HashMap<>(); @@ -999,4 +779,38 @@ public class MusicCore { resultMap.put("keyspace",keyspace); return resultMap; } + + + /** + * Given the time of write for an update in a critical section, this method provides a transformed timestamp + * that ensures that a previous lock holder who is still alive can never corrupt a later critical section. + * The main idea is to us the lock reference to clearly demarcate the timestamps across critical sections. + * @param the UUID lock reference associated with the write. + * @param the long timeOfWrite which is the actual time at which the write took place + * @throws MusicServiceException + * @throws MusicQueryException + */ + private static long v2sTimeStampInMicroseconds(String lockReference, long timeOfWrite) throws MusicServiceException, MusicQueryException{ + long lockEpochMillis = Long.parseLong(lockReference); + + long lockEternityMillis = lockEpochMillis - MusicUtil.MusicEternityEpochMillis; + + long ts = lockEternityMillis * MusicUtil.MaxCriticalSectionDurationMillis + + (timeOfWrite - lockEpochMillis); + + return ts; + +// long test = (lockReferenceUUID.timestamp()-MusicUtil.MusicEternityEpochMillis); +// long timeStamp = (lockReferenceUUID.timestamp()-MusicUtil.MusicEternityEpochMillis)*MusicUtil.MaxCriticalSectionDurationMillis +// +timeOfWrite; +// return timeStamp; + +// return timeOfWrite*1000; + } + + public static void main(String[] args) { + String x = "axe top"; + x = x.replaceFirst("top", "sword"); + System.out.print(x); //returns sword pickaxe + } } diff --git a/src/main/java/org/onap/music/main/MusicUtil.java b/src/main/java/org/onap/music/main/MusicUtil.java index cfad845d..c11b4c7e 100755 --- a/src/main/java/org/onap/music/main/MusicUtil.java +++ b/src/main/java/org/onap/music/main/MusicUtil.java @@ -23,15 +23,12 @@ package org.onap.music.main; import java.io.File; import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; import java.math.BigInteger; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; import java.util.Scanner; import java.util.StringTokenizer; import java.util.UUID; @@ -72,8 +69,7 @@ public class MusicUtil { public static final String UPSERT = "upsert"; public static final String USERID = "userId"; public static final String PASSWORD = "password"; - - public static final String AUTHORIZATION = "Authorization"; + public static final String AUTHORIZATION = "Authorization"; private static final String LOCALHOST = "localhost"; private static final String PROPERTIES_FILE = "/opt/app/music/etc/music.properties"; @@ -85,10 +81,6 @@ public class MusicUtil { private static String myZkHost = LOCALHOST; private static String myCassaHost = LOCALHOST; private static String defaultMusicIp = LOCALHOST; - private static int cassandraPort = 9042; - private static int notifytimeout = 30000; - private static int notifyinterval = 5000; - private static boolean debug = true; private static String version = "2.3.0"; private static String musicRestIp = LOCALHOST; @@ -96,31 +88,22 @@ public class MusicUtil { private static long defaultLockLeasePeriod = 6000; private static final String[] propKeys = new String[] { "zookeeper.host", "cassandra.host", "music.ip", "debug", "version", "music.rest.ip", "music.properties", "lock.lease.period", "id", "all.ids", "public.ip", - "all.pubic.ips", "cassandra.user", "cassandra.password", "aaf.endpoint.url","cassandra.port", "notify.timeout", "notify.interval" }; + "all.pubic.ips", "cassandra.user", "cassandra.password", "aaf.endpoint.url" }; private static String cassName = "cassandra"; private static String cassPwd; private static String aafEndpointUrl = null; - public static final ConcurrentMap<String, Long> zkNodeMap = new ConcurrentHashMap<>(); + public static ConcurrentMap<String, Long> zkNodeMap = new ConcurrentHashMap<>(); + + public static final long MusicEternityEpochMillis = 1533081600000L; // Wednesday, August 1, 2018 12:00:00 AM + + public static final long MaxCriticalSectionDurationMillis = 1L * 24 * 60 * 60 * 1000; // 1 day private MusicUtil() { throw new IllegalStateException("Utility Class"); } - /** - * - * @return cassandra port - */ - public static int getCassandraPort() { - return cassandraPort; - } - - /** - * set cassandra port - * @param cassandraPort - */ - public static void setCassandraPort(int cassandraPort) { - MusicUtil.cassandraPort = cassandraPort; - } + + /** * @return the cassName */ @@ -583,51 +566,5 @@ public class MusicUtil { return authValues; } - - public static void loadProperties() throws Exception { - Properties prop = new Properties(); - InputStream input = null; - try { - // load the properties file - input = MusicUtil.class.getClassLoader().getResourceAsStream("music.properties"); - prop.load(input); - } catch (Exception ex) { - logger.error(EELFLoggerDelegate.errorLogger, "Unable to find properties file."); - throw new Exception(); - } finally { - if (input != null) { - try { - input.close(); - } catch (IOException e) { - logger.error(EELFLoggerDelegate.applicationLogger,"Load properties failed "+e.getMessage(),e); - } - } - } - // get the property value and return it - MusicUtil.setMyCassaHost(prop.getProperty("cassandra.host")); - String zkHosts = prop.getProperty("zookeeper.host"); - MusicUtil.setMyZkHost(zkHosts); - MusicUtil.setCassName(prop.getProperty("cassandra.user")); - MusicUtil.setCassPwd(prop.getProperty("cassandra.password")); - MusicUtil.setCassandraPort(Integer.parseInt(prop.getProperty("cassandra.port"))); - MusicUtil.setNotifyTimeOut(Integer.parseInt(prop.getProperty("notify.timeout"))); - MusicUtil.setNotifyInterval(Integer.parseInt(prop.getProperty("notify.interval"))); - - } - - public static void setNotifyInterval(int notifyinterval) { - MusicUtil.notifyinterval = notifyinterval; - } - public static void setNotifyTimeOut(int notifytimeout) { - MusicUtil.notifytimeout = notifytimeout; - } - - public static int getNotifyInterval() { - return MusicUtil.notifyinterval; - } - - public static int getNotifyTimeout() { - return MusicUtil.notifytimeout; - - } + } diff --git a/src/main/java/org/onap/music/main/PropertiesListener.java b/src/main/java/org/onap/music/main/PropertiesListener.java index 0ed18be2..8b00e473 100755 --- a/src/main/java/org/onap/music/main/PropertiesListener.java +++ b/src/main/java/org/onap/music/main/PropertiesListener.java @@ -119,15 +119,6 @@ public class PropertiesListener implements ServletContextListener { case "aaf.endpoint.url": MusicUtil.setAafEndpointUrl(prop.getProperty(key)); break; - case "cassandra.port": - MusicUtil.setCassandraPort(Integer.parseInt(prop.getProperty(key))); - break; - case "notify.interval": - MusicUtil.setNotifyInterval(Integer.parseInt(prop.getProperty(key))); - break; - case "notify.timeout": - MusicUtil.setNotifyTimeOut(Integer.parseInt(prop.getProperty(key))); - break; default: logger.error(EELFLoggerDelegate.errorLogger, "No case found for " + key); diff --git a/src/main/java/org/onap/music/main/VotingAppJar.java b/src/main/java/org/onap/music/main/VotingAppJar.java new file mode 100644 index 00000000..1c475639 --- /dev/null +++ b/src/main/java/org/onap/music/main/VotingAppJar.java @@ -0,0 +1,115 @@ +package org.onap.music.main; + +import java.util.HashMap; +import java.util.Map; + +import org.onap.music.datastore.PreparedQueryObject; +import org.onap.music.exceptions.MusicLockingException; +import org.onap.music.exceptions.MusicQueryException; +import org.onap.music.exceptions.MusicServiceException; +import org.onap.music.main.MusicCore; + +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Row; + +/** + * + */ +public class VotingAppJar +{ + String keyspaceName; + String tableName; + + public VotingAppJar() throws MusicServiceException { + keyspaceName = "VotingAppForMusic"; + tableName = "votevount"; + + createVotingKeyspace(); + System.out.println("Created keyspaces"); + createVotingTable(); + System.out.println("Created tables"); + + createEntryForCandidate("Popeye"); + createEntryForCandidate("Judy"); + createEntryForCandidate("Flash"); + createEntryForCandidate("Mickey"); + System.out.println("Created candidates"); + } + + private void createVotingKeyspace() throws MusicServiceException { + + Map<String,Object> replicationInfo = new HashMap<String, Object>(); + replicationInfo.put("'class'", "'SimpleStrategy'"); + replicationInfo.put("'replication_factor'", 1); + + PreparedQueryObject queryObject = new PreparedQueryObject(); + queryObject.appendQueryString( + "CREATE KEYSPACE " + keyspaceName + " WITH REPLICATION = " + replicationInfo.toString().replaceAll("=", ":")); + + try { + MusicCore.nonKeyRelatedPut(queryObject, "eventual"); + } catch (MusicServiceException e) { + if (e.getMessage().equals("Keyspace votingappformusic already exists")) { + // ignore + } else { + throw(e); + } + } + } + + private void createVotingTable() throws MusicServiceException { + PreparedQueryObject queryObject = new PreparedQueryObject(); + queryObject.appendQueryString( + "CREATE TABLE " + keyspaceName + "." + tableName + " (name text PRIMARY KEY, count varint);"); + + try { + MusicCore.createTable(keyspaceName, tableName, queryObject, "eventual"); + } catch (MusicServiceException e) { + if (e.getMessage().equals("Table votingappformusic.votevount already exists")) { + //ignore + } else { + throw(e); + } + } + } + + private void createEntryForCandidate(String candidateName) throws MusicServiceException { + PreparedQueryObject queryObject = new PreparedQueryObject(); + queryObject.appendQueryString( + "INSERT INTO " + keyspaceName + "." + tableName + " (name, count) " + + "VALUES ('"+candidateName+"', 0);"); + + MusicCore.nonKeyRelatedPut(queryObject, "eventual"); + } + + public void vote() throws MusicLockingException, MusicQueryException, MusicServiceException { + updateVoteCount("Popeye",5); + updateVoteCount("Judy",7); + updateVoteCount("Mickey",8); + updateVoteCount("Flash",2); + } + + private void updateVoteCount(String candidateName, int numVotes) throws MusicLockingException, MusicQueryException, MusicServiceException { + PreparedQueryObject queryObject = new PreparedQueryObject(); + queryObject.appendQueryString( + "INSERT INTO " + keyspaceName + "." + tableName + " (name, count) " + + "VALUES ('"+candidateName+"', "+numVotes+");"); + MusicCore.atomicPut(keyspaceName, tableName, candidateName, queryObject, null); + } + + private void readAllVotes() throws MusicServiceException { + PreparedQueryObject queryObject = new PreparedQueryObject(); + queryObject.appendQueryString("SELECT * FROM " + keyspaceName + "." + tableName); + ResultSet rs = MusicCore.get(queryObject); + for(Row candidate : rs.all()) { + System.out.println(candidate.getString("name") + " - " + candidate.getVarint("count")); + } + } + + public static void main( String[] args ) throws Exception { + VotingAppJar vHandle = new VotingAppJar(); + vHandle.vote(); + vHandle.readAllVotes(); + } + +} |