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