* org.onap.aaf
* ===========================================================================
* Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
+ *
+ * Modification Copyright (c) 2019 IBM
* ===========================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
import org.onap.aaf.cadi.client.Future;
import org.onap.aaf.cadi.client.Rcli;
import org.onap.aaf.cadi.client.Retryable;
+import org.onap.aaf.cadi.config.Config;
import org.onap.aaf.cadi.http.HMangr;
import org.onap.aaf.misc.env.APIException;
import org.onap.aaf.misc.env.Env;
public static final Map<String,Date[]> info = new ConcurrentHashMap<>();
private static CacheUpdate cacheUpdate;
-
+
// Hold current time stamps from Tables
private final Date startTime;
+ private final boolean cacheNotify;
private PreparedStatement psCheck;
-
+
+ //////////////////////////////////////////
+ // Data Definition, matches Cassandra DM
+ //////////////////////////////////////////
+ private static final int KEYLIMIT = 2;
+
public CacheInfoDAO(AuthzTrans trans, Cluster cluster, String keyspace) throws APIException, IOException {
super(trans, CacheInfoDAO.class.getSimpleName(),cluster,keyspace,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
startTime = new Date();
+ cacheNotify = noK8s(trans);
init(trans);
}
super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
startTime = new Date();
init(trans);
+ cacheNotify = noK8s(trans);
}
-
- //////////////////////////////////////////
- // Data Definition, matches Cassandra DM
- //////////////////////////////////////////
- private static final int KEYLIMIT = 2;
+ /**
+ * Need a different point to point cache clear strategy for K8s...
+ * @param trans
+ * @return
+ */
+ private boolean noK8s(AuthzTrans trans) {
+ String container = trans.getProperty(Config.AAF_LOCATOR_CONTAINER);
+ return ! ("helm".equals(container) || "oom".equals(container));
+ }
/**
* @author Jonathan
*/
public static class Data {
+ public String name;
+ public int seg;
+ public Date touched;
+
public Data() {
name = null;
touched = null;
this.seg = seg;
touched = null;
}
-
- public String name;
- public int seg;
- public Date touched;
}
private static class InfoLoader extends Loader<Data> {
public static final InfoLoader dflt = new InfoLoader(KEYLIMIT);
-
+
public InfoLoader(int keylimit) {
super(keylimit);
}
-
+
@Override
public Data load(Data data, Row row) {
// Int more efficient
}
@Override
- protected void key(Data data, int _idx, Object[] obj) {
- int idx = _idx;
+ protected void key(Data data, int idxParam, Object[] obj) {
+ int idx = idxParam;
obj[idx]=data.name;
obj[++idx]=data.seg;
obj[idx]=data.touched;
}
}
-
- public static<T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
+
+ public static synchronized <T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
if (cacheUpdate==null) {
- Thread t= new Thread(cacheUpdate = new CacheUpdate(env,hman,ss, ip,port),"CacheInfo Update Thread");
+ cacheUpdate = new CacheUpdate(env,hman,ss, ip,port);
+ Thread t= new Thread(cacheUpdate,"CacheInfo Update Thread");
t.setDaemon(true);
t.start();
}
}
}
- private final static class CacheUpdate extends Thread {
+ private static final class CacheUpdate extends Thread {
public static BlockingQueue<Transfer> notifyDQ = new LinkedBlockingQueue<Transfer>(2000);
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";
private SecuritySetter<HttpURLConnection> ss;
private final String authority;
public boolean go = true;
-
+
public CacheUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
this.env = env;
this.hman = hman;
this.ss = ss;
-
+
this.authority = ip+':'+port;
}
-
+
private static class Transfer {
public String table;
public int segs[];
private AuthzTrans trans;
private String type;
private String segs;
-
+
public CacheClear(AuthzTrans trans) {
this.trans = trans;
}
type = es.getKey();
segs = es.getValue().toString();
}
-
+
@Override
public Integer code(Rcli<?> client) throws APIException, CadiException {
URI to = client.getURI();
return total;
}
}
-
+
private class IntHolder {
private int[] raw;
HashSet<Integer> set;
-
+
public IntHolder(int ints[]) {
raw = ints;
set = null;
public void add(int[] ints) {
if (set==null) {
set = new HashSet<>();
-
+
for (int i=0;i<raw.length;++i) {
set.add(raw[i]);
}
return sb.toString();
}
}
-
+
@Override
public void run() {
do {
if (data==null) {
continue;
}
-
+
int count = 0;
CacheClear cc = null;
Map<String,IntHolder> gather = null;
sb.append(count);
if (count==1) {
sb.append(" entry for ");
- } else {
+ } else {
sb.append(" entries for ");
}
int peers = count<=0?0:cc.total/count;
}
private void init(AuthzTrans trans) throws APIException, IOException {
-
String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
psCheck = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE);
/* (non-Javadoc)
* @see org.onap.aaf.auth.dao.cass.CIDAO#touch(org.onap.aaf.auth.env.test.AuthzTrans, java.lang.String, int)
*/
-
+
@Override
public Result<Void> touch(AuthzTrans trans, String name, int ... seg) {
/////////////
// Direct Service Cache Invalidation
/////////////
- // ConcurrentQueues are open-ended. We don't want any Memory leaks
+ // ConcurrentQueues are open-ended. We don't want any Memory leaks
// Note: we keep a separate counter, because "size()" on a Linked Queue is expensive
- if (cacheUpdate!=null) {
+ if (cacheNotify && cacheUpdate!=null) {
try {
if (!CacheUpdate.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {
trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );
sb.append(name);
sb.append("' AND seg = ");
sb.append(s);
- sb.append(";\n");
+ sb.append(";\n");
if (first) {
first =false;
} else {
} finally {
tt.done();
}
-
+
String lastName = null;
Date[] dates = null;
for (Row row : rs.all()) {
}
return Result.ok();
}
-
+
/* (non-Javadoc)
* @see org.onap.aaf.auth.dao.cass.CIDAO#get(java.lang.String, int)
*/
Code Review / aaf / authz.git / blobdiff