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