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.main.MusicCore;
import org.onap.music.main.MusicUtil;
import org.onap.music.main.ResultType;
ReturnType lockAcqResult = MusicCore.acquireLock(fullyQualifiedKey, lockId);
if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
try {
- results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryBank.get(MusicUtil.SELECT));
+ results = MusicCore.getInstanceDSHandle().executeQuorumConsistencyGet(queryBank.get(MusicUtil.SELECT));
} catch (Exception e) {
return new ReturnType(ResultType.FAILURE, e.getMessage());
}
if (results.all().isEmpty()) {
- MusicDataStoreHandle.getDSHandle().executePut(queryBank.get(MusicUtil.INSERT), "critical");
+ MusicCore.getInstanceDSHandle().executePut(queryBank.get(MusicUtil.INSERT), "critical");
return new ReturnType(ResultType.SUCCESS, "insert");
} else {
- MusicDataStoreHandle.getDSHandle().executePut(queryBank.get(MusicUtil.UPDATE), "critical");
+ MusicCore.getInstanceDSHandle().executePut(queryBank.get(MusicUtil.UPDATE), "critical");
return new ReturnType(ResultType.SUCCESS, "update");
}
} else {
try {
ReturnType lockAcqResult = MusicCore.acquireLockWithLease(key, lockId, leasePeriod);
if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
- Row row = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryBank.get(MusicUtil.SELECT)).one();
+ Row row = MusicCore.getInstanceDSHandle().executeQuorumConsistencyGet(queryBank.get(MusicUtil.SELECT)).one();
if(row != null) {
Map<String, String> updatedValues = cascadeColumnUpdateSpecific(row, cascadeColumnValues, casscadeColumnName, planId);
update.addValue(MusicUtil.convertToActualDataType(DataType.text(), vector_ts));
update.addValue(MusicUtil.convertToActualDataType(DataType.text(), primaryKeyValue));
try {
- MusicDataStoreHandle.getDSHandle().executePut(update, "critical");
+ MusicCore.getInstanceDSHandle().executePut(update, "critical");
} catch (Exception ex) {
return new ReturnType(ResultType.FAILURE, ex.getMessage());
}
}else {
return new ReturnType(ResultType.FAILURE,"Cannot find data related to key: "+primaryKey);
}
- MusicDataStoreHandle.getDSHandle().executePut(queryBank.get(MusicUtil.UPSERT), "critical");
+ MusicCore.getInstanceDSHandle().executePut(queryBank.get(MusicUtil.UPSERT), "critical");
return new ReturnType(ResultType.SUCCESS, "update success");
} else {
--- /dev/null
+package org.onap.music.datastore;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.NoHostAvailableException;
+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.main.MusicUtil;
+
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.util.ArrayList;
+import java.util.Enumeration;
+
+public class CassandraClusterBuilder {
+ private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(CassandraClusterBuilder.class);
+
+ private static Cluster createCassandraCluster(String address) throws NoHostAvailableException {
+ Cluster cluster = Cluster.builder().withPort(9042)
+ .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
+ .addContactPoint(address).build();
+ Metadata metadata = cluster.getMetadata();
+ logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
+ + metadata.getClusterName() + " at " + address);
+ return cluster;
+ }
+ /**
+ *
+ * @return
+ */
+ private static ArrayList<String> getAllPossibleLocalIps() {
+ ArrayList<String> allPossibleIps = new ArrayList<String>();
+ try {
+ Enumeration<NetworkInterface> en = NetworkInterface.getNetworkInterfaces();
+ while (en.hasMoreElements()) {
+ NetworkInterface ni = (NetworkInterface) en.nextElement();
+ Enumeration<InetAddress> ee = ni.getInetAddresses();
+ while (ee.hasMoreElements()) {
+ InetAddress ia = (InetAddress) ee.nextElement();
+ allPossibleIps.add(ia.getHostAddress());
+ }
+ }
+ } catch (SocketException e) {
+ logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.CONNCECTIVITYERROR, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
+ }catch(Exception e) {
+ logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), ErrorSeverity.ERROR, ErrorTypes.GENERALSERVICEERROR);
+ }
+ return allPossibleIps;
+ }
+
+ /**
+ * This method iterates through all available local IP addresses and tries to connect to first successful one
+ */
+ public static Cluster connectToLocalCassandraCluster() {
+ ArrayList<String> localAddrs = getAllPossibleLocalIps();
+ localAddrs.add(0, "localhost");
+ logger.info(EELFLoggerDelegate.applicationLogger,
+ "Connecting to cassa cluster: Iterating through possible ips:"
+ + getAllPossibleLocalIps());
+ for (String address: localAddrs) {
+ try {
+ return createCassandraCluster(address);
+ } catch (NoHostAvailableException e) {
+ logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.HOSTUNAVAILABLE, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * This method connects to cassandra cluster on specific address.
+ *
+ * @param address
+ */
+ public static Cluster connectToRemoteCassandraCluster(String address) throws MusicServiceException {
+ try {
+ return createCassandraCluster(address);
+ } catch (Exception ex) {
+ logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY, ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE);
+ throw new MusicServiceException(
+ "Error while connecting to Cassandra cluster.. " + ex.getMessage());
+ }
+ }
+
+ public static Cluster connectSmart(String cassaHost) throws MusicServiceException {
+ if (cassaHost.equals("localhost")) {
+ Cluster cluster = CassandraClusterBuilder.connectToLocalCassandraCluster();
+ return cluster;
+ } else {
+ Cluster cluster = CassandraClusterBuilder.connectToRemoteCassandraCluster(MusicUtil.getMyCassaHost());
+ return cluster;
+ }
+
+ }
+}
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
-
-
- public class Condition {
- Map<String, Object> conditions;
- PreparedQueryObject selectQueryForTheRow;
- private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(Condition.class);
- //private static MusicCoreService musicCore = MusicCassaCore.getInstance();
-
- public Condition(Map<String, Object> conditions, PreparedQueryObject selectQueryForTheRow) {
- this.conditions = conditions;
- this.selectQueryForTheRow = selectQueryForTheRow;
- }
-
- public boolean testCondition() throws Exception {
- // first generate the row
- ResultSet results = MusicCore.quorumGet(selectQueryForTheRow);
- Row row = results.one();
- return MusicDataStoreHandle.getDSHandle().doesRowSatisfyCondition(row, conditions);
- }
+public class Condition {
+ Map<String, Object> conditions;
+ PreparedQueryObject selectQueryForTheRow;
+ private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(Condition.class);
+ //private static MusicCoreService musicCore = MusicCassaCore.getInstance();
+
+ public Condition(Map<String, Object> conditions, PreparedQueryObject selectQueryForTheRow) {
+ this.conditions = conditions;
+ this.selectQueryForTheRow = selectQueryForTheRow;
}
+ public boolean testCondition() throws Exception {
+ // first generate the row
+ ResultSet results = MusicCore.quorumGet(selectQueryForTheRow);
+ Row row = results.one();
+ return MusicDataStore.doesRowSatisfyCondition(row, conditions);
+ }
+}
}
- /**
- * Constructs DataStore by connecting to local Cassandra
- */
- public MusicDataStore() {
- connectToLocalCassandraCluster();
- }
-
/**
* Constructs DataStore by providing existing cluster and session
* @param cluster
this.cluster = cluster;
}
- /**
- * Constructs DataStore by connecting to provided remote Cassandra
- * @param remoteAddress
- * @throws MusicServiceException
- */
- public MusicDataStore(String remoteAddress) {
- try {
- connectToRemoteCassandraCluster(remoteAddress);
- } catch (MusicServiceException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage());
- }
- }
-
- private void createCassandraSession(String address) throws NoHostAvailableException {
- cluster = Cluster.builder().withPort(9042)
- .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
- .addContactPoint(address).build();
- Metadata metadata = cluster.getMetadata();
- logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
- + metadata.getClusterName() + " at " + address);
- session = cluster.connect();
- }
-
- /**
- *
- * @return
- */
- private ArrayList<String> getAllPossibleLocalIps() {
- ArrayList<String> allPossibleIps = new ArrayList<String>();
- try {
- Enumeration<NetworkInterface> en = NetworkInterface.getNetworkInterfaces();
- while (en.hasMoreElements()) {
- NetworkInterface ni = (NetworkInterface) en.nextElement();
- Enumeration<InetAddress> ee = ni.getInetAddresses();
- while (ee.hasMoreElements()) {
- InetAddress ia = (InetAddress) ee.nextElement();
- allPossibleIps.add(ia.getHostAddress());
- }
- }
- } catch (SocketException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.CONNCECTIVITYERROR, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
- }catch(Exception e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), ErrorSeverity.ERROR, ErrorTypes.GENERALSERVICEERROR);
- }
- return allPossibleIps;
- }
-
- /**
- * This method iterates through all available local IP addresses and tries to connect to first successful one
- */
- private void connectToLocalCassandraCluster() {
- ArrayList<String> localAddrs = getAllPossibleLocalIps();
- localAddrs.add(0, "localhost");
- logger.info(EELFLoggerDelegate.applicationLogger,
- "Connecting to cassa cluster: Iterating through possible ips:"
- + getAllPossibleLocalIps());
- for (String address: localAddrs) {
- try {
- createCassandraSession(address);
- break;
- } catch (NoHostAvailableException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.HOSTUNAVAILABLE, ErrorSeverity.ERROR, ErrorTypes.CONNECTIONERROR);
- }
- }
- }
-
- /**
- * This method connects to cassandra cluster on specific address.
- *
- * @param address
- */
- private void connectToRemoteCassandraCluster(String address) throws MusicServiceException {
- try {
- createCassandraSession(address);
- } catch (Exception ex) {
- logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY, ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE);
- throw new MusicServiceException(
- "Error while connecting to Cassandra cluster.. " + ex.getMessage());
- }
- }
-
- /**
- *
- */
- public void close() {
- session.close();
- }
-
/**
*
* @param keyspace
* @param colType
* @return
*/
- public Object getColValue(Row row, String colName, DataType colType) {
+ public static Object getColValue(Row row, String colName, DataType colType) {
switch (colType.getName()) {
case VARCHAR:
return data;
}
- public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
+ public static boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
ColumnDefinitions colInfo = row.getColumnDefinitions();
for (Map.Entry<String, Object> entry : condition.entrySet()) {
TimeMeasureInstance.instance().exit();
}
}
+
+ @Deprecated
+ public void close() {
+ session.close();
+ }
}
import java.util.HashMap;
import java.util.Map;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
import org.onap.music.eelf.logging.EELFLoggerDelegate;
import org.onap.music.exceptions.MusicServiceException;
import org.onap.music.main.MusicUtil;
-import org.onap.music.service.impl.MusicCassaCore;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.TableMetadata;
public static MusicDataStore mDstoreHandle = null;
private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicDataStoreHandle.class);
- /**
- *
- * @param remoteAddress
- * @return
- */
- public static MusicDataStore getDSHandle(String remoteAddress) {
- logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring data store handle");
- long start = System.currentTimeMillis();
- if (mDstoreHandle == null) {
- try {
- MusicUtil.loadProperties();
- } catch (Exception e) {
- logger.error(EELFLoggerDelegate.errorLogger, "No properties file defined. Falling back to default.");
- }
- mDstoreHandle = new MusicDataStore(remoteAddress);
- }
- long end = System.currentTimeMillis();
- logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire data store handle:" + (end - start) + " ms");
- return mDstoreHandle;
- }
-
/**
*
* @return
* @throws MusicServiceException
*/
- public static MusicDataStore getDSHandle() throws MusicServiceException {
+ private static MusicDataStore getDSHandle() throws MusicServiceException {
logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring data store handle");
long start = System.currentTimeMillis();
} catch (Exception e) {
logger.error(EELFLoggerDelegate.errorLogger, "No properties file defined. Falling back to default.");
}
- // Quick Fix - Best to put this into every call to getDSHandle?
- if (MusicUtil.getMyCassaHost().equals("localhost")) {
- mDstoreHandle = new MusicDataStore();
- } else {
- mDstoreHandle = new MusicDataStore(MusicUtil.getMyCassaHost());
- }
+ // Quick Fix - Best to put this into every call to getInstanceDSHandle?
+ Cluster cluster = CassandraClusterBuilder.connectSmart(MusicUtil.getMyCassaHost());
+ Session session = cluster.connect();
+ mDstoreHandle = new MusicDataStore(cluster, session);
+
}
if(mDstoreHandle.getSession() == null) {
String message = "Connection to Cassandra has not been enstablished."
package org.onap.music.lockingservice.cassandra;
+import java.io.IOError;
import java.util.ArrayList;
import java.util.List;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
+import org.onap.music.datastore.CassandraClusterBuilder;
import org.onap.music.datastore.MusicDataStore;
import org.onap.music.datastore.PreparedQueryObject;
import org.onap.music.eelf.logging.EELFLoggerDelegate;
+import org.onap.music.exceptions.MusicLockingException;
import org.onap.music.exceptions.MusicQueryException;
import org.onap.music.exceptions.MusicServiceException;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
+import org.onap.music.main.MusicUtil;
import org.onap.music.util.TimeMeasureInstance;
/*
private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(CassaLockStore.class);
private static String table_prepend_name = "lockQ_";
-
+ private MusicDataStore dsHandle;
+
public class LockObject{
public String lockRef;
public String createTime;
this.lockRef = lockRef;
this.acquireTime = acquireTime;
this.createTime = createTime;
-
}
}
- MusicDataStore dsHandle;
- public CassaLockStore() {
- dsHandle = new MusicDataStore();
- }
-
+
public CassaLockStore(MusicDataStore dsHandle) {
this.dsHandle=dsHandle;
}
-
/**
*
* This method creates a shadow locking table for every main table in Cassandra. This table tracks all information regarding locks.
TimeMeasureInstance.instance().exit();
}
}
-
-
}
package org.onap.music.main;
-import java.io.StringWriter;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.StringTokenizer;
-import org.onap.music.datastore.Condition;
-import org.onap.music.datastore.MusicDataStore;
-import org.onap.music.datastore.MusicDataStoreHandle;
-import org.onap.music.datastore.PreparedQueryObject;
+import com.datastax.driver.core.*;
+import org.onap.music.datastore.*;
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.MusicQueryException;
import org.onap.music.exceptions.MusicServiceException;
import org.onap.music.lockingservice.cassandra.CassaLockStore;
import org.onap.music.lockingservice.cassandra.MusicLockState;
-import org.onap.music.lockingservice.cassandra.CassaLockStore.LockObject;
import org.onap.music.service.MusicCoreService;
import org.onap.music.service.impl.MusicCassaCore;
-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;
-
/**
* This class .....
*/
public class MusicCore {
- public static CassaLockStore mLockHandle = null;
private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicCore.class);
private static boolean unitTestRun=true;
-
-private static MusicCoreService musicCore = MusicCassaCore.getInstance();
-
-
+
+ public static CassaLockStore mLockHandle = null;
+ private static MusicDataStore mDependantDataStoreHandle = null;
+ private static MusicCoreService mMusicCore = null;
+
+ public static MusicCoreService getSingletonInstance() {
+ /* Hacks to make singleton instance backward-compatible
+ Some clients of MusicCore, set the data-store handle directly on MusicDataStoreHandle and
+ lock-store handle on mLockHandle static field. We should watch for changes in these and
+ create new Core instance in case they have changed
+ */
+ if (MusicDataStoreHandle.mDstoreHandle != null &&
+ mDependantDataStoreHandle != MusicDataStoreHandle.mDstoreHandle) {
+ // Use externally set data-store and lock-store handle
+ mDependantDataStoreHandle = MusicDataStoreHandle.mDstoreHandle;
+ mMusicCore = new MusicCassaCore(mDependantDataStoreHandle, mLockHandle);
+ }
+ else if (mMusicCore == null) {
+ String address = MusicUtil.getMyCassaHost();
+ Cluster cluster;
+ try {
+ cluster = CassandraClusterBuilder.connectSmart(MusicUtil.getMyCassaHost());
+ } catch (MusicServiceException e) {
+ logger.error(EELFLoggerDelegate.errorLogger, "Can not connect to Cassandra cluster: " + address);
+ return null;
+ }
+ Metadata metadata = cluster.getMetadata();
+ logger.info(EELFLoggerDelegate.applicationLogger, "Connected to Cassandra cluster "
+ + metadata.getClusterName() + " at " + address);
+ Session session = cluster.connect();
+
+ mMusicCore = new MusicCassaCore(cluster, session);
+ }
+
+ return mMusicCore;
+ }
+
public static ReturnType acquireLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
- return musicCore.acquireLock(fullyQualifiedKey, lockReference);
+ return getSingletonInstance().acquireLock(fullyQualifiedKey, lockReference);
}
public static ReturnType acquireLockWithLease(String key, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
- return musicCore.acquireLockWithLease(key, lockReference, leasePeriod);
+ return getSingletonInstance().acquireLockWithLease(key, lockReference, leasePeriod);
}
public static String createLockReference(String fullyQualifiedKey) {
- return musicCore.createLockReference(fullyQualifiedKey);
+ return getSingletonInstance().createLockReference(fullyQualifiedKey);
}
public static String createLockReference(String fullyQualifiedKey, boolean isWriteLock) {
- return musicCore.createLockReference(fullyQualifiedKey, isWriteLock);
+ return getSingletonInstance().createLockReference(fullyQualifiedKey, isWriteLock);
}
public static MusicLockState forciblyReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicServiceException, MusicQueryException{
- return musicCore.forciblyReleaseLock(fullyQualifiedKey, lockReference);
+ return getSingletonInstance().forciblyReleaseLock(fullyQualifiedKey, lockReference);
}
public static ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
- return musicCore.createTable(keyspace, table, tableQueryObject, consistency);
+ return getSingletonInstance().createTable(keyspace, table, tableQueryObject, consistency);
}
public static ResultSet quorumGet(PreparedQueryObject query) {
- return musicCore.quorumGet(query);
+ return getSingletonInstance().quorumGet(query);
}
public static String whoseTurnIsIt(String fullyQualifiedKey) {
- return musicCore.whoseTurnIsIt(fullyQualifiedKey);
+ return getSingletonInstance().whoseTurnIsIt(fullyQualifiedKey);
}
public static MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) {
- return musicCore.destroyLockRef(fullyQualifiedKey, lockReference);
+ return getSingletonInstance().destroyLockRef(fullyQualifiedKey, lockReference);
}
public static MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException {
- return musicCore.voluntaryReleaseLock(fullyQualifiedKey, lockReference);
+ return getSingletonInstance().voluntaryReleaseLock(fullyQualifiedKey, lockReference);
}
public static ReturnType eventualPut(PreparedQueryObject queryObject) {
- return musicCore.eventualPut(queryObject);
+ return getSingletonInstance().eventualPut(queryObject);
}
public static ReturnType criticalPut(String keyspace, String table, String primaryKeyValue,
PreparedQueryObject queryObject, String lockReference, Condition conditionInfo) {
- return musicCore.criticalPut(keyspace, table, primaryKeyValue, queryObject, lockReference, conditionInfo);
+ return getSingletonInstance().criticalPut(keyspace, table, primaryKeyValue, queryObject, lockReference, conditionInfo);
}
public static ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
- return musicCore.nonKeyRelatedPut(queryObject, consistency);
+ return getSingletonInstance().nonKeyRelatedPut(queryObject, consistency);
}
public static ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException{
- return musicCore.get(queryObject);
+ return getSingletonInstance().get(queryObject);
}
public static ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
PreparedQueryObject queryObject, String lockReference) throws MusicServiceException{
- return musicCore.criticalGet(keyspace, table, primaryKeyValue, queryObject,lockReference);
+ return getSingletonInstance().criticalGet(keyspace, table, primaryKeyValue, queryObject,lockReference);
}
public static ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException
{
- return musicCore.atomicPut(keyspaceName, tableName, primaryKey, queryObject, conditionInfo);
+ return getSingletonInstance().atomicPut(keyspaceName, tableName, primaryKey, queryObject, conditionInfo);
}
public static ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
- return musicCore.atomicGet(keyspaceName, tableName, primaryKey, queryObject);
+ return getSingletonInstance().atomicGet(keyspaceName, tableName, primaryKey, queryObject);
}
public static List<String> getLockQueue(String fullyQualifiedKey)
throws MusicServiceException, MusicQueryException, MusicLockingException{
- return musicCore.getLockQueue(fullyQualifiedKey);
+ return getSingletonInstance().getLockQueue(fullyQualifiedKey);
}
public static long getLockQueueSize(String fullyQualifiedKey)
throws MusicServiceException, MusicQueryException, MusicLockingException {
- return musicCore.getLockQueueSize(fullyQualifiedKey);
+ return getSingletonInstance().getLockQueueSize(fullyQualifiedKey);
}
+ public static MusicDataStore getInstanceDSHandle() {
+ MusicCoreService singletonInstance = getSingletonInstance();
+ if (singletonInstance instanceof MusicCassaCore)
+ return ((MusicCassaCore) singletonInstance).getDataStoreHandle();
+ return null;
+ }
}
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
-import java.util.UUID;
-import org.onap.music.datastore.MusicDataStore;
-import org.onap.music.datastore.MusicDataStoreHandle;
-import org.onap.music.datastore.Condition;
-import org.onap.music.datastore.PreparedQueryObject;
+import com.datastax.driver.core.*;
+import org.onap.music.datastore.*;
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.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 org.onap.music.util.TimeMeasureInstance;
*
*/
public class MusicCassaCore implements MusicCoreService {
+ private CassaLockStore mLockStoreHandle = null;
+ private MusicDataStore mDataStoreHandle = null;
- public static CassaLockStore mLockHandle = null;;
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 MusicCassaCore(Cluster cassandraCluster, Session cassandraSession) {
+ try {
+ MusicUtil.loadProperties();
+ } catch (Exception e) {
+ logger.error(EELFLoggerDelegate.errorLogger, "No properties file defined. Falling back to default.");
+ }
+
+ {
+ logger.info(EELFLoggerDelegate.applicationLogger, "Acquiring data store handle");
+ long start = System.currentTimeMillis();
+ if (mDataStoreHandle == null) {
+ mDataStoreHandle = new MusicDataStore(cassandraCluster, cassandraSession);
+ }
+ long end = System.currentTimeMillis();
+ logger.info(EELFLoggerDelegate.applicationLogger, "Time taken to acquire data store handle:" + (end - start) + " ms");
+ }
+ {
+ logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
+ long start = System.currentTimeMillis();
+
+ try {
+ mLockStoreHandle = new CassaLockStore(getDataStoreHandle());
+ } catch (Exception e) {
+ logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
+ }
+ long end = System.currentTimeMillis();
+ logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
+ }
}
-
- public static CassaLockStore getLockingServiceHandle() throws MusicLockingException {
- logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
- long start = System.currentTimeMillis();
- if (mLockHandle == null) {
+ public MusicCassaCore(MusicDataStore dataStore, CassaLockStore lockStore) {
+ mDataStoreHandle = dataStore;
+ mLockStoreHandle = lockStore;
+
+ if (lockStore == null)
+ {
+ logger.info(EELFLoggerDelegate.applicationLogger,"Acquiring lock store handle");
+ long start = System.currentTimeMillis();
+
try {
- mLockHandle = new CassaLockStore(MusicDataStoreHandle.getDSHandle());
+ mLockStoreHandle = new CassaLockStore(getDataStoreHandle());
} catch (Exception e) {
- logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
- throw new MusicLockingException("Failed to acquire Local store handle " + e);
+ logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKHANDLE,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
}
+ long end = System.currentTimeMillis();
+ logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
}
- long end = System.currentTimeMillis();
- logger.info(EELFLoggerDelegate.applicationLogger,"Time taken to acquire lock store handle:" + (end - start) + " ms");
- return mLockHandle;
}
+ public MusicDataStore getDataStoreHandle() {
+ return mDataStoreHandle;
+ }
+
+ public CassaLockStore getLockingServiceHandle() {
+ return mLockStoreHandle;
+ }
public String createLockReference(String fullyQualifiedKey) {
return createLockReference(fullyQualifiedKey, true);
}
- public String createLockReference(String fullyQualifiedKey, boolean isWriteLock) {
+ public String createLockReference(String fullyQualifiedKey, boolean isWriteLock) {
TimeMeasureInstance.instance().enter("createLockReference");
try {
String[] splitString = fullyQualifiedKey.split("\\.");
String lockReference = null;
try {
lockReference = "" + getLockingServiceHandle().genLockRefandEnQueue(keyspace, table, lockName, isWriteLock);
- } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
+ } catch (MusicServiceException | MusicQueryException e) {
logger.info(EELFLoggerDelegate.applicationLogger, "Failed to create lock reference for" + lockName);
return null;
}
}
- public ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ public ReturnType acquireLockWithLease(String fullyQualifiedKey, String lockReference, long leasePeriod) throws MusicLockingException, MusicQueryException, MusicServiceException {
evictExpiredLockHolder(fullyQualifiedKey,leasePeriod);
return acquireLock(fullyQualifiedKey, lockReference);
}
}
}
- private static ReturnType isTopOfLockStore(String keyspace, String table,
+ private ReturnType isTopOfLockStore(String keyspace, String table,
String primaryKeyValue, String lockReference)
throws MusicLockingException, MusicQueryException, MusicServiceException {
TimeMeasureInstance.instance().enter("isTopOfLockStore");
}
}
- public ReturnType acquireLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
+ public ReturnType acquireLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException, MusicQueryException, MusicServiceException {
TimeMeasureInstance.instance().enter("acquireLock");
try {
String[] splitString = fullyQualifiedKey.split("\\.");
String query = "select * from " + syncTable + " where key='" + fullyQualifiedKey + "';";
PreparedQueryObject readQueryObject = new PreparedQueryObject();
readQueryObject.appendQueryString(query);
- ResultSet results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(readQueryObject);
+ ResultSet results = getDataStoreHandle().executeQuorumConsistencyGet(readQueryObject);
if (results.all().size() != 0) {
logger.info("In acquire lock: Since there was a forcible release, need to sync quorum!");
try {
String cleanQuery = "delete from music_internal.unsynced_keys where key='" + fullyQualifiedKey + "';";
PreparedQueryObject deleteQueryObject = new PreparedQueryObject();
deleteQueryObject.appendQueryString(cleanQuery);
- MusicDataStoreHandle.getDSHandle().executePut(deleteQueryObject, "critical");
+ getDataStoreHandle().executePut(deleteQueryObject, "critical");
}
getLockingServiceHandle().updateLockAcquireTime(keyspace, table, primaryKeyValue, lockReference);
*
*
*/
- public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
+ public ResultType createTable(String keyspace, String table, PreparedQueryObject tableQueryObject, String consistency) throws MusicServiceException {
boolean result = false;
try {
queryObject.appendQueryString(tabQuery);
result = false;
- result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, "eventual");
+ result = getDataStoreHandle().executePut(queryObject, "eventual");
//create actual table
- result = MusicDataStoreHandle.getDSHandle().executePut(tableQueryObject, consistency);
- } catch (MusicQueryException | MusicServiceException | MusicLockingException ex) {
+ result = getDataStoreHandle().executePut(tableQueryObject, consistency);
+ } catch (MusicQueryException | MusicServiceException 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 {
+ private void syncQuorum(String keyspace, String table, String primaryKeyValue) throws Exception {
logger.info(EELFLoggerDelegate.applicationLogger,"Performing sync operation---");
PreparedQueryObject selectQuery = new PreparedQueryObject();
PreparedQueryObject updateQuery = new PreparedQueryObject();
selectQuery.addValue(cqlFormattedPrimaryKeyValue);
ResultSet results = null;
try {
- results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(selectQuery);
+ results = getDataStoreHandle().executeQuorumConsistencyGet(selectQuery);
// write it back to a quorum
Row row = results.one();
ColumnDefinitions colInfo = row.getColumnDefinitions();
if (colName.equals(primaryKeyName))
continue;
DataType colType = definition.getType();
- Object valueObj = MusicDataStoreHandle.getDSHandle().getColValue(row, colName, colType);
+ Object valueObj = getDataStoreHandle().getColValue(row, colName, colType);
Object valueString = MusicUtil.convertToActualDataType(colType, valueObj);
fieldValueString.append(colName + " = ?");
updateQuery.addValue(valueString);
+ fieldValueString + " WHERE " + primaryKeyName + "= ? " + ";");
updateQuery.addValue(cqlFormattedPrimaryKeyValue);
- MusicDataStoreHandle.getDSHandle().executePut(updateQuery, "critical");
+ getDataStoreHandle().executePut(updateQuery, "critical");
} catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.QUERYERROR +""+updateQuery ,ErrorSeverity.MAJOR, ErrorTypes.QUERYERROR);
}
}
-
-
-
/**
*
* @param query
* @return ResultSet
*/
- public ResultSet quorumGet(PreparedQueryObject query) {
+ public ResultSet quorumGet(PreparedQueryObject query) {
ResultSet results = null;
try {
- results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(query);
+ results = getDataStoreHandle().executeQuorumConsistencyGet(query);
} catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.MAJOR, ErrorTypes.GENERALSERVICEERROR);
try {
LockObject lockObject = getLockingServiceHandle().peekLockQueue(keyspace, table, primaryKeyValue);
return lockObject != null ? lockObject.lockRef : null;
- } catch (MusicLockingException | MusicServiceException | MusicQueryException e) {
+ } catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.LOCKINGERROR+fullyQualifiedKey ,ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
}
return null;
return st.nextToken("$");
}
- public MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) {
+ public MusicLockState destroyLockRef(String fullyQualifiedKey, String lockReference) {
TimeMeasureInstance.instance().enter("destroyLockRef");
try {
long start = System.currentTimeMillis();
String primaryKeyValue = splitString[2];
try {
getLockingServiceHandle().deQueueLockRef(keyspace, table, primaryKeyValue, lockReference);
- } catch (MusicServiceException | MusicQueryException | MusicLockingException e) {
+ } catch (MusicServiceException | MusicQueryException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.DESTROYLOCK + " " + primaryKeyValue + " " + lockReference, ErrorSeverity.CRITICAL, ErrorTypes.LOCKINGERROR);
logger.info(EELFLoggerDelegate.applicationLogger, "Lock not destroyed " + primaryKeyValue + " " + lockReference + " " + " :" + e.getMessage());
return new MusicLockState(MusicLockState.LockStatus.LOCKED, "");
}
}
- public MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException{
+ public MusicLockState voluntaryReleaseLock(String fullyQualifiedKey, String lockReference) throws MusicLockingException{
TimeMeasureInstance.instance().enter("voluntaryReleaseLock");
try {
return destroyLockRef(fullyQualifiedKey, lockReference);
queryObject.addValue(fullyQualifiedKey);
String insQuery = "insert into "+syncTable+" (key) values "+values+";";
queryObject.appendQueryString(insQuery);
- MusicDataStoreHandle.getDSHandle().executePut(queryObject, "critical");
+ getDataStoreHandle().executePut(queryObject, "critical");
//now release the lock
return destroyLockRef(fullyQualifiedKey, lockReference);
* @param lockName
* @throws MusicLockingException
*/
- public void deleteLock(String lockName) throws MusicLockingException {
+ public void deleteLock(String lockName) throws MusicLockingException {
//deprecated
}
* @return ReturnType
* @throws MusicServiceException
*/
- public ReturnType eventualPut(PreparedQueryObject queryObject) {
+ public ReturnType eventualPut(PreparedQueryObject queryObject) {
boolean result = false;
try {
- result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, MusicUtil.EVENTUAL);
+ result = getDataStoreHandle().executePut(queryObject, MusicUtil.EVENTUAL);
} catch (MusicServiceException | MusicQueryException ex) {
logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage(), "[ERR512E] Failed to get ZK Lock Handle " ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() + " " + ex.getCause() + " " + ex);
return new ReturnType(ResultType.FAILURE, "Failure");
}
}
-
+
/**
*
* @param keyspace
* @param lockReference
* @return
*/
- public ReturnType criticalPut(String keyspace, String table, String primaryKeyValue,
+ public ReturnType criticalPut(String keyspace, String table, String primaryKeyValue,
PreparedQueryObject queryObject, String lockReference, Condition conditionInfo) {
TimeMeasureInstance.instance().enter("criticalPut");
try {
// String query = queryObject.getQuery();
// query = query.replaceFirst("SET", "USING TIMESTAMP " + ts + " SET");
// queryObject.replaceQueryString(query);
- MusicDataStore dsHandle = MusicDataStoreHandle.getDSHandle();
+ MusicDataStore dsHandle = getDataStoreHandle();
dsHandle.executePut(queryObject, MusicUtil.CRITICAL, lockOrdinal);
long end = System.currentTimeMillis();
logger.info(EELFLoggerDelegate.applicationLogger, "Time taken for the critical put " + primaryKeyValue + " " + lockReference + " :" + (end - start) + " ms");
*
*
*/
- public ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
+ public ResultType nonKeyRelatedPut(PreparedQueryObject queryObject, String consistency) throws MusicServiceException {
// this is mainly for some functions like keyspace creation etc which does not
// really need the bells and whistles of Music locking.
boolean result = false;
try {
- result = MusicDataStoreHandle.getDSHandle().executePut(queryObject, consistency);
+ result = getDataStoreHandle().executePut(queryObject, consistency);
} catch (MusicQueryException | MusicServiceException ex) {
logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(), AppMessages.UNKNOWNERROR,
ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
* @return ResultSet
* @throws MusicServiceException
*/
- public ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
+ public ResultSet get(PreparedQueryObject queryObject) throws MusicServiceException {
ResultSet results = null;
try {
- results = MusicDataStoreHandle.getDSHandle().executeOneConsistencyGet(queryObject);
+ results = getDataStoreHandle().executeOneConsistencyGet(queryObject);
} catch (MusicQueryException | MusicServiceException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
throw new MusicServiceException(e.getMessage());
* @param lockReference lock ID to check if the resource is free to perform the operation.
* @return ResultSet
*/
- public ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
+ public ResultSet criticalGet(String keyspace, String table, String primaryKeyValue,
PreparedQueryObject queryObject, String lockReference) throws MusicServiceException {
ResultSet results = null;
ReturnType result = isTopOfLockStore(keyspace, table, primaryKeyValue, lockReference);
if(result.getResult().equals(ResultType.FAILURE))
return null; // not top of the lock store q
- results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryObject);
+ results = getDataStoreHandle().executeQuorumConsistencyGet(queryObject);
} catch (MusicQueryException | MusicServiceException | MusicLockingException e) {
logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), AppMessages.UNKNOWNERROR ,ErrorSeverity.WARN, ErrorTypes.MUSICSERVICEERROR);
}
* @throws MusicServiceException
* @throws MusicQueryException
*/
- public ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
+ public ReturnType atomicPut(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject, Condition conditionInfo) throws MusicLockingException, MusicQueryException, MusicServiceException {
TimeMeasureInstance.instance().enter("atomicPut");
try {
* @throws MusicLockingException
* @throws MusicQueryException
*/
- public ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
+ public ResultSet atomicGet(String keyspaceName, String tableName, String primaryKey,
PreparedQueryObject queryObject) throws MusicServiceException, MusicLockingException, MusicQueryException {
String fullyQualifiedKey = keyspaceName + "." + tableName + "." + primaryKey;
String lockReference = createLockReference(fullyQualifiedKey);
testObject.appendQueryString(CassandraCQL.dropKeyspace);
dataStore.executePut(testObject, "eventual");
dataStore.close();
-
}
@Test
import java.util.Map;
import java.util.UUID;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
+import org.onap.music.datastore.CassandraClusterBuilder;
+import org.onap.music.datastore.MusicDataStore;
import org.onap.music.datastore.PreparedQueryObject;
import org.onap.music.exceptions.MusicQueryException;
import org.onap.music.exceptions.MusicServiceException;
import org.onap.music.lockingservice.cassandra.CassaLockStore;
import org.onap.music.main.MusicCore;
+import org.onap.music.main.MusicUtil;
public class TestCassaLockStore {
try {
- CassaLockStore lockStore = new CassaLockStore();
+ Cluster cluster = CassandraClusterBuilder.connectSmart(MusicUtil.getMyCassaHost());
+ Session session = cluster.connect();
+ CassaLockStore lockStore = new CassaLockStore(new MusicDataStore(cluster, session));
String keyspace = "ks_testLockStore";
String table = "table_testLockStore";
Code Review / music.git / commitdiff