Misc updates to collector

Change-Id: I7a3b52c454442d6a9f93331142b78bf671ffcd0a
Signed-off-by: VENKATESH KUMAR <vv770d@att.com>
Issue-ID: DCAEGEN2-373

2 files changed, 120 insertions, 50 deletions

diff --git a/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java b/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java
index 7b2e0b3b..db4a5ad7 100644
--- a/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java
+++ b/src/main/java/org/onap/dcae/controller/FetchDynamicConfig.java
@@ -23,10 +23,20 @@ package org.onap.dcae.controller;
 import org.json.JSONArray;

 import org.json.JSONObject;

 import org.json.JSONTokener;

+import org.onap.dcae.commonFunction.CommonStartup;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 

+import com.fasterxml.jackson.databind.JsonNode;

+import com.fasterxml.jackson.databind.ObjectMapper;

+import com.google.gson.JsonElement;

+import com.google.gson.JsonObject;

+import com.google.gson.JsonParser;

+

 import java.io.BufferedReader;

+import java.io.File;

+import java.io.FileNotFoundException;

+import java.io.FileReader;

 import java.io.FileWriter;

 import java.io.IOException;

 import java.io.InputStreamReader;

@@ -39,48 +49,83 @@ public class FetchDynamicConfig {
 	public static String configFile = "/opt/app/KV-Configuration.json";

 	static String url;

 	public static String retString;

-	public static String retCBSString;		

+	public static String retCBSString;

 	public static Map<String, String> env;

-	

+

 	public FetchDynamicConfig() {

 	}

 

 	public static void main(String[] args) {

-		

-		//Call consul api and identify the CBS Service address and port

+		Boolean areEqual = false;

+		// Call consul api and identify the CBS Service address and port

 		getconsul();

-		//Construct and invoke CBS API to get application Configuration

+		// Construct and invoke CBS API to get application Configuration

 		getCBS();

-		//Write data returned into configFile for LoadDynamicConfig process to pickup

-		FetchDynamicConfig fc= new FetchDynamicConfig();

-		fc.writefile(retCBSString);

+		// Verify if data has changed

+		areEqual = verifyConfigChange();

+		// If new config then write data returned into configFile for

+		// LoadDynamicConfig process

+		if (! areEqual) {

+			FetchDynamicConfig fc = new FetchDynamicConfig();

+			fc.writefile(retCBSString);

+		} else {

+			log.info("New config pull results identical -  " + configFile + " NOT refreshed");

+		}

 	}

-	

-	public  static void getconsul()

-	{

-		

+

+	public static void getconsul() {

+

 		env = System.getenv();

 		for (Map.Entry<String, String> entry : env.entrySet()) {

-			log.info( entry.getKey() + ":"+ entry.getValue());

+			log.info(entry.getKey() + ":" + entry.getValue());

 		}

 

 		if (env.containsKey("CONSUL_HOST") && env.containsKey("CONFIG_BINDING_SERVICE")) {

-//				&& env.containsKey("HOSTNAME")) {

+			// && env.containsKey("HOSTNAME")) {

 			log.info(">>>Dynamic configuration to be fetched from ConfigBindingService");

 			url = env.get("CONSUL_HOST") + ":8500/v1/catalog/service/" + env.get("CONFIG_BINDING_SERVICE");

 

 			retString = executecurl(url);

-			

-			

+

 		} else {

 			log.info(">>>Static configuration to be used");

 		}

 

-		

 	}

 

-	public static void getCBS()

-	{

+	public static boolean verifyConfigChange() {

+

+		boolean areEqual = false;

+		// Read current data

+		try {

+			File f = new File(configFile);

+			if (f.exists() && !f.isDirectory()) {

+

+				String jsonData = LoadDynamicConfig.readFile(configFile);

+				JSONObject jsonObject = new JSONObject(jsonData);

+

+				ObjectMapper mapper = new ObjectMapper();

+

+				JsonNode tree1 = mapper.readTree(jsonObject.toString());

+				JsonNode tree2 = mapper.readTree(retCBSString.toString());

+				areEqual = tree1.equals(tree2);

+				log.info("Comparison value:" + areEqual);

+			} else {

+				log.info("First time config file read: " + configFile);

+				// To allow first time file creation

+				areEqual = false;

+			}

+

+		} catch (IOException e) {

+			log.error("Comparison with new fetched data failed" + e.getMessage());

+

+		}

+

+		return areEqual;

+

+	}

+

+	public static void getCBS() {

 

 		env = System.getenv();

 		// consul return as array

@@ -93,40 +138,33 @@ public class FetchDynamicConfig {
 		}

 

 		log.info("CONFIG_BINDING_SERVICE DNS RESOLVED:" + urlPart1);

-		

-		if (env.containsKey("HOSTNAME"))

-		{

+

+		if (env.containsKey("HOSTNAME")) {

 			url = urlPart1 + "/service_component/" + env.get("HOSTNAME");

 			retCBSString = executecurl(url);

-		}

-		else if (env.containsKey("SERVICE_NAME"))

-		{

+		} else if (env.containsKey("SERVICE_NAME")) {

 			url = urlPart1 + "/service_component/" + env.get("SERVICE_NAME");

 			retCBSString = executecurl(url);

-		}

-		else

-		{

+		} else {

 			log.error("Service name environment variable - HOSTNAME/SERVICE_NAME not found within container ");

 		}

-	

+

 	}

-	

-	public void writefile (String retCBSString)

-	{

-		log.info("URL to fetch configuration:" + url  + " Return String:" + retCBSString);

-

-		

-		String indentedretstring=(new JSONObject(retCBSString)).toString(4);

-		

+

+	public void writefile(String retCBSString) {

+		log.info("URL to fetch configuration:" + url + " Return String:" + retCBSString);

+

+		String indentedretstring = (new JSONObject(retCBSString)).toString(4);

+

 		try (FileWriter file = new FileWriter(FetchDynamicConfig.configFile)) {

 			file.write(indentedretstring);

 

 			log.info("Successfully Copied JSON Object to file " + configFile);

 		} catch (IOException e) {

-			log.error("Error in writing configuration into file " + configFile  + retString +  e.getMessage());

+			log.error("Error in writing configuration into file " + configFile + retString + e.getMessage());

 			e.printStackTrace();

 		}

-		

+

 	}

 

 	public static String executecurl(String url) {

diff --git a/src/main/java/org/onap/dcae/restapi/endpoints/EventReceipt.java b/src/main/java/org/onap/dcae/restapi/endpoints/EventReceipt.java
index f45f60c3..204e5058 100644
--- a/src/main/java/org/onap/dcae/restapi/endpoints/EventReceipt.java
+++ b/src/main/java/org/onap/dcae/restapi/endpoints/EventReceipt.java
@@ -44,6 +44,8 @@ import org.slf4j.LoggerFactory;
 import java.io.FileReader;

 import java.io.IOException;

 import java.io.InputStream;

+import java.nio.charset.Charset;

+import java.util.Base64;

 import java.util.UUID;

 import java.util.regex.Matcher;

 import java.util.regex.Pattern;

@@ -68,11 +70,11 @@ public class EventReceipt extends NsaBaseEndpoint {
 		InputStream istr = null;

 		int arrayFlag = 0;

 		String vesVersion = null;

+		String userId=null;

 

 		try {

 

 

-			log.debug("Request recieved :" + ctx.request().getRemoteAddress());

 			istr = ctx.request().getBodyStream();

 			jsonObject = new JSONObject(new JSONTokener(istr));

 

@@ -107,16 +109,22 @@ public class EventReceipt extends NsaBaseEndpoint {
 

 			try {

 				if (CommonStartup.authflag == 1) {

+					userId = getUser (ctx);

 					retkey = NsaBaseEndpoint.getAuthenticatedUser(ctx);

 				}

 			} catch (NullPointerException x) {

-				log.info("Invalid user request " + ctx.request().getContentType() + MESSAGE + jsonObject);

-				CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Unauthorized user" + x);

+				log.info("Invalid user request " + userId + ctx.request().getContentType() + MESSAGE + jsonObject);

+				CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Unauthorized user" + userId +  x);

 				respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Invalid user");

 				return;

 			}

 			

-			schemaCheck( retkey,  arrayFlag, jsonObject,  vesVersion,  ctx,  uuid);

+			Boolean ErrorStatus = false;

+			ErrorStatus = schemaCheck( retkey,  arrayFlag, jsonObject,  vesVersion,  ctx,  uuid);

+			if (ErrorStatus)

+			{

+				return;

+			}

 

 		} catch (JSONException | NullPointerException | IOException x) {

 			log.error(String.format("Couldn't parse JSON Array - HttpStatusCodes.k400_badRequest%d%s%s",

@@ -142,11 +150,29 @@ public class EventReceipt extends NsaBaseEndpoint {
 		ctx.response().sendErrorAndBody(HttpStatusCodes.k200_ok, "Message Accepted", MimeTypes.kAppJson);

 	}

 	

-	public static void schemaCheck(NsaSimpleApiKey retkey, int arrayFlag,JSONObject jsonObject, String vesVersion,  DrumlinRequestContext ctx, UUID uuid) throws JSONException, QueueFullException, IOException

+	

+	public static String getUser( DrumlinRequestContext ctx){

+		String authorization = null;

+		authorization = ctx.request().getFirstHeader("Authorization");

+		 if (authorization != null && authorization.startsWith("Basic")) {

+		        // Authorization: Basic base64credentials

+		        String base64Credentials = authorization.substring("Basic".length()).trim();

+		        String credentials = new String(Base64.getDecoder().decode(base64Credentials),

+		                Charset.forName("UTF-8"));

+		        // credentials = username:password

+		        final String[] values = credentials.split(":",2);

+		        log.debug("User:" + values[0].toString() + " Pwd:" + values[1].toString());

+		        return values[0].toString();

+		 }

+		 return null;

+		

+	}

+	public static Boolean schemaCheck(NsaSimpleApiKey retkey, int arrayFlag,JSONObject jsonObject, String vesVersion,  DrumlinRequestContext ctx, UUID uuid) throws JSONException, QueueFullException, IOException

 	{

 		JSONArray jsonArray;

 		JSONArray jsonArrayMod = new JSONArray();

 		JSONObject event;

+		Boolean ErrorStatus=false;

 		FileReader fr;

 		if (retkey != null || CommonStartup.authflag == 0) {

 			if (CommonStartup.schemaValidatorflag > 0) {

@@ -162,17 +188,20 @@ public class EventReceipt extends NsaBaseEndpoint {
 						log.info("Validation failed");

 						respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,

 								"Schema validation failed");

-						return;

+						ErrorStatus=true;

+						return ErrorStatus;

 					} else {

 						log.error("Validation errored" + valresult);

 						respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,

 								"Couldn't parse JSON object");

-						return;

+						ErrorStatus=true;

+						return ErrorStatus;

 					}

 				} else {

 					log.info("Validation failed");

 					respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest, "Schema validation failed");

-					return;

+					ErrorStatus=true;

+					return ErrorStatus;

 				}

 				if (arrayFlag == 1) {

 					jsonArray = jsonObject.getJSONArray("eventList");

@@ -197,16 +226,19 @@ public class EventReceipt extends NsaBaseEndpoint {
 						ctx.request().getContentType(), jsonObject));

 				respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,

 						"Incorrect message content-type; only accepts application/json messages");

-				return;

+				ErrorStatus=true;

+				return ErrorStatus;

 			}

 

 			CommonStartup.handleEvents(jsonArrayMod);

 		} else {

-			log.info(String.format("Unauthorized request %s%s%s", ctx.request().getContentType(), MESSAGE,

+			log.info(String.format("Unauthorized request %s%s%s%s", getUser(ctx), ctx.request().getContentType(), MESSAGE,

 					jsonObject));

 			respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Unauthorized user");

-			return;

+			ErrorStatus=true;

+			return ErrorStatus;

 		}

+		return ErrorStatus;

 	}

 

 	public static void respondWithCustomMsginJson(DrumlinRequestContext ctx, int sc, String msg) {