import java.util.*;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
-import org.onap.aai.restclient.PropertyPasswordConfiguration;
import org.apache.tinkerpop.gremlin.process.traversal.P;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.structure.Edge;
public class HistoryTruncate {
private static Logger LOGGER = LoggerFactory.getLogger(HistoryTruncate.class);
-
+
/* Using realtime d */
private static final String REALTIME_DB = "realtime";
-
+
private static final String LOG_ONLY_MODE = "LOG_ONLY";
private static final String DELETE_AND_LOG_MODE = "DELETE_AND_LOG";
- private static final String SILENT_DELETE_MODE = "SILENT_DELETE";
+ private static final String SILENT_DELETE_MODE = "SILENT_DELETE";
static ArrayList <String> VALIDMODES = new <String> ArrayList ();
static {
VALIDMODES.add(LOG_ONLY_MODE);
VALIDMODES.add(SILENT_DELETE_MODE);
}
- private static final int batchCommitSize = 500;
-
+ private static final int batchCommitSize = 500;
+
private static boolean historyEnabled;
private static String defaultTruncateMode;
private static Integer defaultTruncateWindowDays;
-
+
/**
* The main method.
*
*/
public static void main(String[] args) {
-
+
// Set the logging file properties to be used by EELFManager
System.setProperty("aai.service.name", HistoryTruncate.class.getSimpleName());
Properties props = System.getProperties();
props.setProperty(Configuration.PROPERTY_LOGGING_FILE_NAME, AAIConstants.AAI_LOGBACK_PROPS);
props.setProperty(Configuration.PROPERTY_LOGGING_FILE_PATH, AAIConstants.AAI_HOME_BUNDLECONFIG);
-
- AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
- PropertyPasswordConfiguration initializer = new PropertyPasswordConfiguration();
- initializer.initialize(ctx);
- try {
- ctx.scan(
- "org.onap.aai.config",
- "org.onap.aai.setup"
- );
- ctx.refresh();
- } catch (Exception e) {
- LOGGER.error("Error - Could not initialize context beans for HistoryTruncate. ");
- AAISystemExitUtil.systemExitCloseAAIGraph(1);
- }
-
- historyEnabled = Boolean.parseBoolean(ctx.getEnvironment().getProperty("history.enabled","false"));
- if( !historyEnabled ) {
- String emsg = "Error - HistoryTruncate may only be used when history.enabled=true. ";
- System.out.println(emsg);
- LOGGER.error(emsg);
- AAISystemExitUtil.systemExitCloseAAIGraph(1);
+
+ try (AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext()) {
+ try {
+ ctx.refresh();
+ } catch (Exception e) {
+ LOGGER.error("Error - Could not initialize context beans for HistoryTruncate. ");
+ AAISystemExitUtil.systemExitCloseAAIGraph(1);
+ }
+
+ historyEnabled = Boolean.parseBoolean(ctx.getEnvironment().getProperty("history.enabled","false"));
+ if( !historyEnabled ) {
+ String emsg = "Error - HistoryTruncate may only be used when history.enabled=true. ";
+ System.out.println(emsg);
+ LOGGER.error(emsg);
+ AAISystemExitUtil.systemExitCloseAAIGraph(1);
+ }
+
+ defaultTruncateWindowDays = Integer.parseInt(ctx.getEnvironment().getProperty("history.truncate.window.days","999"));
+ defaultTruncateMode = ctx.getEnvironment().getProperty("history.truncate.mode",LOG_ONLY_MODE);
}
-
- defaultTruncateWindowDays = Integer.parseInt(ctx.getEnvironment().getProperty("history.truncate.window.days","999"));
- defaultTruncateMode = ctx.getEnvironment().getProperty("history.truncate.mode",LOG_ONLY_MODE);
-
- HistoryTruncate histTrunc = new HistoryTruncate();
+ HistoryTruncate histTrunc = new HistoryTruncate();
boolean success = histTrunc.executeCommand(args);
if(success){
AAISystemExitUtil.systemExitCloseAAIGraph(0);
public boolean executeCommand(String[] args) {
boolean successStatus = true;
- // If they passed in args on the command line, then we should
+ // If they passed in args on the command line, then we should
// use those in place of the default ones we got from environment variables.
// "-truncateMode","LOG_ONLY","-truncateWindow","999"
String truncateMode = defaultTruncateMode;
int truncateWindowDays = defaultTruncateWindowDays;
-
- if (args != null && args.length > 0) {
- // They passed some arguments in that will affect processing
- for (int i = 0; i < args.length; i++) {
- String thisArg = args[i];
- if (thisArg.equals("-truncateMode")) {
- i++;
- if (i >= args.length) {
- LOGGER.error(" No value passed with -truncateMode option. ");
- return false;
- }
- if( !VALIDMODES.contains(args[i]) ) {
- LOGGER.error(" Unrecognized -truncateMode value passed: [" +
- args[i] + "]. Valid values = " + VALIDMODES.toString() );
- return false;
- }
- truncateMode = args[i];
- } else if (thisArg.equals("-truncateWindowDays")) {
- i++;
- if (i >= args.length) {
- LOGGER.error("No value passed with -truncateWindowDays option.");
- return false;
- }
- String nextArg = args[i];
- try {
- truncateWindowDays = Integer.parseInt(nextArg);
- } catch (Exception e) {
- LOGGER.error("Bad value passed with -truncateWindowDays option: ["
- + nextArg + "]");
- return false;
- }
- } else {
- LOGGER.error(" Unrecognized argument passed to HistoryTruncate: ["
- + thisArg + "]. ");
- LOGGER.error(" Valid values are: -truncateMode -truncateWindowDays ");
- return false;
- }
- }
- }
+
+ if (args != null && args.length > 0) {
+ // They passed some arguments in that will affect processing
+ for (int i = 0; i < args.length; i++) {
+ String thisArg = args[i];
+ if (thisArg.equals("-truncateMode")) {
+ i++;
+ if (i >= args.length) {
+ LOGGER.error(" No value passed with -truncateMode option. ");
+ return false;
+ }
+ if( !VALIDMODES.contains(args[i]) ) {
+ LOGGER.error(" Unrecognized -truncateMode value passed: [" +
+ args[i] + "]. Valid values = " + VALIDMODES.toString() );
+ return false;
+ }
+ truncateMode = args[i];
+ } else if (thisArg.equals("-truncateWindowDays")) {
+ i++;
+ if (i >= args.length) {
+ LOGGER.error("No value passed with -truncateWindowDays option.");
+ return false;
+ }
+ String nextArg = args[i];
+ try {
+ truncateWindowDays = Integer.parseInt(nextArg);
+ } catch (Exception e) {
+ LOGGER.error("Bad value passed with -truncateWindowDays option: ["
+ + nextArg + "]");
+ return false;
+ }
+ } else {
+ LOGGER.error(" Unrecognized argument passed to HistoryTruncate: ["
+ + thisArg + "]. ");
+ LOGGER.error(" Valid values are: -truncateMode -truncateWindowDays ");
+ return false;
+ }
+ }
+ }
LOGGER.debug(" Running HistoryTruncate with: truncateMode = " + truncateMode +
", truncateWindowDays = " + truncateWindowDays );
-
+
Long truncateEndTs = calculateTruncWindowEndTimeStamp(truncateWindowDays);
JanusGraph jgraph = null;
long scriptStartTime = System.currentTimeMillis();
Boolean doLogging = doLoggingOrNot( truncateMode );
Boolean doDelete = doDeleteOrNot( truncateMode );
-
+
try {
AAIConfig.init();
ErrorLogHelper.loadProperties();
verifyGraph(AAIGraph.getInstance().getGraph());
jgraph = AAIGraph.getInstance().getGraph();
LOGGER.debug(" ---- got the new graph instance. ");
-
+
// Note - process edges first so they get logged as they are deleted since
// edges connected to vertices being deleted would get auto-deleted by the db.
long timeA = System.nanoTime();
long secCount = TimeUnit.NANOSECONDS.toSeconds(diffTime) - (60 * minCount);
LOGGER.debug(" Took this long to process the Edges: " +
minCount + " minutes, " + secCount + " seconds " );
-
+
processVerts(jgraph, truncateEndTs, doLogging, doDelete);
long timeC = System.nanoTime();
diffTime = timeC - timeB;
secCount = TimeUnit.NANOSECONDS.toSeconds(diffTime) - (60 * minCount);
LOGGER.debug(" Took this long to process the Vertices: " +
minCount + " minutes, " + secCount + " seconds " );
-
+
} catch (AAIException e) {
ErrorLogHelper.logError("AAI_6128", e.getMessage());
LOGGER.error("Encountered an exception during the historyTruncate: ", e);
return successStatus;
}
-
-
- public void processVerts(JanusGraph jgraph, Long truncBeforeTs,
+
+
+ public void processVerts(JanusGraph jgraph, Long truncBeforeTs,
Boolean doLogging, Boolean doDelete ) {
Graph g = jgraph.newTransaction();
GraphTraversalSource gts = g.traversal();
//Iterator <Vertex> vertItr = gts.V().has(AAIProperties.END_TS, P.lt(truncBeforeTs));
- Iterator <Vertex> vertItr = gts.V().has("end-ts", P.lt(truncBeforeTs));
- ArrayList <Long> vidList = new ArrayList <Long> ();
- while( vertItr.hasNext() ) {
- Vertex tmpV = vertItr.next();
- Long tmpVid = Long.valueOf(tmpV.id().toString());
- vidList.add(tmpVid);
- }
+ List<Long> vidList = gts.V().has("end-ts", P.lt(truncBeforeTs))
+ .toStream()
+ .map(v -> v.id().toString())
+ .map(Long::valueOf)
+ .collect(Collectors.toList());
int vTotalCount = vidList.size();
int batchCount = vTotalCount / batchCommitSize;
if((batchCount * batchCommitSize) < vTotalCount){
batchCount++;
}
-
+
LOGGER.info( " Vertex TotalCount = " + vTotalCount +
- ", we get batchCount = " + batchCount +
+ ", we get batchCount = " + batchCount +
", using commit size = " + batchCommitSize );
-
- int vIndex = 0;
+
+ int vIndex = 0;
for(int batchNo=1; batchNo<=batchCount; batchNo++){
- ArrayList <Long> batchVids = new ArrayList <Long> ();
- int thisBVCount = 0;
- while( (thisBVCount < batchCommitSize) && (vIndex < vTotalCount) ) {
+ ArrayList <Long> batchVids = new ArrayList<Long>();
+ int thisBVCount = 0;
+ while((thisBVCount < batchCommitSize) && (vIndex < vTotalCount)) {
batchVids.add(vidList.get(vIndex));
thisBVCount++;
vIndex++;
}
// now process this batch
LOGGER.info( "Process vertex batch # " + batchNo +
- ", which contains " + batchVids.size() + " ids. ");
+ ", which contains " + batchVids.size() + " ids. ");
processVertBatch(jgraph, doLogging, doDelete, batchVids);
}
}
-
- private void processVertBatch(JanusGraph jgraph, Boolean doLogging,
+
+ private void processVertBatch(JanusGraph jgraph, Boolean doLogging,
Boolean doDelete, ArrayList <Long> vidList ) {
Graph g = jgraph.newTransaction();
int delFailCount = 0;
int vCount = 0;
int delCount = 0;
-
- Iterator <Vertex> vertItr = gts.V(vidList);
- while( vertItr.hasNext() ) {
+
+ Iterator<Vertex> vertItr = gts.V(vidList);
+ while(vertItr.hasNext()) {
vCount++;
Vertex tmpV = vertItr.next();
String tmpVid = tmpV.id().toString();
String tmpPropsStr = "";
- if( doLogging ) {
+ if(doLogging) {
Iterator<VertexProperty<Object>> pI = tmpV.properties();
while( pI.hasNext() ){
VertexProperty<Object> tp = pI.next();
}
LOGGER.info(" vid = " + tmpVid + ", props: (" + tmpPropsStr + ") " );
}
-
- if( doDelete ) {
+
+ if(doDelete) {
LOGGER.info("Removing vid = " + tmpVid );
try {
tmpV.remove();
}
}
}
-
- if( doDelete ) {
+
+ if(doDelete) {
LOGGER.info("Calling commit on delete of Vertices." );
try {
g.tx().commit();
} catch ( Exception e ) {
- LOGGER.error("ERROR trying to commit Vertex Deletes for this batch. " +
+ LOGGER.error("ERROR trying to commit Vertex Deletes for this batch. " +
LogFormatTools.getStackTop(e) );
- LOGGER.info( vCount + " candidate vertices processed. "
+ LOGGER.info( vCount + " candidate vertices processed. "
+ " vertex deletes - COMMIT FAILED. ");
return;
}
}
-
- if( doDelete ) {
+
+ if(doDelete) {
LOGGER.info( vCount + " candidate vertices processed. " +
delFailCount + " delete attempts failed, " +
delCount + " deletes successful. ");
}
}
-
- public void processEdges(JanusGraph jgraph, Long truncBeforeTs,
+
+ public void processEdges(JanusGraph jgraph, Long truncBeforeTs,
Boolean doLogging, Boolean doDelete ) {
Graph g = jgraph.newTransaction();
Edge tmpE = edgeItr.next();
String tmpEid = tmpE.id().toString();
eidList.add(tmpEid);
- }
+ }
int eTotalCount = eidList.size();
int batchCount = eTotalCount / batchCommitSize;
if((batchCount * batchCommitSize) < eTotalCount){
batchCount++;
}
-
+
LOGGER.info( " Edge TotalCount = " + eTotalCount +
- ", we get batchCount = " + batchCount +
+ ", we get batchCount = " + batchCount +
", using commit size = " + batchCommitSize );
-
- int eIndex = 0;
+
+ int eIndex = 0;
for(int batchNo=1; batchNo<=batchCount; batchNo++){
ArrayList <String> batchEids = new ArrayList <String> ();
- int thisBECount = 0;
+ int thisBECount = 0;
while( (thisBECount < batchCommitSize) && (eIndex < eTotalCount) ) {
batchEids.add(eidList.get(eIndex));
thisBECount++;
}
// now process this batch
LOGGER.info( "Process edge batch # " + batchNo +
- ", which contains " + batchEids.size() + " ids. ");
+ ", which contains " + batchEids.size() + " ids. ");
processEdgeBatch(jgraph, doLogging, doDelete, batchEids);
}
}
-
- private void processEdgeBatch(JanusGraph jgraph, Boolean doLogging,
+
+ private void processEdgeBatch(JanusGraph jgraph, Boolean doLogging,
Boolean doDelete, ArrayList <String> eidList ) {
Graph g = jgraph.newTransaction();
int delFailCount = 0;
int eCount = 0;
int delCount = 0;
-
+
Iterator <Edge> edgeItr = gts.E(eidList);
while( edgeItr.hasNext() ) {
eCount++;
Vertex conV = conVtxs.next();
tmpConVs = tmpConVs + "[" + conV.id().toString() + "] ";
}
- LOGGER.info(" eid = " + tmpEid
+ LOGGER.info(" eid = " + tmpEid
+ ", Connecting vids = " + tmpConVs
+ ", props: (" + tmpEProps + "). " );
}
-
+
if( doDelete ) {
LOGGER.info("Removing Edge eid = " + tmpEid );
try {
}
}
}
-
+
if( doDelete ) {
LOGGER.info("Calling commit on delete of Edges." );
try {
g.tx().commit();
} catch ( Exception e ) {
- LOGGER.error("ERROR trying to commit Edge Deletes for this batch. " +
+ LOGGER.error("ERROR trying to commit Edge Deletes for this batch. " +
LogFormatTools.getStackTop(e) );
- LOGGER.info( eCount + " candidate edges processed. "
+ LOGGER.info( eCount + " candidate edges processed. "
+ " edge deletes - COMMIT FAILED. ");
return;
}
}
-
+
if( doDelete ) {
LOGGER.info( eCount + " candidate edges processed. " +
delFailCount + " delete attempts failed, " +
LOGGER.info( eCount + " candidate edges processed in this batch. " );
}
}
-
-
+
+
public int getCandidateVertexCount(JanusGraph jgraph, int windowDaysVal) {
Graph g = jgraph.newTransaction();
GraphTraversalSource gts = g.traversal();
- Long truncTs = calculateTruncWindowEndTimeStamp(windowDaysVal);
+ Long truncTs = calculateTruncWindowEndTimeStamp(windowDaysVal);
//int candVCount = gts.V().has(AAIProperties.END_TS, P.lt(truncTs)).count().next().intValue();
int candVCount = gts.V().has("end-ts", P.lt(truncTs)).count().next().intValue();
- LOGGER.info( " for the timeStamp = " + truncTs
- + ", which corresponds to the passed truncateWindowDays = "
- + windowDaysVal
- + ", found " + candVCount
+ LOGGER.info( " for the timeStamp = " + truncTs
+ + ", which corresponds to the passed truncateWindowDays = "
+ + windowDaysVal
+ + ", found " + candVCount
+ " candidate vertices. ");
return candVCount;
}
-
+
public int getCandidateEdgeCount(JanusGraph jgraph, int windowDaysVal) {
Graph g = jgraph.newTransaction();
GraphTraversalSource gts = g.traversal();
- Long truncTs = calculateTruncWindowEndTimeStamp(windowDaysVal);
+ Long truncTs = calculateTruncWindowEndTimeStamp(windowDaysVal);
//int candECount = gts.E().has(AAIProperties.END_TS, P.lt(truncTs)).count().next().intValue();
int candECount = gts.E().has("end-ts", P.lt(truncTs)).count().next().intValue();
- LOGGER.info( " for the timeStamp = " + truncTs
- + ", which corresponds to the passed truncateWindowDays = "
- + windowDaysVal
- + ", found " + candECount
+ LOGGER.info( " for the timeStamp = " + truncTs
+ + ", which corresponds to the passed truncateWindowDays = "
+ + windowDaysVal
+ + ", found " + candECount
+ " candidate Edges. ");
return candECount;
}
-
+
public static void verifyGraph(JanusGraph graph) {
if (graph == null) {
- String emsg = "Not able to get a graph object in DataSnapshot.java\n";
+ String emsg = "Not able to get a graph object in HistoryTruncate.java\n";
LOGGER.debug(emsg);
AAISystemExitUtil.systemExitCloseAAIGraph(1);
}
}
public long calculateTruncWindowEndTimeStamp( int timeWindowDays ){
- // Given a window size in days, calculate the timestamp that
+ // Given a window size in days, calculate the timestamp that
// represents the early-edge of that window.
-
+
long unixTimeNow = System.currentTimeMillis();
if( timeWindowDays <= 0 ){
// This just means that they want to truncate all the way up to the current time
return unixTimeNow;
}
-
+
long windowInMillis = timeWindowDays * 24 * 60 * 60L * 1000;
long windowEdgeTimeStampInMs = unixTimeNow - windowInMillis;
return windowEdgeTimeStampInMs;
-
- }
+
+ }
private Boolean doLoggingOrNot( String truncMode ){
if( truncMode.equals(SILENT_DELETE_MODE) ){
else {
return true;
}
- }
-
+ }
+
private Boolean doDeleteOrNot( String truncMode ){
if( truncMode.equals(LOG_ONLY_MODE) ){
return false;
else {
return true;
}
- }
-
-
-}
\ No newline at end of file
+ }
+
+
+}
Code Review / aai / graphadmin.git / commitdiff