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