https://jenkins.onap.org/job/dcaegen2-collectors-ves-master-merge-java/33/console
Change-Id: I90693665dcf650028186a47ed69eaf15bc2f0d15
Issue-ID:DCAEGEN2-82
Signed-off-by: Vijay <vv770d@att.com>
package org.onap.dcae.commonFunction;
-
import com.att.nsa.apiServer.ApiServer;
import com.att.nsa.apiServer.ApiServerConnector;
import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;
public static final String KSETTING_PORT = "collector.service.port";
public static final int KDEFAULT_PORT = 8080;
-
+
public static final String KSETTING_SECUREPORT = "collector.service.secure.port";
- public static final int KDEFAULT_SECUREPORT = -1;
-
+ public static final int KDEFAULT_SECUREPORT = -1;
+
public static final String KSETTING_KEYSTOREPASSFILE = "collector.keystore.passwordfile";
- public static final String KDEFAULT_KEYSTOREPASSFILE = "../etc/passwordfile";
+ public static final String KDEFAULT_KEYSTOREPASSFILE = "../etc/passwordfile";
public static final String KSETTING_KEYSTOREFILE = "collector.keystore.file.location";
public static final String KDEFAULT_KEYSTOREFILE = "../etc/keystore";
public static final String KSETTING_KEYALIAS = "collector.keystore.alias";
protected static final String[] KDEFAULT_DMAAPCONFIGS = new String[] { "/etc/DmaapConfig.json" };
public static final String KSETTING_MAXQUEUEDEVENTS = "collector.inputQueue.maxPending";
- public static final int KDEFAULT_MAXQUEUEDEVENTS = 1024*4;
-
+ public static final int KDEFAULT_MAXQUEUEDEVENTS = 1024 * 4;
+
public static final String KSETTING_SCHEMAVALIDATOR = "collector.schema.checkflag";
- public static final int KDEFAULT_SCHEMAVALIDATOR = -1;
-
+ public static final int KDEFAULT_SCHEMAVALIDATOR = -1;
+
public static final String KSETTING_SCHEMAFILE = "collector.schema.file";
public static final String KDEFAULT_SCHEMAFILE = "{\"v5\":\"./etc/CommonEventFormat_28.3.json\"}";
public static final String KSETTING_EXCEPTIONCONFIG = "exceptionConfig";
-
+
public static final String KSETTING_DMAAPSTREAMID = "collector.dmaap.streamid";
-
+
public static final String KSETTING_AUTHFLAG = "header.authflag";
public static final int KDEFAULT_AUTHFLAG = 0;
-
- public static final String kSetting_authid = "header.authid";
- public static final String kSetting_authpwd = "header.authpwd";
- public static final String kSetting_authstore = "header.authstore";
+
public static final String kSetting_authlist = "header.authlist";
-
+
public static final String KSETTING_EVENTTRANSFORMFLAG = "event.transform.flag";
public static final int KDEFAULT_EVENTTRANSFORMFLAG = 1;
- public static final Logger inlog = LoggerFactory
- .getLogger("org.onap.dcae.commonFunction.input");
- public static final Logger oplog = LoggerFactory
- .getLogger("org.onap.dcae.commonFunction.output");
- public static final Logger eplog = LoggerFactory
- .getLogger("org.onap.dcae.commonFunction.error");
- public static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
-
- public static int schema_Validatorflag = -1;
- public static int authflag = 1;
- public static int eventTransformFlag = 1;
- public static String schemaFile;
- public static JSONObject schemaFileJson;
- public static String exceptionConfig;
- public static String cambriaConfigFile;
- private boolean listnerstatus;
- public static String streamid;
-
-
- private CommonStartup(rrNvReadable settings)
- throws loadException, IOException, rrNvReadable.missingReqdSetting, rrNvReadable.invalidSettingValue, ServletException, InterruptedException {
- final List<ApiServerConnector> connectors = new LinkedList<ApiServerConnector>();
-
- if (settings.getInt(KSETTING_PORT, KDEFAULT_PORT) > 0) {
- // http service
- connectors.add(
- new ApiServerConnector.Builder(settings.getInt(KSETTING_PORT, KDEFAULT_PORT))
- .secure(false)
- .build()
- );
- }
-
- // optional https service
- final int securePort = settings.getInt(KSETTING_SECUREPORT, KDEFAULT_SECUREPORT);
- final String keystoreFile = settings
- .getString(KSETTING_KEYSTOREFILE, KDEFAULT_KEYSTOREFILE);
- final String keystorePasswordFile = settings
- .getString(KSETTING_KEYSTOREPASSFILE, KDEFAULT_KEYSTOREPASSFILE);
- final String keyAlias = settings.getString(KSETTING_KEYALIAS, KDEFAULT_KEYALIAS);
-
- if (securePort > 0) {
- final String KSETTING_KEYSTOREPASS = readFile(keystorePasswordFile,
- Charset.defaultCharset());
- connectors.add(new ApiServerConnector.Builder(securePort)
- .secure(true)
- .keystorePassword(KSETTING_KEYSTOREPASS)
- .keystoreFile(keystoreFile)
- .keyAlias(keyAlias)
- .build());
-
- }
-
- //Reading other config properties
-
- schema_Validatorflag = settings.getInt(KSETTING_SCHEMAVALIDATOR, KDEFAULT_SCHEMAVALIDATOR);
- if (schema_Validatorflag > 0) {
- schemaFile = settings.getString(KSETTING_SCHEMAFILE, KDEFAULT_SCHEMAFILE);
- //System.out.println("SchemaFile:" + schemaFile);
- schemaFileJson = new JSONObject(schemaFile);
-
- }
- exceptionConfig = settings.getString(KSETTING_EXCEPTIONCONFIG, null);
- authflag = settings
- .getInt(CommonStartup.KSETTING_AUTHFLAG, CommonStartup.KDEFAULT_AUTHFLAG);
- String[] currentconffile = settings
- .getStrings(CommonStartup.KSETTING_DMAAPCONFIGS, CommonStartup.KDEFAULT_DMAAPCONFIGS);
- cambriaConfigFile = currentconffile[0];
- streamid = settings.getString(KSETTING_DMAAPSTREAMID, null);
- eventTransformFlag = settings
- .getInt(KSETTING_EVENTTRANSFORMFLAG, KDEFAULT_EVENTTRANSFORMFLAG);
-
- fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings))
- .encodeSlashes(true)
- .name("collector")
- .build();
-
- //Load override exception map
- CustomExceptionLoader.LoadMap();
- setListnerstatus(true);
- }
-
- public static void main(String[] args) {
- ExecutorService executor = null;
- try {
- // process command line arguments
- final Map<String, String> argMap = NsaCommandLineUtil.processCmdLine(args, true);
- final String config = NsaCommandLineUtil
- .getSetting(argMap, KCONFIG, "collector.properties");
- final URL settingStream = DrumlinServlet.findStream(config, CommonStartup.class);
-
- final nvReadableStack settings = new nvReadableStack();
- settings.push(new nvPropertiesFile(settingStream));
- settings.push(new nvReadableTable(argMap));
-
- fProcessingInputQueue = new LinkedBlockingQueue<JSONObject>(
- CommonStartup.KDEFAULT_MAXQUEUEDEVENTS);
-
- VESLogger.setUpEcompLogging();
-
- CommonStartup cs = new CommonStartup(settings);
-
- Thread csmain = new Thread(cs);
- csmain.start();
-
- EventProcessor ep = new EventProcessor();
- //Thread epThread=new Thread(ep);
- //epThread.start();
- executor = Executors.newFixedThreadPool(20);
- executor.execute(ep);
-
- } catch (loadException | missingReqdSetting | IOException | invalidSettingValue |
- ServletException | InterruptedException e) {
- CommonStartup.eplog.error("FATAL_STARTUP_ERROR" + e.getMessage());
- throw new RuntimeException(e);
- } finally {
- // This will make the executor accept no new threads
- // and finish all existing threads in the queue
- if (executor != null) {
- executor.shutdown();
- }
-
- }
- }
-
- public void run() {
- try {
- fTomcatServer.start();
- } catch (LifecycleException | IOException e) {
-
- LOG.error("lifecycle or IO: ", e);
- }
- fTomcatServer.await();
- }
-
- public boolean isListnerstatus() {
- return listnerstatus;
- }
-
- public void setListnerstatus(boolean listnerstatus) {
- this.listnerstatus = listnerstatus;
- }
-
- public static Queue<JSONObject> getProcessingInputQueue() {
- return fProcessingInputQueue;
- }
-
- public static class QueueFullException extends Exception {
-
- private static final long serialVersionUID = 1L;
- }
-
-
- public static void handleEvents(JSONArray a)
- throws QueueFullException, JSONException, IOException {
- final Queue<JSONObject> queue = getProcessingInputQueue();
- try {
-
- CommonStartup.metriclog.info("EVENT_PUBLISH_START");
- for (int i = 0; i < a.length(); i++) {
- if (!queue.offer(a.getJSONObject(i))) {
- throw new QueueFullException();
- }
-
- }
- LOG.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
- CommonStartup.metriclog.info("EVENT_PUBLISH_END");
- //ecomplogger.debug(secloggerMessageEnum.SEC_COLLECT_AND_PULIBISH_SUCCESS);
-
- } catch (JSONException e) {
- throw e;
-
- }
- }
-
-
- static String readFile(String path, Charset encoding)
- throws IOException {
- byte[] encoded = Files.readAllBytes(Paths.get(path));
- String pwd = new String(encoded);
- return pwd.substring(0, pwd.length() - 1);
- }
-
-
- public static String schemavalidate(String jsonData, String jsonSchema) {
- ProcessingReport report;
- String result = "false";
-
- try {
- //System.out.println("Applying schema: @<@<"+jsonSchema+">@>@ to data: #<#<"+jsonData+">#>#");
- LOG.trace("Schema validation for event:" + jsonData);
- JsonNode schemaNode = JsonLoader.fromString(jsonSchema);
- JsonNode data = JsonLoader.fromString(jsonData);
- JsonSchemaFactory factory = JsonSchemaFactory.byDefault();
- JsonSchema schema = factory.getJsonSchema(schemaNode);
- report = schema.validate(data);
- } catch (JsonParseException e) {
- LOG.error("schemavalidate:JsonParseException for event:" + jsonData);
- return e.getMessage().toString();
- } catch (ProcessingException e) {
- LOG.error("schemavalidate:Processing exception for event:" + jsonData);
- return e.getMessage().toString();
- } catch (IOException e) {
- LOG.error(
- "schemavalidate:IO exception; something went wrong trying to read json data for event:"
- + jsonData);
- return e.getMessage().toString();
- }
- if (report != null) {
- Iterator<ProcessingMessage> iter = report.iterator();
- while (iter.hasNext()) {
- ProcessingMessage pm = iter.next();
- LOG.trace("Processing Message: " + pm.getMessage());
- }
- result = String.valueOf(report.isSuccess());
- }
- try {
- LOG.debug("Validation Result:" + result + " Validation report:" + report);
- } catch (NullPointerException e) {
- LOG.error("schemavalidate:NullpointerException on report");
- }
- return result;
- }
-
- public static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;
- private static ApiServer fTomcatServer = null;
- private static final Logger LOG = LoggerFactory.getLogger(CommonStartup.class);
+ public static final Logger inlog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.input");
+ public static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.output");
+ public static final Logger eplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.error");
+ public static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
+
+ public static int schema_Validatorflag = -1;
+ public static int authflag = 1;
+ public static int eventTransformFlag = 1;
+ public static String schemaFile;
+ public static JSONObject schemaFileJson;
+ public static String exceptionConfig;
+ public static String cambriaConfigFile;
+ private boolean listnerstatus;
+ public static String streamid;
+
+ private CommonStartup(rrNvReadable settings) throws loadException, IOException, rrNvReadable.missingReqdSetting,
+ rrNvReadable.invalidSettingValue, ServletException, InterruptedException {
+ final List<ApiServerConnector> connectors = new LinkedList<ApiServerConnector>();
+
+ if (settings.getInt(KSETTING_PORT, KDEFAULT_PORT) > 0) {
+ // http service
+ connectors.add(new ApiServerConnector.Builder(settings.getInt(KSETTING_PORT, KDEFAULT_PORT)).secure(false)
+ .build());
+ }
+
+ // optional https service
+ final int securePort = settings.getInt(KSETTING_SECUREPORT, KDEFAULT_SECUREPORT);
+ final String keystoreFile = settings.getString(KSETTING_KEYSTOREFILE, KDEFAULT_KEYSTOREFILE);
+ final String keystorePasswordFile = settings.getString(KSETTING_KEYSTOREPASSFILE, KDEFAULT_KEYSTOREPASSFILE);
+ final String keyAlias = settings.getString(KSETTING_KEYALIAS, KDEFAULT_KEYALIAS);
+
+ if (securePort > 0) {
+ final String KSETTING_KEYSTOREPASS = readFile(keystorePasswordFile, Charset.defaultCharset());
+ connectors.add(new ApiServerConnector.Builder(securePort).secure(true)
+ .keystorePassword(KSETTING_KEYSTOREPASS).keystoreFile(keystoreFile).keyAlias(keyAlias).build());
+
+ }
+
+ // Reading other config properties
+
+ schema_Validatorflag = settings.getInt(KSETTING_SCHEMAVALIDATOR, KDEFAULT_SCHEMAVALIDATOR);
+ if (schema_Validatorflag > 0) {
+ schemaFile = settings.getString(KSETTING_SCHEMAFILE, KDEFAULT_SCHEMAFILE);
+ // System.out.println("SchemaFile:" + schemaFile);
+ schemaFileJson = new JSONObject(schemaFile);
+
+ }
+ exceptionConfig = settings.getString(KSETTING_EXCEPTIONCONFIG, null);
+ authflag = settings.getInt(CommonStartup.KSETTING_AUTHFLAG, CommonStartup.KDEFAULT_AUTHFLAG);
+ String[] currentconffile = settings.getStrings(CommonStartup.KSETTING_DMAAPCONFIGS,
+ CommonStartup.KDEFAULT_DMAAPCONFIGS);
+ cambriaConfigFile = currentconffile[0];
+ streamid = settings.getString(KSETTING_DMAAPSTREAMID, null);
+ eventTransformFlag = settings.getInt(KSETTING_EVENTTRANSFORMFLAG, KDEFAULT_EVENTTRANSFORMFLAG);
+
+ fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings)).encodeSlashes(true)
+ .name("collector").build();
+
+ // Load override exception map
+ CustomExceptionLoader.LoadMap();
+ setListnerstatus(true);
+ }
+
+ public static void main(String[] args) {
+ ExecutorService executor = null;
+ try {
+ // process command line arguments
+ final Map<String, String> argMap = NsaCommandLineUtil.processCmdLine(args, true);
+ final String config = NsaCommandLineUtil.getSetting(argMap, KCONFIG, "collector.properties");
+ final URL settingStream = DrumlinServlet.findStream(config, CommonStartup.class);
+
+ final nvReadableStack settings = new nvReadableStack();
+ settings.push(new nvPropertiesFile(settingStream));
+ settings.push(new nvReadableTable(argMap));
+
+ fProcessingInputQueue = new LinkedBlockingQueue<JSONObject>(CommonStartup.KDEFAULT_MAXQUEUEDEVENTS);
+
+ VESLogger.setUpEcompLogging();
+
+ CommonStartup cs = new CommonStartup(settings);
+
+ Thread csmain = new Thread(cs);
+ csmain.start();
+
+ EventProcessor ep = new EventProcessor();
+ // Thread epThread=new Thread(ep);
+ // epThread.start();
+ executor = Executors.newFixedThreadPool(20);
+ executor.execute(ep);
+
+ } catch (loadException | missingReqdSetting | IOException | invalidSettingValue | ServletException
+ | InterruptedException e) {
+ CommonStartup.eplog.error("FATAL_STARTUP_ERROR" + e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ // This will make the executor accept no new threads
+ // and finish all existing threads in the queue
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ }
+ }
+
+ public void run() {
+ try {
+ fTomcatServer.start();
+ } catch (LifecycleException | IOException e) {
+
+ LOG.error("lifecycle or IO: ", e);
+ }
+ fTomcatServer.await();
+ }
+
+ public boolean isListnerstatus() {
+ return listnerstatus;
+ }
+
+ public void setListnerstatus(boolean listnerstatus) {
+ this.listnerstatus = listnerstatus;
+ }
+
+ public static Queue<JSONObject> getProcessingInputQueue() {
+ return fProcessingInputQueue;
+ }
+
+ public static class QueueFullException extends Exception {
+
+ private static final long serialVersionUID = 1L;
+ }
+
+ public static void handleEvents(JSONArray a) throws QueueFullException, JSONException, IOException {
+ final Queue<JSONObject> queue = getProcessingInputQueue();
+ try {
+
+ CommonStartup.metriclog.info("EVENT_PUBLISH_START");
+ for (int i = 0; i < a.length(); i++) {
+ if (!queue.offer(a.getJSONObject(i))) {
+ throw new QueueFullException();
+ }
+
+ }
+ LOG.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
+ CommonStartup.metriclog.info("EVENT_PUBLISH_END");
+ // ecomplogger.debug(secloggerMessageEnum.SEC_COLLECT_AND_PULIBISH_SUCCESS);
+
+ } catch (JSONException e) {
+ throw e;
+
+ }
+ }
+
+ static String readFile(String path, Charset encoding) throws IOException {
+ byte[] encoded = Files.readAllBytes(Paths.get(path));
+ String pwd = new String(encoded);
+ return pwd.substring(0, pwd.length() - 1);
+ }
+
+ public static String schemavalidate(String jsonData, String jsonSchema) {
+ ProcessingReport report;
+ String result = "false";
+
+ try {
+ // System.out.println("Applying schema: @<@<"+jsonSchema+">@>@ to
+ // data: #<#<"+jsonData+">#>#");
+ LOG.trace("Schema validation for event:" + jsonData);
+ JsonNode schemaNode = JsonLoader.fromString(jsonSchema);
+ JsonNode data = JsonLoader.fromString(jsonData);
+ JsonSchemaFactory factory = JsonSchemaFactory.byDefault();
+ JsonSchema schema = factory.getJsonSchema(schemaNode);
+ report = schema.validate(data);
+ } catch (JsonParseException e) {
+ LOG.error("schemavalidate:JsonParseException for event:" + jsonData);
+ return e.getMessage().toString();
+ } catch (ProcessingException e) {
+ LOG.error("schemavalidate:Processing exception for event:" + jsonData);
+ return e.getMessage().toString();
+ } catch (IOException e) {
+ LOG.error(
+ "schemavalidate:IO exception; something went wrong trying to read json data for event:" + jsonData);
+ return e.getMessage().toString();
+ }
+ if (report != null) {
+ Iterator<ProcessingMessage> iter = report.iterator();
+ while (iter.hasNext()) {
+ ProcessingMessage pm = iter.next();
+ LOG.trace("Processing Message: " + pm.getMessage());
+ }
+ result = String.valueOf(report.isSuccess());
+ }
+ try {
+ LOG.debug("Validation Result:" + result + " Validation report:" + report);
+ } catch (NullPointerException e) {
+ LOG.error("schemavalidate:NullpointerException on report");
+ }
+ return result;
+ }
+
+ public static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;
+ private static ApiServer fTomcatServer = null;
+ private static final Logger LOG = LoggerFactory.getLogger(CommonStartup.class);
}
+
package org.onap.dcae.commonFunction;
+
import java.text.DecimalFormat;
import org.json.JSONArray;
import org.json.JSONObject;
}
}
else //if new array
- setEventObjectVal(field + "[0]", new JSONObject(value));
+ setEventObjectVal(field + "[0]", new JSONObject(value), "JArray");
}
else //oldfield is jsonArray
setEventObjectVal(field, new JSONArray(value));
}
else if (fieldType.equals("integer") && value instanceof String)
((JSONObject)keySeriesObj).put(keySet[keySet.length -1], Integer.valueOf((String) value));
+ else if (fieldType.equals("JArray"))
+ ((JSONArray)keySeriesObj).put( value);
else
((JSONObject)keySeriesObj).put(keySet[keySet.length -1], value);
+
}
private JSONObject event = new JSONObject();
}
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
-
public class CustomExceptionLoader {
protected static HashMap<String, JsonArray> map = null;
}
log.debug("CustomExceptionLoader.LoadMap --> Map loaded - " + map);
- } catch (JsonIOException e) {
- e.printStackTrace();
- } catch (JsonSyntaxException e) {
- e.printStackTrace();
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- } catch (Exception e) {
- e.printStackTrace();
+ } catch (JsonIOException|JsonSyntaxException|FileNotFoundException e) {
+ log.error("Exception in LoadMap:" + e.getMessage());
+ //e.printStackTrace();
+ map = null;
}
finally {
if (fr != null) {
fr.close();
} catch (IOException e) {
log.error("Error closing file reader stream : " +e.toString());
+ map = null;
}
}
}
}
}
-
package org.onap.dcae.commonFunction;
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonIOException;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
-import com.google.gson.JsonSyntaxException;
+import java.io.FileNotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonIOException;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonSyntaxException;
+
public class DmaapPropertyReader {
- private static DmaapPropertyReader instance = null;
-
- private static final Logger log = LoggerFactory.getLogger(DmaapPropertyReader.class);
- private static final String TOPIC = "TOPIC:";
- private static final String HOST_URL = " HOST-URL:";
- private static final String CAMBRIA_URL = "cambria.url";
- private static final String CAMBRIA_HOSTS = "cambria.hosts";
- private static final String USER = " USER:";
- private static final String PWD = " PWD:";
- private static final String BA_PWD = "basicAuthPassword";
- private static final String BA_UNAME = "basicAuthUsername";
-
- public HashMap<String, String> dmaap_hash = new HashMap<String, String>();
-
- public DmaapPropertyReader(String cambriaConfigFile) {
-
- FileReader fr = null;
- try {
- JsonElement root;
- fr = new FileReader(cambriaConfigFile);
- root = new JsonParser().parse(fr);
-
- //Check if dmaap config is handled by legacy controller/service manager
- if (root.getAsJsonObject().has("channels")) {
- JsonArray jsonObject = (JsonArray) root.getAsJsonObject().get("channels");
-
- for (int i = 0; i < jsonObject.size(); i++) {
- log.debug(String.format("%s%s%s%s HOSTS:%s%s%s%s%s NAME:%s", TOPIC,
- jsonObject.get(i).getAsJsonObject().get("cambria.topic"), HOST_URL,
- jsonObject.get(i).getAsJsonObject().get(CAMBRIA_URL),
- jsonObject.get(i).getAsJsonObject().get(CAMBRIA_HOSTS), PWD,
- jsonObject.get(i).getAsJsonObject().get(BA_PWD), USER,
- jsonObject.get(i).getAsJsonObject().get(BA_UNAME),
- jsonObject.get(i).getAsJsonObject().get("name")));
-
- String convertedname = jsonObject.get(i).getAsJsonObject().get("name")
- .toString().replace("\"", "");
- dmaap_hash.put(convertedname + ".cambria.topic",
- jsonObject.get(i).getAsJsonObject().get("cambria.topic").toString()
- .replace("\"", ""));
-
- if (jsonObject.get(i).getAsJsonObject().get(CAMBRIA_HOSTS) != null) {
- dmaap_hash.put(convertedname + ".cambria.hosts",
- jsonObject.get(i).getAsJsonObject().get(CAMBRIA_HOSTS).toString()
- .replace("\"", ""));
- }
- if (jsonObject.get(i).getAsJsonObject().get(CAMBRIA_URL) != null) {
- dmaap_hash.put(convertedname + ".cambria.url",
- jsonObject.get(i).getAsJsonObject().get(CAMBRIA_URL).toString()
- .replace("\"", ""));
- }
- if (jsonObject.get(i).getAsJsonObject().get(BA_PWD) != null) {
- dmaap_hash.put(convertedname + ".basicAuthPassword",
- jsonObject.get(i).getAsJsonObject()
- .get(BA_PWD).toString().replace("\"", ""));
- }
- if (jsonObject.get(i).getAsJsonObject().get(BA_UNAME) != null) {
- dmaap_hash.put(convertedname + ".basicAuthUsername",
- jsonObject.get(i).getAsJsonObject()
- .get(BA_UNAME).toString().replace("\"", ""));
- }
-
- }
- } else {
-
- //Handing new format from controllergen2/config_binding_service
- JsonObject jsonObject = root.getAsJsonObject();
- Set<Map.Entry<String, JsonElement>> entries = jsonObject.entrySet();
-
- for (Map.Entry<String, JsonElement> entry : entries) {
-
- JsonElement topicurl = entry.getValue().getAsJsonObject().get("dmaap_info")
- .getAsJsonObject().get("topic_url");
- String[] urlParts = dmaapUrlSplit(topicurl.toString().replace("\"", ""));
-
- String mrTopic = null;
- String mrUrl = null;
- String[] hostport = null;
- String username = null;
- String userpwd = null;
-
- try {
-
- if (null != urlParts) {
- mrUrl = urlParts[2];
-
- // DCAE internal dmaap topic convention
- if (urlParts[3].equals("events")) {
- mrTopic = urlParts[4];
- } else {
- // ONAP dmaap topic convention
- mrTopic = urlParts[3];
- hostport = mrUrl.split(":");
- }
-
- }
- } catch (NullPointerException e) {
- log.error("NullPointerException " + e.getLocalizedMessage(), e);
- }
-
- if (entry.getValue().getAsJsonObject().has("aaf_username")) {
- username = entry.getValue().getAsJsonObject().get("aaf_username").toString()
- .replace("\"", "");
- }
- if (entry.getValue().getAsJsonObject().has("aaf_password")) {
- userpwd = entry.getValue().getAsJsonObject().get("aaf_password").toString()
- .replace("\"", "");
- }
- if (hostport == null) {
- log.debug(
- String.format("%s%s%s%s%s%s%s%s", TOPIC, mrTopic, HOST_URL, mrUrl, PWD,
- userpwd, USER, username));
- } else {
- log.debug(
- String.format("%s%s%s%s HOSTS:%s%s%s%s%s NAME:%s", TOPIC, mrTopic,
- HOST_URL, mrUrl, hostport[0], PWD, userpwd, USER, username,
- entry.getKey()));
- }
-
- dmaap_hash.put(entry.getKey() + ".cambria.topic", mrTopic);
-
- if (hostport != null) {
- dmaap_hash.put(entry.getKey() + ".cambria.hosts", hostport[0]);
- }
-
- if (mrUrl != null) {
- dmaap_hash.put(entry.getKey() + ".cambria.url", mrUrl);
- }
-
- if (username != null) {
- dmaap_hash.put(entry.getKey() + ".basicAuthUsername", username);
- }
-
- if (userpwd != null) {
- dmaap_hash.put(entry.getKey() + ".basicAuthPassword", userpwd);
- }
-
- }
-
- }
-
- fr.close();
- } catch (JsonIOException | JsonSyntaxException | IOException e1) {
- log.error("Problem loading Dmaap Channel configuration file: " +
- e1.getLocalizedMessage(), e1);
- } finally {
- if (fr != null) {
- try {
- fr.close();
- } catch (IOException e) {
- log.error("Error closing file reader stream : " + e);
- }
- }
- }
-
- }
-
- /***
- * Dmaap url structure pub - https://<dmaaphostname>:<port>/events/
- * <namespace>.<dmaapcluster>.<topic>, sub - https://<dmaaphostname>:
- * <port>/events/<namespace>.<dmaapcluster>.<topic>/G1/u1";
- *
- * Onap url structure pub - http://<dmaaphostname>:<port>/<unauthenticated>.
- * <topic>,
- */
-
- private String[] dmaapUrlSplit(String dmUrl) {
- String[] multUrls = dmUrl.split(",");
-
- StringBuilder newUrls = new StringBuilder();
- String[] urlParts = null;
- for (int i = 0; i < multUrls.length; i++) {
- urlParts = multUrls[i].split("/");
- if (i == 0) {
- newUrls = newUrls.append(urlParts[2]);
- } else {
- newUrls = newUrls.append(",").append(urlParts[2]);
- }
- }
- return urlParts;
- }
-
- public static synchronized DmaapPropertyReader getInstance(String channelConfig) {
- if (instance == null) {
- instance = new DmaapPropertyReader(channelConfig);
- }
- return instance;
- }
-
- public String getKeyValue(String hashKey) {
- return dmaap_hash.get(hashKey);
- }
+ private static DmaapPropertyReader instance = null;
+
+ private static final Logger log = LoggerFactory.getLogger(DmaapPropertyReader.class);
+
+ public HashMap<String, String> dmaap_hash = new HashMap<String, String>();
+
+ public DmaapPropertyReader(String CambriaConfigFile) {
+
+ FileReader fr = null;
+ try {
+ JsonElement root = null;
+ fr = new FileReader(CambriaConfigFile);
+ root = new JsonParser().parse(fr);
+
+ //Check if dmaap config is handled by legacy controller/service manager
+ if (root.getAsJsonObject().has("channels")) {
+ JsonArray jsonObject = (JsonArray) root.getAsJsonObject().get("channels");
+
+ for (int i = 0; i < jsonObject.size(); i++) {
+ log.debug("TOPIC:" + jsonObject.get(i).getAsJsonObject().get("cambria.topic") + " HOST-URL:"
+ + jsonObject.get(i).getAsJsonObject().get("cambria.url") + " HOSTS:"
+ + jsonObject.get(i).getAsJsonObject().get("cambria.hosts") + " PWD:"
+ + jsonObject.get(i).getAsJsonObject().get("basicAuthPassword") + " USER:"
+ + jsonObject.get(i).getAsJsonObject().get("basicAuthUsername") + " NAME:"
+ + jsonObject.get(i).getAsJsonObject().get("name"));
+
+ String convertedname = jsonObject.get(i).getAsJsonObject().get("name").toString().replace("\"", "");
+ dmaap_hash.put(convertedname + ".cambria.topic",
+ jsonObject.get(i).getAsJsonObject().get("cambria.topic").toString().replace("\"", ""));
+
+ if (jsonObject.get(i).getAsJsonObject().get("cambria.hosts") != null) {
+ dmaap_hash.put(convertedname + ".cambria.hosts",
+ jsonObject.get(i).getAsJsonObject().get("cambria.hosts").toString().replace("\"", ""));
+ }
+ if (jsonObject.get(i).getAsJsonObject().get("cambria.url") != null) {
+ dmaap_hash.put(convertedname + ".cambria.url",
+ jsonObject.get(i).getAsJsonObject().get("cambria.url").toString().replace("\"", ""));
+ }
+ if (jsonObject.get(i).getAsJsonObject().get("basicAuthPassword") != null) {
+ dmaap_hash.put(convertedname + ".basicAuthPassword", jsonObject.get(i).getAsJsonObject()
+ .get("basicAuthPassword").toString().replace("\"", ""));
+ }
+ if (jsonObject.get(i).getAsJsonObject().get("basicAuthUsername") != null) {
+ dmaap_hash.put(convertedname + ".basicAuthUsername", jsonObject.get(i).getAsJsonObject()
+ .get("basicAuthUsername").toString().replace("\"", ""));
+ }
+
+ }
+ } else {
+
+ //Handing new format from controllergen2/config_binding_service
+ JsonObject jsonObject = root.getAsJsonObject();
+ Set<Map.Entry<String, JsonElement>> entries = jsonObject.entrySet();
+
+ for (Map.Entry<String, JsonElement> entry : entries) {
+
+ JsonElement topicurl = entry.getValue().getAsJsonObject().get("dmaap_info").getAsJsonObject().get("topic_url");
+ String[] urlParts = dmaapUrlSplit(topicurl.toString().replace("\"", ""));
+
+ String mrTopic = null;
+ String mrUrl = null;
+ String[] hostport = null;
+ String username = null;
+ String userpwd = null;
+
+ try {
+
+ if (null != urlParts) {
+ mrUrl = urlParts[2];
+
+ // DCAE internal dmaap topic convention
+ if (urlParts[3].equals("events")) {
+ mrTopic = urlParts[4];
+ } else {
+ // ONAP dmaap topic convention
+ mrTopic = urlParts[3];
+ hostport = mrUrl.split(":");
+ }
+
+ }
+ } catch (NullPointerException e) {
+ System.out.println("NullPointerException");
+ e.getMessage();
+ }
+
+ if (entry.getValue().getAsJsonObject().has("aaf_username")) {
+ username = entry.getValue().getAsJsonObject().get("aaf_username").toString().replace("\"", "");
+ }
+ if (entry.getValue().getAsJsonObject().has("aaf_password")) {
+ userpwd = entry.getValue().getAsJsonObject().get("aaf_password").toString().replace("\"", "");
+ }
+ if (hostport == null) {
+ log.debug("TOPIC:" + mrTopic + " HOST-URL:" + mrUrl + " PWD:" + userpwd + " USER:" + username);
+ } else {
+ log.debug("TOPIC:" + mrTopic + " HOST-URL:" + mrUrl + " HOSTS:" + hostport[0] + " PWD:"
+ + userpwd + " USER:" + username + " NAME:" + entry.getKey());
+ }
+
+ dmaap_hash.put(entry.getKey() + ".cambria.topic", mrTopic);
+
+ if (!(hostport == null)) {
+ dmaap_hash.put(entry.getKey() + ".cambria.hosts", hostport[0]);
+ }
+
+ if (!(mrUrl == null)) {
+ dmaap_hash.put(entry.getKey() + ".cambria.url", mrUrl);
+ }
+
+ if (!(username == null)) {
+ dmaap_hash.put(entry.getKey() + ".basicAuthUsername", username);
+ }
+
+ if (!(userpwd == null)) {
+ dmaap_hash.put(entry.getKey() + ".basicAuthPassword", userpwd);
+ }
+
+ }
+
+ }
+
+ } catch (JsonIOException | JsonSyntaxException |
+
+ FileNotFoundException e1) {
+ e1.printStackTrace();
+ log.error("Problem loading Dmaap Channel configuration file: " + e1.toString());
+ } finally {
+ if (fr != null) {
+ try {
+ fr.close();
+ } catch (IOException e) {
+ log.error("Error closing file reader stream : " + e.toString());
+ }
+ }
+ }
+
+ }
+
+ /***
+ * Dmaap url structure pub - https://<dmaaphostname>:<port>/events/
+ * <namespace>.<dmaapcluster>.<topic>, sub - https://<dmaaphostname>:
+ * <port>/events/<namespace>.<dmaapcluster>.<topic>/G1/u1";
+ *
+ * Onap url structure pub - http://<dmaaphostname>:<port>/<unauthenticated>.
+ * <topic>,
+ */
+
+ private String[] dmaapUrlSplit(String dmUrl) {
+ String[] multUrls = dmUrl.split(",");
+
+ StringBuffer newUrls = new StringBuffer();
+ String urlParts[] = null;
+ for (int i = 0; i < multUrls.length; i++) {
+ urlParts = multUrls[i].split("/");
+ if (i == 0) {
+ newUrls = newUrls.append(urlParts[2]);
+ } else {
+ newUrls = newUrls.append(",").append(urlParts[2]);
+ }
+ }
+ return urlParts;
+ }
+
+ public static synchronized DmaapPropertyReader getInstance(String ChannelConfig) {
+ if (instance == null) {
+ instance = new DmaapPropertyReader(ChannelConfig);
+ }
+ return instance;
+ }
+
+ public String getKeyValue(String HashKey) {
+ return this.dmaap_hash.get(HashKey);
+ }
}
import com.att.nsa.security.authenticators.SimpleAuthenticator;
import com.att.nsa.security.db.simple.NsaSimpleApiKey;
+
public class RestfulCollectorServlet extends CommonServlet
{
- String authid;
- String authpwd;
+ String authid=null;
+ String authpwd=null;
public String authlist;
public RestfulCollectorServlet ( rrNvReadable settings ) throws loadException, missingReqdSetting
{
super ( settings, "collector", false );
- authid = settings.getString(CommonStartup.kSetting_authid,null);
- if (authid != null)
+ //authid = settings.getString(CommonStartup.kSetting_authid,null);
+ /*if (authid != null)
{
String authpwdtemp = settings.getString(CommonStartup.kSetting_authpwd,null);
authpwd = new String(Base64.decodeBase64(authpwdtemp));
- }
+ }*/
authlist = settings.getString(CommonStartup.kSetting_authlist,null);
}
private static final long serialVersionUID = 1L;
private static final Logger log = LoggerFactory.getLogger ( RestfulCollectorServlet.class );
}
-
import java.io.FileReader;
import java.io.IOException;
-
import org.json.JSONObject;
import com.google.gson.JsonParser;
-
import com.google.gson.JsonObject;
import org.junit.Test;
}
return jsonObject;
}
-
@Test
public void testAttrMap(){
System.out.println("responseData==" + responseData);
assertEquals (alarmAdditionalInformation, responseData);
}
-
+ @Test
+ public void testJobjMaptoArray(){
+
+ final JSONObject jsonObject = getFileAsJsonObject();
+ //final String receiveDiscards = (((jsonObject.getJSONObject("event")).getJSONObject("faultFields")).get("errors")).get("receiveDiscards").toString();
+ System.out.println("event==" + jsonObject.toString());
+ //System.out.println("alarmAdditionalInformation==" + alarmAdditionalInformation);
+ final JSONObject jsonArgs = new JSONObject ( "{\"field\": \"event.faultFields.vNicPerformanceArray[]\",\"oldField\": \"event.faultFields.errors\",\"attrMap\":{\"receiveDiscards\":\"receivedDiscardedPacketsAccumulated\"}}" );
+ ConfigProcessors cpEvent = new ConfigProcessors(jsonObject);
+ final String receiveDiscards = cpEvent.getEventObjectVal("event.faultFields.errors.receiveDiscards").toString();
+ System.out.println("receiveDiscards==" + receiveDiscards);
+ cpEvent.map(jsonArgs);
+ final String responseData = cpEvent.getEventObjectVal("event.faultFields.vNicPerformanceArray[0].receivedDiscardedPacketsAccumulated").toString();
+ System.out.println("modified event==" + jsonObject.toString());
+ System.out.println("responseData==" + responseData);
+ assertEquals (receiveDiscards, responseData);
+ }
@Test
public void testAttrAdd(){
assertEquals ("2.0", responseData);
}
+ @Test
+ public void testAttrUpdate(){
+
+ final JSONObject jsonObject = getFileAsJsonObject();
+ //final String functionRole = (jsonObject.getJSONObject("event")).getJSONObject("commonEventHeader").get("functionalRole").toString();
+ System.out.println("event==" + jsonObject.toString());
+ //System.out.println("functionRole==" + functionRole);
+ final JSONObject jsonArgs = new JSONObject ( "{\"field\": \"event.faultFields.version\",\"value\": \"2.0\",\"fieldType\": \"number\"}" );
+ ConfigProcessors cpEvent = new ConfigProcessors(jsonObject);
+ cpEvent.updateAttribute(jsonArgs);
+ final String responseData = cpEvent.getEventObjectVal("event.faultFields.version").toString();
+ System.out.println("modified event==" + jsonObject.toString());
+ System.out.println("responseData==" + responseData);
+ assertEquals ("2.0", responseData);
+ }
+
@Test
public void testAttrConcatenate(){
import com.att.nsa.drumlin.till.nv.impl.nvReadableStack;
import com.att.nsa.drumlin.till.nv.impl.nvReadableTable;
+
public class TestCommonStartup {
String payload = null;
// process command line arguments
payload = new JsonParser().parse(new FileReader("src/test/resources/VES_valid.txt")).toString();
- CommonStartup.fProcessingInputQueue = new LinkedBlockingQueue<JSONObject> (CommonStartup.kDefault_MaxQueuedEvents);
+ CommonStartup.fProcessingInputQueue = new LinkedBlockingQueue<JSONObject> (CommonStartup.KDEFAULT_MAXQUEUEDEVENTS);
}
@After
{
+
"event": {
"commonEventHeader": {
"sourceId": "owb-rdm-003",
"sourceName": "owb-rdm-003"
},
"faultFields": {
+"errors": {
+ "receiveDiscards": 87665,
+ "transmitDiscards": 99999999,
+ "receiveErrors": 888888888,
+ "transmitErrors": 9000000001111
+},
"eventSeverity": "CRITICAL",
"alarmCondition": "lossOfSignal",
"faultFieldsVersion": 1.0,
Code Review / dcaegen2 / collectors / ves.git / commitdiff