music-rest/src/main/java/org/onap/music/conductor/conditionals/MusicConditional.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  * ===================================================================
   7  *  Modifications Copyright (C) 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.conductor.conditionals;
  27
  28 import java.io.PrintWriter;
  29 import java.io.StringWriter;
  30 import java.util.HashMap;
  31 import java.util.Map;
  32
  33 import org.codehaus.jettison.json.JSONObject;
  34 import org.onap.music.datastore.MusicDataStoreHandle;
  35 import org.onap.music.datastore.PreparedQueryObject;
  36 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  37 import org.onap.music.eelf.logging.format.AppMessages;
  38 import org.onap.music.eelf.logging.format.ErrorSeverity;
  39 import org.onap.music.eelf.logging.format.ErrorTypes;
  40 import org.onap.music.exceptions.MusicLockingException;
  41 import org.onap.music.exceptions.MusicQueryException;
  42 import org.onap.music.exceptions.MusicServiceException;
  43 import org.onap.music.main.MusicCore;
  44 import org.onap.music.main.MusicUtil;
  45 import org.onap.music.main.ResultType;
  46 import org.onap.music.main.ReturnType;
  47 import org.onap.music.rest.RestMusicDataAPI;
  48
  49 import com.datastax.driver.core.ColumnDefinitions;
  50 import com.datastax.driver.core.DataType;
  51 import com.datastax.driver.core.ResultSet;
  52 import com.datastax.driver.core.Row;
  53 import com.datastax.driver.core.TableMetadata;
  54
  55 public class MusicConditional {
  56     private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(RestMusicDataAPI.class);
  57
  58     public static ReturnType conditionalInsert(String keyspace, String tablename, String casscadeColumnName,
  59             Map<String, Object> casscadeColumnData, String primaryKey, Map<String, Object> valuesMap,
  60             Map<String, String> status) throws Exception {
  61
  62         Map<String, PreparedQueryObject> queryBank = new HashMap<>();
  63         TableMetadata tableInfo = null;
  64         tableInfo = MusicDataStoreHandle.returnColumnMetadata(keyspace, tablename);
  65         DataType primaryIdType = tableInfo.getPrimaryKey().get(0).getType();
  66         String primaryId = tableInfo.getPrimaryKey().get(0).getName();
  67         DataType casscadeColumnType = tableInfo.getColumn(casscadeColumnName).getType();
  68         String vector = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
  69
  70         PreparedQueryObject select = new PreparedQueryObject();
  71         select.appendQueryString("SELECT * FROM " + keyspace + "." + tablename + " where " + primaryId + " = ?");
  72         select.addValue(MusicUtil.convertToActualDataType(primaryIdType, primaryKey));
  73         queryBank.put(MusicUtil.SELECT, select);
  74
  75         PreparedQueryObject update = new PreparedQueryObject();
  76         //casscade column values
  77         Map<String, String> updateColumnvalues = getValues(true, casscadeColumnData, status);
  78         Object formatedValues = MusicUtil.convertToActualDataType(casscadeColumnType, updateColumnvalues);
  79         update.appendQueryString("UPDATE " + keyspace + "." + tablename + " SET " + casscadeColumnName + " ="
  80                 + casscadeColumnName + " + ? , vector_ts = ?" + " WHERE " + primaryId + " = ? ");
  81         update.addValue(formatedValues);
  82         update.addValue(MusicUtil.convertToActualDataType(DataType.text(), vector));
  83         update.addValue(MusicUtil.convertToActualDataType(primaryIdType, primaryKey));
  84         queryBank.put(MusicUtil.UPDATE, update);
  85
  86
  87         //casscade column values
  88         Map<String, String> insertColumnvalues = getValues(false, casscadeColumnData, status);
  89         formatedValues = MusicUtil.convertToActualDataType(casscadeColumnType, insertColumnvalues);
  90         PreparedQueryObject insert = extractQuery(valuesMap, tableInfo, tablename, keyspace, primaryId, primaryKey,casscadeColumnName,formatedValues);
  91         queryBank.put(MusicUtil.INSERT, insert);
  92
  93
  94         String key = keyspace + "." + tablename + "." + primaryKey;
  95         String lockId;
  96         try {
  97             lockId = MusicCore.createLockReferenceAtomic(key);
  98         } catch (MusicLockingException e) {
  99             return new ReturnType(ResultType.FAILURE, e.getMessage());
 100         }
 101         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 102         ReturnType lockAcqResult = MusicCore.acquireLockWithLease(key, lockId, leasePeriod);
 103
 104         try {
 105             if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 106                 ReturnType criticalPutResult = conditionalInsertAtomic(lockId, keyspace, tablename, primaryKey,
 107                         queryBank);
 108                 MusicCore.destroyLockRef(lockId);
 109                 if (criticalPutResult.getMessage().contains("insert"))
 110                     criticalPutResult
 111                             .setMessage("Insert values: ");
 112                 else if (criticalPutResult.getMessage().contains("update"))
 113                     criticalPutResult
 114                             .setMessage("Update values: " + updateColumnvalues);
 115                 return criticalPutResult;
 116
 117             } else {
 118                 MusicCore.destroyLockRef(lockId);
 119                 return lockAcqResult;
 120             }
 121         } catch (Exception e) {
 122             logger.error(EELFLoggerDelegate.applicationLogger, e);
 123             MusicCore.destroyLockRef(lockId);
 124             return new ReturnType(ResultType.FAILURE, e.getMessage());
 125         }
 126
 127     }
 128
 129     public static ReturnType conditionalInsertAtomic(String lockId, String keyspace, String tableName,
 130             String primaryKey, Map<String, PreparedQueryObject> queryBank) {
 131
 132         ResultSet results = null;
 133
 134         try {
 135             String fullyQualifiedKey = keyspace + "." + tableName + "." + primaryKey;
 136             ReturnType lockAcqResult = MusicCore.acquireLock(fullyQualifiedKey, lockId);
 137             if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 138                 try {
 139                     results = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(queryBank.get(MusicUtil.SELECT));
 140                 } catch (Exception e) {
 141                     logger.error(EELFLoggerDelegate.applicationLogger, e);
 142                     return new ReturnType(ResultType.FAILURE, e.getMessage());
 143                 }
 144                 if (results.all().isEmpty()) {
 145                     PreparedQueryObject qObject = queryBank.get(MusicUtil.INSERT);
 146                     qObject.setOperation(MusicUtil.INSERT);
 147                     logger.info(EELFLoggerDelegate.debugLogger,"### Conditional Insert");
 148                     MusicCore.criticalPut(keyspace, tableName, primaryKey, qObject, lockId, null);
 149                     //MusicDataStoreHandle.getDSHandle().executePut(queryBank.get(MusicUtil.INSERT), "critical");
 150                     return new ReturnType(ResultType.SUCCESS, MusicUtil.INSERT);
 151
 152                 } else {
 153                     PreparedQueryObject qObject = queryBank.get(MusicUtil.UPDATE);
 154                     qObject.setOperation(MusicUtil.UPDATE);
 155                     logger.info(EELFLoggerDelegate.debugLogger,"### Condition Update");
 156                     MusicCore.criticalPut(keyspace, tableName, primaryKey, qObject, lockId, null);
 157                     //MusicDataStoreHandle.getDSHandle().executePut(queryBank.get(MusicUtil.UPDATE), "critical");
 158                     return new ReturnType(ResultType.SUCCESS, MusicUtil.UPDATE);
 159                 }
 160             } else {
 161                 return new ReturnType(ResultType.FAILURE,
 162                         "Cannot perform operation since you are the not the lock holder");
 163             }
 164
 165         } catch (Exception e) {
 166             StringWriter sw = new StringWriter();
 167             e.printStackTrace(new PrintWriter(sw));
 168             String exceptionAsString = sw.toString();
 169             logger.error(EELFLoggerDelegate.applicationLogger, e);
 170             return new ReturnType(ResultType.FAILURE,
 171                     "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 172                             + exceptionAsString);
 173         }
 174
 175     }
 176
 177     public static ReturnType update(UpdateDataObject dataObj)
 178             throws MusicLockingException, MusicQueryException, MusicServiceException {
 179
 180         String key = dataObj.getKeyspace() + "." + dataObj.getTableName() + "." + dataObj.getPrimaryKeyValue();
 181         String lockId = MusicCore.createLockReferenceAtomic(key);
 182         long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 183         ReturnType lockAcqResult = MusicCore.acquireLockWithLease(key, lockId, leasePeriod);
 184
 185         try {
 186
 187             if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 188                 ReturnType criticalPutResult = updateAtomic(new UpdateDataObject().setLockId(lockId)
 189                         .setKeyspace(dataObj.getKeyspace())
 190                         .setTableName( dataObj.getTableName())
 191                         .setPrimaryKey(dataObj.getPrimaryKey())
 192                         .setPrimaryKeyValue(dataObj.getPrimaryKeyValue())
 193                         .setQueryBank(dataObj.getQueryBank())
 194                         .setPlanId(dataObj.getPlanId())
 195                         .setCascadeColumnValues(dataObj.getCascadeColumnValues())
 196                         .setCascadeColumnName(dataObj.getCascadeColumnName()));
 197
 198                 MusicCore.destroyLockRef(lockId);
 199                 return criticalPutResult;
 200             } else {
 201                 MusicCore.destroyLockRef(lockId);
 202                 return lockAcqResult;
 203             }
 204
 205         } catch (Exception e) {
 206             MusicCore.destroyLockRef(lockId);
 207             logger.error(EELFLoggerDelegate.applicationLogger, e);
 208             return new ReturnType(ResultType.FAILURE, e.getMessage());
 209
 210         }
 211     }
 212
 213     public static ReturnType updateAtomic(UpdateDataObject dataObj) {
 214         try {
 215             String fullyQualifiedKey = dataObj.getKeyspace() + "." + dataObj.getTableName() + "." + dataObj.getPrimaryKeyValue();
 216             ReturnType lockAcqResult = MusicCore.acquireLock(fullyQualifiedKey, dataObj.getLockId());
 217
 218             if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 219                 Row row  = MusicDataStoreHandle.getDSHandle().executeQuorumConsistencyGet(dataObj.getQueryBank().get(MusicUtil.SELECT)).one();
 220
 221                 if(row != null) {
 222                     Map<String, String> updatedValues = cascadeColumnUpdateSpecific(row, dataObj.getCascadeColumnValues(), dataObj.getCascadeColumnName(), dataObj.getPlanId());
 223                     JSONObject json = new JSONObject(updatedValues);
 224                     PreparedQueryObject update = new PreparedQueryObject();
 225                     String vector_ts = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
 226                     update.appendQueryString("UPDATE " + dataObj.getKeyspace() + "." + dataObj.getTableName() + " SET "
 227                             + dataObj.getCascadeColumnName() + "['" + dataObj.getPlanId()
 228                             + "'] = ?, vector_ts = ? WHERE " + dataObj.getPrimaryKey() + " = ?");
 229                     update.addValue(MusicUtil.convertToActualDataType(DataType.text(), json.toString()));
 230                     update.addValue(MusicUtil.convertToActualDataType(DataType.text(), vector_ts));
 231                     update.addValue(MusicUtil.convertToActualDataType(DataType.text(), dataObj.getPrimaryKeyValue()));
 232                     try {
 233                         update.setOperation(MusicUtil.UPDATE);
 234                         MusicCore.criticalPut(dataObj.keyspace, dataObj.tableName, dataObj.primaryKeyValue, update, dataObj.lockId, null);
 235                     } catch (Exception ex) {
 236                         logger.error(EELFLoggerDelegate.applicationLogger, ex);
 237                         return new ReturnType(ResultType.FAILURE, ex.getMessage());
 238                     }
 239                 }else {
 240                     return new ReturnType(ResultType.FAILURE,"Cannot find data related to key: "+dataObj.getPrimaryKey());
 241                 }
 242                 PreparedQueryObject qObject = dataObj.getQueryBank().get(MusicUtil.UPSERT);
 243                 qObject.setOperation(MusicUtil.INSERT);
 244                 MusicCore.criticalPut(dataObj.keyspace, dataObj.tableName, dataObj.primaryKeyValue, qObject, dataObj.lockId, null);
 245                 return new ReturnType(ResultType.SUCCESS, "update success");
 246             } else {
 247                 return new ReturnType(ResultType.FAILURE,
 248                         "Cannot perform operation since you are the not the lock holder");
 249             }
 250
 251         } catch (Exception e) {
 252             StringWriter sw = new StringWriter();
 253             e.printStackTrace(new PrintWriter(sw));
 254             String exceptionAsString = sw.toString();
 255             logger.error(EELFLoggerDelegate.applicationLogger, e);
 256             return new ReturnType(ResultType.FAILURE,
 257                     "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 258                             + exceptionAsString);
 259         }
 260
 261     }
 262
 263     @SuppressWarnings("unchecked")
 264     public static Map<String, String> getValues(boolean isExists, Map<String, Object> casscadeColumnData,
 265             Map<String, String> status) {
 266
 267         Map<String, String> returnMap = new HashMap<>();
 268         Object key = casscadeColumnData.get("key");
 269         String setStatus = "";
 270         Map<String, String> value = (Map<String, String>) casscadeColumnData.get("value");
 271
 272         if (isExists)
 273             setStatus = status.get("exists");
 274         else
 275             setStatus = status.get("nonexists");
 276
 277         value.put("status", setStatus);
 278         JSONObject valueJson = new JSONObject(value);
 279         returnMap.put(key.toString(), valueJson.toString());
 280         return returnMap;
 281
 282     }
 283
 284     public static PreparedQueryObject extractQuery(Map<String, Object> valuesMap, TableMetadata tableInfo, String tableName,
 285             String keySpaceName,String primaryKeyName,String primaryKey,String casscadeColumn,Object casscadeColumnValues) throws Exception {
 286
 287         PreparedQueryObject queryObject = new PreparedQueryObject();
 288         StringBuilder fieldsString = new StringBuilder("(vector_ts"+",");
 289         StringBuilder valueString = new StringBuilder("(" + "?" + ",");
 290         String vector = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
 291         String localPrimaryKey;
 292         queryObject.addValue(vector);
 293         if(casscadeColumn!=null && casscadeColumnValues!=null) {
 294             fieldsString.append(casscadeColumn).append(" ,");
 295             valueString.append("?,");
 296             queryObject.addValue(casscadeColumnValues);
 297         }
 298
 299         int counter = 0;
 300         for (Map.Entry<String, Object> entry : valuesMap.entrySet()) {
 301
 302             fieldsString.append(entry.getKey());
 303             Object valueObj = entry.getValue();
 304             if (primaryKeyName.equals(entry.getKey())) {
 305                 localPrimaryKey = entry.getValue() + "";
 306                 localPrimaryKey = localPrimaryKey.replace("'", "''");
 307             }
 308             DataType colType = null;
 309             try {
 310                 colType = tableInfo.getColumn(entry.getKey()).getType();
 311             } catch(NullPointerException ex) {
 312                 logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() +" Invalid column name : "+entry.getKey(),
 313                     AppMessages.INCORRECTDATA  ,ErrorSeverity.CRITICAL, ErrorTypes.DATAERROR, ex);
 314             }
 315
 316             Object formattedValue = null;
 317             try {
 318                 formattedValue = MusicUtil.convertToActualDataType(colType, valueObj);
 319             } catch (Exception e) {
 320                 logger.error(EELFLoggerDelegate.errorLogger,e.getMessage(), e);
 321             }
 322
 323             valueString.append("?");
 324             queryObject.addValue(formattedValue);
 325
 326
 327             if (counter == valuesMap.size() - 1) {
 328                 fieldsString.append(")");
 329                 valueString.append(")");
 330             } else {
 331                 fieldsString.append(",");
 332                 valueString.append(",");
 333             }
 334             counter = counter + 1;
 335         }
 336         queryObject.appendQueryString("INSERT INTO " + keySpaceName + "." + tableName + " "
 337                 + fieldsString + " VALUES " + valueString + ";");
 338         return queryObject;
 339     }
 340
 341     public static Object getColValue(Row row, String colName, DataType colType) {
 342         switch (colType.getName()) {
 343         case VARCHAR:
 344             return row.getString(colName);
 345         case UUID:
 346             return row.getUUID(colName);
 347         case VARINT:
 348             return row.getVarint(colName);
 349         case BIGINT:
 350             return row.getLong(colName);
 351         case INT:
 352             return row.getInt(colName);
 353         case FLOAT:
 354             return row.getFloat(colName);
 355         case DOUBLE:
 356             return row.getDouble(colName);
 357         case BOOLEAN:
 358             return row.getBool(colName);
 359         case MAP:
 360             return row.getMap(colName, String.class, String.class);
 361         default:
 362             return null;
 363         }
 364     }
 365
 366     @SuppressWarnings("unchecked")
 367     public static Map<String, String> cascadeColumnUpdateSpecific(Row row, Map<String, String> changeOfStatus,
 368             String cascadeColumnName, String planId) {
 369
 370         ColumnDefinitions colInfo = row.getColumnDefinitions();
 371         DataType colType = colInfo.getType(cascadeColumnName);
 372         Object columnValue = getColValue(row, cascadeColumnName, colType);
 373
 374         Map<String, String> finalValues = new HashMap<>();
 375         Map<String, String> values = (Map<String, String>) columnValue;
 376         if (values != null && values.keySet().contains(planId)) {
 377             String valueString = values.get(planId);
 378             String tempValueString = valueString.replaceAll("\\{", "").replaceAll("\"", "").replaceAll("\\}", "");
 379             String[] elements = tempValueString.split(",");
 380             for (String str : elements) {
 381                 String[] keyValue = str.split(":");
 382                 if ((changeOfStatus.keySet().contains(keyValue[0].replaceAll("\\s", ""))))
 383                 keyValue[1] = changeOfStatus.get(keyValue[0].replaceAll("\\s", ""));
 384                 finalValues.put(keyValue[0], keyValue[1]);
 385             }
 386         }
 387         return finalValues;
 388
 389     }
 390
 391 }