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.config.Config;
51 import org.onap.aaf.cadi.http.HMangr;
52 import org.onap.aaf.misc.env.APIException;
53 import org.onap.aaf.misc.env.Env;
54 import org.onap.aaf.misc.env.TimeTaken;
55 import org.onap.aaf.misc.env.Trans;
57 import com.datastax.driver.core.BoundStatement;
58 import com.datastax.driver.core.Cluster;
59 import com.datastax.driver.core.PreparedStatement;
60 import com.datastax.driver.core.ResultSet;
61 import com.datastax.driver.core.Row;
62 import com.datastax.driver.core.exceptions.DriverException;
64 public class CacheInfoDAO extends CassDAOImpl<AuthzTrans,CacheInfoDAO.Data> implements CIDAO<AuthzTrans> {
66 private static final String TABLE = "cache";
67 public static final Map<String,Date[]> info = new ConcurrentHashMap<>();
69 private static CacheUpdate cacheUpdate;
71 // Hold current time stamps from Tables
72 private final Date startTime;
73 private final boolean cacheNotify;
74 private PreparedStatement psCheck;
76 public CacheInfoDAO(AuthzTrans trans, Cluster cluster, String keyspace) throws APIException, IOException {
77 super(trans, CacheInfoDAO.class.getSimpleName(),cluster,keyspace,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
78 startTime = new Date();
79 cacheNotify = noK8s(trans);
83 public CacheInfoDAO(AuthzTrans trans, AbsCassDAO<AuthzTrans,?> aDao) throws APIException, IOException {
84 super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
85 startTime = new Date();
87 cacheNotify = noK8s(trans);
91 * Need a different point to point cache clear strategy for K8s...
95 private boolean noK8s(AuthzTrans trans) {
96 String container = trans.getProperty(Config.AAF_LOCATOR_CONTAINER);
97 return ! ("helm".equals(container) || "oom".equals(container));
100 //////////////////////////////////////////
101 // Data Definition, matches Cassandra DM
102 //////////////////////////////////////////
103 private static final int KEYLIMIT = 2;
107 public static class Data {
112 public Data(String name, int seg) {
123 private static class InfoLoader extends Loader<Data> {
124 public static final InfoLoader dflt = new InfoLoader(KEYLIMIT);
126 public InfoLoader(int keylimit) {
131 public Data load(Data data, Row row) {
132 // Int more efficient
133 data.name = row.getString(0);
134 data.seg = row.getInt(1);
135 data.touched = row.getTimestamp(2);
140 protected void key(Data data, int _idx, Object[] obj) {
148 protected void body(Data data, int idx, Object[] obj) {
149 obj[idx]=data.touched;
153 public static synchronized <T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
154 if (cacheUpdate==null) {
155 Thread t= new Thread(cacheUpdate = new CacheUpdate(env,hman,ss, ip,port),"CacheInfo Update Thread");
161 public static<T extends Trans> void stopUpdate() {
162 if (cacheUpdate!=null) {
163 cacheUpdate.go=false;
167 private final static class CacheUpdate extends Thread {
168 public static BlockingQueue<Transfer> notifyDQ = new LinkedBlockingQueue<Transfer>(2000);
170 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";
171 private AuthzEnv env;
173 private SecuritySetter<HttpURLConnection> ss;
174 private final String authority;
175 public boolean go = true;
177 public CacheUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
182 this.authority = ip+':'+port;
185 private static class Transfer {
188 public Transfer(String table, int[] segs) {
193 private class CacheClear extends Retryable<Integer> {
195 private AuthzTrans trans;
199 public CacheClear(AuthzTrans trans) {
203 public void set(Entry<String, IntHolder> es) {
205 segs = es.getValue().toString();
209 public Integer code(Rcli<?> client) throws APIException, CadiException {
210 URI to = client.getURI();
211 if (!to.getAuthority().equals(authority)) {
212 Future<Void> f = client.delete("/mgmt/cache/"+type+'/'+segs,VOID_CT);
213 if (f.get(hman.readTimeout())) {
216 trans.error().log("Error During AAF Peer Notify",f.code(),f.body());
223 private class IntHolder {
225 HashSet<Integer> set;
227 public IntHolder(int ints[]) {
231 public void add(int[] ints) {
233 set = new HashSet<>();
235 for (int i=0;i<raw.length;++i) {
239 for (int i=0;i<ints.length;++i) {
245 public String toString() {
246 StringBuilder sb = new StringBuilder();
247 boolean first = true;
258 for (Integer i : set) {
267 return sb.toString();
275 Transfer data = notifyDQ.poll(4,TimeUnit.SECONDS);
281 CacheClear cc = null;
282 Map<String,IntHolder> gather = null;
283 AuthzTrans trans = null;
285 // Do a block poll first
288 start = System.nanoTime();
289 trans = env.newTransNoAvg();
290 cc = new CacheClear(trans);
291 gather = new HashMap<>();
293 IntHolder prev = gather.get(data.table);
295 gather.put(data.table,new IntHolder(data.segs));
299 // continue while there is data
300 } while ((data = notifyDQ.poll())!=null);
302 for (Entry<String, IntHolder> es : gather.entrySet()) {
305 if (hman.all(ss, cc, false)!=null) {
308 } catch (Exception e) {
309 trans.error().log(e, "Error on Cache Update");
312 if (env.debug().isLoggable()) {
313 float millis = (System.nanoTime()-start)/1000000f;
314 StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");
315 sb.append("Updated ");
318 sb.append(" entry for ");
320 sb.append(" entries for ");
322 int peers = count<=0?0:cc.total/count;
324 sb.append(" client");
331 trans.auditTrail(0, sb, Env.REMOTE);
335 } catch (InterruptedException e1) {
337 Thread.currentThread().interrupt();
343 private void init(AuthzTrans trans) throws APIException, IOException {
344 String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
345 psCheck = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE);
347 disable(CRUD.create);
348 disable(CRUD.delete);
352 * @see org.onap.aaf.auth.dao.cass.CIDAO#touch(org.onap.aaf.auth.env.test.AuthzTrans, java.lang.String, int)
356 public Result<Void> touch(AuthzTrans trans, String name, int ... seg) {
358 // Direct Service Cache Invalidation
360 // ConcurrentQueues are open-ended. We don't want any Memory leaks
361 // Note: we keep a separate counter, because "size()" on a Linked Queue is expensive
362 if (cacheNotify && cacheUpdate!=null) {
364 if (!CacheUpdate.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {
365 trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );
367 } catch (InterruptedException e) {
368 trans.error().log("Cache Notify Queue posting was interrupted" );
369 Thread.currentThread().interrupt();
374 // Table Based Cache Invalidation (original)
376 // Note: Save time with multiple Sequence Touches, but PreparedStmt doesn't support IN
377 StringBuilder start = new StringBuilder("CacheInfoDAO Touch segments ");
380 StringBuilder sb = new StringBuilder("BEGIN BATCH\n");
381 boolean first = true;
383 sb.append(UPDATE_SP);
385 sb.append(" SET touched=dateof(now()) WHERE name = '");
387 sb.append("' AND seg = ");
397 sb.append("APPLY BATCH;");
398 TimeTaken tt = trans.start(start.toString(),Env.REMOTE);
400 getSession(trans).executeAsync(sb.toString());
401 } catch (DriverException | APIException | IOException e) {
402 reportPerhapsReset(trans,e);
403 return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
411 * @see org.onap.aaf.auth.dao.cass.CIDAO#check(org.onap.aaf.auth.env.test.AuthzTrans)
414 public Result<Void> check(AuthzTrans trans) {
416 TimeTaken tt = trans.start("Check Table Timestamps",Env.REMOTE);
418 rs = getSession(trans).execute(new BoundStatement(psCheck));
419 } catch (DriverException | APIException | IOException e) {
420 reportPerhapsReset(trans,e);
421 return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);
426 String lastName = null;
428 for (Row row : rs.all()) {
429 String name = row.getString(0);
430 int seg = row.getInt(1);
431 if (!name.equals(lastName)) {
432 dates = info.get(name);
436 dates=new Date[seg+1];
437 info.put(name,dates);
438 } else if (dates.length<=seg) {
439 Date[] temp = new Date[seg+1];
440 System.arraycopy(dates, 0, temp, 0, dates.length);
442 info.put(name, dates);
444 Date temp = row.getTimestamp(2);
445 if (dates[seg]==null || dates[seg].before(temp)) {
453 * @see org.onap.aaf.auth.dao.cass.CIDAO#get(java.lang.String, int)
456 public Date get(AuthzTrans trans, String table, int seg) {
457 Date[] dates = info.get(table);
459 dates = new Date[seg+1];
460 touch(trans,table, seg);
461 } else if (dates.length<=seg) {
462 Date[] temp = new Date[seg+1];
463 System.arraycopy(dates, 0, temp, 0, dates.length);
466 Date rv = dates[seg];
468 rv=dates[seg]=startTime;
474 protected void wasModified(AuthzTrans trans, CRUD modified, Data data, String ... override) {