2 * ============LICENSE_START====================================================
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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====================================================
22 package org.onap.aaf.auth.dao.cass;
24 import java.io.IOException;
25 import java.net.HttpURLConnection;
27 import java.util.Date;
28 import java.util.HashMap;
29 import java.util.HashSet;
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;
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;
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;
63 public class CacheInfoDAO extends CassDAOImpl<AuthzTrans,CacheInfoDAO.Data> implements CIDAO<AuthzTrans> {
65 private static final String TABLE = "cache";
66 public static final Map<String,Date[]> info = new ConcurrentHashMap<>();
68 private static CacheUpdate cacheUpdate;
70 // Hold current time stamps from Tables
71 private final Date startTime;
72 private PreparedStatement psCheck;
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();
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();
87 //////////////////////////////////////////
88 // Data Definition, matches Cassandra DM
89 //////////////////////////////////////////
90 private static final int KEYLIMIT = 2;
94 public static class Data {
99 public Data(String name, int seg) {
110 private static class InfoLoader extends Loader<Data> {
111 public static final InfoLoader dflt = new InfoLoader(KEYLIMIT);
113 public InfoLoader(int keylimit) {
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);
127 protected void key(Data data, int _idx, Object[] obj) {
135 protected void body(Data data, int idx, Object[] obj) {
136 obj[idx]=data.touched;
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");
148 public static<T extends Trans> void stopUpdate() {
149 if (cacheUpdate!=null) {
150 cacheUpdate.go=false;
154 private final static class CacheUpdate extends Thread {
155 public static BlockingQueue<Transfer> notifyDQ = new LinkedBlockingQueue<Transfer>(2000);
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;
160 private SecuritySetter<HttpURLConnection> ss;
161 private final String authority;
162 public boolean go = true;
164 public CacheUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
169 this.authority = ip+':'+port;
172 private static class Transfer {
175 public Transfer(String table, int[] segs) {
180 private class CacheClear extends Retryable<Integer> {
182 private AuthzTrans trans;
186 public CacheClear(AuthzTrans trans) {
190 public void set(Entry<String, IntHolder> es) {
192 segs = es.getValue().toString();
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())) {
203 trans.error().log("Error During AAF Peer Notify",f.code(),f.body());
210 private class IntHolder {
212 HashSet<Integer> set;
214 public IntHolder(int ints[]) {
218 public void add(int[] ints) {
220 set = new HashSet<>();
222 for (int i=0;i<raw.length;++i) {
226 for (int i=0;i<ints.length;++i) {
232 public String toString() {
233 StringBuilder sb = new StringBuilder();
234 boolean first = true;
245 for (Integer i : set) {
254 return sb.toString();
262 Transfer data = notifyDQ.poll(4,TimeUnit.SECONDS);
268 CacheClear cc = null;
269 Map<String,IntHolder> gather = null;
270 AuthzTrans trans = null;
272 // Do a block poll first
275 start = System.nanoTime();
276 trans = env.newTransNoAvg();
277 cc = new CacheClear(trans);
278 gather = new HashMap<>();
280 IntHolder prev = gather.get(data.table);
282 gather.put(data.table,new IntHolder(data.segs));
286 // continue while there is data
287 } while ((data = notifyDQ.poll())!=null);
289 for (Entry<String, IntHolder> es : gather.entrySet()) {
292 if (hman.all(ss, cc, false)!=null) {
295 } catch (Exception e) {
296 trans.error().log(e, "Error on Cache Update");
299 if (env.debug().isLoggable()) {
300 float millis = (System.nanoTime()-start)/1000000f;
301 StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");
302 sb.append("Updated ");
305 sb.append(" entry for ");
307 sb.append(" entries for ");
309 int peers = count<=0?0:cc.total/count;
311 sb.append(" client");
318 trans.auditTrail(0, sb, Env.REMOTE);
322 } catch (InterruptedException e1) {
324 Thread.currentThread().interrupt();
330 private void init(AuthzTrans trans) throws APIException, IOException {
332 String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
333 psCheck = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE);
335 disable(CRUD.create);
336 disable(CRUD.delete);
340 * @see org.onap.aaf.auth.dao.cass.CIDAO#touch(org.onap.aaf.auth.env.test.AuthzTrans, java.lang.String, int)
344 public Result<Void> touch(AuthzTrans trans, String name, int ... seg) {
346 // Direct Service Cache Invalidation
348 // ConcurrentQueues are open-ended. We don't want any Memory leaks
349 // Note: we keep a separate counter, because "size()" on a Linked Queue is expensive
350 if (cacheUpdate!=null) {
352 if (!CacheUpdate.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {
353 trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );
355 } catch (InterruptedException e) {
356 trans.error().log("Cache Notify Queue posting was interrupted" );
357 Thread.currentThread().interrupt();
362 // Table Based Cache Invalidation (original)
364 // Note: Save time with multiple Sequence Touches, but PreparedStmt doesn't support IN
365 StringBuilder start = new StringBuilder("CacheInfoDAO Touch segments ");
368 StringBuilder sb = new StringBuilder("BEGIN BATCH\n");
369 boolean first = true;
371 sb.append(UPDATE_SP);
373 sb.append(" SET touched=dateof(now()) WHERE name = '");
375 sb.append("' AND seg = ");
385 sb.append("APPLY BATCH;");
386 TimeTaken tt = trans.start(start.toString(),Env.REMOTE);
388 getSession(trans).executeAsync(sb.toString());
389 } catch (DriverException | APIException | IOException e) {
390 reportPerhapsReset(trans,e);
391 return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
399 * @see org.onap.aaf.auth.dao.cass.CIDAO#check(org.onap.aaf.auth.env.test.AuthzTrans)
402 public Result<Void> check(AuthzTrans trans) {
404 TimeTaken tt = trans.start("Check Table Timestamps",Env.REMOTE);
406 rs = getSession(trans).execute(new BoundStatement(psCheck));
407 } catch (DriverException | APIException | IOException e) {
408 reportPerhapsReset(trans,e);
409 return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
414 String lastName = null;
416 for (Row row : rs.all()) {
417 String name = row.getString(0);
418 int seg = row.getInt(1);
419 if (!name.equals(lastName)) {
420 dates = info.get(name);
424 dates=new Date[seg+1];
425 info.put(name,dates);
426 } else if (dates.length<=seg) {
427 Date[] temp = new Date[seg+1];
428 System.arraycopy(dates, 0, temp, 0, dates.length);
430 info.put(name, dates);
432 Date temp = row.getTimestamp(2);
433 if (dates[seg]==null || dates[seg].before(temp)) {
441 * @see org.onap.aaf.auth.dao.cass.CIDAO#get(java.lang.String, int)
444 public Date get(AuthzTrans trans, String table, int seg) {
445 Date[] dates = info.get(table);
447 dates = new Date[seg+1];
448 touch(trans,table, seg);
449 } else if (dates.length<=seg) {
450 Date[] temp = new Date[seg+1];
451 System.arraycopy(dates, 0, temp, 0, dates.length);
454 Date rv = dates[seg];
456 rv=dates[seg]=startTime;
462 protected void wasModified(AuthzTrans trans, CRUD modified, Data data, String ... override) {