auth/auth-cass/src/main/java/org/onap/aaf/auth/dao/AbsCassDAO.java

   1 /**
   2  * ============LICENSE_START====================================================
   3  * org.onap.aaf
   4  * ===========================================================================
   5  * Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
   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  * ============LICENSE_END====================================================
  19  *
  20  */
  21
  22 package org.onap.aaf.auth.dao;
  23
  24 import java.io.IOException;
  25 import java.util.ArrayList;
  26 import java.util.Deque;
  27 import java.util.List;
  28 import java.util.concurrent.ConcurrentLinkedDeque;
  29
  30 import org.onap.aaf.auth.dao.cass.Status;
  31 import org.onap.aaf.auth.layer.Result;
  32 import org.onap.aaf.misc.env.APIException;
  33 import org.onap.aaf.misc.env.Env;
  34 import org.onap.aaf.misc.env.TimeTaken;
  35 import org.onap.aaf.misc.env.TransStore;
  36
  37 import com.datastax.driver.core.BoundStatement;
  38 import com.datastax.driver.core.Cluster;
  39 import com.datastax.driver.core.ConsistencyLevel;
  40 import com.datastax.driver.core.PreparedStatement;
  41 import com.datastax.driver.core.ResultSet;
  42 import com.datastax.driver.core.ResultSetFuture;
  43 import com.datastax.driver.core.Row;
  44 import com.datastax.driver.core.Session;
  45 import com.datastax.driver.core.exceptions.DriverException;
  46
  47 public abstract class AbsCassDAO<TRANS extends TransStore,DATA> {
  48         protected static final char DOT = '.';
  49         protected static final char DOT_PLUS_ONE = '.'+1;
  50         protected static final String FIRST_CHAR = Character.toString((char)0);
  51         protected static final String LAST_CHAR = Character.toString((char)Character.MAX_VALUE);
  52         protected static final int FIELD_COMMAS = 0;
  53         protected static final int QUESTION_COMMAS = 1;
  54         protected static final int ASSIGNMENT_COMMAS = 2;
  55         protected static final int WHERE_ANDS = 3;
  56
  57         private Cluster cluster;
  58         /*
  59          * From DataStax
  60          * com.datastax.driver.core.Session
  61                 A session holds connections to a Cassandra cluster, allowing it to be queried. Each session maintains multiple connections to the cluster nodes,
  62                 provides policies to choose which node to use for each query (round-robin on all nodes of the cluster by default), and handles retries for
  63                 failed query (when it makes sense), etc...
  64                 Session instances are thread-safe and usually a single instance is enough per application. However, a given session can only be set to one
  65                 keyspace at a time, so one instance per keyspace is necessary.
  66          */
  67         private Session session;
  68         private final String keyspace;
  69         // If this is null, then we own session
  70         private final AbsCassDAO<TRANS,?> owningDAO;
  71         protected Class<DATA> dataClass;
  72         private final String name;
  73 //      private static Slot sessionSlot; // not used since 2015
  74         private static final ArrayList<AbsCassDAO<? extends TransStore,?>.PSInfo> psinfos = new ArrayList<>();
  75         private static final List<Object> EMPTY = new ArrayList<>(0);
  76         private static final Deque<ResetRequest> resetDeque = new ConcurrentLinkedDeque<ResetRequest>();
  77         private static boolean resetTrigger = false;
  78         private static long nextAvailableReset = 0;
  79
  80         public AbsCassDAO(TRANS trans, String name, Cluster cluster, String keyspace, Class<DATA> dataClass) {
  81                 this.name = name;
  82                 this.cluster = cluster;
  83                 this.keyspace = keyspace;
  84                 owningDAO = null;  // we own session
  85                 session = null;
  86                 this.dataClass = dataClass;
  87         }
  88
  89         public AbsCassDAO(TRANS trans, String name, AbsCassDAO<TRANS,?> aDao, Class<DATA> dataClass) {
  90                 this.name = name;
  91                 cluster = aDao.cluster;
  92                 keyspace = aDao.keyspace;
  93                 session = null;
  94                 // We do not own session
  95                 owningDAO = aDao;
  96                 this.dataClass = dataClass;
  97         }
  98
  99 // Not used since 2015
 100 //      public static void setSessionSlot(Slot slot) {
 101 //              sessionSlot = slot;
 102 //      }
 103
 104         //Note: Lower case ON PURPOSE. These names used to create History Messages
 105         public enum CRUD {
 106                 create,read,update,delete;
 107         }
 108
 109         public class PSInfo {
 110                 private PreparedStatement ps;
 111                 private final int size;
 112                 private final Loader<DATA> loader;
 113                 private final CRUD crud; // Store CRUD, because it makes a difference in Object Order, see Loader
 114                 private final String cql;
 115                 private final ConsistencyLevel consistency;
 116
 117
 118                 /**
 119                  * Create a PSInfo and create Prepared Statement
 120                  *
 121                  * @param trans
 122                  * @param theCQL
 123                  * @param loader
 124                  */
 125                 public PSInfo(TRANS trans, String theCQL, Loader<DATA> loader, ConsistencyLevel consistency) {
 126                         this.loader = loader;
 127                         this.consistency=consistency;
 128                         psinfos.add(this);
 129
 130                         cql = theCQL.trim().toUpperCase();
 131                         if(cql.startsWith("INSERT")) {
 132                                 crud = CRUD.create;
 133                         } else if(cql.startsWith("UPDATE")) {
 134                                 crud = CRUD.update;
 135                         } else if(cql.startsWith("DELETE")) {
 136                                 crud = CRUD.delete;
 137                         } else {
 138                                 crud = CRUD.read;
 139                         }
 140
 141                         int idx = 0, count=0;
 142                         while((idx=cql.indexOf('?',idx))>=0) {
 143                                 ++idx;
 144                                 ++count;
 145                         }
 146                         size=count;
 147                 }
 148
 149                 public synchronized void reset() {
 150                         ps = null;
 151                 }
 152
 153                 private synchronized BoundStatement ps(TransStore trans) throws APIException, IOException {
 154                         /* From Datastax
 155                                 You should prepare only once, and cache the PreparedStatement in your application (it is thread-safe).
 156                                 If you call prepare multiple times with the same query string, the driver will log a warning.
 157                         */
 158                         if(ps==null) {
 159                                 TimeTaken tt = trans.start("Preparing PSInfo " + crud.toString().toUpperCase() + " on " + name,Env.SUB);
 160                                 try {
 161                                         ps = getSession(trans).prepare(cql);
 162                                         ps.setConsistencyLevel(consistency);
 163                                 } catch (DriverException e) {
 164                                         reportPerhapsReset(trans,e);
 165                                         throw e;
 166                                 } finally {
 167                                         tt.done();
 168                                 }
 169                         }
 170                         // BoundStatements are NOT threadsafe... need a new one each time.
 171                         return new BoundStatement(ps);
 172                 }
 173
 174                 /**
 175                  * Execute a Prepared Statement by extracting from DATA object
 176                  *
 177                  * @param trans
 178                  * @param text
 179                  * @param data
 180                  * @return
 181                  */
 182                 public Result<ResultSetFuture> execAsync(TRANS trans, String text, DATA data) {
 183                         TimeTaken tt = trans.start(text, Env.REMOTE);
 184                         try {
 185                                 return Result.ok(getSession(trans).executeAsync(
 186                                                 ps(trans).bind(loader.extract(data, size, crud))));
 187                         } catch (DriverException | APIException | IOException e) {
 188                                 AbsCassDAO.this.reportPerhapsReset(trans,e);
 189                                 return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
 190                         } finally {
 191                                 tt.done();
 192                         }
 193                 }
 194
 195                 /**
 196                  * Execute a Prepared Statement on Object[] key
 197                  *
 198                  * @param trans
 199                  * @param text
 200                  * @param objs
 201                  * @return
 202                  */
 203                 public Result<ResultSetFuture> execAsync(TRANS trans, String text, Object ... objs) {
 204                         TimeTaken tt = trans.start(text, Env.REMOTE);
 205                         try {
 206                                 return Result.ok(getSession(trans).executeAsync(ps(trans).bind(objs)));
 207                         } catch (DriverException | APIException | IOException e) {
 208                                 AbsCassDAO.this.reportPerhapsReset(trans,e);
 209                                 return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
 210                         } finally {
 211                                 tt.done();
 212                         }
 213                 }
 214
 215                 /*
 216                  * Note:
 217                  *
 218                  */
 219
 220                 /**
 221                  * Execute a Prepared Statement by extracting from DATA object
 222                  *
 223                  * @param trans
 224                  * @param text
 225                  * @param data
 226                  * @return
 227                  */
 228                 public Result<ResultSet> exec(TRANS trans, String text, DATA data) {
 229                         TimeTaken tt = trans.start(text, Env.REMOTE);
 230                         try {
 231                                 /*
 232                                  * "execute" (and executeAsync)
 233                                  * Executes the provided query.
 234                                         This method blocks until at least some result has been received from the database. However,
 235                                         for SELECT queries, it does not guarantee that the result has been received in full. But it
 236                                         does guarantee that some response has been received from the database, and in particular
 237                                         guarantee that if the request is invalid, an exception will be thrown by this method.
 238
 239                                         Parameters:
 240                                         statement - the CQL query to execute (that can be any Statement).
 241                                         Returns:
 242                                                 the result of the query. That result will never be null but can be empty (and will
 243                                                 be for any non SELECT query).
 244                                  */
 245                                 return Result.ok(getSession(trans).execute(
 246                                                 ps(trans).bind(loader.extract(data, size, crud))));
 247                         } catch (DriverException | APIException | IOException e) {
 248                                 AbsCassDAO.this.reportPerhapsReset(trans,e);
 249                                 return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
 250                         } finally {
 251                                 tt.done();
 252                         }
 253                 }
 254
 255                 /**
 256                  * Execute a Prepared Statement on Object[] key
 257                  *
 258                  * @param trans
 259                  * @param text
 260                  * @param objs
 261                  * @return
 262                  */
 263                 public Result<ResultSet> exec(TRANS trans, String text, Object ... objs) {
 264                         TimeTaken tt = trans.start(text, Env.REMOTE);
 265                         try {
 266                                 return Result.ok(getSession(trans).execute(ps(trans).bind(objs)));
 267                         } catch (DriverException | APIException | IOException e) {
 268                                 AbsCassDAO.this.reportPerhapsReset(trans,e);
 269                                 return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
 270                         } finally {
 271                                 tt.done();
 272                         }
 273                 }
 274
 275                 /**
 276                  * Read the Data from Cassandra given a Prepared Statement (defined by the
 277                  * DAO Instance)
 278                  *
 279                  * This is common behavior among all DAOs.
 280                  * @throws DAOException
 281                  */
 282                 public Result<List<DATA>> read(TRANS trans, String text, Object[] key) {
 283                         TimeTaken tt = trans.start(text,Env.REMOTE);
 284
 285                         ResultSet rs;
 286                         try {
 287                                 rs = getSession(trans).execute(key==null?ps(trans):ps(trans).bind(key));
 288 /// TEST CODE for Exception
 289 //                              boolean force = true;
 290 //                              if(force) {
 291 //                                      Map<InetSocketAddress, Throwable> misa = new HashMap<>();
 292 //                                      //misa.put(new InetSocketAddress(444),new Exception("no host was tried"));
 293 //                                      misa.put(new InetSocketAddress(444),new Exception("Connection has been closed"));
 294 //                                      throw new com.datastax.driver.core.exceptions.NoHostAvailableException(misa);
 295 ////                                    throw new com.datastax.driver.core.exceptions.AuthenticationException(new InetSocketAddress(9999),"no host was tried");
 296 //                              }
 297 //// END TEST CODE
 298                         } catch (DriverException | APIException | IOException e) {
 299                                 AbsCassDAO.this.reportPerhapsReset(trans,e);
 300                                 return Result.err(Status.ERR_Backend,"%s-%s executing %s",e.getClass().getName(),e.getMessage(), cql);
 301                         } finally {
 302                                 tt.done();
 303                         }
 304
 305                         return extract(loader,rs,null /*let Array be created if necessary*/,dflt);
 306                 }
 307
 308                 public Result<List<DATA>> read(TRANS trans, String text, DATA data) {
 309                         return read(trans,text, loader.extract(data, size, crud));
 310                 }
 311
 312                 public Object[] keyFrom(DATA data) {
 313                         return loader.extract(data, size, CRUD.delete); // Delete is key only
 314                 }
 315
 316                 /*
 317                  * Note: in case PSInfos are deleted, we want to remove them from list.  This is not expected,
 318                  * but we don't want a data leak if it does.  Finalize doesn't have to happen quickly
 319                  */
 320                 @Override
 321                 protected void finalize() throws Throwable {
 322                         psinfos.remove(this);
 323                 }
 324         }
 325
 326         protected final Accept<DATA> dflt = new Accept<DATA>() {
 327                 @Override
 328                 public boolean ok(DATA data) {
 329                         return true;
 330                 }
 331         };
 332
 333
 334         @SuppressWarnings("unchecked")
 335     protected final Result<List<DATA>> extract(Loader<DATA> loader, ResultSet rs, List<DATA> indata, Accept<DATA> accept) {
 336                 List<Row> rows = rs.all();
 337                 if(rows.isEmpty()) {
 338                         return Result.ok((List<DATA>)EMPTY); // Result sets now .emptyList(true);
 339                 } else {
 340                         DATA d;
 341                         List<DATA> data = indata==null?new ArrayList<>(rows.size()):indata;
 342
 343                         for(Row row : rows) {
 344                                 try {
 345                                         d = loader.load(dataClass.newInstance(),row);
 346                                         if(accept.ok(d)) {
 347                                                 data.add(d);
 348                                         }
 349                                 } catch(Exception e) {
 350                                         return Result.err(e);
 351                                 }
 352                         }
 353                         return Result.ok(data);
 354                 }
 355     }
 356
 357         private static final String NEW_CASSANDRA_SESSION_CREATED = "New Cassandra Session Created";
 358         private static final String NEW_CASSANDRA_CLUSTER_OBJECT_CREATED = "New Cassandra Cluster Object Created";
 359         private static final String NEW_CASSANDRA_SESSION = "New Cassandra Session";
 360         private static final Object LOCK = new Object();
 361
 362         private static class ResetRequest {
 363                 //package on purpose
 364                 Session session;
 365                 long timestamp;
 366
 367                 public ResetRequest(Session session) {
 368                         this.session = session;
 369                         timestamp = System.currentTimeMillis();
 370                 }
 371         }
 372
 373
 374         public static final void primePSIs(TransStore trans) throws APIException, IOException {
 375                 for(AbsCassDAO<? extends TransStore, ?>.PSInfo psi : psinfos) {
 376                         if(psi.ps==null) {
 377                                 psi.ps(trans);
 378                         }
 379                 }
 380         }
 381
 382         public final Session getSession(TransStore trans) throws APIException, IOException {
 383                 // SessionFilter unused since 2015
 384                 // Try to use Trans' session, if exists
 385 //              if(sessionSlot!=null) { // try to get from Trans
 386 //                      Session sess = trans.get(sessionSlot, null);
 387 //                      if(sess!=null) {
 388 //                              return sess;
 389 //                      }
 390 //              }
 391
 392                 // If there's an owning DAO, use it's session
 393                 if(owningDAO!=null) {
 394                         return owningDAO.getSession(trans);
 395                 }
 396
 397                 // OK, nothing else works... get our own.
 398                 if(session==null || resetTrigger) {
 399                         Cluster tempCluster = null;
 400                         Session tempSession = null;
 401                         try {
 402                                 synchronized(LOCK) {
 403                                         boolean reset = false;
 404                                         for(ResetRequest r : resetDeque) {
 405                                                 if(r.session == session) {
 406                                                         if(r.timestamp>nextAvailableReset) {
 407                                                                 reset=true;
 408                                                                 nextAvailableReset = System.currentTimeMillis() + 60000;
 409                                                                 tempCluster = cluster;
 410                                                                 tempSession = session;
 411                                                                 break;
 412                                                         } else {
 413                                                                 trans.warn().log("Cassandra Connection Reset Ignored: Recent Reset");
 414                                                         }
 415                                                 }
 416                                         }
 417
 418                                         if(reset || session == null) {
 419                                                 TimeTaken tt = trans.start(NEW_CASSANDRA_SESSION, Env.SUB);
 420                                                 try {
 421                                                         // Note: Maitrayee recommended not closing the cluster, just
 422                                                         // overwrite it. Jonathan 9/30/2016 assuming same for Session
 423                                                         // This was a bad idea.  Ran out of File Handles as I suspected, Jonathan
 424                                                         if(reset) {
 425                                                                 for(AbsCassDAO<? extends TransStore, ?>.PSInfo psi : psinfos) {
 426                                                                         psi.reset();
 427                                                                 }
 428                                                         }
 429                                                         if(reset || cluster==null) {
 430                                                                 cluster = CassAccess.cluster(trans, keyspace);
 431                                                                 trans.warn().log(NEW_CASSANDRA_CLUSTER_OBJECT_CREATED);
 432                                                         }
 433                                                         if(reset || session==null) {
 434                                                                 session = cluster.connect(keyspace);
 435                                                                 trans.warn().log(NEW_CASSANDRA_SESSION_CREATED);
 436                                                         }
 437                                                 } finally {
 438                                                         resetTrigger=false;
 439                                                         tt.done();
 440                                                 }
 441                                         }
 442                                 }
 443                         } finally {
 444                                 TimeTaken tt = trans.start("Clear Reset Deque", Env.SUB);
 445                                 try {
 446                                         resetDeque.clear();
 447                                         // Not clearing Session/Cluster appears to kill off FileHandles
 448                                         if(tempSession!=null && !tempSession.isClosed()) {
 449                                                 tempSession.close();
 450                                         }
 451                                         if(tempCluster!=null && !tempCluster.isClosed()) {
 452                                                 tempCluster.close();
 453                                         }
 454                                 } finally {
 455                                         tt.done();
 456                                 }
 457                         }
 458                 }
 459                 return session;
 460         }
 461
 462         public final boolean reportPerhapsReset(TransStore trans, Exception e) {
 463                 if(owningDAO!=null) {
 464                         return owningDAO.reportPerhapsReset(trans, e);
 465                 } else {
 466                         boolean rv = false;
 467                         if(CassAccess.isResetException(e)) {
 468                                 trans.warn().printf("Session Reset called for %s by %s ",session==null?"":session,e==null?"Mgmt Command":e.getClass().getName());
 469                                 resetDeque.addFirst(new ResetRequest(session));
 470                                 rv = resetTrigger = true;
 471                         }
 472                         trans.error().log(e);
 473                         return rv;
 474                 }
 475         }
 476
 477         public void close(TransStore trans) {
 478                 if(owningDAO==null) {
 479                         if(session!=null) {
 480                                 TimeTaken tt = trans.start("Cassandra Session Close", Env.SUB);
 481                                 try {
 482                                         session.close();
 483                                 } finally {
 484                                         tt.done();
 485                                 }
 486                                 session = null;
 487                         } else {
 488                                 trans.debug().log("close called(), Session already closed");
 489                         }
 490                 } else {
 491                         owningDAO.close(trans);
 492                 }
 493         }
 494
 495         protected void wasModified(TRANS trans, CRUD modified, DATA data, String ... override) {
 496         }
 497
 498         protected interface Accept<DATA> {
 499                 public boolean ok(DATA data);
 500         }
 501
 502 }
 503
 504
 505