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