music-core/src/main/java/org/onap/music/datastore/MusicDataStore.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  * ===================================================================
   7  *  Modifications Copyright (c) 2018-2019 IBM
   8  *  Modifications Copyright (c) 2019 Samsung
   9  * ===================================================================
  10  *  Licensed under the Apache License, Version 2.0 (the "License");
  11  *  you may not use this file except in compliance with the License.
  12  *  You may obtain a copy of the License at
  13  *
  14  *     http://www.apache.org/licenses/LICENSE-2.0
  15  *
  16  *  Unless required by applicable law or agreed to in writing, software
  17  *  distributed under the License is distributed on an "AS IS" BASIS,
  18  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  19  *  See the License for the specific language governing permissions and
  20  *  limitations under the License.
  21  *
  22  * ============LICENSE_END=============================================
  23  * ====================================================================
  24  */
  25
  26 package org.onap.music.datastore;
  27
  28 import java.nio.ByteBuffer;
  29 import java.util.HashMap;
  30 import java.util.Map;
  31
  32 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  33 import org.onap.music.eelf.logging.format.AppMessages;
  34 import org.onap.music.eelf.logging.format.ErrorSeverity;
  35 import org.onap.music.eelf.logging.format.ErrorTypes;
  36 import org.onap.music.exceptions.MusicQueryException;
  37 import org.onap.music.exceptions.MusicServiceException;
  38 import org.onap.music.lockingservice.cassandra.LockType;
  39 import org.onap.music.main.CipherUtil;
  40 import org.onap.music.main.MusicUtil;
  41 import com.datastax.driver.core.Cluster;
  42 import com.datastax.driver.core.ColumnDefinitions;
  43 import com.datastax.driver.core.ColumnDefinitions.Definition;
  44 import com.datastax.driver.core.ConsistencyLevel;
  45 import com.datastax.driver.core.DataType;
  46 import com.datastax.driver.core.HostDistance;
  47 import com.datastax.driver.core.KeyspaceMetadata;
  48 import com.datastax.driver.core.Metadata;
  49 import com.datastax.driver.core.PoolingOptions;
  50 import com.datastax.driver.core.ResultSet;
  51 import com.datastax.driver.core.Row;
  52 import com.datastax.driver.core.Session;
  53 import com.datastax.driver.core.SimpleStatement;
  54 import com.datastax.driver.core.SocketOptions;
  55 import com.datastax.driver.core.TableMetadata;
  56 import com.datastax.driver.core.exceptions.AlreadyExistsException;
  57 import com.datastax.driver.core.exceptions.InvalidQueryException;
  58 import com.datastax.driver.extras.codecs.enums.EnumNameCodec;
  59
  60 /**
  61  * @author nelson24
  62  *
  63  */
  64 public class MusicDataStore {
  65
  66     public static final String CONSISTENCY_LEVEL_ONE = "ONE";
  67     public static final String CONSISTENCY_LEVEL_QUORUM = "QUORUM";
  68     public static final String CONSISTENCY_LEVEL_LOCAL_QUORUM = "LOCAL_QUORUM";
  69     private Session session;
  70     private Cluster cluster;
  71
  72
  73     /**
  74      * Connect to default Cassandra address
  75      */
  76     public MusicDataStore() {
  77         try {
  78             connectToCassaCluster(MusicUtil.getMyCassaHost());
  79         } catch (MusicServiceException e) {
  80             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), e);
  81         }
  82     }
  83
  84
  85     /**
  86      * @param cluster
  87      * @param session
  88      */
  89     public MusicDataStore(Cluster cluster, Session session) {
  90         this.session = session;
  91         setCluster(cluster);
  92     }
  93
  94
  95     /**
  96      * @param session
  97      */
  98     public void setSession(Session session) {
  99         this.session = session;
 100     }
 101
 102     /**
 103      * @param session
 104      */
 105     public Session getSession() {
 106         return session;
 107     }
 108
 109     /**
 110      * @param cluster
 111      */
 112     public void setCluster(Cluster cluster) {
 113         EnumNameCodec<LockType> lockTypeCodec = new EnumNameCodec<LockType>(LockType.class);
 114         cluster.getConfiguration().getCodecRegistry().register(lockTypeCodec);
 115
 116         this.cluster = cluster;
 117     }
 118
 119     public Cluster getCluster() {
 120         return this.cluster;
 121     }
 122
 123
 124     private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicDataStore.class);
 125
 126
 127     /**
 128      *
 129      * @param remoteIp
 130      * @throws MusicServiceException
 131      */
 132     public MusicDataStore(String remoteIp) {
 133         try {
 134             connectToCassaCluster(remoteIp);
 135         } catch (MusicServiceException e) {
 136             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), e);
 137         }
 138     }
 139
 140     /**
 141      *
 142      */
 143     public void close() {
 144         session.close();
 145     }
 146
 147     /**
 148      * This method connects to cassandra cluster on specific address.
 149      *
 150      * @param address
 151      */
 152     private void connectToCassaCluster(String address) throws MusicServiceException {
 153         String[] addresses = null;
 154         addresses = address.split(",");
 155         PoolingOptions poolingOptions = new PoolingOptions();
 156         poolingOptions
 157         .setConnectionsPerHost(HostDistance.LOCAL,  4, 10)
 158         .setConnectionsPerHost(HostDistance.REMOTE, 2, 4);
 159
 160         Cluster cluster;
 161         if(MusicUtil.getCassName() != null && MusicUtil.getCassPwd() != null) {
 162             String cassPwd;
 163             if (MusicUtil.getCipherEncKey() != null && !("").equals(MusicUtil.getCipherEncKey())) {
 164                 cassPwd = CipherUtil.decryptPKC(MusicUtil.getCassPwd());
 165             } else {
 166                 cassPwd = MusicUtil.getCassPwd();
 167             }
 168             logger.info(EELFLoggerDelegate.applicationLogger,
 169                     "Building with credentials "+MusicUtil.getCassName()+" & "+ MusicUtil.getCassPwd());
 170             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 171                     .withCredentials(MusicUtil.getCassName(), cassPwd)
 172                     //.withLoadBalancingPolicy(new RoundRobinPolicy())
 173                     .withoutJMXReporting()
 174                     .withPoolingOptions(poolingOptions)
 175                     .withSocketOptions(
 176                             new SocketOptions().setConnectTimeoutMillis(MusicUtil.getCassandraConnectTimeOutMS())
 177                             .setReadTimeoutMillis(MusicUtil.getCassandraReadTimeOutMS()))
 178                     .addContactPoints(addresses).build();
 179         } else {
 180             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 181                     .withoutJMXReporting()
 182                     .withPoolingOptions(poolingOptions)
 183                     .withSocketOptions(new SocketOptions()
 184                             .setConnectTimeoutMillis(MusicUtil.getCassandraConnectTimeOutMS())
 185                             .setReadTimeoutMillis(MusicUtil.getCassandraReadTimeOutMS()))
 186                     .addContactPoints(addresses)
 187                     .build();
 188         }
 189
 190         this.setCluster(cluster);
 191         Metadata metadata = this.cluster.getMetadata();
 192         logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 193                 + metadata.getClusterName() + " at " + address);
 194
 195         try {
 196             session = this.cluster.connect();
 197         } catch (Exception ex) {
 198             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY,
 199                     ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE, ex);
 200             throw new MusicServiceException(
 201                     "Error while connecting to Cassandra cluster.. " + ex.getMessage());
 202         }
 203     }
 204
 205     /**
 206      *
 207      * @param keyspace
 208      * @param tableName
 209      * @param columnName
 210      * @return DataType
 211      */
 212     public DataType returnColumnDataType(String keyspace, String tableName, String columnName) {
 213         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 214         TableMetadata table = ks.getTable(tableName);
 215         return table.getColumn(columnName).getType();
 216
 217     }
 218
 219     /**
 220      *
 221      * @param keyspace
 222      * @param tableName
 223      * @return TableMetadata
 224      */
 225     public TableMetadata returnColumnMetadata(String keyspace, String tableName) {
 226         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 227         return ks.getTable(tableName);
 228     }
 229
 230     /**
 231      *
 232      * @param keyspace
 233      * @param tableName
 234      * @return TableMetadata
 235      */
 236     public KeyspaceMetadata returnKeyspaceMetadata(String keyspace) {
 237         return cluster.getMetadata().getKeyspace(keyspace);
 238     }
 239
 240
 241     /**
 242      * Utility function to return the Java specific object type.
 243      *
 244      * @param row
 245      * @param colName
 246      * @param colType
 247      * @return
 248      */
 249     public Object getColValue(Row row, String colName, DataType colType) {
 250
 251         switch (colType.getName()) {
 252             case VARCHAR:
 253                 return row.getString(colName);
 254             case UUID:
 255                 return row.getUUID(colName);
 256             case VARINT:
 257                 return row.getVarint(colName);
 258             case BIGINT:
 259                 return row.getLong(colName);
 260             case INT:
 261                 return row.getInt(colName);
 262             case FLOAT:
 263                 return row.getFloat(colName);
 264             case DOUBLE:
 265                 return row.getDouble(colName);
 266             case BOOLEAN:
 267                 return row.getBool(colName);
 268             case MAP:
 269                 return row.getMap(colName, String.class, String.class);
 270             case LIST:
 271                 return row.getList(colName, String.class);
 272             default:
 273                 return null;
 274         }
 275     }
 276
 277     public byte[] getBlobValue(Row row, String colName, DataType colType) {
 278         ByteBuffer bb = row.getBytes(colName);
 279         return bb.array();
 280     }
 281
 282     public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
 283         ColumnDefinitions colInfo = row.getColumnDefinitions();
 284
 285         for (Map.Entry<String, Object> entry : condition.entrySet()) {
 286             String colName = entry.getKey();
 287             DataType colType = colInfo.getType(colName);
 288             Object columnValue = getColValue(row, colName, colType);
 289             Object conditionValue = MusicUtil.convertToActualDataType(colType, entry.getValue());
 290             if (columnValue.equals(conditionValue) == false)
 291                 return false;
 292         }
 293         return true;
 294     }
 295
 296     /**
 297      * Utility function to store ResultSet values in to a MAP for output.
 298      *
 299      * @param results
 300      * @return MAP
 301      */
 302     public Map<String, HashMap<String, Object>> marshalData(ResultSet results) {
 303         Map<String, HashMap<String, Object>> resultMap =
 304                 new HashMap<>();
 305         int counter = 0;
 306         for (Row row : results) {
 307             ColumnDefinitions colInfo = row.getColumnDefinitions();
 308             HashMap<String, Object> resultOutput = new HashMap<>();
 309             for (Definition definition : colInfo) {
 310                 if (!(("vector_ts").equals(definition.getName()))) {
 311                     if(definition.getType().toString().toLowerCase().contains("blob")) {
 312                         resultOutput.put(definition.getName(),
 313                                 getBlobValue(row, definition.getName(), definition.getType()));
 314                     } else {
 315                         resultOutput.put(definition.getName(),
 316                                 getColValue(row, definition.getName(), definition.getType()));
 317                     }
 318                 }
 319             }
 320             resultMap.put("row " + counter, resultOutput);
 321             counter++;
 322         }
 323         return resultMap;
 324     }
 325
 326
 327     // Prepared Statements 1802 additions
 328
 329     public boolean executePut(PreparedQueryObject queryObject, String consistency)
 330             throws MusicServiceException, MusicQueryException {
 331         return executePut(queryObject, consistency, 0);
 332     }
 333     /**
 334      * This Method performs DDL and DML operations on Cassandra using specified consistency level
 335      *
 336      * @param queryObject Object containing cassandra prepared query and values.
 337      * @param consistency Specify consistency level for data synchronization across cassandra
 338      *        replicas
 339      * @return Boolean Indicates operation success or failure
 340      * @throws MusicServiceException
 341      * @throws MusicQueryException
 342      */
 343     public boolean executePut(PreparedQueryObject queryObject, String consistency,long timeSlot)
 344             throws MusicServiceException, MusicQueryException {
 345
 346         boolean result = false;
 347         long timeOfWrite = System.currentTimeMillis();
 348         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 349             logger.error(EELFLoggerDelegate.errorLogger, queryObject.getQuery(),AppMessages.QUERYERROR, ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 350             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 351                     + queryObject.getQuery() + "]");
 352         }
 353         logger.debug(EELFLoggerDelegate.applicationLogger,
 354                 "In preprared Execute Put: the actual insert query:"
 355                         + queryObject.getQuery() + "; the values"
 356                         + queryObject.getValues());
 357         SimpleStatement preparedInsert = null;
 358
 359         try {
 360             preparedInsert = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 361             if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
 362                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing critical put query");
 363                 preparedInsert.setConsistencyLevel(ConsistencyLevel.QUORUM);
 364             } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
 365                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing simple put query");
 366                 if(queryObject.getConsistency() == null)
 367                     preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 368                 else
 369                     preparedInsert.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 370             } else if (consistency.equalsIgnoreCase(MusicUtil.ONE)) {
 371                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 372             }  else if (consistency.equalsIgnoreCase(MusicUtil.QUORUM)) {
 373                 preparedInsert.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
 374             } else if (consistency.equalsIgnoreCase(MusicUtil.ALL)) {
 375                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ALL);
 376             }
 377             long timestamp = MusicUtil.v2sTimeStampInMicroseconds(timeSlot, timeOfWrite);
 378             preparedInsert.setDefaultTimestamp(timestamp);
 379
 380             ResultSet rs = session.execute(preparedInsert);
 381             result = rs.wasApplied();
 382         } catch (AlreadyExistsException ae) {
 383             throw new MusicServiceException("Already Exists Exception: " + ae.getMessage());
 384         } catch (InvalidQueryException e) {
 385             if (e.getMessage().contains("unconfigured table")) {
 386                 throw new MusicServiceException("Invalid Query Exception: " + e.getMessage());
 387             } else {
 388                 logger.info(EELFLoggerDelegate.applicationLogger, "Query Exception: " + e.getMessage(),
 389                         AppMessages.SESSIONFAILED + " [" + queryObject.getQuery() + "]", ErrorSeverity.INFO,
 390                         ErrorTypes.QUERYERROR, e);
 391                 throw new MusicServiceException("Query Exception: " + e.getMessage());
 392             }
 393         } catch (Exception e) {
 394             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),
 395                     AppMessages.SESSIONFAILED + " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR,
 396                     ErrorTypes.QUERYERROR, e);
 397             throw new MusicServiceException("Executing Session Failure for Request = " + "[" + queryObject.getQuery()
 398                     + "]" + " Reason = " + e.getMessage());
 399         }
 400
 401         return result;
 402     }
 403
 404     /*   *//**
 405      * This method performs DDL operations on Cassandra using consistency level ONE.
 406      *
 407      * @param queryObject Object containing cassandra prepared query and values.
 408      * @return ResultSet
 409      * @throws MusicServiceException
 410      * @throws MusicQueryException
 411      *//*
 412     public ResultSet executeEventualGet(PreparedQueryObject queryObject)
 413                     throws MusicServiceException, MusicQueryException {
 414         CacheAccess<String, PreparedStatement> queryBank = CachingUtil.getStatementBank();
 415         PreparedStatement preparedEventualGet = null;
 416         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 417             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 418             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 419                             + queryObject.getQuery() + "]");
 420         }
 421         logger.info(EELFLoggerDelegate.applicationLogger,
 422                         "Executing Eventual  get query:" + queryObject.getQuery());
 423
 424         ResultSet results = null;
 425         try {
 426             if(queryBank.get(queryObject.getQuery()) != null )
 427                 preparedEventualGet=queryBank.get(queryObject.getQuery());
 428             else {
 429                 preparedEventualGet = session.prepare(queryObject.getQuery());
 430                 CachingUtil.updateStatementBank(queryObject.getQuery(), preparedEventualGet);
 431             }
 432             if(queryObject.getConsistency() == null) {
 433                 preparedEventualGet.setConsistencyLevel(ConsistencyLevel.ONE);
 434             } else {
 435                 preparedEventualGet.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 436             }
 437             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 438
 439         } catch (Exception ex) {
 440             logger.error("Exception", ex);
 441             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 442             throw new MusicServiceException(ex.getMessage());
 443         }
 444         return results;
 445     }
 446
 447       *//**
 448       *
 449       * This method performs DDL operation on Cassandra using consistency level QUORUM.
 450       *
 451       * @param queryObject Object containing cassandra prepared query and values.
 452       * @return ResultSet
 453       * @throws MusicServiceException
 454       * @throws MusicQueryException
 455       *//*
 456     public ResultSet executeCriticalGet(PreparedQueryObject queryObject)
 457                     throws MusicServiceException, MusicQueryException {
 458         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 459             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 460             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 461                             + queryObject.getQuery() + "]");
 462         }
 463         logger.info(EELFLoggerDelegate.applicationLogger,
 464                         "Executing Critical get query:" + queryObject.getQuery());
 465         PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
 466         preparedEventualGet.setConsistencyLevel(ConsistencyLevel.QUORUM);
 467         ResultSet results = null;
 468         try {
 469             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 470         } catch (Exception ex) {
 471             logger.error("Exception", ex);
 472             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 473             throw new MusicServiceException(ex.getMessage());
 474         }
 475         return results;
 476
 477     }
 478        */
 479     public ResultSet executeGet(PreparedQueryObject queryObject,String consistencyLevel) throws MusicQueryException, MusicServiceException {
 480         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 481             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 482             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 483                     + queryObject.getQuery() + "]");
 484         }
 485         ResultSet results = null;
 486         try {
 487             SimpleStatement statement = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 488             if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_ONE)) {
 489                     statement.setConsistencyLevel(ConsistencyLevel.ONE);
 490             } else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_QUORUM)) {
 491                 statement.setConsistencyLevel(ConsistencyLevel.QUORUM);
 492             } else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_LOCAL_QUORUM)) {
 493                 statement.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
 494             }
 495
 496             results = session.execute(statement);
 497
 498         } catch (Exception ex) {
 499             logger.error(EELFLoggerDelegate.errorLogger, "Execute Get Error" + ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject
 500                     .getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR, ex);
 501             throw new MusicServiceException("Execute Get Error" + ex.getMessage());
 502         }
 503
 504         return results;
 505
 506     }
 507
 508     /**
 509      * This method performs DDL operations on Cassandra using consistency level ONE.
 510      *
 511      * @param queryObject Object containing cassandra prepared query and values.
 512      */
 513     public ResultSet executeOneConsistencyGet(PreparedQueryObject queryObject)
 514             throws MusicServiceException, MusicQueryException {
 515         return executeGet(queryObject, CONSISTENCY_LEVEL_ONE);
 516     }
 517
 518     /**
 519      *
 520      * This method performs DDL operation on Cassandra using consistency level LOCAL_QUORUM.
 521      *
 522      * @param queryObject Object containing cassandra prepared query and values.
 523      */
 524     public ResultSet executeLocalQuorumConsistencyGet(PreparedQueryObject queryObject)
 525             throws MusicServiceException, MusicQueryException {
 526         return executeGet(queryObject, CONSISTENCY_LEVEL_LOCAL_QUORUM);
 527     }
 528
 529     /**
 530      *
 531      * This method performs DDL operation on Cassandra using consistency level QUORUM.
 532      *
 533      * @param queryObject Object containing cassandra prepared query and values.
 534      */
 535     public ResultSet executeQuorumConsistencyGet(PreparedQueryObject queryObject)
 536             throws MusicServiceException, MusicQueryException {
 537         return executeGet(queryObject, CONSISTENCY_LEVEL_QUORUM);
 538     }
 539
 540 }