Fix warning and retry

git-svn-id: https://svn.d4science.research-infrastructures.eu/gcube/trunk/accounting/accounting-aggregator-se-plugin@141773 82a268e6-3cf1-43bd-a215-b396298e98cf

pom.xml

@@ -8,7 +8,7 @@
 	</parent>
 	<groupId>org.gcube.accounting</groupId>
 	<artifactId>accounting-aggregator-se-plugin</artifactId>
-	<version>1.0.0-SNAPSHOT</version>
+	<version>1.0.1-SNAPSHOT</version>
 	<name>Accounting Aggregator</name>
 	<description>Accounting Aggregator Smart Executor Plugin</description>
 

src/main/java/org/gcube/accounting/aggregator/configuration/Constant.java

@@ -25,7 +25,9 @@ public class Constant {
 	
 	
 	
-	public static final Integer CONNECTION_TIMEOUT=10;
+	public static final Integer CONNECTION_TIMEOUT=15;
-	public static final Integer NUM_RETRY=5;
+	public static final Integer NUM_RETRY=6;
-	public static final Integer CONNECTION_TIMEOUT_BUCKET=10;
+	public static final Integer CONNECTION_TIMEOUT_BUCKET=15;
+	public static final Integer VIEW_TIMEOUT_BUCKET=120;
+	public static final Integer MAX_REQUEST_LIFE_TIME=120;
 }

src/main/java/org/gcube/accounting/aggregator/madeaggregation/Aggregation.java

@@ -33,7 +33,11 @@ public  class Aggregation  {
 	protected int totalBufferedRecords;
 
 	//list Aggregate record
-	protected Map<String, List<AggregatedRecord<?,?>>>  bufferedRecords = new HashMap<String, List<AggregatedRecord<?,?>>>();
+	//TODO RIMETTERE A PROTECTED
+	public Map<String, List<AggregatedRecord<?,?>>>  bufferedRecords = new HashMap<String, List<AggregatedRecord<?,?>>>();
+
+	
+	
 	
 	public Aggregation() {
 		super();
@@ -70,22 +74,19 @@ public  class Aggregation  {
 		List<AggregatedRecord<?,?>> records;
 		if(this.bufferedRecords.containsKey(recordType)){
 			records = this.bufferedRecords.get(recordType);
-			//logger.debug("value endtime{}, type endtime{}",record.getEndTime());
 			boolean found = false;
 			for(AggregatedRecord bufferedRecord : records){
 				if(!(bufferedRecord instanceof AggregatedRecord)){
 					continue;
 				}
-
 				AggregationUtility util = new AggregationUtility(bufferedRecord);
 				//verify a record is aggregable			
-				
 				//logger.debug("record: {}",record.toString());
 				if (util.isAggregable(record)){
 					try {
-
 						AggregatedRecord bufferedAggregatedRecord = (AggregatedRecord) bufferedRecord;
 						// TODO check compatibility using getAggregable
+						logger.debug("if -- madeAggregation aggregate");
 						bufferedAggregatedRecord.aggregate((AggregatedRecord) record);						
 						//patch for not changed a creation time
 						//bufferedAggregatedRecord.setCreationTime(bufferedAggregatedRecord.getStartTime());
@@ -93,34 +94,32 @@ public  class Aggregation  {
 						found = true;
 						break;
 					} catch(NotAggregatableRecordsExceptions e) {
-						logger.trace("{} is not usable for aggregation", bufferedRecord);
+						logger.debug("{} is not usable for aggregation", bufferedRecord);
 					} 
 				}
-				
 			}
 
 			if(!found){
 				//logger.debug("Aggregated Record not found with execption");
+				logger.debug("if -- madeAggregation  not found with execption add");
 				records.add(record);
 				totalBufferedRecords++;
 				return;
 			}
 
 		}else{
-
 			//logger.debug("else if record contains  "+recordType);
 			records = new ArrayList<AggregatedRecord<?,?>>();
 			try {
+				logger.debug("else -- add  getAggregatedRecord");
 				records.add(getAggregatedRecord(record));
 			} catch (Exception e) {
+				logger.debug("else -- add  Exception");				
 				records.add(record);
 			}
 			totalBufferedRecords++;
 			this.bufferedRecords.put(recordType, records);
 		}		
-
-
-
 	}
 
 

src/main/java/org/gcube/accounting/aggregator/plugin/AccountingAggregatorPlugin.java

@@ -64,8 +64,13 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 
 	public static Integer countInsert=0;
 	public static Integer countDelete=0;
-	public static Integer RecoveryMode=0;
+	public static Integer recoveryMode=0;
+	public Boolean backup=true;
+	//value if 0 PersistTo.MASTER if 1 PersistTo.ONE
+	public static Integer typePersisted=0;
 
+
+	protected PersistTo persisted ;
 	/**
 	 * @param runningPluginEvolution
 	 */
@@ -77,8 +82,11 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 	protected static final CouchbaseEnvironment ENV = 
 			DefaultCouchbaseEnvironment.builder()
 			.connectTimeout(Constant.CONNECTION_TIMEOUT * 1000) 
-			.queryTimeout(Constant.CONNECTION_TIMEOUT * 1000)
+			.maxRequestLifetime(Constant.MAX_REQUEST_LIFE_TIME * 1000)
+			.queryTimeout(Constant.CONNECTION_TIMEOUT * 1000)	//15 Seconds in milliseconds
+			.viewTimeout(Constant.VIEW_TIMEOUT_BUCKET * 1000)//120 Seconds in milliseconds			
 			.keepAliveInterval(3600 * 1000) // 3600 Seconds in milliseconds
+			.kvTimeout(5000) //in ms
 			.build();
 
 	/**{@inheritDoc}*/
@@ -98,6 +106,10 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		AggregationType aggType =AggregationType.valueOf((String)inputs.get("type"));
 		Integer interval=(Integer)inputs.get("interval")* aggType.getMultiplierFactor();
 
+		//new feature for not elaborate the full range but a set of small intervals
+		if (inputs.containsKey("intervalStep"))
+			interval=(Integer)inputs.get("intervalStep");
+
 		Integer inputStartTime=null;
 		if (inputs.containsKey("startTime"))
 			inputStartTime=(Integer)inputs.get("startTime");
@@ -116,9 +128,33 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 			Constant.user="service.aggregatorAccounting";
 
 		if (inputs.containsKey("recovery"))
-			RecoveryMode=(Integer)inputs.get("recovery");
+			recoveryMode=(Integer)inputs.get("recovery");
 
-		logger.debug("Launch with Type:{}, Interval:{}, startTime:{}, Scope:{}, Recovery:{}",aggType.toString(),interval,inputStartTime,scope,RecoveryMode);
+		if (inputs.containsKey("backup"))
+			backup=(Boolean)inputs.get("backup");
+
+		if (inputs.containsKey("typePersisted"))
+			typePersisted=(Integer)inputs.get("typePersisted");
+		switch(typePersisted) {
+			case 0:
+				persisted=PersistTo.MASTER;
+				break;
+			case 1:
+				persisted=PersistTo.ONE;
+				break;
+			default:
+				persisted=PersistTo.MASTER;
+		}
+
+		logger.debug("Launch with Type:{}, Interval:{}, startTime:{}, Scope:{}, Recovery:{}",aggType.toString(),interval,inputStartTime,scope,recoveryMode);
+		logger.debug("persist:{} backup:{}",persisted.toString(),backup);
+		if(!backup){
+			logger.warn("Attention backup disabled");
+			Thread.sleep(20000);
+		}
+		if (inputs.containsKey("intervalStep")){
+			logger.debug("Interval is not considered, aggregate only :{} step",interval);
+		}
 		
 		//Get Configuration from service end point
 		String url=null;
@@ -152,12 +188,12 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		initFolder();		
 
 
-		if ((RecoveryMode==2)||(RecoveryMode==0)){
+		if ((recoveryMode==2)||(recoveryMode==0)){
 			logger.debug("Recovery mode enabled");
 			RecoveryRecord.searchFile(cluster,configuration);
 		}
 
-		if (RecoveryMode!=2){
+		if (recoveryMode!=2){
 			for (String bucket:listBucket){
 				logger.trace("OpenBucket:{}",bucket);
 				accountingBucket = cluster.openBucket(bucket,password);
@@ -238,12 +274,16 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		String endAllKeyString = format.format(now.getTime());
 		String endKeyString = format.format(now.getTime());
 
-
 		//save a record modified into a file and save into a workspace 
 		nowTemp.add(aggType.getCalendarField(), -1*interval);
 		String startAllKeyString = format.format(nowTemp.getTime());
+		if (backup){
+			logger.debug("Start Backup");
 			WorkSpaceManagement.onSaveBackupFile(accountingBucket,bucket,scope,startAllKeyString, endAllKeyString,aggType);
 			//logger.debug("Backup complete startKeyString{}, endKeyString{}",startAllKeyString,endAllKeyString);
+		}
+		else
+			logger.debug("No Backup required");
 
 		List<JsonDocument> documentElaborate=new ArrayList<JsonDocument>();
 
@@ -269,25 +309,23 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 			query.endKey(endKey);
 			query.reduce(false);
 			query.inclusiveEnd(false);
-			logger.debug("View Query: startKey:{} - endKey:{} designDocId:{} - viewName:{}",startKey, endKey,designDocId,viewName);
+			logger.debug("--{}/{} View Query: startKey:{} - endKey:{} designDocId:{} - viewName:{}",i,interval,startKey, endKey,designDocId,viewName);
-
 			ViewResult viewResult = null;
 			try {
 				viewResult = accountingBucket.query(query);
 
 			} catch (Exception e) {
-				logger.error("ERROR VIEW",e.getLocalizedMessage());
+				logger.error("Exception error VIEW",e.getLocalizedMessage(),e);
 				//throw e;
 			}
 
 			// Iterate through the returned ViewRows
 			aggregate = new Aggregation();
 			documentElaborate.clear();
-
+			logger.debug("Start elaborate row");
 			for (ViewRow row : viewResult) 
 				elaborateRow(row,documentElaborate);
-
+			logger.debug("End elaborate row");
-
 			//File backup have a name with scope e 
 			String nameFileBackup="";
 			if (scope!=null)
@@ -311,39 +349,46 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 	 * @throws Exception
 	 */	
 	protected boolean elaborateRow(ViewRow row ,List<JsonDocument>  documentElaborate) throws Exception{
-
+		int i=0;
+		int size=aggregate.bufferedRecords.size();
 		try {
 			//patch for field of long type  
 			String document=row.value().toString().replace("\":", "=").replace("\"", "");
-			
+			i=1;//1
 			Map<String,? extends Serializable> map = getMapFromString(document);
+			i=2;//2
+			//prepare a document for elaborate
+			String identifier=(String) row.document().content().get("id");
+			i=5;//5
+			JsonDocument documentJson = JsonDocument.create(identifier, row.document().content());
+			i=6;//6
 
 			@SuppressWarnings("rawtypes")			
 			AggregatedRecord record = (AggregatedRecord)RecordUtility.getRecord(map);
-			
+			i=3;//3
 			aggregate.aggregate(record);
-			
+			i=4;//4			
 			//insert an elaborate row into list JsonDocument for memory document elaborate
-			
-			String identifier=(String) row.document().content().get("id");
-			
-			JsonDocument documentJson = JsonDocument.create(identifier, row.document().content());
-			
 			documentElaborate.add(documentJson);
-			
+			i=7;//7
 			return true;
 		} 
 		catch(InvalidValueException ex){
-			logger.warn("Record is not valid. Anyway, it will be persisted");
+			logger.warn("InvalidValueException - Record is not valid. Anyway, it will be persisted i:{}",i);
+			return true;
+		}
+		catch(RuntimeException exr){
+			logger.warn("Runtime Exception -Record is not valid. Anyway, it will be persisted i:{}",i);
 			return true;
 		}
 		catch (Exception e) {
-			
+			logger.error("record is not elaborated:"+row.toString()+" but it will be persisted");
-			logger.error("Error elaborateRow", e,e.getLocalizedMessage());			
+			logger.error("error elaborateRow", e);
+			logger.error("i:{}",i);			
+			logger.error("size before:{}, after buffer size:{}",size,aggregate.bufferedRecords.size());
 			//throw e;
 			return false;
 		}
-
 	}
 
 
@@ -357,7 +402,6 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		if(!serializedMap.startsWith(LINE_FREFIX) && !serializedMap.endsWith(LINE_SUFFIX)){
 			return null;
 		}
-    	
 		/* Cleaning prefix and suffix to parse line */
 		serializedMap = serializedMap.replace(LINE_FREFIX, "");
 		serializedMap = serializedMap.replace(LINE_SUFFIX, "");
@@ -368,7 +412,6 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		for (int i=0;i<pairs.length;i++) {
 			String pair = pairs[i];
 			pair.trim();
-            
 			String[] keyValue = pair.split(KEY_VALUE_LINKER);
 			String key = keyValue[0].trim();
 			Serializable value = keyValue[1].trim();
@@ -392,7 +435,7 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 		if (docs.size()!=0){
 			Integer index=0;
 			boolean succesfulDelete=false;
-
+			logger.trace("Start a delete document:{}",docs.size());
 			//before elaborate a record, create a backup file 
 			List<JsonDocument> notDeleted = docs;
 			List<JsonDocument> notInserted = aggregate.reallyFlush();
@@ -406,18 +449,30 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 				for (JsonDocument doc: notDeleted){	
 					countDelete ++;
 					try{
-						accountingBucket.remove(doc.id(),PersistTo.MASTER,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
+
+						//accountingBucket.remove(doc.id(),PersistTo.MASTER,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
+						accountingBucket.remove(doc.id(),persisted,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
 					}catch(Exception e){
 						logger.trace("doc:{} not deleted retry:{}",doc.id(),index);
+						Thread.sleep(1500);
 						try{
 							if (accountingBucket.exists(doc.id()))
 								notDeletedTemp.add(doc);
 						}
+						catch(Exception ext){
+							logger.warn("doc:{} not verify for delete because timeout, retry",doc.id());
+							Thread.sleep(6000);
+							try{
+								if (!accountingBucket.exists(doc.id()))
+									notDeletedTemp.add(doc);
+							}
 							catch(Exception ex)	{
-							logger.warn("doc:{} not verify for delete",doc.id());
+								logger.error("doc:{} not delete ({}), problem with exist bucket",doc.id(),doc.toString(),ex);
+								logger.error("force insert into list for delete");
+								notDeletedTemp.add(doc);
+							}
 						}						
 					}
-
 				}
 				if (notDeletedTemp.isEmpty()){
 					succesfulDelete=true;					
@@ -431,10 +486,8 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 			if (!succesfulDelete){
 				logger.error("Error Delete record");
 			}
-			logger.debug("Delete complete {}, Start a insert aggregated document",countDelete);
+			logger.debug("Delete complete:{}, Start a insert aggregated document:{}",countDelete,notInserted.size());
-			/**
+			// delete all record and ready for insert a new aggregated record			 
-			 * delete all record and ready for insert a new aggregated record
-			 */
 			if (succesfulDelete){
 				//if successful record delete, delete backup file 
 				ManagementFileBackup.getInstance().onDeleteFile(Constant.FILE_RECORD_NO_AGGREGATE+"_"+nameFile,false);
@@ -445,17 +498,29 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 					for (JsonDocument document: notInserted){
 						countInsert ++;
 						try{
-							@SuppressWarnings("unused")
+							//JsonDocument response = accountingBucket.upsert(document,PersistTo.MASTER,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
-							JsonDocument response = accountingBucket.upsert(document,PersistTo.MASTER,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
+							JsonDocument response = accountingBucket.upsert(document,persisted,Constant.CONNECTION_TIMEOUT_BUCKET, TimeUnit.SECONDS);
 						}
 						catch(Exception e){
 							logger.trace("record:{} not insert retry:{} ",document.id(),index);
+							Thread.sleep(1500);
+							try{
+
+								if (!accountingBucket.exists(document.id()))
+									notInsertedTemp.add(document);									
+							}
+							catch(Exception ext){
+								logger.warn("doc:{} not verify for insert because timeout, retry",document.id(),ext);
+								Thread.sleep(3000);
 								try{
 									if (!accountingBucket.exists(document.id()))
 										notInsertedTemp.add(document);
 								}
 								catch(Exception ex)	{
-								logger.warn("doc:{} not verify for inset",document.id());
+									logger.error("doc:{} not insert ({}), problem with exist bucket",document.id(),document.toString(),ex);
+									logger.error("force insert into list for insert");									
+									notInsertedTemp.add(document);
+								}
 							}							
 						}
 					}
@@ -477,6 +542,7 @@ public class AccountingAggregatorPlugin extends Plugin<AccountingAggregatorPlugi
 				}
 
 			}
+			logger.trace("Insert complete");
 		}
 		return true;
 	}

src/main/java/org/gcube/accounting/aggregator/plugin/WorkSpaceManagement.java

@@ -97,14 +97,31 @@ public class WorkSpaceManagement {
 
 			Thread.sleep(500);
 			filebackup = new BufferedWriter(new FileWriter(logFile));
+			int count = 0;
+			int maxTries = 3;
+			boolean exitRetry=false;
 			for (ViewRow row : viewResult){
+				while(!exitRetry) {
+					try {
 						if (row.document()!=null){
 							if (!row.document().content().toString().isEmpty()){
 								filebackup.write(row.document().content().toString());
 								filebackup.newLine();
 							}
 						}
+						exitRetry=true;
+					} catch (Exception e) {
+						logger.error("retry:{}",count);
+						logger.error(e.getMessage());
+						if (++count == maxTries){ 
+							filebackup.close();	
+							throw e;
 						}
+					}
+				}
+			}
+
+
 			filebackup.close();			
 			//create a zip file
 			byte[] buffer = new byte[1024];