Merge pull request #7 from pagopa/NOD-836-fdr-gestire-gli-errori-di-r…

…iversamento

[NOD-836] feat: handle errors during data migration

README.md

@@ -1,7 +1,7 @@
 # pagoPA Functions fdr-json-to-xml-fn
 
 Java fdr-json-to-xml Azure Function.
-The function aims to dump RE sent via Azure Event Hub to a CosmosDB, with a TTL of 120 days, and to an Azure Table Storage with a TTL of 10 years.
+The function aims to convert FdR reports from JSON format to XML format.
 
 [![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=pagopa_pagopa-fdr-re-to-datastore&metric=alert_status)](https://sonarcloud.io/dashboard?id=pagopa_pagopa-fdr-re-to-datastore)
 
@@ -18,10 +18,13 @@ The function aims to dump RE sent via Azure Event Hub to a CosmosDB, with a TTL
 
 ## Run locally with Maven
 
+In order to autogenerate the required classes, please run the command:  
 `mvn clean package`
 
+In order to test the Azure Function in local environment, please run the command:
 `mvn azure-functions:run`
 
+
 ### Test
 `curl http://localhost:7071/example`
 

host.json

@@ -5,23 +5,28 @@
     "version": "[4.0.0, 5.0.0)"
   },
   "extensions": {
+    "tracing": {
+      "traceInputsAndOutputs": false,
+      "traceReplayEvents": false
+    },
     "http": {
       "routePrefix": ""
     }
   },
   "functions": [ "Info", "QueueFdrJsonToXmlEventProcessor", "JsonErrorRetry" ],
   "logging": {
+    "logLevel": {
+      "default": "Error",
+      "Function.QueueFdrJsonToXmlEventProcessor": "Information",
+      "Function.JsonErrorRetry": "Information"
+    },
     "applicationInsights": {
       "samplingSettings": {
-        "isEnabled": false
+        "isEnabled": true,
+        "maxTelemetryItemsPerSecond": 5,
+        "includedTypes": "PageView;Trace;Dependency;Request",
+        "excludedTypes": "Exception;Event;CustomEvent"
       }
-    },
-    "fileLoggingMode": "always",
-    "logLevel": {
-      "default": "Information",
-      "Host.Results": "Error",
-      "Function": "Information",
-      "Host.Aggregator": "Trace"
     }
   }
 }

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>it.gov.pagopa</groupId>
     <artifactId>fdrjsontoxml</artifactId>
-    <version>0.2.0</version>
+    <version>0.2.0-2-NOD-836-fdr-gestire-gli-errori-di-riversamento</version>
     <packaging>jar</packaging>
 
     <name>FDR JSON to XML Fn</name>

src/main/java/it/gov/pagopa/fdrjsontoxml/FdrJsonToXml.java

@@ -4,8 +4,11 @@
 import com.azure.data.tables.TableServiceClient;
 import com.azure.data.tables.TableServiceClientBuilder;
 import com.azure.data.tables.models.TableEntity;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.microsoft.azure.functions.ExecutionContext;
+import com.microsoft.azure.functions.annotation.ExponentialBackoffRetry;
 import com.microsoft.azure.functions.annotation.FunctionName;
 import com.microsoft.azure.functions.annotation.QueueTrigger;
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -19,12 +22,15 @@
 import java.util.*;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.regex.Pattern;
 
 /**
  * Azure Functions with Azure Queue trigger.
  */
 public class FdrJsonToXml {
 
+	private static final Integer MAX_RETRY_COUNT = 10;
+
 	private static final String fdrFase1BaseUrl = System.getenv("FDR_FASE1_BASE_URL");
 	private static final String fdrFase1NewApiKey = System.getenv("FDR_FASE1_API_KEY");
 
@@ -56,6 +62,7 @@ private static TableServiceClient getTableServiceClient(){
 
 
     @FunctionName("QueueFdrJsonToXmlEventProcessor")
+	@ExponentialBackoffRetry(maxRetryCount = 10, maximumInterval = "01:30:00", minimumInterval = "00:00:10")
     public void processNodoReEvent (
 			@QueueTrigger(
 					name = "jsonTrigger",
@@ -64,34 +71,62 @@ public void processNodoReEvent (
 			String b64Message,
 			final ExecutionContext context) {
 
+		String errorCause = null;
+		boolean isPersistenceOk = true;
+		Throwable causedBy = null;
+		int retryIndex = context.getRetryContext() == null ? -1 : context.getRetryContext().getRetrycount();
+
 		Logger logger = context.getLogger();
+		if (retryIndex == MAX_RETRY_COUNT) {
+			logger.log(Level.WARNING, () -> String.format("[ALERT][FdrJsonToXml][LAST_RETRY] Performing last retry for event ingestion: InvocationId [%s]", context.getInvocationId()));
+		}
+
 		String message = new String(Base64.getDecoder().decode(b64Message), StandardCharsets.UTF_8);
-		logger.info("Queue message: " + message);
 
 		try {
+
 			// read json
 			ObjectMapper objectMapper = new ObjectMapper();
 			FdrMessage fdrMessage = objectMapper.readValue(message, FdrMessage.class);
-			logger.info("Process fdr=["+fdrMessage.getFdr()+"], pspId=["+fdrMessage.getPspId()+"]");
+			logger.log(Level.INFO, () -> String.format("Performing event ingestion: InvocationId [%s], Retry Attempt [%d], File name:[%s]", context.getInvocationId(), retryIndex, fdrMessage.getFdr()));
 
 			// create body for notify FDR
 			Fdr fdr = getFdr(fdrMessage);
+
 			// call notify FDR
-			logger.info("Calling... notify");
-			manageHttpError(logger, message, () ->
+			manageHttpError(message, () ->
 					getFdrFase1Api().notifyFdrToConvert(fdr)
 			);
 
-			logger.info("Done processing events");
+		} catch (JsonProcessingException e) {
+			logger.log(Level.INFO, () -> String.format("Performing event ingestion: InvocationId [%s], Retry Attempt [%d], File name:[UNKNOWN]", context.getInvocationId(), retryIndex));
+
+			Instant now = Instant.now();
+			ErrorEnum error = ErrorEnum.GENERIC_ERROR;
+			sendGenericError(now, message, error, e);
+
+			isPersistenceOk = false;
+			errorCause = getErrorMessage(error, message, now);
+			causedBy = e;
 		} catch (AppException e) {
-			logger.info("Failure processing events");
+			isPersistenceOk = false;
+			errorCause = e.getMessage();
+			causedBy = e;
 		} catch (Exception e) {
 			Instant now = Instant.now();
 			ErrorEnum error = ErrorEnum.GENERIC_ERROR;
-			logger.log(Level.SEVERE, getErrorMessage(error, message, now), e);
-			sendGenericError(logger, now, message, error, e);
-			logger.info("Failure processing events");
+			sendGenericError(now, message, error, e);
+
+			isPersistenceOk = false;
+			errorCause = getErrorMessage(error, message, now);
+			causedBy = e;
         }
+
+		if (!isPersistenceOk) {
+			String finalErrorCause = errorCause;
+			logger.log(Level.SEVERE, () -> finalErrorCause);
+			throw new AppException(errorCause, causedBy);
+		}
     }
 
 	private Fdr getFdr(FdrMessage fdrMessage){
@@ -109,41 +144,39 @@ public interface SupplierWithApiException<T> {
 		T get() throws ApiException;
 	}
 	private static String getErrorMessage(ErrorEnum errorEnum, String message, Instant now){
-		return "[ALERT] [error="+errorEnum.name()+"] [message="+message+"] Http error at "+ now;
+		return "[ALERT][FdrJsonToXml][error="+errorEnum.name()+"] [message="+message+"] Http error at "+ now;
 	}
-	private static<T> void manageHttpError(Logger logger, String message, SupplierWithApiException<T> fn){
+	private static<T> void manageHttpError(String message, SupplierWithApiException<T> fn){
 		try {
 			fn.get();
 		} catch (ApiException e) {
-			String errorResposne = e.getResponseBody();
-
+			String errorResponse = e.getResponseBody();
 			Instant now = Instant.now();
 			ErrorEnum error = ErrorEnum.HTTP_ERROR;
-			logger.log(Level.SEVERE, getErrorMessage(error, message, now), e);
-			sendHttpError(logger, now, message, error, errorResposne, e);
-			throw new AppException(message, e);
+			String errorMessage = getErrorMessage(error, message, now);
+			sendHttpError(now, message, error, errorResponse, e);
+			throw new AppException(errorMessage, e);
 		}
 	}
-	private static void sendGenericError(Logger logger, Instant now, String message, ErrorEnum errorEnum, Exception e){
-		_sendToErrorTable(logger, now, message, errorEnum,   Optional.empty(), e);
+	private static void sendGenericError(Instant now, String message, ErrorEnum errorEnum, Exception e){
+		_sendToErrorTable(now, message, errorEnum,   Optional.empty(), e);
 	}
-	private static void sendHttpError(Logger logger, Instant now, String message, ErrorEnum errorEnum, String httpErrorResposne, Exception e){
-		_sendToErrorTable(logger, now, message, errorEnum, Optional.ofNullable(httpErrorResposne), e);
+	private static void sendHttpError(Instant now, String message, ErrorEnum errorEnum, String httpErrorResposne, Exception e){
+		_sendToErrorTable(now, message, errorEnum, Optional.ofNullable(httpErrorResposne), e);
 	}
 
-	private static void _sendToErrorTable(Logger logger, Instant now, String message, ErrorEnum errorEnum, Optional<String> httpErrorResposne, Exception e){
+	private static void _sendToErrorTable(Instant now, String message, ErrorEnum errorEnum, Optional<String> httpErrorResponse, Exception e){
 		String id = UUID.randomUUID().toString();
 		Map<String,Object> errorMap = new LinkedHashMap<>();
 		errorMap.put(AppConstant.columnFieldId, id);
 		errorMap.put(AppConstant.columnFieldCreated, now);
 		errorMap.put(AppConstant.columnFieldMessage, message);
 		errorMap.put(AppConstant.columnFieldErrorType, errorEnum.name());
-		httpErrorResposne.ifPresent(a -> errorMap.put(AppConstant.columnFieldHttpErrorResposne, a));
+		httpErrorResponse.ifPresent(a -> errorMap.put(AppConstant.columnFieldHttpErrorResposne, a));
 		errorMap.put(AppConstant.columnFieldStackTrace, ExceptionUtils.getStackTrace(e));
 
 		String partitionKey =  now.toString().substring(0,10);
 
-		logger.info("Send to "+tableName+" record with "+AppConstant.columnFieldId+"="+id);
 		TableClient tableClient = getTableServiceClient().getTableClient(tableName);
 		TableEntity entity = new TableEntity(partitionKey, id);
 		entity.setProperties(errorMap);