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