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  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *
  11  *     http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *
  19  * ============LICENSE_END=============================================
  20  * ====================================================================
  21  */
  22 package org.onap.music.datastore;
  23
  24 import java.net.InetAddress;
  25 import java.net.NetworkInterface;
  26 import java.net.SocketException;
  27 import java.util.ArrayList;
  28 import java.util.Enumeration;
  29 import java.util.HashMap;
  30 import java.util.Iterator;
  31 import java.util.Map;
  32 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  33 import org.onap.music.exceptions.MusicQueryException;
  34 import org.onap.music.exceptions.MusicServiceException;
  35 import org.onap.music.main.MusicUtil;
  36 import org.onap.music.main.ResultType;
  37
  38 import com.datastax.driver.core.Cluster;
  39 import com.datastax.driver.core.ColumnDefinitions;
  40 import com.datastax.driver.core.ColumnDefinitions.Definition;
  41 import com.datastax.driver.core.ConsistencyLevel;
  42 import com.datastax.driver.core.DataType;
  43 import com.datastax.driver.core.KeyspaceMetadata;
  44 import com.datastax.driver.core.Metadata;
  45 import com.datastax.driver.core.PreparedStatement;
  46 import com.datastax.driver.core.ResultSet;
  47 import com.datastax.driver.core.Row;
  48 import com.datastax.driver.core.Session;
  49 import com.datastax.driver.core.TableMetadata;
  50 import com.datastax.driver.core.exceptions.AlreadyExistsException;
  51 import com.datastax.driver.core.exceptions.InvalidQueryException;
  52 import com.datastax.driver.core.exceptions.NoHostAvailableException;
  53
  54 /**
  55  * @author nelson24
  56  *
  57  */
  58 public class MusicDataStore {
  59
  60     private Session session;
  61     private Cluster cluster;
  62
  63
  64
  65     /**
  66      * @param session
  67      */
  68     public void setSession(Session session) {
  69         this.session = session;
  70     }
  71
  72     /**
  73      * @param session
  74      */
  75     public Session getSession() {
  76         return session;
  77     }
  78
  79     /**
  80      * @param cluster
  81      */
  82     public void setCluster(Cluster cluster) {
  83         this.cluster = cluster;
  84     }
  85
  86
  87
  88     private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MusicDataStore.class);
  89
  90     /**
  91      *
  92      */
  93     public MusicDataStore() {
  94         connectToCassaCluster();
  95     }
  96
  97
  98     /**
  99      * @param cluster
 100      * @param session
 101      */
 102     public MusicDataStore(Cluster cluster, Session session) {
 103         this.session = session;
 104         this.cluster = cluster;
 105     }
 106
 107     /**
 108      *
 109      * @param remoteIp
 110      * @throws MusicServiceException
 111      */
 112     public MusicDataStore(String remoteIp) {
 113         try {
 114             connectToCassaCluster(remoteIp);
 115         } catch (MusicServiceException e) {
 116             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage());
 117         }
 118     }
 119
 120     /**
 121      *
 122      * @return
 123      */
 124     private ArrayList<String> getAllPossibleLocalIps() {
 125         ArrayList<String> allPossibleIps = new ArrayList<String>();
 126         try {
 127             Enumeration<NetworkInterface> en = NetworkInterface.getNetworkInterfaces();
 128             while (en.hasMoreElements()) {
 129                 NetworkInterface ni = (NetworkInterface) en.nextElement();
 130                 Enumeration<InetAddress> ee = ni.getInetAddresses();
 131                 while (ee.hasMoreElements()) {
 132                     InetAddress ia = (InetAddress) ee.nextElement();
 133                     allPossibleIps.add(ia.getHostAddress());
 134                 }
 135             }
 136         } catch (SocketException e) {
 137             logger.error(EELFLoggerDelegate.errorLogger, e.getMessage());
 138         }
 139         return allPossibleIps;
 140     }
 141
 142     /**
 143      * This method iterates through all available IP addresses and connects to multiple cassandra
 144      * clusters.
 145      */
 146     private void connectToCassaCluster() {
 147         Iterator<String> it = getAllPossibleLocalIps().iterator();
 148         String address = "localhost";
 149         logger.info(EELFLoggerDelegate.applicationLogger,
 150                         "Connecting to cassa cluster: Iterating through possible ips:"
 151                                         + getAllPossibleLocalIps());
 152         while (it.hasNext()) {
 153             try {
 154                 cluster = Cluster.builder().withPort(9042)
 155                                 .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
 156                                 .addContactPoint(address).build();
 157                 Metadata metadata = cluster.getMetadata();
 158                 logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 159                                 + metadata.getClusterName() + " at " + address);
 160                 session = cluster.connect();
 161
 162                 break;
 163             } catch (NoHostAvailableException e) {
 164                 address = it.next();
 165                 logger.error(EELFLoggerDelegate.errorLogger, e.getMessage());
 166             }
 167         }
 168     }
 169
 170     /**
 171      *
 172      */
 173     public void close() {
 174         session.close();
 175     }
 176
 177     /**
 178      * This method connects to cassandra cluster on specific address.
 179      *
 180      * @param address
 181      */
 182     private void connectToCassaCluster(String address) throws MusicServiceException {
 183         cluster = Cluster.builder().withPort(9042)
 184                         .withCredentials(MusicUtil.getCassName(), MusicUtil.getCassPwd())
 185                         .addContactPoint(address).build();
 186         Metadata metadata = cluster.getMetadata();
 187         logger.info(EELFLoggerDelegate.applicationLogger, "Connected to cassa cluster "
 188                         + metadata.getClusterName() + " at " + address);
 189         try {
 190             session = cluster.connect();
 191         } catch (Exception ex) {
 192             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage());
 193             throw new MusicServiceException(
 194                             "Error while connecting to Cassandra cluster.. " + ex.getMessage());
 195         }
 196     }
 197
 198     /**
 199      *
 200      * @param keyspace
 201      * @param tableName
 202      * @param columnName
 203      * @return DataType
 204      */
 205     public DataType returnColumnDataType(String keyspace, String tableName, String columnName) {
 206         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 207         TableMetadata table = ks.getTable(tableName);
 208         return table.getColumn(columnName).getType();
 209
 210     }
 211
 212     /**
 213      *
 214      * @param keyspace
 215      * @param tableName
 216      * @return TableMetadata
 217      */
 218     public TableMetadata returnColumnMetadata(String keyspace, String tableName) {
 219         KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(keyspace);
 220         return ks.getTable(tableName);
 221     }
 222
 223
 224     /**
 225      * Utility function to return the Java specific object type.
 226      *
 227      * @param row
 228      * @param colName
 229      * @param colType
 230      * @return
 231      */
 232     public Object getColValue(Row row, String colName, DataType colType) {
 233
 234         switch (colType.getName()) {
 235             case VARCHAR:
 236                 return row.getString(colName);
 237             case UUID:
 238                 return row.getUUID(colName);
 239             case VARINT:
 240                 return row.getVarint(colName);
 241             case BIGINT:
 242                 return row.getLong(colName);
 243             case INT:
 244                 return row.getInt(colName);
 245             case FLOAT:
 246                 return row.getFloat(colName);
 247             case DOUBLE:
 248                 return row.getDouble(colName);
 249             case BOOLEAN:
 250                 return row.getBool(colName);
 251             case MAP:
 252                 return row.getMap(colName, String.class, String.class);
 253             default:
 254                 return null;
 255         }
 256     }
 257
 258     public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition) throws Exception {
 259         ColumnDefinitions colInfo = row.getColumnDefinitions();
 260
 261         for (Map.Entry<String, Object> entry : condition.entrySet()) {
 262             String colName = entry.getKey();
 263             DataType colType = colInfo.getType(colName);
 264             Object columnValue = getColValue(row, colName, colType);
 265             Object conditionValue = MusicUtil.convertToActualDataType(colType, entry.getValue());
 266             if (columnValue.equals(conditionValue) == false)
 267                 return false;
 268         }
 269         return true;
 270     }
 271
 272     /**
 273      * Utility function to store ResultSet values in to a MAP for output.
 274      *
 275      * @param results
 276      * @return MAP
 277      */
 278     public Map<String, HashMap<String, Object>> marshalData(ResultSet results) {
 279         Map<String, HashMap<String, Object>> resultMap =
 280                         new HashMap<String, HashMap<String, Object>>();
 281         int counter = 0;
 282         for (Row row : results) {
 283             ColumnDefinitions colInfo = row.getColumnDefinitions();
 284             HashMap<String, Object> resultOutput = new HashMap<String, Object>();
 285             for (Definition definition : colInfo) {
 286                 if (!definition.getName().equals("vector_ts"))
 287                     resultOutput.put(definition.getName(),
 288                                     getColValue(row, definition.getName(), definition.getType()));
 289             }
 290             resultMap.put("row " + counter, resultOutput);
 291             counter++;
 292         }
 293         return resultMap;
 294     }
 295
 296
 297     // Prepared Statements 1802 additions
 298     /**
 299      * This Method performs DDL and DML operations on Cassandra using specified consistency level
 300      *
 301      * @param queryObject Object containing cassandra prepared query and values.
 302      * @param consistency Specify consistency level for data synchronization across cassandra
 303      *        replicas
 304      * @return Boolean Indicates operation success or failure
 305      * @throws MusicServiceException
 306      * @throws MusicQueryException
 307      */
 308     public boolean executePut(PreparedQueryObject queryObject, String consistency)
 309                     throws MusicServiceException, MusicQueryException {
 310
 311         boolean result = false;
 312
 313         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 314             logger.error(EELFLoggerDelegate.errorLogger,
 315                             "Error while processing prepared query object");
 316             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 317                             + queryObject.getQuery() + "]");
 318         }
 319         logger.info(EELFLoggerDelegate.applicationLogger,
 320                         "In preprared Execute Put: the actual insert query:"
 321                                         + queryObject.getQuery() + "; the values"
 322                                         + queryObject.getValues());
 323         PreparedStatement preparedInsert = null;
 324         try {
 325                 preparedInsert = session.prepare(queryObject.getQuery());
 326         } catch(InvalidQueryException iqe) {
 327                 logger.error(EELFLoggerDelegate.errorLogger, iqe.getMessage());
 328                 throw new MusicQueryException(iqe.getMessage());
 329         }
 330
 331         try {
 332             if (consistency.equalsIgnoreCase(MusicUtil.CRITICAL)) {
 333                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing critical put query");
 334                 preparedInsert.setConsistencyLevel(ConsistencyLevel.QUORUM);
 335             } else if (consistency.equalsIgnoreCase(MusicUtil.EVENTUAL)) {
 336                 logger.info(EELFLoggerDelegate.applicationLogger, "Executing simple put query");
 337                 preparedInsert.setConsistencyLevel(ConsistencyLevel.ONE);
 338             }
 339
 340             ResultSet rs = session.execute(preparedInsert.bind(queryObject.getValues().toArray()));
 341             result = rs.wasApplied();
 342
 343         }
 344         catch (AlreadyExistsException ae) {
 345                 logger.error(EELFLoggerDelegate.errorLogger, "Executing Session Failure for Request = "
 346                     + "[" + queryObject.getQuery() + "]" + " Reason = " + ae.getMessage());
 347                 throw new MusicServiceException(ae.getMessage());
 348         }
 349         catch (Exception e) {
 350             logger.error(EELFLoggerDelegate.errorLogger, "Executing Session Failure for Request = "
 351                             + "[" + queryObject.getQuery() + "]" + " Reason = " + e.getMessage());
 352             throw new MusicServiceException("Executing Session Failure for Request = " + "["
 353                             + queryObject.getQuery() + "]" + " Reason = " + e.getMessage());
 354         }
 355
 356
 357         return result;
 358     }
 359
 360     /**
 361      * This method performs DDL operations on Cassandra using consistency level ONE.
 362      *
 363      * @param queryObject Object containing cassandra prepared query and values.
 364      * @return ResultSet
 365      * @throws MusicServiceException
 366      * @throws MusicQueryException
 367      */
 368     public ResultSet executeEventualGet(PreparedQueryObject queryObject)
 369                     throws MusicServiceException, MusicQueryException {
 370
 371         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 372             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 373                             + queryObject.getQuery() + "]");
 374         }
 375         logger.info(EELFLoggerDelegate.applicationLogger,
 376                         "Executing Eventual  get query:" + queryObject.getQuery());
 377         PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
 378         preparedEventualGet.setConsistencyLevel(ConsistencyLevel.ONE);
 379         ResultSet results = null;
 380         try {
 381             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 382
 383         } catch (Exception ex) {
 384             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage());
 385             throw new MusicServiceException(ex.getMessage());
 386         }
 387         return results;
 388     }
 389
 390     /**
 391      *
 392      * This method performs DDL operation on Cassandra using consistency level QUORUM.
 393      *
 394      * @param queryObject Object containing cassandra prepared query and values.
 395      * @return ResultSet
 396      * @throws MusicServiceException
 397      * @throws MusicQueryException
 398      */
 399     public ResultSet executeCriticalGet(PreparedQueryObject queryObject)
 400                     throws MusicServiceException, MusicQueryException {
 401         if (!MusicUtil.isValidQueryObject(!queryObject.getValues().isEmpty(), queryObject)) {
 402             logger.error(EELFLoggerDelegate.errorLogger, "Error processing Prepared Query Object");
 403             throw new MusicQueryException("Ill formed queryObject for the request = " + "["
 404                             + queryObject.getQuery() + "]");
 405         }
 406         logger.info(EELFLoggerDelegate.applicationLogger,
 407                         "Executing Critical get query:" + queryObject.getQuery());
 408         PreparedStatement preparedEventualGet = session.prepare(queryObject.getQuery());
 409         preparedEventualGet.setConsistencyLevel(ConsistencyLevel.QUORUM);
 410         ResultSet results = null;
 411         try {
 412             results = session.execute(preparedEventualGet.bind(queryObject.getValues().toArray()));
 413         } catch (Exception ex) {
 414             logger.error(EELFLoggerDelegate.errorLogger, ex.getMessage());
 415             throw new MusicServiceException(ex.getMessage());
 416         }
 417         return results;
 418
 419     }
 420
 421 }