* org.onap.music
* ===================================================================
* Copyright (c) 2017 AT&T Intellectual Property
+ * Modifications Copyright (c) 2018 IBM.
* ===================================================================
+ * Modifications Copyright (c) 2019 Samsung
+ * ===================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
package org.onap.music.service.impl;
import java.io.StringWriter;
- import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
import org.onap.music.main.ReturnType;
import org.onap.music.service.MusicCoreService;
- import com.datastax.driver.core.ColumnDefinitions;
- import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.TableMetadata;
- import com.google.common.util.concurrent.Monitor.Guard;
import org.onap.music.datastore.*;
private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicCassaCore.class);
private static boolean unitTestRun=true;
private static MusicCassaCore musicCassaCoreInstance = null;
-
+
private MusicCassaCore() {
-
+
}
public static MusicCassaCore getInstance() {
-
+
if(musicCassaCoreInstance == null) {
musicCassaCoreInstance = new MusicCassaCore();
}
return musicCassaCoreInstance;
}
-
+
public static CassaLockStore getLockingServiceHandle() throws MusicLockingException {
logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
long start = System.currentTimeMillis();
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
+
+ /* 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);
logger.info(EELFLoggerDelegate.applicationLogger, lockReference+" is not the lock holder yet");
return new ReturnType(ResultType.FAILURE, lockReference+" is not the lock holder yet");
}
-
+
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 ReturnType acquireLock(String fullyQualifiedKey, String lockId)
throws MusicLockingException, MusicQueryException, MusicServiceException {
String[] splitString = lockId.split("\\.");
String keyspace = splitString[0].substring(1);//remove '$'
String table = splitString[1];
String primaryKeyValue = splitString[2].substring(0, splitString[2].lastIndexOf("$"));
- fullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
+ String localFullyQualifiedKey = lockId.substring(1, lockId.lastIndexOf("$"));
String lockRef = lockId.substring(lockId.lastIndexOf("$")+1); //lockRef is "$" to end
-
+
ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockRef);
-
+
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+"';";
+ String query = "select * from "+syncTable+" where key='"+localFullyQualifiedKey+"';";
PreparedQueryObject readQueryObject = new PreparedQueryObject();
readQueryObject.appendQueryString(query);
- ResultSet results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(readQueryObject);
+ ResultSet results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(readQueryObject);
if (results.all().size() != 0) {
logger.info("In acquire lock: Since there was a forcible release, need to sync quorum!");
try {
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);
+ return new ReturnType(ResultType.FAILURE, "Exception thrown while syncing key:\n" + exceptionAsString);
}
- String cleanQuery = "delete from music_internal.unsynced_keys where key='"+fullyQualifiedKey+"';";
+ String cleanQuery = "delete from music_internal.unsynced_keys where key='"+localFullyQualifiedKey+"';";
PreparedQueryObject deleteQueryObject = new PreparedQueryObject();
deleteQueryObject.appendQueryString(cleanQuery);
MusicDataStoreHandle.getDSHandle().executePut(deleteQueryObject, "critical");
}
-
+
getLockingServiceHandle().updateLockAcquireTime(keyspace, table, primaryKeyValue, lockRef);
-
+
return new ReturnType(ResultType.SUCCESS, lockRef+" is the lock holder for the key");
}
/**
- *
+ *
* @param tableQueryObject
* @param consistency
* @return Boolean Indicates success or failure
- * @throws MusicServiceException
- *
- *
+ * @throws MusicServiceException
+ *
+ *
*/
public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
boolean result = false;
-
+
try {
- //create shadow locking table
+ //create shadow locking table
result = getLockingServiceHandle().createLockQueue(keyspace, table);
- if(result == false)
+ if(result == false)
return ResultType.FAILURE;
-
+
result = false;
-
+
//create table to track unsynced_keys
- table = "unsyncedKeys_"+table;
-
+ 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();
-
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+
queryObject.appendQueryString(tabQuery);
result = false;
result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
-
+
//create actual table
result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
} catch (MusicQueryException | MusicServiceException | MusicLockingException ex) {
selectQuery.appendQueryString("SELECT * FROM " + keyspace + "." + table + " WHERE "
+ primaryKeyName + "= ?" + ";");
selectQuery.addValue(cqlFormattedPrimaryKeyValue);
- ResultSet results = null;
- try {
- results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(selectQuery);
- // write it back to a quorum
- Row row = results.one();
- ColumnDefinitions colInfo = row.getColumnDefinitions();
- int totalColumns = colInfo.size();
- int counter = 1;
- StringBuilder fieldValueString = new StringBuilder("");
- for (Definition definition : colInfo) {
- String colName = definition.getName();
- if (colName.equals(primaryKeyName))
- continue;
- DataType colType = definition.getType();
- Object valueObj = MusicDataStoreHandle.getDSHandle().getColValue(row, colName, colType);
- Object valueString = MusicUtil.convertToActualDataType(colType, valueObj);
- fieldValueString.append(colName + " = ?");
- updateQuery.addValue(valueString);
- if (counter != (totalColumns - 1))
- fieldValueString.append(",");
- counter = counter + 1;
- }
- updateQuery.appendQueryString("UPDATE " + keyspace + "." + table + " SET "
- + fieldValueString + " WHERE " + primaryKeyName + "= ? " + ";");
- updateQuery.addValue(cqlFormattedPrimaryKeyValue);
+ MusicUtil.writeBackToQuorum(selectQuery, primaryKeyName, updateQuery, keyspace, table,
+ cqlFormattedPrimaryKeyValue);
- MusicDataStoreHandle.getDSHandle().executePut(updateQuery, "critical");
- } catch (MusicServiceException | MusicQueryException e) {
- logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.QUERYERROR +""+updateQuery ,ErrorSeverity.MAJOR, ErrorTypes.QUERYERROR);
- }
}
-
-
-
/**
- *
+ *
* @param query
* @return ResultSet
*/
results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
} catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR);
-
+
}
return results;
}
-
-
/**
- *
+ *
* @param fullyQualifiedKey lockName
* @return
*/
String table = splitString[1];
String primaryKeyValue = splitString[2];
try {
- return "$" + fullyQualifiedKey + "$"
+ return "$" + fullyQualifiedKey + "$"
+ getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue).lockRef;
} catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
}
/**
- *
+ *
* @param lockReference
* @return
*/
getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockRef);
} catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.DESTROYLOCK+lockRef ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
- }
+ }
long end = System.currentTimeMillis();
logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to destroy lock reference:" + (end - start) + " ms");
}
-
+
public MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) {
long start = System.currentTimeMillis();
String[] splitString = fullyQualifiedKey.split("\\.");
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 new MusicLockState(LockStatus.UNLOCKED, "");
return forciblyReleaseLock(fullyQualifiedKey, lockRef);
}
}
-
+
public MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) {
return destroyLockRef(fullyQualifiedKey, lockReference);
}
String table = splitString[1];
//leave a signal that this key could potentially be unsynchronized
- String syncTable = keyspace+".unsyncedKeys_"+table;
+ String syncTable = keyspace+".unsyncedKeys_"+table;
PreparedQueryObject queryObject = new PreparedQueryObject();
String values = "(?)";
queryObject.addValue(fullyQualifiedKey);
try {
MusicDataStoreHandle.getDSHandle().executePut(queryObject, "critical");
} catch (Exception e) {
- logger.error("Cannot forcibly release lock: " + fullyQualifiedKey + " " + lockReference + ". "
+ logger.error("Cannot forcibly release lock: " + fullyQualifiedKey + " " + lockReference + ". "
+ e.getMessage());
}
-
+
//now release the lock
return destroyLockRef(fullyQualifiedKey, lockReference);
}
/**
- *
+ *
* @param lockName
- * @throws MusicLockingException
+ * @throws MusicLockingException
*/
public void deleteLock(String lockName) throws MusicLockingException {
//deprecated
// Prepared Query Additions.
/**
- *
+ *
* @param queryObject
* @return ReturnType
* @throws MusicServiceException
return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
}
}
-
+
/**
- *
+ *
* @param queryObject
* @return ReturnType
* @throws MusicServiceException
}
queryObject.replaceQueryString(query);
}
-
+
} catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.applicationLogger,e.getMessage());
- }
+ }
try {
result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
} catch (MusicServiceException | MusicQueryException ex) {
return new ReturnType(ResultType.FAILURE, "Eventual Operation failed to perform");
}
}
-
+
/**
- *
+ *
* @param keyspace
* @param table
* @param primaryKeyValue
"Exception thrown while checking the condition, check its sanctity:\n"
+ e.getMessage());
}
-
+
String query = queryObject.getQuery();
long timeOfWrite = System.currentTimeMillis();
long lockOrdinal = Long.parseLong(lockId.substring(lockId.lastIndexOf("$")+1));
return new ReturnType(ResultType.SUCCESS, "Update performed");
}
-
+
/**
- *
+ *
* @param queryObject
* @param consistency
* @return Boolean Indicates success or failure
- * @throws MusicServiceException
- *
- *
+ * @throws MusicServiceException
+ *
+ *
*/
public ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
// this is mainly for some functions like keyspace creation etc which does not
/**
* This method performs DDL operation on cassandra.
- *
+ *
* @param queryObject query object containing prepared query and values
* @return ResultSet
- * @throws MusicServiceException
+ * @throws MusicServiceException
*/
public ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
ResultSet results = null;
/**
* 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 keyspace name of the keyspace
* @param table name of the table
* @param primaryKeyValue primary key value
public ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
PreparedQueryObject queryObject, String lockId) throws MusicServiceException {
ResultSet results = null;
-
+
try {
- ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue,
+ ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue,
lockId.substring(lockId.lastIndexOf("$")+1));
if(result.getResult().equals(ResultType.FAILURE))
return null;//not top of the lock store q
/**
* This method performs DML operation on cassandra, when the lock of the dd is acquired.
- *
+ *
* @param keyspaceName name of the keyspace
* @param tableName name of the table
* @param primaryKey primary key value
* @param queryObject query object containing prepared query and values
* @return ReturnType
- * @throws MusicLockingException
- * @throws MusicServiceException
- * @throws MusicQueryException
+ * @throws MusicLockingException
+ * @throws MusicServiceException
+ * @throws MusicQueryException
*/
public ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException {
criticalPutResult.setTimingInfo(timingInfo);
return criticalPutResult;
}
-
+
/**
* This method performs DDL operation on cassasndra, when the lock for the resource is acquired.
- *
+ *
* @param keyspaceName name of the keyspace
* @param tableName name of the table
* @param primaryKey primary key value
* @param queryObject query object containing prepared query and values
* @return ResultSet
* @throws MusicServiceException
- * @throws MusicLockingException
- * @throws MusicQueryException
+ * @throws MusicLockingException
+ * @throws MusicQueryException
*/
public ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
}
}
-
-
+
+
/**
* @param lockName
* @return
*/
public Map<String, Object> validateLock(String lockName) {
- Map<String, Object> resultMap = new HashMap<>();
- String[] locks = lockName.split("\\.");
- if(locks.length < 3) {
- resultMap.put("Error", "Invalid lock. Please make sure lock is of the type keyspaceName.tableName.primaryKey");
- return resultMap;
- }
- String keyspace= locks[0];
- if(keyspace.startsWith("$"))
- keyspace = keyspace.substring(1);
- resultMap.put("keyspace",keyspace);
- return resultMap;
+ return MusicUtil.validateLock(lockName);
}
-
-
- public static void main(String[] args) {
- String x = "axe top";
- x = x.replaceFirst("top", "sword");
- System.out.print(x); //returns sword pickaxe
- }
-
-
@Override
public ReturnType atomicPutWithDeleteLock(String keyspaceName, String tableName, String primaryKey,
Code Review / music.git / commitdiff