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.client.Future;
import org.onap.aaf.cadi.client.Rcli;
import org.onap.aaf.cadi.client.Retryable;
import org.onap.aaf.cadi.http.HMangr;
import org.onap.aaf.misc.env.APIException;
import org.onap.aaf.misc.env.Env;
import org.onap.aaf.cadi.http.HMangr;
import org.onap.aaf.misc.env.APIException;
import org.onap.aaf.misc.env.Env;
private PreparedStatement psCheck;
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();
private PreparedStatement psCheck;
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();
- public CacheInfoDAO(AuthzTrans trans, AbsCassDAO<AuthzTrans,?> aDao) throws APIException, IOException {
+ public CacheInfoDAO(AuthzTrans trans, AbsCassDAO<AuthzTrans,?> aDao) throws APIException, IOException {
super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
startTime = new Date();
init(trans);
super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));
startTime = new Date();
init(trans);
+ /**
+ * 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));
+ }
//////////////////////////////////////////
// Data Definition, matches Cassandra DM
//////////////////////////////////////////
// Data Definition, matches Cassandra DM
- public static<T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {
- if(cacheUpdate==null) {
+ 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");
t.setDaemon(true);
t.start();
Thread t= new Thread(cacheUpdate = new CacheUpdate(env,hman,ss, ip,port),"CacheInfo Update Thread");
t.setDaemon(true);
t.start();
@Override
public Integer code(Rcli<?> client) throws APIException, CadiException {
URI to = client.getURI();
@Override
public Integer code(Rcli<?> client) throws APIException, CadiException {
URI to = client.getURI();
start = System.nanoTime();
trans = env.newTransNoAvg();
cc = new CacheClear(trans);
gather = new HashMap<>();
}
IntHolder prev = gather.get(data.table);
start = System.nanoTime();
trans = env.newTransNoAvg();
cc = new CacheClear(trans);
gather = new HashMap<>();
}
IntHolder prev = gather.get(data.table);
gather.put(data.table,new IntHolder(data.segs));
} else {
prev.add(data.segs);
}
// continue while there is data
gather.put(data.table,new IntHolder(data.segs));
} else {
prev.add(data.segs);
}
// continue while there is data
- } while((data = notifyDQ.poll())!=null);
- if(gather!=null) {
- for(Entry<String, IntHolder> es : gather.entrySet()) {
+ } while ((data = notifyDQ.poll())!=null);
+ if (gather!=null) {
+ for (Entry<String, IntHolder> es : gather.entrySet()) {
float millis = (System.nanoTime()-start)/1000000f;
StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");
sb.append("Updated ");
sb.append(count);
float millis = (System.nanoTime()-start)/1000000f;
StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");
sb.append("Updated ");
sb.append(count);
String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
psCheck = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE);
String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);
psCheck = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE);
/////////////
// 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
/////////////
// 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.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {
+ 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" );
}
} catch (InterruptedException e) {
trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );
}
} catch (InterruptedException e) {
Date[] temp = new Date[seg+1];
System.arraycopy(dates, 0, temp, 0, dates.length);
dates = temp;
info.put(name, dates);
}
Date temp = row.getTimestamp(2);
Date[] temp = new Date[seg+1];
System.arraycopy(dates, 0, temp, 0, dates.length);
dates = temp;
info.put(name, dates);
}
Date temp = row.getTimestamp(2);
@Override
public Date get(AuthzTrans trans, String table, int seg) {
Date[] dates = info.get(table);
@Override
public Date get(AuthzTrans trans, String table, int seg) {
Date[] dates = info.get(table);
Date[] temp = new Date[seg+1];
System.arraycopy(dates, 0, temp, 0, dates.length);
dates = temp;
}
Date rv = dates[seg];
Date[] temp = new Date[seg+1];
System.arraycopy(dates, 0, temp, 0, dates.length);
dates = temp;
}
Date rv = dates[seg];