auth/auth-cass/src/main/java/org/onap/aaf/auth/dao/cass/CacheInfoDAO.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.cass;
  23
  24 import java.io.IOException;
  25 import java.net.HttpURLConnection;
  26 import java.net.URI;
  27 import java.util.Date;
  28 import java.util.HashMap;
  29 import java.util.HashSet;
  30 import java.util.Map;
  31 import java.util.Map.Entry;
  32 import java.util.concurrent.BlockingQueue;
  33 import java.util.concurrent.ConcurrentHashMap;
  34 import java.util.concurrent.LinkedBlockingQueue;
  35 import java.util.concurrent.TimeUnit;
  36
  37 import org.onap.aaf.auth.dao.AbsCassDAO;
  38 import org.onap.aaf.auth.dao.CIDAO;
  39 import org.onap.aaf.auth.dao.CassAccess;
  40 import org.onap.aaf.auth.dao.CassDAOImpl;
  41 import org.onap.aaf.auth.dao.Loader;
  42 import org.onap.aaf.auth.env.AuthzEnv;
  43 import org.onap.aaf.auth.env.AuthzTrans;
  44 import org.onap.aaf.auth.layer.Result;
  45 import org.onap.aaf.cadi.CadiException;
  46 import org.onap.aaf.cadi.SecuritySetter;
  47 import org.onap.aaf.cadi.client.Future;
  48 import org.onap.aaf.cadi.client.Rcli;
  49 import org.onap.aaf.cadi.client.Retryable;
  50 import org.onap.aaf.cadi.http.HMangr;
  51 import org.onap.aaf.misc.env.APIException;
  52 import org.onap.aaf.misc.env.Env;
  53 import org.onap.aaf.misc.env.TimeTaken;
  54 import org.onap.aaf.misc.env.Trans;
  55
  56 import com.datastax.driver.core.BoundStatement;
  57 import com.datastax.driver.core.Cluster;
  58 import com.datastax.driver.core.PreparedStatement;
  59 import com.datastax.driver.core.ResultSet;
  60 import com.datastax.driver.core.Row;
  61 import com.datastax.driver.core.exceptions.DriverException;
  62
  63 public class CacheInfoDAO extends CassDAOImpl<AuthzTrans,CacheInfoDAO.Data> implements CIDAO<AuthzTrans> {
  64
  65         private static final String TABLE = "cache";
  66         public static final Map<String,Date[]> info = new ConcurrentHashMap<String,Date[]>();
  67
  68         private static CacheUpdate cacheUpdate;
  69
  70         // Hold current time stamps from Tables
  71         private final Date startTime;
  72         private PreparedStatement psCheck;
  73
  74         public CacheInfoDAO(AuthzTrans trans, Cluster cluster, String keyspace) throws APIException, IOException {
  75                 super(trans, CacheInfoDAO.class.getSimpleName(),cluster,keyspace,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
  76                 startTime = new Date();
  77                 init(trans);
  78         }
  79
  80         public CacheInfoDAO(AuthzTrans trans, AbsCassDAO<AuthzTrans,?> aDao) throws APIException, IOException {
  81                 super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
  82                 startTime = new Date();
  83                 init(trans);
  84         }
  85
  86
  87     //////////////////////////////////////////
  88     // Data Definition, matches Cassandra DM
  89     //////////////////////////////////////////
  90     private static final int KEYLIMIT = 2;
  91         /**
  92      * @author Jonathan
  93      */
  94         public static class Data {
  95                 public Data() {
  96                         name = null;
  97                         touched = null;
  98                 }
  99                 public Data(String name, int seg) {
 100                         this.name = name;
 101                         this.seg = seg;
 102                         touched = null;
 103                 }
 104
 105                 public String           name;
 106                 public int                      seg;
 107                 public Date                     touched;
 108     }
 109
 110     private static class InfoLoader extends Loader<Data> {
 111         public static final InfoLoader dflt = new InfoLoader(KEYLIMIT);
 112
 113                 public InfoLoader(int keylimit) {
 114                         super(keylimit);
 115                 }
 116
 117                 @Override
 118                 public Data load(Data data, Row row) {
 119                         // Int more efficient
 120                         data.name = row.getString(0);
 121                         data.seg = row.getInt(1);
 122                         data.touched = row.getTimestamp(2);
 123                         return data;
 124                 }
 125
 126                 @Override
 127                 protected void key(Data data, int _idx, Object[] obj) {
 128                         int idx = _idx;
 129
 130                         obj[idx]=data.name;
 131                         obj[++idx]=data.seg;
 132                 }
 133
 134                 @Override
 135                 protected void body(Data data, int idx, Object[] obj) {
 136                         obj[idx]=data.touched;
 137                 }
 138     }
 139
 140         public static<T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
 141                 if(cacheUpdate==null) {
 142                         Thread t= new Thread(cacheUpdate = new CacheUpdate(env,hman,ss, ip,port),"CacheInfo Update Thread");
 143                         t.setDaemon(true);
 144                         t.start();
 145                 }
 146         }
 147
 148         public static<T extends Trans> void stopUpdate() {
 149                 if(cacheUpdate!=null) {
 150                         cacheUpdate.go=false;
 151                 }
 152         }
 153
 154         private final static class CacheUpdate extends Thread {
 155                 public static BlockingQueue<Transfer> notifyDQ = new LinkedBlockingQueue<Transfer>(2000);
 156
 157                 private static final String VOID_CT="application/Void+json;q=1.0;charset=utf-8;version=2.0,application/json;q=1.0;version=2.0,*/*;q=1.0";
 158                 private AuthzEnv env;
 159                 private HMangr hman;
 160                 private SecuritySetter<HttpURLConnection> ss;
 161                 private final String authority;
 162                 public boolean go = true;
 163
 164                 public CacheUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
 165                         this.env = env;
 166                         this.hman = hman;
 167                         this.ss = ss;
 168
 169                         this.authority = ip+':'+port;
 170                 }
 171
 172                 private static class Transfer {
 173                         public String table;
 174                         public int segs[];
 175                         public Transfer(String table, int[] segs)  {
 176                                 this.table = table;
 177                                 this.segs = segs;
 178                         }
 179                 }
 180                 private class CacheClear extends Retryable<Integer> {
 181                         public int total=0;
 182                         private AuthzTrans trans;
 183                         private String type;
 184                         private String segs;
 185
 186                         public CacheClear(AuthzTrans trans) {
 187                                 this.trans = trans;
 188                         }
 189
 190                         public void set(Entry<String, IntHolder> es) {
 191                                 type = es.getKey();
 192                                 segs = es.getValue().toString();
 193                         }
 194
 195                 @Override
 196                         public Integer code(Rcli<?> client) throws APIException, CadiException {
 197                                 URI to = client.getURI();
 198                                 if(!to.getAuthority().equals(authority)) {
 199                                         Future<Void> f = client.delete("/mgmt/cache/"+type+'/'+segs,VOID_CT);
 200                                         if(f.get(hman.readTimeout())) {
 201                                             ++total;
 202                                         } else {
 203                                             trans.error().log("Error During AAF Peer Notify",f.code(),f.body());
 204                                         }
 205                                 }
 206                                 return total;
 207                         }
 208                 }
 209
 210                 private class IntHolder {
 211                         private int[] raw;
 212                         HashSet<Integer> set;
 213
 214                         public IntHolder(int ints[]) {
 215                                 raw = ints;
 216                                 set = null;
 217                         }
 218                         public void add(int[] ints) {
 219                                 if(set==null) {
 220                                         set = new HashSet<Integer>();
 221
 222                                         for(int i=0;i<raw.length;++i) {
 223                                                 set.add(raw[i]);
 224                                         }
 225                                 }
 226                                 for(int i=0;i<ints.length;++i) {
 227                                         set.add(ints[i]);
 228                                 }
 229                         }
 230
 231                         @Override
 232                         public String toString() {
 233                                 StringBuilder sb = new StringBuilder();
 234                                 boolean first = true;
 235                                 if(set==null) {
 236                                         for(int i : raw) {
 237                                                 if(first) {
 238                                                         first=false;
 239                                                 } else {
 240                                                         sb.append(',');
 241                                                 }
 242                                                 sb.append(i);
 243                                         }
 244                                 } else {
 245                                         for(Integer i : set) {
 246                                                 if(first) {
 247                                                         first=false;
 248                                                 } else {
 249                                                         sb.append(',');
 250                                                 }
 251                                                 sb.append(i);
 252                                         }
 253                                 }
 254                                 return sb.toString();
 255                         }
 256                 }
 257
 258                 @Override
 259                 public void run() {
 260                         do {
 261                                 try {
 262                                         Transfer data = notifyDQ.poll(4,TimeUnit.SECONDS);
 263                                         if(data==null) {
 264                                                 continue;
 265                                         }
 266
 267                                         int count = 0;
 268                                         CacheClear cc = null;
 269                                         Map<String,IntHolder> gather = null;
 270                                         AuthzTrans trans = null;
 271                                         long start=0;
 272                                         // Do a block poll first
 273                                         do {
 274                                                 if(gather==null) {
 275                                                         start = System.nanoTime();
 276                                                         trans = env.newTransNoAvg();
 277                                                         cc = new CacheClear(trans);
 278                                                         gather = new HashMap<String,IntHolder>();
 279                                                 }
 280                                                 IntHolder prev = gather.get(data.table);
 281                                                 if(prev==null) {
 282                                                         gather.put(data.table,new IntHolder(data.segs));
 283                                                 } else {
 284                                                         prev.add(data.segs);
 285                                                 }
 286                                                 // continue while there is data
 287                                         } while((data = notifyDQ.poll())!=null);
 288                                         if(gather!=null) {
 289                                                 for(Entry<String, IntHolder> es : gather.entrySet()) {
 290                                                         cc.set(es);
 291                                                         try {
 292                                                                 if(hman.all(ss, cc, false)!=null) {
 293                                                                         ++count;
 294                                                                 }
 295                                                         } catch (Exception e) {
 296                                                                 trans.error().log(e, "Error on Cache Update");
 297                                                         }
 298                                                 }
 299                                                 if(env.debug().isLoggable()) {
 300                                                         float millis = (System.nanoTime()-start)/1000000f;
 301                                                         StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");
 302                                                         sb.append("Updated ");
 303                                                         sb.append(count);
 304                                                         if(count==1) {
 305                                                                 sb.append(" entry for ");
 306                                                         } else {
 307                                                                 sb.append(" entries for ");
 308                                                         }
 309                                                         int peers = count<=0?0:cc.total/count;
 310                                                         sb.append(peers);
 311                                                         sb.append(" client");
 312                                                         if(peers!=1) {
 313                                                                 sb.append('s');
 314                                                         }
 315                                                         sb.append(" in ");
 316                                                         sb.append(millis);
 317                                                         sb.append("ms");
 318                                                         trans.auditTrail(0, sb, Env.REMOTE);
 319                                                         env.debug().log(sb);
 320                                                 }
 321                                         }
 322                                 } catch (InterruptedException e1) {
 323                                         go = false;
 324                                 }
 325                         } while(go);
 326                 }
 327         }
 328
 329         private void init(AuthzTrans trans) throws APIException, IOException {
 330
 331                 String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
 332                 psCheck = getSession(trans).prepare(SELECT_SP +  helpers[FIELD_COMMAS] + " FROM " + TABLE);
 333
 334                 disable(CRUD.create);
 335                 disable(CRUD.delete);
 336         }
 337
 338         /* (non-Javadoc)
 339          * @see org.onap.aaf.auth.dao.cass.CIDAO#touch(org.onap.aaf.auth.env.test.AuthzTrans, java.lang.String, int)
 340          */
 341
 342         @Override
 343         public Result<Void> touch(AuthzTrans trans, String name, int ... seg) {
 344                 /////////////
 345                 // Direct Service Cache Invalidation
 346                 /////////////
 347                 // ConcurrentQueues are open-ended.  We don't want any Memory leaks
 348                 // Note: we keep a separate counter, because "size()" on a Linked Queue is expensive
 349                 if(cacheUpdate!=null) {
 350                         try {
 351                                 if(!CacheUpdate.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {
 352                                         trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );
 353                                 }
 354                         } catch (InterruptedException e) {
 355                                 trans.error().log("Cache Notify Queue posting was interrupted" );
 356                         }
 357                 }
 358
 359                 /////////////
 360                 // Table Based Cache Invalidation (original)
 361                 /////////////
 362                 // Note: Save time with multiple Sequence Touches, but PreparedStmt doesn't support IN
 363                 StringBuilder start = new StringBuilder("CacheInfoDAO Touch segments ");
 364                 start.append(name);
 365                 start.append(": ");
 366                 StringBuilder sb = new StringBuilder("BEGIN BATCH\n");
 367                 boolean first = true;
 368                 for(int s : seg) {
 369                         sb.append(UPDATE_SP);
 370                         sb.append(TABLE);
 371                         sb.append(" SET touched=dateof(now()) WHERE name = '");
 372                         sb.append(name);
 373                         sb.append("' AND seg = ");
 374                         sb.append(s);
 375                         sb.append(";\n");
 376                         if(first) {
 377                                 first =false;
 378                         } else {
 379                                 start.append(',');
 380                         }
 381                         start.append(s);
 382                 }
 383                 sb.append("APPLY BATCH;");
 384                 TimeTaken tt = trans.start(start.toString(),Env.REMOTE);
 385                 try {
 386                         getSession(trans).executeAsync(sb.toString());
 387                 } catch (DriverException | APIException | IOException e) {
 388                         reportPerhapsReset(trans,e);
 389                         return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
 390                 } finally {
 391                         tt.done();
 392                 }
 393                 return Result.ok();
 394         }
 395
 396         /* (non-Javadoc)
 397          * @see org.onap.aaf.auth.dao.cass.CIDAO#check(org.onap.aaf.auth.env.test.AuthzTrans)
 398          */
 399         @Override
 400         public Result<Void> check(AuthzTrans trans) {
 401                 ResultSet rs;
 402                 TimeTaken tt = trans.start("Check Table Timestamps",Env.REMOTE);
 403                 try {
 404                         rs = getSession(trans).execute(new BoundStatement(psCheck));
 405                 } catch (DriverException | APIException | IOException e) {
 406                         reportPerhapsReset(trans,e);
 407                         return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
 408                 } finally {
 409                         tt.done();
 410                 }
 411
 412                 String lastName = null;
 413                 Date[] dates = null;
 414                 for(Row row : rs.all()) {
 415                         String name = row.getString(0);
 416                         int seg = row.getInt(1);
 417                         if(!name.equals(lastName)) {
 418                                 dates = info.get(name);
 419                                 lastName=name;
 420                         }
 421                         if(dates==null) {
 422                                 dates=new Date[seg+1];
 423                                 info.put(name,dates);
 424                         } else if(dates.length<=seg) {
 425                                 Date[] temp = new Date[seg+1];
 426                                 System.arraycopy(dates, 0, temp, 0, dates.length);
 427                                 dates = temp;
 428                                 info.put(name, dates);
 429                         }
 430                         Date temp = row.getTimestamp(2);
 431                         if(dates[seg]==null || dates[seg].before(temp)) {
 432                                 dates[seg]=temp;
 433                         }
 434                 }
 435                 return Result.ok();
 436         }
 437
 438     /* (non-Javadoc)
 439          * @see org.onap.aaf.auth.dao.cass.CIDAO#get(java.lang.String, int)
 440          */
 441     @Override
 442         public Date get(AuthzTrans trans, String table, int seg) {
 443                 Date[] dates = info.get(table);
 444                 if(dates==null) {
 445                         dates = new Date[seg+1];
 446                         touch(trans,table, seg);
 447                 } else if(dates.length<=seg) {
 448                         Date[] temp = new Date[seg+1];
 449                         System.arraycopy(dates, 0, temp, 0, dates.length);
 450                         dates = temp;
 451                 }
 452                 Date rv = dates[seg];
 453                 if(rv==null) {
 454                         rv=dates[seg]=startTime;
 455                 }
 456                 return rv;
 457         }
 458
 459         @Override
 460         protected void wasModified(AuthzTrans trans, CRUD modified, Data data, String ... override) {
 461                 // Do nothing
 462         }
 463
 464 }