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 = CipherUtil.decryptPKC(MusicUtil.getCassPwd());
 163             logger.info(EELFLoggerDelegate.applicationLogger,
 164                     "Building with credentials "+MusicUtil.getCassName()+" & "+ MusicUtil.getCassPwd());
 165             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 166                     .withCredentials(MusicUtil.getCassName(), cassPwd)
 167                     //.withLoadBalancingPolicy(new RoundRobinPolicy())
 168                     .withoutJMXReporting()
 169                     .withPoolingOptions(poolingOptions)
 170                     .withSocketOptions(
 171                             new SocketOptions().setConnectTimeoutMillis(MusicUtil.getCassandraConnectTimeOutMS())
 172                             .setReadTimeoutMillis(MusicUtil.getCassandraReadTimeOutMS()))
 173                     .addContactPoints(addresses).build();
 174         } else {
 175             cluster = Cluster.builder().withPort(MusicUtil.getCassandraPort())
 176                     .withoutJMXReporting()
 177                     .withPoolingOptions(poolingOptions)
 178                     .withSocketOptions(new SocketOptions()
 179                             .setConnectTimeoutMillis(MusicUtil.getCassandraConnectTimeOutMS())
 180                             .setReadTimeoutMillis(MusicUtil.getCassandraReadTimeOutMS()))
 181                     .addContactPoints(addresses)
 182                     .build();
 183         }
 184
 185         this.setCluster(cluster);
 186         Metadata metadata = this.cluster.getMetadata();
 187         logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 188                 + metadata.getClusterName() + " at " + address);
 189
 190         try {
 191             session = this.cluster.connect();
 192         } catch (Exception ex) {
 193             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.CASSANDRACONNECTIVITY,
 194                     ErrorSeverity.ERROR, ErrorTypes.SERVICEUNAVAILABLE, ex);
 195             throw new MusicServiceException(
 196                     "Error while connecting to Cassandra cluster.. " + ex.getMessage());
 197         }
 198     }
 199
 200     /**
 201      *
 202      * @param keyspace
 203      * @param tableName
 204      * @param columnName
 205      * @return DataType
 206      */
 207     public DataType returnColumnDataType(String keyspace, String tableName, String columnName) {
 208         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 209         TableMetadata table = ks.getTable(tableName);
 210         return table.getColumn(columnName).getType();
 211
 212     }
 213
 214     /**
 215      *
 216      * @param keyspace
 217      * @param tableName
 218      * @return TableMetadata
 219      */
 220     public TableMetadata returnColumnMetadata(String keyspace, String tableName) {
 221         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 222         return ks.getTable(tableName);
 223     }
 224
 225     /**
 226      *
 227      * @param keyspace
 228      * @param tableName
 229      * @return TableMetadata
 230      */
 231     public KeyspaceMetadata returnKeyspaceMetadata(String keyspace) {
 232         return cluster.getMetadata().getKeyspace(keyspace);
 233     }
 234
 235
 236     /**
 237      * Utility function to return the Java specific object type.
 238      *
 239      * @param row
 240      * @param colName
 241      * @param colType
 242      * @return
 243      */
 244     public Object getColValue(Row row, String colName, DataType colType) {
 245
 246         switch (colType.getName()) {
 247             case VARCHAR:
 248                 return row.getString(colName);
 249             case UUID:
 250                 return row.getUUID(colName);
 251             case VARINT:
 252                 return row.getVarint(colName);
 253             case BIGINT:
 254                 return row.getLong(colName);
 255             case INT:
 256                 return row.getInt(colName);
 257             case FLOAT:
 258                 return row.getFloat(colName);
 259             case DOUBLE:
 260                 return row.getDouble(colName);
 261             case BOOLEAN:
 262                 return row.getBool(colName);
 263             case MAP:
 264                 return row.getMap(colName, String.class, String.class);
 265             case LIST:
 266                 return row.getList(colName, String.class);
 267             default:
 268                 return null;
 269         }
 270     }
 271
 272     public byte[] getBlobValue(Row row, String colName, DataType colType) {
 273         ByteBuffer bb = row.getBytes(colName);
 274         return bb.array();
 275     }
 276
 277     public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
 278         ColumnDefinitions colInfo = row.getColumnDefinitions();
 279
 280         for (Map.Entry<String, Object> entry : condition.entrySet()) {
 281             String colName = entry.getKey();
 282             DataType colType = colInfo.getType(colName);
 283             Object columnValue = getColValue(row, colName, colType);
 284             Object conditionValue = MusicUtil.convertToActualDataType(colType, entry.getValue());
 285             if (columnValue.equals(conditionValue) == false)
 286                 return false;
 287         }
 288         return true;
 289     }
 290
 291     /**
 292      * Utility function to store ResultSet values in to a MAP for output.
 293      *
 294      * @param results
 295      * @return MAP
 296      */
 297     public Map<String, HashMap<String, Object>> marshalData(ResultSet results) {
 298         Map<String, HashMap<String, Object>> resultMap =
 299                 new HashMap<>();
 300         int counter = 0;
 301         for (Row row : results) {
 302             ColumnDefinitions colInfo = row.getColumnDefinitions();
 303             HashMap<String, Object> resultOutput = new HashMap<>();
 304             for (Definition definition : colInfo) {
 305                 if (!(("vector_ts").equals(definition.getName()))) {
 306                     if(definition.getType().toString().toLowerCase().contains("blob")) {
 307                         resultOutput.put(definition.getName(),
 308                                 getBlobValue(row, definition.getName(), definition.getType()));
 309                     } else {
 310                         resultOutput.put(definition.getName(),
 311                                 getColValue(row, definition.getName(), definition.getType()));
 312                     }
 313                 }
 314             }
 315             resultMap.put("row " + counter, resultOutput);
 316             counter++;
 317         }
 318         return resultMap;
 319     }
 320
 321
 322     // Prepared Statements 1802 additions
 323
 324     public boolean executePut(PreparedQueryObject queryObject, String consistency)
 325             throws MusicServiceException, MusicQueryException {
 326         return executePut(queryObject, consistency, 0);
 327     }
 328     /**
 329      * This Method performs DDL and DML operations on Cassandra using specified consistency level
 330      *
 331      * @param queryObject Object containing cassandra prepared query and values.
 332      * @param consistency Specify consistency level for data synchronization across cassandra
 333      *        replicas
 334      * @return Boolean Indicates operation success or failure
 335      * @throws MusicServiceException
 336      * @throws MusicQueryException
 337      */
 338     public boolean executePut(PreparedQueryObject queryObject, String consistency,long timeSlot)
 339             throws MusicServiceException, MusicQueryException {
 340
 341         boolean result = false;
 342         long timeOfWrite = System.currentTimeMillis();
 343         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 344             logger.error(EELFLoggerDelegate.errorLogger, queryObject.getQuery(),AppMessages.QUERYERROR, ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 345             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 346                     + queryObject.getQuery() + "]");
 347         }
 348         logger.debug(EELFLoggerDelegate.applicationLogger,
 349                 "In preprared Execute Put: the actual insert query:"
 350                         + queryObject.getQuery() + "; the values"
 351                         + queryObject.getValues());
 352         SimpleStatement preparedInsert = null;
 353
 354         try {
 355             preparedInsert = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 356             if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
 357                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing critical put query");
 358                 preparedInsert.setConsistencyLevel(ConsistencyLevel.QUORUM);
 359             } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
 360                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing simple put query");
 361                 if(queryObject.getConsistency() == null)
 362                     preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 363                 else
 364                     preparedInsert.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 365             } else if (consistency.equalsIgnoreCase(MusicUtil.ONE)) {
 366                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 367             }  else if (consistency.equalsIgnoreCase(MusicUtil.QUORUM)) {
 368                 preparedInsert.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
 369             } else if (consistency.equalsIgnoreCase(MusicUtil.ALL)) {
 370                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ALL);
 371             }
 372             long timestamp = MusicUtil.v2sTimeStampInMicroseconds(timeSlot, timeOfWrite);
 373             preparedInsert.setDefaultTimestamp(timestamp);
 374
 375             ResultSet rs = session.execute(preparedInsert);
 376             result = rs.wasApplied();
 377         } catch (AlreadyExistsException ae) {
 378             throw new MusicServiceException("Already Exists Exception: " + ae.getMessage());
 379         } catch (InvalidQueryException e) {
 380             if (e.getMessage().contains("unconfigured table")) {
 381                 throw new MusicServiceException("Invalid Query Exception: " + e.getMessage());
 382             } else {
 383                 logger.info(EELFLoggerDelegate.applicationLogger, "Query Exception: " + e.getMessage(),
 384                         AppMessages.SESSIONFAILED + " [" + queryObject.getQuery() + "]", ErrorSeverity.INFO,
 385                         ErrorTypes.QUERYERROR, e);
 386                 throw new MusicServiceException("Query Exception: " + e.getMessage());
 387             }
 388         } catch (Exception e) {
 389             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),
 390                     AppMessages.SESSIONFAILED + " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR,
 391                     ErrorTypes.QUERYERROR, e);
 392             throw new MusicServiceException("Executing Session Failure for Request = " + "[" + queryObject.getQuery()
 393                     + "]" + " Reason = " + e.getMessage());
 394         }
 395
 396         return result;
 397     }
 398
 399     /*   *//**
 400      * This method performs DDL operations on Cassandra using consistency level ONE.
 401      *
 402      * @param queryObject Object containing cassandra prepared query and values.
 403      * @return ResultSet
 404      * @throws MusicServiceException
 405      * @throws MusicQueryException
 406      *//*
 407     public ResultSet executeEventualGet(PreparedQueryObject queryObject)
 408                     throws MusicServiceException, MusicQueryException {
 409         CacheAccess<String, PreparedStatement> queryBank = CachingUtil.getStatementBank();
 410         PreparedStatement preparedEventualGet = null;
 411         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 412             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 413             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 414                             + queryObject.getQuery() + "]");
 415         }
 416         logger.info(EELFLoggerDelegate.applicationLogger,
 417                         "Executing Eventual  get query:" + queryObject.getQuery());
 418
 419         ResultSet results = null;
 420         try {
 421             if(queryBank.get(queryObject.getQuery()) != null )
 422                 preparedEventualGet=queryBank.get(queryObject.getQuery());
 423             else {
 424                 preparedEventualGet = session.prepare(queryObject.getQuery());
 425                 CachingUtil.updateStatementBank(queryObject.getQuery(), preparedEventualGet);
 426             }
 427             if(queryObject.getConsistency() == null) {
 428                 preparedEventualGet.setConsistencyLevel(ConsistencyLevel.ONE);
 429             } else {
 430                 preparedEventualGet.setConsistencyLevel(MusicUtil.getConsistencyLevel(queryObject.getConsistency()));
 431             }
 432             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 433
 434         } catch (Exception ex) {
 435             logger.error("Exception", ex);
 436             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 437             throw new MusicServiceException(ex.getMessage());
 438         }
 439         return results;
 440     }
 441
 442       *//**
 443       *
 444       * This method performs DDL operation on Cassandra using consistency level QUORUM.
 445       *
 446       * @param queryObject Object containing cassandra prepared query and values.
 447       * @return ResultSet
 448       * @throws MusicServiceException
 449       * @throws MusicQueryException
 450       *//*
 451     public ResultSet executeCriticalGet(PreparedQueryObject queryObject)
 452                     throws MusicServiceException, MusicQueryException {
 453         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 454             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 455             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 456                             + queryObject.getQuery() + "]");
 457         }
 458         logger.info(EELFLoggerDelegate.applicationLogger,
 459                         "Executing Critical get query:" + queryObject.getQuery());
 460         PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
 461         preparedEventualGet.setConsistencyLevel(ConsistencyLevel.QUORUM);
 462         ResultSet results = null;
 463         try {
 464             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 465         } catch (Exception ex) {
 466             logger.error("Exception", ex);
 467             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 468             throw new MusicServiceException(ex.getMessage());
 469         }
 470         return results;
 471
 472     }
 473        */
 474     public ResultSet executeGet(PreparedQueryObject queryObject,String consistencyLevel) throws MusicQueryException, MusicServiceException {
 475         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 476             logger.error(EELFLoggerDelegate.errorLogger, "",AppMessages.QUERYERROR+ " [" + queryObject.getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR);
 477             throw new MusicQueryException("Error processing Prepared Query Object for the request = " + "["
 478                     + queryObject.getQuery() + "]");
 479         }
 480         ResultSet results = null;
 481         try {
 482             SimpleStatement statement = new SimpleStatement(queryObject.getQuery(), queryObject.getValues().toArray());
 483             if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_ONE)) {
 484                     statement.setConsistencyLevel(ConsistencyLevel.ONE);
 485             } else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_QUORUM)) {
 486                 statement.setConsistencyLevel(ConsistencyLevel.QUORUM);
 487             } else if (consistencyLevel.equalsIgnoreCase(CONSISTENCY_LEVEL_LOCAL_QUORUM)) {
 488                 statement.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
 489             }
 490
 491             results = session.execute(statement);
 492
 493         } catch (Exception ex) {
 494             logger.error(EELFLoggerDelegate.errorLogger, "Execute Get Error" + ex.getMessage(),AppMessages.UNKNOWNERROR+ "[" + queryObject
 495                     .getQuery() + "]", ErrorSeverity.ERROR, ErrorTypes.QUERYERROR, ex);
 496             throw new MusicServiceException("Execute Get Error" + ex.getMessage());
 497         }
 498
 499         return results;
 500
 501     }
 502
 503     /**
 504      * This method performs DDL operations on Cassandra using consistency level ONE.
 505      *
 506      * @param queryObject Object containing cassandra prepared query and values.
 507      */
 508     public ResultSet executeOneConsistencyGet(PreparedQueryObject queryObject)
 509             throws MusicServiceException, MusicQueryException {
 510         return executeGet(queryObject, CONSISTENCY_LEVEL_ONE);
 511     }
 512
 513     /**
 514      *
 515      * This method performs DDL operation on Cassandra using consistency level LOCAL_QUORUM.
 516      *
 517      * @param queryObject Object containing cassandra prepared query and values.
 518      */
 519     public ResultSet executeLocalQuorumConsistencyGet(PreparedQueryObject queryObject)
 520             throws MusicServiceException, MusicQueryException {
 521         return executeGet(queryObject, CONSISTENCY_LEVEL_LOCAL_QUORUM);
 522     }
 523
 524     /**
 525      *
 526      * This method performs DDL operation on Cassandra using consistency level QUORUM.
 527      *
 528      * @param queryObject Object containing cassandra prepared query and values.
 529      */
 530     public ResultSet executeQuorumConsistencyGet(PreparedQueryObject queryObject)
 531             throws MusicServiceException, MusicQueryException {
 532         return executeGet(queryObject, CONSISTENCY_LEVEL_QUORUM);
 533     }
 534
 535 }