+++ /dev/null
-/*******************************************************************************\r
- * ============LICENSE_START====================================================\r
- * * org.onap.aaf\r
- * * ===========================================================================\r
- * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
- * * ===========================================================================\r
- * * Licensed under the Apache License, Version 2.0 (the "License");\r
- * * you may not use this file except in compliance with the License.\r
- * * You may obtain a copy of the License at\r
- * * \r
- * * http://www.apache.org/licenses/LICENSE-2.0\r
- * * \r
- * * Unless required by applicable law or agreed to in writing, software\r
- * * distributed under the License is distributed on an "AS IS" BASIS,\r
- * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * * See the License for the specific language governing permissions and\r
- * * limitations under the License.\r
- * * ============LICENSE_END====================================================\r
- * *\r
- * * ECOMP is a trademark and service mark of AT&T Intellectual Property.\r
- * *\r
- ******************************************************************************/\r
-package com.att.dao.aaf.cass;\r
-\r
-import java.io.IOException;\r
-import java.net.HttpURLConnection;\r
-import java.net.URI;\r
-import java.util.Date;\r
-import java.util.HashMap;\r
-import java.util.HashSet;\r
-import java.util.Map;\r
-import java.util.Map.Entry;\r
-import java.util.concurrent.BlockingQueue;\r
-import java.util.concurrent.ConcurrentHashMap;\r
-import java.util.concurrent.LinkedBlockingQueue;\r
-import java.util.concurrent.TimeUnit;\r
-\r
-import com.att.authz.env.AuthzEnv;\r
-import com.att.authz.env.AuthzTrans;\r
-import com.att.authz.layer.Result;\r
-import com.att.cadi.CadiException;\r
-import com.att.cadi.SecuritySetter;\r
-import com.att.cadi.client.Future;\r
-import com.att.cadi.client.Rcli;\r
-import com.att.cadi.client.Retryable;\r
-import com.att.cadi.http.HMangr;\r
-import com.att.dao.AbsCassDAO;\r
-import com.att.dao.CIDAO;\r
-import com.att.dao.CassAccess;\r
-import com.att.dao.CassDAOImpl;\r
-import com.att.dao.Loader;\r
-import com.att.inno.env.APIException;\r
-import com.att.inno.env.Env;\r
-import com.att.inno.env.TimeTaken;\r
-import com.att.inno.env.Trans;\r
-import com.datastax.driver.core.BoundStatement;\r
-import com.datastax.driver.core.Cluster;\r
-import com.datastax.driver.core.ResultSet;\r
-import com.datastax.driver.core.Row;\r
-import com.datastax.driver.core.exceptions.DriverException;\r
-\r
-public class CacheInfoDAO extends CassDAOImpl<AuthzTrans,CacheInfoDAO.Data> implements CIDAO<AuthzTrans> {\r
-\r
- private static final String TABLE = "cache";\r
- public static final Map<String,Date[]> info = new ConcurrentHashMap<String,Date[]>();\r
-\r
- private static CacheUpdate cacheUpdate;\r
- \r
- \r
- private BoundStatement check;\r
- // Hold current time stamps from Tables\r
- private final Date startTime;\r
- \r
- public CacheInfoDAO(AuthzTrans trans, Cluster cluster, String keyspace) throws APIException, IOException {\r
- super(trans, CacheInfoDAO.class.getSimpleName(),cluster,keyspace,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));\r
- startTime = new Date();\r
- init(trans);\r
- }\r
-\r
- public CacheInfoDAO(AuthzTrans trans, AbsCassDAO<AuthzTrans,?> aDao) throws APIException, IOException {\r
- super(trans, CacheInfoDAO.class.getSimpleName(),aDao,Data.class,TABLE,readConsistency(trans,TABLE), writeConsistency(trans,TABLE));\r
- startTime = new Date();\r
- init(trans);\r
- }\r
-\r
-\r
- //////////////////////////////////////////\r
- // Data Definition, matches Cassandra DM\r
- //////////////////////////////////////////\r
- private static final int KEYLIMIT = 2;\r
- /**\r
- */\r
- public static class Data {\r
- public Data() {\r
- name = null;\r
- touched = null;\r
- }\r
- public Data(String name, int seg) {\r
- this.name = name;\r
- this.seg = seg;\r
- touched = null;\r
- }\r
- \r
- public String name;\r
- public int seg;\r
- public Date touched;\r
- }\r
-\r
- private static class InfoLoader extends Loader<Data> {\r
- public static final InfoLoader dflt = new InfoLoader(KEYLIMIT);\r
- \r
- public InfoLoader(int keylimit) {\r
- super(keylimit);\r
- }\r
- \r
- @Override\r
- public Data load(Data data, Row row) {\r
- // Int more efficient\r
- data.name = row.getString(0);\r
- data.seg = row.getInt(1);\r
- data.touched = row.getDate(2);\r
- return data;\r
- }\r
-\r
- @Override\r
- protected void key(Data data, int _idx, Object[] obj) {\r
- int idx = _idx;\r
-\r
- obj[idx]=data.name;\r
- obj[++idx]=data.seg;\r
- }\r
-\r
- @Override\r
- protected void body(Data data, int idx, Object[] obj) {\r
- obj[idx]=data.touched;\r
- }\r
- }\r
- \r
- public static<T extends Trans> void startUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {\r
- if(cacheUpdate==null) {\r
- Thread t= new Thread(cacheUpdate = new CacheUpdate(env,hman,ss, ip,port),"CacheInfo Update Thread");\r
- t.setDaemon(true);\r
- t.start();\r
- }\r
- }\r
-\r
- public static<T extends Trans> void stopUpdate() {\r
- if(cacheUpdate!=null) {\r
- cacheUpdate.go=false;\r
- }\r
- }\r
-\r
- private final static class CacheUpdate extends Thread {\r
- public static BlockingQueue<Transfer> notifyDQ = new LinkedBlockingQueue<Transfer>(2000);\r
-\r
- 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";\r
- private AuthzEnv env;\r
- private HMangr hman;\r
- private SecuritySetter<HttpURLConnection> ss;\r
- private final String authority;\r
- public boolean go = true;\r
- \r
- public CacheUpdate(AuthzEnv env, HMangr hman, SecuritySetter<HttpURLConnection> ss, String ip, int port) {\r
- this.env = env;\r
- this.hman = hman;\r
- this.ss = ss;\r
- \r
- this.authority = ip+':'+port;\r
- }\r
- \r
- private static class Transfer {\r
- public String table;\r
- public int segs[];\r
- public Transfer(String table, int[] segs) {\r
- this.table = table;\r
- this.segs = segs;\r
- }\r
- }\r
- private class CacheClear extends Retryable<Integer> {\r
- public int total=0;\r
- private AuthzTrans trans;\r
- private String type;\r
- private String segs;\r
- \r
- public CacheClear(AuthzTrans trans) {\r
- this.trans = trans;\r
- }\r
-\r
- public void set(Entry<String, IntHolder> es) {\r
- type = es.getKey();\r
- segs = es.getValue().toString();\r
- }\r
- \r
- @Override\r
- public Integer code(Rcli<?> client) throws APIException, CadiException {\r
- URI to = client.getURI();\r
- if(!to.getAuthority().equals(authority)) {\r
- Future<Void> f = client.delete("/mgmt/cache/"+type+'/'+segs,VOID_CT);\r
- if(f.get(hman.readTimeout())) {\r
- ++total;\r
- } else {\r
- trans.error().log("Error During AAF Peer Notify",f.code(),f.body());\r
- }\r
- }\r
- return total;\r
- }\r
- }\r
- \r
- private class IntHolder {\r
- private int[] raw;\r
- HashSet<Integer> set;\r
- \r
- public IntHolder(int ints[]) {\r
- raw = ints;\r
- set = null;\r
- }\r
- public void add(int[] ints) {\r
- if(set==null) {\r
- set = new HashSet<Integer>();\r
- \r
- for(int i=0;i<raw.length;++i) {\r
- set.add(raw[i]);\r
- }\r
- }\r
- for(int i=0;i<ints.length;++i) {\r
- set.add(ints[i]);\r
- }\r
- }\r
-\r
- @Override\r
- public String toString() {\r
- StringBuilder sb = new StringBuilder();\r
- boolean first = true;\r
- if(set==null) {\r
- for(int i : raw) {\r
- if(first) {\r
- first=false;\r
- } else {\r
- sb.append(',');\r
- }\r
- sb.append(i);\r
- }\r
- } else {\r
- for(Integer i : set) {\r
- if(first) {\r
- first=false;\r
- } else {\r
- sb.append(',');\r
- }\r
- sb.append(i);\r
- }\r
- }\r
- return sb.toString();\r
- }\r
- }\r
- \r
- @Override\r
- public void run() {\r
- do {\r
- try {\r
- Transfer data = notifyDQ.poll(4,TimeUnit.SECONDS);\r
- if(data==null) {\r
- continue;\r
- }\r
- \r
- int count = 0;\r
- CacheClear cc = null;\r
- Map<String,IntHolder> gather = null;\r
- AuthzTrans trans = null;\r
- long start=0;\r
- // Do a block poll first\r
- do {\r
- if(gather==null) {\r
- start = System.nanoTime();\r
- trans = env.newTransNoAvg();\r
- cc = new CacheClear(trans);\r
- gather = new HashMap<String,IntHolder>();\r
- }\r
- IntHolder prev = gather.get(data.table);\r
- if(prev==null) {\r
- gather.put(data.table,new IntHolder(data.segs));\r
- } else {\r
- prev.add(data.segs);\r
- }\r
- // continue while there is data\r
- } while((data = notifyDQ.poll())!=null);\r
- if(gather!=null) {\r
- for(Entry<String, IntHolder> es : gather.entrySet()) {\r
- cc.set(es);\r
- try {\r
- if(hman.all(ss, cc, false)!=null) {\r
- ++count;\r
- }\r
- } catch (Exception e) {\r
- trans.error().log(e, "Error on Cache Update");\r
- }\r
- }\r
- if(env.debug().isLoggable()) {\r
- float millis = (System.nanoTime()-start)/1000000f;\r
- StringBuilder sb = new StringBuilder("Direct Cache Refresh: ");\r
- sb.append("Updated ");\r
- sb.append(count);\r
- if(count==1) {\r
- sb.append(" entry for ");\r
- } else { \r
- sb.append(" entries for ");\r
- }\r
- int peers = count<=0?0:cc.total/count;\r
- sb.append(peers);\r
- sb.append(" client");\r
- if(peers!=1) {\r
- sb.append('s');\r
- }\r
- sb.append(" in ");\r
- sb.append(millis);\r
- sb.append("ms");\r
- trans.auditTrail(0, sb, Env.REMOTE);\r
- env.debug().log(sb);\r
- }\r
- }\r
- } catch (InterruptedException e1) {\r
- go = false;\r
- }\r
- } while(go);\r
- }\r
- }\r
-\r
- private void init(AuthzTrans trans) throws APIException, IOException {\r
- \r
- String[] helpers = setCRUD(trans, TABLE, Data.class, InfoLoader.dflt);\r
- check = getSession(trans).prepare(SELECT_SP + helpers[FIELD_COMMAS] + " FROM " + TABLE).bind();\r
-\r
- disable(CRUD.create);\r
- disable(CRUD.delete);\r
- }\r
-\r
- /* (non-Javadoc)\r
- * @see com.att.dao.aaf.cass.CIDAO#touch(com.att.authz.env.AuthzTrans, java.lang.String, int)\r
- */\r
- \r
- @Override\r
- public Result<Void> touch(AuthzTrans trans, String name, int ... seg) {\r
- /////////////\r
- // Direct Service Cache Invalidation\r
- /////////////\r
- // ConcurrentQueues are open-ended. We don't want any Memory leaks \r
- // Note: we keep a separate counter, because "size()" on a Linked Queue is expensive\r
- if(cacheUpdate!=null) {\r
- try {\r
- if(!CacheUpdate.notifyDQ.offer(new CacheUpdate.Transfer(name, seg),2,TimeUnit.SECONDS)) {\r
- trans.error().log("Cache Notify Queue is not accepting messages, bouncing may be appropriate" );\r
- }\r
- } catch (InterruptedException e) {\r
- trans.error().log("Cache Notify Queue posting was interrupted" );\r
- }\r
- }\r
-\r
- /////////////\r
- // Table Based Cache Invalidation (original)\r
- /////////////\r
- // Note: Save time with multiple Sequence Touches, but PreparedStmt doesn't support IN\r
- StringBuilder start = new StringBuilder("CacheInfoDAO Touch segments ");\r
- start.append(name);\r
- start.append(": ");\r
- StringBuilder sb = new StringBuilder("BEGIN BATCH\n");\r
- boolean first = true;\r
- for(int s : seg) {\r
- sb.append(UPDATE_SP);\r
- sb.append(TABLE);\r
- sb.append(" SET touched=dateof(now()) WHERE name = '");\r
- sb.append(name);\r
- sb.append("' AND seg = ");\r
- sb.append(s);\r
- sb.append(";\n"); \r
- if(first) {\r
- first =false;\r
- } else {\r
- start.append(',');\r
- }\r
- start.append(s);\r
- }\r
- sb.append("APPLY BATCH;");\r
- TimeTaken tt = trans.start(start.toString(),Env.REMOTE);\r
- try {\r
- getSession(trans).executeAsync(sb.toString());\r
- } catch (DriverException | APIException | IOException e) {\r
- reportPerhapsReset(trans,e);\r
- return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);\r
- } finally {\r
- tt.done();\r
- }\r
- return Result.ok();\r
- }\r
-\r
- /* (non-Javadoc)\r
- * @see com.att.dao.aaf.cass.CIDAO#check(com.att.authz.env.AuthzTrans)\r
- */\r
- @Override\r
- public Result<Void> check(AuthzTrans trans) {\r
- ResultSet rs;\r
- TimeTaken tt = trans.start("Check Table Timestamps",Env.REMOTE);\r
- try {\r
- rs = getSession(trans).execute(check);\r
- } catch (DriverException | APIException | IOException e) {\r
- reportPerhapsReset(trans,e);\r
- return Result.err(Result.ERR_Backend, CassAccess.ERR_ACCESS_MSG);\r
- } finally {\r
- tt.done();\r
- }\r
- \r
- String lastName = null;\r
- Date[] dates = null;\r
- for(Row row : rs.all()) {\r
- String name = row.getString(0);\r
- int seg = row.getInt(1);\r
- if(!name.equals(lastName)) {\r
- dates = info.get(name);\r
- lastName=name;\r
- }\r
- if(dates==null) {\r
- dates=new Date[seg+1];\r
- info.put(name,dates);\r
- } else if(dates.length<=seg) {\r
- Date[] temp = new Date[seg+1];\r
- System.arraycopy(dates, 0, temp, 0, dates.length);\r
- dates = temp;\r
- info.put(name, dates);\r
- }\r
- Date temp = row.getDate(2);\r
- if(dates[seg]==null || dates[seg].before(temp)) {\r
- dates[seg]=temp;\r
- }\r
- }\r
- return Result.ok();\r
- }\r
- \r
- /* (non-Javadoc)\r
- * @see com.att.dao.aaf.cass.CIDAO#get(java.lang.String, int)\r
- */\r
- @Override\r
- public Date get(AuthzTrans trans, String table, int seg) {\r
- Date[] dates = info.get(table);\r
- if(dates==null) {\r
- dates = new Date[seg+1];\r
- touch(trans,table, seg);\r
- } else if(dates.length<=seg) {\r
- Date[] temp = new Date[seg+1];\r
- System.arraycopy(dates, 0, temp, 0, dates.length);\r
- dates = temp;\r
- }\r
- Date rv = dates[seg];\r
- if(rv==null) {\r
- rv=dates[seg]=startTime;\r
- }\r
- return rv;\r
- }\r
-\r
- @Override\r
- protected void wasModified(AuthzTrans trans, CRUD modified, Data data, String ... override) {\r
- // Do nothing\r
- }\r
-\r
-}\r
Code Review / aaf / authz.git / blobdiff