+++ /dev/null
-/*
- * ============LICENSE_START==========================================
- * org.onap.music
- * ===================================================================
- * Copyright (c) 2017 AT&T Intellectual Property
- * ===================================================================
- * Modifications Copyright (c) 2018-2019 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
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- * ============LICENSE_END=============================================
- * ====================================================================
- */
-
-package org.onap.music.datastore;
-
-import java.net.InetAddress;
-import java.net.NetworkInterface;
-import java.net.SocketException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Enumeration;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-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.MusicQueryException;
-import org.onap.music.exceptions.MusicServiceException;
-import org.onap.music.lockingservice.cassandra.LockType;
-import org.onap.music.main.CipherUtil;
-import org.onap.music.main.MusicUtil;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ColumnDefinitions;
-import com.datastax.driver.core.ColumnDefinitions.Definition;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.DataType;
-import com.datastax.driver.core.HostDistance;
-import com.datastax.driver.core.KeyspaceMetadata;
-import com.datastax.driver.core.Metadata;
-import com.datastax.driver.core.PoolingOptions;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.SimpleStatement;
-import com.datastax.driver.core.TableMetadata;
-import com.datastax.driver.core.TypeCodec;
-import com.datastax.driver.core.exceptions.AlreadyExistsException;
-import com.datastax.driver.core.exceptions.InvalidQueryException;
-import com.datastax.driver.core.exceptions.NoHostAvailableException;
-import com.datastax.driver.extras.codecs.enums.EnumNameCodec;
-import com.datastax.driver.extras.codecs.enums.EnumOrdinalCodec;
-
-/**
- * @author nelson24
- *
- */
-public class MusicDataStore {
-
- public static final String CONSISTENCY_LEVEL_ONE = "ONE";
- public static final String CONSISTENCY_LEVEL_QUORUM = "QUORUM";
- private Session session;
- private Cluster cluster;
-
-
- /**
- * @param session
- */
- public void setSession(Session session) {
- this.session = session;
- }
-
- /**
- * @param session
- */
- public Session getSession() {
- return session;
- }
-
- /**
- * @param cluster
- */
- public void setCluster(Cluster cluster) {
- this.cluster = cluster;
- }
-
- public Cluster getCluster() {
- return this.cluster;
- }
-
-
- private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicDataStore.class);
-
- /**
- * Connect to default Cassandra address
- */
- public MusicDataStore() {
- try {
- connectToCassaCluster(MusicUtil.getMyCassaHost());
- } catch (MusicServiceException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), e);
- }
- }
-
-
- /**
- * @param cluster
- * @param session
- */
- public MusicDataStore(Cluster cluster, Session session) {
- this.session = session;
- this.cluster = cluster;
- }
-
- /**
- *
- * @param remoteIp
- * @throws MusicServiceException
- */
- public MusicDataStore(String remoteIp) {
- try {
- connectToCassaCluster(remoteIp);
- } catch (MusicServiceException e) {
- logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), e);
- }
- }
-
- /**
- *
- */
- public void close() {
- session.close();
- }
-
- /**
- * This method connects to cassandra cluster on specific address.
- *
- * @param address
- */
- private void connectToCassaCluster(String address) throws MusicServiceException {
- String[] addresses = null;
- addresses = address.split(",");
- PoolingOptions poolingOptions = new PoolingOptions();
- poolingOptions
- .setConnectionsPerHost(HostDistance.LOCAL, 4, 10)
- .setConnectionsPerHost(HostDistance.REMOTE, 2, 4);
-
- if(MusicUtil.getCassName() != null && MusicUtil.getCassPwd() != null) {
- String cassPwd = CipherUtil.decryptPKC(MusicUtil.getCassPwd());
- logger.info(EELFLoggerDelegate.applicationLogger,
- "Building with credentials "+MusicUtil.getCassName()+" & "+ MusicUtil.getCassPwd());
- cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
- .withCredentials(MusicUtil.getCassName(), cassPwd)
- //.withLoadBalancingPolicy(new RoundRobinPolicy())
- .withoutJMXReporting()
- .withPoolingOptions(poolingOptions)
- .addContactPoints(addresses).build();
- } else {
- cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
- .withoutJMXReporting()
- .withPoolingOptions(poolingOptions)
- .addContactPoints(addresses)
- .build();
- }
-
-
- Metadata metadata = cluster.getMetadata();
- logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
- + metadata.getClusterName() + " at " + address);
-
- EnumNameCodec<LockType> lockTypeCodec = new EnumNameCodec<LockType>(LockType.class);
- cluster.getConfiguration().getCodecRegistry().register(lockTypeCodec);
-
- try {
- session = cluster.connect();
- } catch (Exception ex) {
- logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY,
- ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE, ex);
- throw new MusicServiceException(
- "Error while connecting to Cassandra cluster.. " + ex.getMessage());
- }
- }
-
- /**
- *
- * @param keyspace
- * @param tableName
- * @param columnName
- * @return DataType
- */
- public DataType returnColumnDataType(String keyspace, String tableName, String columnName) {
- KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
- TableMetadata table = ks.getTable(tableName);
- return table.getColumn(columnName).getType();
-
- }
-
- /**
- *
- * @param keyspace
- * @param tableName
- * @return TableMetadata
- */
- public TableMetadata returnColumnMetadata(String keyspace, String tableName) {
- KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
- return ks.getTable(tableName);
- }
-
- /**
- *
- * @param keyspace
- * @param tableName
- * @return TableMetadata
- */
- public KeyspaceMetadata returnKeyspaceMetadata(String keyspace) {
- return cluster.getMetadata().getKeyspace(keyspace);
- }
-
-
- /**
- * Utility function to return the Java specific object type.
- *
- * @param row
- * @param colName
- * @param colType
- * @return
- */
- public Object getColValue(Row row, String colName, DataType colType) {
-
- switch (colType.getName()) {
- case VARCHAR:
- return row.getString(colName);
- case UUID:
- return row.getUUID(colName);
- case VARINT:
- return row.getVarint(colName);
- case BIGINT:
- return row.getLong(colName);
- case INT:
- return row.getInt(colName);
- case FLOAT:
- return row.getFloat(colName);
- case DOUBLE:
- return row.getDouble(colName);
- case BOOLEAN:
- return row.getBool(colName);
- case MAP:
- return row.getMap(colName, String.class, String.class);
- case LIST:
- return row.getList(colName, String.class);
- default:
- return null;
- }
- }
-
- public byte[] getBlobValue(Row row, String colName, DataType colType) {
- ByteBuffer bb = row.getBytes(colName);
- return bb.array();
- }
-
- public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
- ColumnDefinitions colInfo = row.getColumnDefinitions();
-
- for (Map.Entry<String, Object> entry : condition.entrySet()) {
- String colName = entry.getKey();
- DataType colType = colInfo.getType(colName);
- Object columnValue = getColValue(row, colName, colType);
- Object conditionValue = MusicUtil.convertToActualDataType(colType, entry.getValue());
- if (columnValue.equals(conditionValue) == false)
- return false;
- }
- return true;
- }
-
- /**
- * Utility function to store ResultSet values in to a MAP for output.
- *
- * @param results
- * @return MAP
- */
- public Map<String, HashMap<String, Object>> marshalData(ResultSet results) {
- Map<String, HashMap<String, Object>> resultMap =
- new HashMap<>();
- int counter = 0;
- for (Row row : results) {
- ColumnDefinitions colInfo = row.getColumnDefinitions();
- HashMap<String, Object> resultOutput = new HashMap<>();
- for (Definition definition : colInfo) {
- if (!(("vector_ts").equals(definition.getName()))) {
- if(definition.getType().toString().toLowerCase().contains("blob")) {
- resultOutput.put(definition.getName(),
- getBlobValue(row, definition.getName(), definition.getType()));
- } else {
- resultOutput.put(definition.getName(),
- getColValue(row, definition.getName(), definition.getType()));
- }
- }
- }
- resultMap.put("row " + counter, resultOutput);
- counter++;
- }
- return resultMap;
- }
-
-
- // Prepared Statements 1802 additions
-
- public boolean executePut(PreparedQueryObject queryObject, String consistency)
- throws MusicServiceException, MusicQueryException {
- return executePut(queryObject, consistency, 0);
- }
- /**
- * This Method performs DDL and DML operations on Cassandra using specified consistency level
- *
- * @param queryObject Object containing cassandra prepared query and values.
- * @param consistency Specify consistency level for data synchronization across cassandra
- * replicas
- * @return Boolean Indicates operation success or failure
- * @throws MusicServiceException
- * @throws MusicQueryException
- */
- public boolean executePut(PreparedQueryObject queryObject, String consistency,long timeSlot)
- throws MusicServiceException, MusicQueryException {
-
- boolean result = false;
- long timeOfWrite = System.currentTimeMillis();
- if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
- logger.error(EELFLoggerDelegate.errorLogger, queryObject.getQuery(),AppMessages.QUERYERROR, ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("Ill formed queryObject for the request = " + "["
- + queryObject.getQuery() + "]");
- }
- logger.debug(EELFLoggerDelegate.applicationLogger,
- "In preprared Execute Put: the actual insert query:"
- + queryObject.getQuery() + "; the values"
- + queryObject.getValues());
- SimpleStatement preparedInsert = null;
-
- try {
- preparedInsert = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
- if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
- logger.info(EELFLoggerDelegate.applicationLogger, "Executing critical put query");
- preparedInsert.setConsistencyLevel(ConsistencyLevel.QUORUM);
- } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
- logger.info(EELFLoggerDelegate.applicationLogger, "Executing simple put query");
- if(queryObject.getConsistency() == null)
- preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
- else
- preparedInsert.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
- } else if (consistency.equalsIgnoreCase(MusicUtil.ONE)) {
- preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
- } else if (consistency.equalsIgnoreCase(MusicUtil.QUORUM)) {
- preparedInsert.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
- } else if (consistency.equalsIgnoreCase(MusicUtil.ALL)) {
- preparedInsert.setConsistencyLevel(ConsistencyLevel.ALL);
- }
- long timestamp = MusicUtil.v2sTimeStampInMicroseconds(timeSlot, timeOfWrite);
- preparedInsert.setDefaultTimestamp(timestamp);
-
- ResultSet rs = session.execute(preparedInsert);
- result = rs.wasApplied();
-
- }
- catch (AlreadyExistsException ae) {
- // logger.error(EELFLoggerDelegate.errorLogger,"AlreadExistsException: " + ae.getMessage(),AppMessages.QUERYERROR,
- // ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("AlreadyExistsException: " + ae.getMessage(),ae);
- } catch ( InvalidQueryException e ) {
- // logger.error(EELFLoggerDelegate.errorLogger,"InvalidQueryException: " + e.getMessage(),AppMessages.SESSIONFAILED + " ["
- // + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("InvalidQueryException: " + e.getMessage(),e);
- } catch (Exception e) {
- // logger.error(EELFLoggerDelegate.errorLogger,e.getClass().toString() + ":" + e.getMessage(),AppMessages.SESSIONFAILED + " ["
- // + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR, e);
- throw new MusicServiceException("Executing Session Failure for Request = " + "["
- + queryObject.getQuery() + "]" + " Reason = " + e.getMessage(),e);
- }
- return result;
- }
-
- /* *//**
- * This method performs DDL operations on Cassandra using consistency level ONE.
- *
- * @param queryObject Object containing cassandra prepared query and values.
- * @return ResultSet
- * @throws MusicServiceException
- * @throws MusicQueryException
- *//*
- public ResultSet executeEventualGet(PreparedQueryObject queryObject)
- throws MusicServiceException, MusicQueryException {
- CacheAccess<String, PreparedStatement> queryBank = CachingUtil.getStatementBank();
- PreparedStatement preparedEventualGet = null;
- if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
- logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("Ill formed queryObject for the request = " + "["
- + queryObject.getQuery() + "]");
- }
- logger.info(EELFLoggerDelegate.applicationLogger,
- "Executing Eventual get query:" + queryObject.getQuery());
-
- ResultSet results = null;
- try {
- if(queryBank.get(queryObject.getQuery()) != null )
- preparedEventualGet=queryBank.get(queryObject.getQuery());
- else {
- preparedEventualGet = session.prepare(queryObject.getQuery());
- CachingUtil.updateStatementBank(queryObject.getQuery(), preparedEventualGet);
- }
- if(queryObject.getConsistency() == null) {
- preparedEventualGet.setConsistencyLevel(ConsistencyLevel.ONE);
- } else {
- preparedEventualGet.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
- }
- results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
-
- } catch (Exception ex) {
- logger.error("Exception", ex);
- logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicServiceException(ex.getMessage());
- }
- return results;
- }
-
- *//**
- *
- * This method performs DDL operation on Cassandra using consistency level QUORUM.
- *
- * @param queryObject Object containing cassandra prepared query and values.
- * @return ResultSet
- * @throws MusicServiceException
- * @throws MusicQueryException
- *//*
- public ResultSet executeCriticalGet(PreparedQueryObject queryObject)
- throws MusicServiceException, MusicQueryException {
- if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
- logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
- + queryObject.getQuery() + "]");
- }
- logger.info(EELFLoggerDelegate.applicationLogger,
- "Executing Critical get query:" + queryObject.getQuery());
- PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
- preparedEventualGet.setConsistencyLevel(ConsistencyLevel.QUORUM);
- ResultSet results = null;
- try {
- results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
- } catch (Exception ex) {
- logger.error("Exception", ex);
- logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicServiceException(ex.getMessage());
- }
- return results;
-
- }
- */
- public ResultSet executeGet(PreparedQueryObject queryObject,String consistencyLevel) throws MusicQueryException, MusicServiceException {
- if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
- logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
- throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
- + queryObject.getQuery() + "]");
- }
- ResultSet results = null;
- try {
- SimpleStatement statement = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
-
- if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_ONE)) {
- if(queryObject.getConsistency() == null) {
- statement.setConsistencyLevel(ConsistencyLevel.ONE);
- } else {
- statement.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
- }
- }
- else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_QUORUM)) {
- statement.setConsistencyLevel(ConsistencyLevel.QUORUM);
- }
-
- results = session.execute(statement);
-
- } catch (Exception ex) {
- logger.error(EELFLoggerDelegate.errorLogger, "Execute Get Error" + ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject
- .getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR, ex);
- throw new MusicServiceException("Execute Get Error" + ex.getMessage());
- }
-
- return results;
-
- }
-
- /**
- * This method performs DDL operations on Cassandra using consistency level ONE.
- *
- * @param queryObject Object containing cassandra prepared query and values.
- */
- public ResultSet executeOneConsistencyGet(PreparedQueryObject queryObject)
- throws MusicServiceException, MusicQueryException {
- return executeGet(queryObject, CONSISTENCY_LEVEL_ONE);
- }
-
- /**
- *
- * This method performs DDL operation on Cassandra using consistency level QUORUM.
- *
- * @param queryObject Object containing cassandra prepared query and values.
- */
- public ResultSet executeQuorumConsistencyGet(PreparedQueryObject queryObject)
- throws MusicServiceException, MusicQueryException {
- return executeGet(queryObject, CONSISTENCY_LEVEL_QUORUM);
- }
-
-}
Code Review / music.git / blobdiff