diff options
Diffstat (limited to 'src/main/java/org/onap/dcae/commonFunction')
11 files changed, 695 insertions, 904 deletions
diff --git a/src/main/java/org/onap/dcae/commonFunction/AnyNode.java b/src/main/java/org/onap/dcae/commonFunction/AnyNode.java index 97d73ddd..7be45b0c 100644 --- a/src/main/java/org/onap/dcae/commonFunction/AnyNode.java +++ b/src/main/java/org/onap/dcae/commonFunction/AnyNode.java @@ -19,16 +19,17 @@ */ package org.onap.dcae.commonFunction; -import static io.vavr.API.Set; - import io.vavr.collection.List; import io.vavr.collection.Set; import io.vavr.control.Option; -import java.util.stream.StreamSupport; import org.json.JSONArray; import org.json.JSONException; import org.json.JSONObject; +import java.util.stream.StreamSupport; + +import static io.vavr.API.Set; + /** * This class is a wrapper for 2 most used entities of org.json lib: JSONArray and JSONObject and comprises utility * methods for fast access of json structures without need to explicitly coerce between them. While using this, bear in @@ -109,5 +110,4 @@ public class AnyNode { return (JSONObject) this.obj; } - } diff --git a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java b/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java deleted file mode 100644 index 36713aa4..00000000 --- a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java +++ /dev/null @@ -1,204 +0,0 @@ -/*-
- * ============LICENSE_START=======================================================
- * PROJECT
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.dcae.commonFunction;
-
-import com.att.nsa.apiServer.ApiServer;
-import com.att.nsa.apiServer.ApiServerConnector;
-import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;
-import com.att.nsa.drumlin.till.nv.rrNvReadable;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.github.fge.jackson.JsonLoader;
-import com.github.fge.jsonschema.core.exceptions.ProcessingException;
-import com.github.fge.jsonschema.core.report.ProcessingMessage;
-import com.github.fge.jsonschema.core.report.ProcessingReport;
-import com.github.fge.jsonschema.main.JsonSchema;
-import com.github.fge.jsonschema.main.JsonSchemaFactory;
-import org.apache.catalina.LifecycleException;
-import org.json.JSONArray;
-import org.json.JSONException;
-import org.json.JSONObject;
-import org.onap.dcae.ApplicationSettings;
-import org.onap.dcae.CLIUtils;
-import org.onap.dcae.commonFunction.event.publishing.DMaaPConfigurationParser;
-import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
-import org.onap.dcae.restapi.RestfulCollectorServlet;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-
-public class CommonStartup extends NsaBaseEndpoint implements Runnable {
-
- private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
- public static final Logger inlog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.input");
- static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.output");
- public static final Logger eplog = LoggerFactory.getLogger("org.onap.dcae.commonFunction.error");
-
- static int maxQueueEvent = 1024 * 4;
- public static boolean schemaValidatorflag = false;
- public static boolean authflag = false;
- static boolean eventTransformFlag = true;
- public static JSONObject schemaFileJson;
- static String cambriaConfigFile;
- public static io.vavr.collection.Map<String , String [] > streamID;
-
- static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;
- private static ApiServer fTomcatServer = null;
- private static final Logger log = LoggerFactory.getLogger(CommonStartup.class);
-
- private CommonStartup(ApplicationSettings settings) throws loadException, IOException, rrNvReadable.missingReqdSetting {
- final List<ApiServerConnector> connectors = new LinkedList<>();
-
- if (!settings.authorizationEnabled()) {
- connectors.add(new ApiServerConnector.Builder(settings.httpPort()).secure(false).build());
- }
-
- final int securePort = settings.httpsPort();
- final String keystoreFile = settings.keystoreFileLocation();
- final String keystorePasswordFile = settings.keystorePasswordFileLocation();
- final String keyAlias = settings.keystoreAlias();
-
- if (settings.authorizationEnabled()) {
- String keystorePassword = readFile(keystorePasswordFile);
- connectors.add(new ApiServerConnector.Builder(securePort).secure(true)
- .keystorePassword(keystorePassword).keystoreFile(keystoreFile).keyAlias(keyAlias).build());
-
- }
-
- schemaValidatorflag = settings.jsonSchemaValidationEnabled();
- maxQueueEvent = settings.maximumAllowedQueuedEvents();
- if (schemaValidatorflag) {
- schemaFileJson = settings.jsonSchema();
-
- }
- authflag = settings.authorizationEnabled();
- cambriaConfigFile = settings.cambriaConfigurationFileLocation();
- streamID = settings.dMaaPStreamsMapping();
- eventTransformFlag = settings.eventTransformingEnabled();
-
- fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings)).encodeSlashes(true)
- .name("collector").build();
- }
-
- public static void main(String[] args) {
- try {
-
- fProcessingInputQueue = new LinkedBlockingQueue<>(CommonStartup.maxQueueEvent);
-
- VESLogger.setUpEcompLogging();
-
- CommonStartup cs = new CommonStartup(new ApplicationSettings(args, CLIUtils::processCmdLine));
-
- Thread commonStartupThread = new Thread(cs);
- commonStartupThread.start();
-
- EventProcessor ep = new EventProcessor(EventPublisher.createPublisher(oplog,
- DMaaPConfigurationParser
- .parseToDomainMapping(Paths.get(cambriaConfigFile))
- .get()));
- ExecutorService executor = Executors.newFixedThreadPool(20);
- for (int i = 0; i < 20; ++i) {
- executor.execute(ep);
- }
- } catch (Exception e) {
- CommonStartup.eplog.error("Fatal error during application startup", e);
- throw new RuntimeException(e);
- }
- }
-
- public void run() {
- try {
- fTomcatServer.start();
- fTomcatServer.await();
- } catch (LifecycleException | IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- public static class QueueFullException extends Exception {
-
- private static final long serialVersionUID = 1L;
- }
-
- public static void handleEvents(JSONArray a) throws QueueFullException, JSONException {
- CommonStartup.metriclog.info("EVENT_PUBLISH_START");
- for (int i = 0; i < a.length(); i++) {
- if (!fProcessingInputQueue.offer(a.getJSONObject(i))) {
- throw new QueueFullException();
- }
- }
- log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
- CommonStartup.metriclog.info("EVENT_PUBLISH_END");
- }
-
- private static String readFile(String path) throws IOException {
- byte[] encoded = Files.readAllBytes(Paths.get(path));
- String pwd = new String(encoded);
- return pwd.substring(0, pwd.length() - 1);
- }
-
- public static String validateAgainstSchema(String jsonData, String jsonSchema) {
- ProcessingReport report;
- String result = "false";
-
- try {
- 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("validateAgainstSchema:JsonParseException for event:" + jsonData);
- return e.getMessage();
- } catch (ProcessingException e) {
- log.error("validateAgainstSchema:Processing exception for event:" + jsonData);
- return e.getMessage();
- } catch (IOException e) {
- log.error(
- "validateAgainstSchema:IO exception; something went wrong trying to read json data for event:" + jsonData);
- return e.getMessage();
- }
- if (report != null) {
- for (ProcessingMessage pm : report) {
- 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("validateAgainstSchema:NullpointerException on report");
- }
- return result;
- }
-
-
-}
diff --git a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java index a6de0fc8..e3d59098 100644 --- a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java +++ b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java @@ -8,9 +8,9 @@ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -21,483 +21,483 @@ package org.onap.dcae.commonFunction; -import java.text.DecimalFormat; import org.json.JSONArray; import org.json.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class ConfigProcessors { - - private static final Logger log = LoggerFactory.getLogger(ConfigProcessors.class); - private static final String FIELD = "field"; - private static final String OLD_FIELD = "oldField"; - private static final String FILTER = "filter"; - private static final String VALUE = "value"; - private static final String REGEX = "\\[\\]"; - private static final String OBJECT_NOT_FOUND = "ObjectNotFound"; - private static final String FILTER_NOT_MET = "Filter not met"; - private static final String COMP_FALSE = "==false"; - - private final JSONObject event; - - public ConfigProcessors(JSONObject eventJson) { - event = eventJson; - } - - public void getValue(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - - if (filter == null || isFilterMet(filter)) { - getEventObjectVal(field); - } else - log.info(FILTER_NOT_MET); - } +import java.text.DecimalFormat; +public class ConfigProcessors { - public void setValue(JSONObject jsonObject) { - final String field = jsonObject.getString(FIELD); - final String value = jsonObject.getString(VALUE); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - if (filter == null || isFilterMet(filter)) { - setEventObjectVal(field, value); - } else - log.info(FILTER_NOT_MET); - } + private static final Logger log = LoggerFactory.getLogger(ConfigProcessors.class); + private static final String FIELD = "field"; + private static final String OLD_FIELD = "oldField"; + private static final String FILTER = "filter"; + private static final String VALUE = "value"; + private static final String REGEX = "\\[\\]"; + private static final String OBJECT_NOT_FOUND = "ObjectNotFound"; + private static final String FILTER_NOT_MET = "Filter not met"; + private static final String COMP_FALSE = "==false"; + private final JSONObject event; + public ConfigProcessors(JSONObject eventJson) { + event = eventJson; + } - private String evaluate(String str) { - String value = str; - if (str.startsWith("$")) { - value = (String) getEventObjectVal(str.substring(1)); + public void getValue(JSONObject jsonObject) { - } - return value; - } + final String field = jsonObject.getString(FIELD); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + getEventObjectVal(field); + } else + log.info(FILTER_NOT_MET); + } - public void suppressEvent(JSONObject jsonObject) { - final JSONObject filter = jsonObject.optJSONObject(FILTER); - if (filter == null || isFilterMet(filter)) { - setEventObjectVal("suppressEvent", "true"); - } else - log.info(FILTER_NOT_MET); - } + public void setValue(JSONObject jsonObject) { + final String field = jsonObject.getString(FIELD); + final String value = jsonObject.getString(VALUE); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + setEventObjectVal(field, value); + } else + log.info(FILTER_NOT_MET); + } - public void addAttribute(JSONObject jsonObject) { + private String evaluate(String str) { + String value = str; + if (str.startsWith("$")) { + value = (String) getEventObjectVal(str.substring(1)); + + } + return value; + } - final String field = jsonObject.getString(FIELD); - final String value = evaluate(jsonObject.getString(VALUE)); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - final String fieldType = jsonObject.optString("fieldType", "string").toLowerCase(); - if (filter == null || isFilterMet(filter)) { - setEventObjectVal(field, value, fieldType); - } else - log.info(FILTER_NOT_MET); - } + public void suppressEvent(JSONObject jsonObject) { + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + setEventObjectVal("suppressEvent", "true"); + } else + log.info(FILTER_NOT_MET); + } - public void updateAttribute(JSONObject jsonObject) { - final String field = jsonObject.getString(FIELD); - final String value = evaluate(jsonObject.getString(VALUE)); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - if (filter == null || isFilterMet(filter)) { - setEventObjectVal(field, value); - } else - log.info(FILTER_NOT_MET); - } + public void addAttribute(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final String value = evaluate(jsonObject.getString(VALUE)); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + final String fieldType = jsonObject.optString("fieldType", "string").toLowerCase(); + + if (filter == null || isFilterMet(filter)) { + setEventObjectVal(field, value, fieldType); + } else + log.info(FILTER_NOT_MET); + } - public void removeAttribute(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - - if (filter == null || isFilterMet(filter)) { - removeEventKey(field); - } else - log.info(FILTER_NOT_MET); - } - - - private void renameArrayInArray(JSONObject jsonObject) // map - { - log.info("renameArrayInArray"); - final String field = jsonObject.getString(FIELD); - final String oldField = jsonObject.getString(OLD_FIELD); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - - if (filter == null || isFilterMet(filter)) { - - final String[] fsplit = field.split(REGEX, field.length()); - final String[] oldfsplit = oldField.split(REGEX, oldField.length()); - - final String oldValue = getEventObjectVal(oldfsplit[0]).toString(); - if (!oldValue.equals(OBJECT_NOT_FOUND)) { - final String oldArrayName = oldfsplit[1].substring(1); - final String newArrayName = fsplit[1].substring(1); - final String value = oldValue.replaceAll(oldArrayName, newArrayName); - - log.info("oldValue ==" + oldValue); - log.info("value ==" + value); - JSONArray ja = new JSONArray(value); - removeEventKey(oldfsplit[0]); - setEventObjectVal(fsplit[0], ja); - } - } else - log.info(FILTER_NOT_MET); - } - - - public void map(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - if (field.contains("[]")) { - if (field.matches(".*\\[\\]\\..*\\[\\]")) - renameArrayInArray(jsonObject); - else - mapToJArray(jsonObject); - } else - mapAttribute(jsonObject); - } - - private String performOperation(String operation, String value) { - log.info("performOperation"); - if ("convertMBtoKB".equals(operation)) { - float kbValue = Float.parseFloat(value) * 1024; - value = String.valueOf(kbValue); - } - return value; - } - - - public void mapAttribute(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - final String oldField = jsonObject.getString(OLD_FIELD); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - final String operation = jsonObject.optString("operation"); - String value; - if (filter == null || isFilterMet(filter)) { - - value = getEventObjectVal(oldField).toString(); - if (!value.equals(OBJECT_NOT_FOUND)) { - if (operation != null && !operation.isEmpty()) - value = performOperation(operation, value); - - setEventObjectVal(field, value); - - removeEventKey(oldField); - } - } else - log.info(FILTER_NOT_MET); - } - - - private void mapToJArray(JSONObject jsonObject) { - log.info("mapToJArray"); - String field = jsonObject.getString(FIELD); - String oldField = jsonObject.getString(OLD_FIELD); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - final JSONObject attrMap = jsonObject.optJSONObject("attrMap"); - oldField = oldField.replaceAll(REGEX, ""); - field = field.replaceAll(REGEX, ""); - - if (filter == null || isFilterMet(filter)) { - - String value = getEventObjectVal(oldField).toString(); - if (!value.equals(OBJECT_NOT_FOUND)) { - log.info("old value ==" + value); - // update old value based on attrMap - if (attrMap != null) { - // loop thru attrMap and update attribute name to new name - for (String key : attrMap.keySet()) { - value = value.replaceAll(key, attrMap.getString(key)); - } - } - - log.info("new value ==" + value); - char c = value.charAt(0); - if (c != '[') { - // oldfield is JsonObject - JSONObject valueJO = new JSONObject(value); - // if the array already exists - String existingValue = getEventObjectVal(field).toString(); - if (!existingValue.equals(OBJECT_NOT_FOUND)) { - JSONArray ja = new JSONArray(existingValue); - JSONObject jo = ja.optJSONObject(0); - if (jo != null) { - for (String key : valueJO.keySet()) { - jo.put(key, valueJO.get(key)); - - } - ja.put(0, jo); - - setEventObjectVal(field, ja); - } - } else // if new array - setEventObjectVal(field + "[0]", new JSONObject(value), "JArray"); - } else // oldfield is jsonArray - setEventObjectVal(field, new JSONArray(value)); - - removeEventKey(oldField); - } - } else - log.info(FILTER_NOT_MET); - } - - /** - * example - { "functionName": "concatenateValue", "args":{ "filter": - * {"event.commonEventHeader.event":"heartbeat"}, - * FIELD:"event.commonEventHeader.eventName", "concatenate": - * ["event.commonEventHeader.domain","event.commonEventHeader.eventType","event.commonEventHeader.alarmCondition"], - * "delimiter":"_" } } - **/ - public void concatenateValue(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - final String delimiter = jsonObject.getString("delimiter"); - final JSONArray values = jsonObject.getJSONArray("concatenate"); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - if (filter == null || isFilterMet(filter)) { - StringBuilder value = new StringBuilder(); - for (int i = 0; i < values.length(); i++) { - - String tempVal = evaluate(values.getString(i)); - if (!tempVal.equals(OBJECT_NOT_FOUND)) { - if (i == 0) - value.append(tempVal); - else - value.append(delimiter).append(tempVal); - } - } - - setEventObjectVal(field, value.toString()); - } else - log.info(FILTER_NOT_MET); - } - - public void subtractValue(JSONObject jsonObject) { - - final String field = jsonObject.getString(FIELD); - final JSONArray values = jsonObject.getJSONArray("subtract"); - final JSONObject filter = jsonObject.optJSONObject(FILTER); - if (filter == null || isFilterMet(filter)) { - float value = 0; - for (int i = 0; i < values.length(); i++) { - log.info(values.getString(i)); - String tempVal = evaluate(values.getString(i)); - log.info("tempVal==" + tempVal); - if (!tempVal.equals(OBJECT_NOT_FOUND)) { - if (i == 0) - value = value + Float.valueOf(tempVal); - else - value = value - Float.valueOf(tempVal); - } - } - log.info("value ==" + value); - setEventObjectVal(field, value, "number"); - } else - log.info(FILTER_NOT_MET); - } - - - private void removeEventKey(String field) { - String[] keySet = field.split("\\.", field.length()); - JSONObject keySeries = event; - for (int i = 0; i < (keySet.length - 1); i++) { - - keySeries = keySeries.getJSONObject(keySet[i]); - } - - keySeries.remove(keySet[keySet.length - 1]); - } - - - private boolean checkFilter(JSONObject jo, String key, String logicKey) { - String filterValue = jo.getString(key); - if (filterValue.contains(":")) { - String[] splitVal = filterValue.split(":"); - if ("matches".equals(splitVal[0])) { - if ("not".equals(logicKey)) { - if (getEventObjectVal(key).toString().matches(splitVal[1])) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } else { - if (!(getEventObjectVal(key).toString().matches(splitVal[1]))) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } - - } - if ("contains".equals(splitVal[0])) { - if ("not".equals(logicKey)) { - if (getEventObjectVal(key).toString().contains(splitVal[1])) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } else { - if (!(getEventObjectVal(key).toString().contains(splitVal[1]))) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } - - } - } else { - if ("not".equals(logicKey)) { - if (getEventObjectVal(key).toString().equals(filterValue)) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } else { - if (!(getEventObjectVal(key).toString().equals(filterValue))) { - log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); - return false; - } - } - } - return true; - } - - - public boolean isFilterMet(JSONObject jo) { - for (String key : jo.keySet()) { - if ("not".equals(key)) { - JSONObject njo = jo.getJSONObject(key); - for (String njoKey : njo.keySet()) { - if (!checkFilter(njo, njoKey, key)) - return false; - } - } else { - if (!checkFilter(jo, key, key)) - return false; - } - } - return true; - } - - /** - * returns a string or JSONObject or JSONArray - **/ - public Object getEventObjectVal(String keySeriesStr) { - keySeriesStr = keySeriesStr.replaceAll("\\[", "."); - keySeriesStr = keySeriesStr.replaceAll("\\]", "."); - if (keySeriesStr.contains("..")) { - keySeriesStr = keySeriesStr.replaceAll("\\.\\.", "."); - } - - if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1) - keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1); - String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length()); - Object keySeriesObj = event; - for (String aKeySet : keySet) { - if (keySeriesObj != null) { - if (keySeriesObj instanceof String) { - - log.info("STRING==" + keySeriesObj); - } else if (keySeriesObj instanceof JSONArray) { - keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(aKeySet)); - - } else if (keySeriesObj instanceof JSONObject) { - keySeriesObj = ((JSONObject) keySeriesObj).opt(aKeySet); - - } else { - log.info("unknown object==" + keySeriesObj); - } - } - } - - if (keySeriesObj == null) - return OBJECT_NOT_FOUND; - return keySeriesObj; - } - - public void setEventObjectVal(String keySeriesStr, Object value) { - setEventObjectVal(keySeriesStr, value, "string"); - } - - /** - * returns a string or JSONObject or JSONArray - **/ - public void setEventObjectVal(String keySeriesStr, Object value, String fieldType) { - keySeriesStr = keySeriesStr.replaceAll("\\[", "."); - keySeriesStr = keySeriesStr.replaceAll("\\]", "."); - if (keySeriesStr.contains("..")) { - keySeriesStr = keySeriesStr.replaceAll("\\.\\.", "."); - } - log.info("fieldType==" + fieldType); - - if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1) - keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1); - String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length()); - Object keySeriesObj = event; - for (int i = 0; i < (keySet.length - 1); i++) { - - if (keySeriesObj instanceof JSONArray) { - - if (((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])) == null) // if - // the - // object - // is - // not - // there - // then - // add - // it - { - log.info("Object is null, must add it"); - if (keySet[i + 1].matches("[0-9]*")) // if index then array - ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONArray()); - else - ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONObject()); - } - keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])); - - } else if (keySeriesObj instanceof JSONObject) { - if (((JSONObject) keySeriesObj).opt(keySet[i]) == null) // if - // the - // object - // is - // not - // there - // then - // add - // it - { - if (keySet[i + 1].matches("[0-9]*")) // if index then array - ((JSONObject) keySeriesObj).put(keySet[i], new JSONArray()); - else - ((JSONObject) keySeriesObj).put(keySet[i], new JSONObject()); - log.info("Object is null, must add it"); - } - keySeriesObj = ((JSONObject) keySeriesObj).opt(keySet[i]); - } else { - log.info("unknown object==" + keySeriesObj); - } - } - if ("number".equals(fieldType)) { - DecimalFormat df = new DecimalFormat("#.0"); - if (value instanceof String) - ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], - Float.valueOf(df.format(Float.valueOf((String) value)))); - else - ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Float.valueOf(df.format(value))); - } else if ("integer".equals(fieldType) && value instanceof String) - ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Integer.valueOf((String) value)); - else if ("JArray".equals(fieldType)) - ((JSONArray) keySeriesObj).put(value); - else - ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], value); - - } + public void updateAttribute(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final String value = evaluate(jsonObject.getString(VALUE)); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + setEventObjectVal(field, value); + } else + log.info(FILTER_NOT_MET); + } + + + public void removeAttribute(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + + if (filter == null || isFilterMet(filter)) { + removeEventKey(field); + } else + log.info(FILTER_NOT_MET); + } + + + private void renameArrayInArray(JSONObject jsonObject) // map + { + log.info("renameArrayInArray"); + final String field = jsonObject.getString(FIELD); + final String oldField = jsonObject.getString(OLD_FIELD); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + + if (filter == null || isFilterMet(filter)) { + + final String[] fsplit = field.split(REGEX, field.length()); + final String[] oldfsplit = oldField.split(REGEX, oldField.length()); + + final String oldValue = getEventObjectVal(oldfsplit[0]).toString(); + if (!oldValue.equals(OBJECT_NOT_FOUND)) { + final String oldArrayName = oldfsplit[1].substring(1); + final String newArrayName = fsplit[1].substring(1); + final String value = oldValue.replaceAll(oldArrayName, newArrayName); + + log.info("oldValue ==" + oldValue); + log.info("value ==" + value); + JSONArray ja = new JSONArray(value); + removeEventKey(oldfsplit[0]); + setEventObjectVal(fsplit[0], ja); + } + } else + log.info(FILTER_NOT_MET); + } + + + public void map(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + if (field.contains("[]")) { + if (field.matches(".*\\[\\]\\..*\\[\\]")) + renameArrayInArray(jsonObject); + else + mapToJArray(jsonObject); + } else + mapAttribute(jsonObject); + } + + private String performOperation(String operation, String value) { + log.info("performOperation"); + if ("convertMBtoKB".equals(operation)) { + float kbValue = Float.parseFloat(value) * 1024; + value = String.valueOf(kbValue); + } + return value; + } + + + public void mapAttribute(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final String oldField = jsonObject.getString(OLD_FIELD); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + final String operation = jsonObject.optString("operation"); + String value; + if (filter == null || isFilterMet(filter)) { + + value = getEventObjectVal(oldField).toString(); + if (!value.equals(OBJECT_NOT_FOUND)) { + if (operation != null && !operation.isEmpty()) + value = performOperation(operation, value); + + setEventObjectVal(field, value); + + removeEventKey(oldField); + } + } else + log.info(FILTER_NOT_MET); + } + + + private void mapToJArray(JSONObject jsonObject) { + log.info("mapToJArray"); + String field = jsonObject.getString(FIELD); + String oldField = jsonObject.getString(OLD_FIELD); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + final JSONObject attrMap = jsonObject.optJSONObject("attrMap"); + oldField = oldField.replaceAll(REGEX, ""); + field = field.replaceAll(REGEX, ""); + + if (filter == null || isFilterMet(filter)) { + + String value = getEventObjectVal(oldField).toString(); + if (!value.equals(OBJECT_NOT_FOUND)) { + log.info("old value ==" + value); + // update old value based on attrMap + if (attrMap != null) { + // loop thru attrMap and update attribute name to new name + for (String key : attrMap.keySet()) { + value = value.replaceAll(key, attrMap.getString(key)); + } + } + + log.info("new value ==" + value); + char c = value.charAt(0); + if (c != '[') { + // oldfield is JsonObject + JSONObject valueJO = new JSONObject(value); + // if the array already exists + String existingValue = getEventObjectVal(field).toString(); + if (!existingValue.equals(OBJECT_NOT_FOUND)) { + JSONArray ja = new JSONArray(existingValue); + JSONObject jo = ja.optJSONObject(0); + if (jo != null) { + for (String key : valueJO.keySet()) { + jo.put(key, valueJO.get(key)); + + } + ja.put(0, jo); + + setEventObjectVal(field, ja); + } + } else // if new array + setEventObjectVal(field + "[0]", new JSONObject(value), "JArray"); + } else // oldfield is jsonArray + setEventObjectVal(field, new JSONArray(value)); + + removeEventKey(oldField); + } + } else + log.info(FILTER_NOT_MET); + } + + /** + * example - { "functionName": "concatenateValue", "args":{ "filter": + * {"event.commonEventHeader.event":"heartbeat"}, + * FIELD:"event.commonEventHeader.eventName", "concatenate": + * ["event.commonEventHeader.domain","event.commonEventHeader.eventType","event.commonEventHeader.alarmCondition"], + * "delimiter":"_" } } + **/ + public void concatenateValue(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final String delimiter = jsonObject.getString("delimiter"); + final JSONArray values = jsonObject.getJSONArray("concatenate"); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + StringBuilder value = new StringBuilder(); + for (int i = 0; i < values.length(); i++) { + + String tempVal = evaluate(values.getString(i)); + if (!tempVal.equals(OBJECT_NOT_FOUND)) { + if (i == 0) + value.append(tempVal); + else + value.append(delimiter).append(tempVal); + } + } + + setEventObjectVal(field, value.toString()); + } else + log.info(FILTER_NOT_MET); + } + + public void subtractValue(JSONObject jsonObject) { + + final String field = jsonObject.getString(FIELD); + final JSONArray values = jsonObject.getJSONArray("subtract"); + final JSONObject filter = jsonObject.optJSONObject(FILTER); + if (filter == null || isFilterMet(filter)) { + float value = 0; + for (int i = 0; i < values.length(); i++) { + log.info(values.getString(i)); + String tempVal = evaluate(values.getString(i)); + log.info("tempVal==" + tempVal); + if (!tempVal.equals(OBJECT_NOT_FOUND)) { + if (i == 0) + value = value + Float.valueOf(tempVal); + else + value = value - Float.valueOf(tempVal); + } + } + log.info("value ==" + value); + setEventObjectVal(field, value, "number"); + } else + log.info(FILTER_NOT_MET); + } + + + private void removeEventKey(String field) { + String[] keySet = field.split("\\.", field.length()); + JSONObject keySeries = event; + for (int i = 0; i < (keySet.length - 1); i++) { + + keySeries = keySeries.getJSONObject(keySet[i]); + } + + keySeries.remove(keySet[keySet.length - 1]); + } + + + private boolean checkFilter(JSONObject jo, String key, String logicKey) { + String filterValue = jo.getString(key); + if (filterValue.contains(":")) { + String[] splitVal = filterValue.split(":"); + if ("matches".equals(splitVal[0])) { + if ("not".equals(logicKey)) { + if (getEventObjectVal(key).toString().matches(splitVal[1])) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } else { + if (!(getEventObjectVal(key).toString().matches(splitVal[1]))) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } + + } + if ("contains".equals(splitVal[0])) { + if ("not".equals(logicKey)) { + if (getEventObjectVal(key).toString().contains(splitVal[1])) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } else { + if (!(getEventObjectVal(key).toString().contains(splitVal[1]))) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } + + } + } else { + if ("not".equals(logicKey)) { + if (getEventObjectVal(key).toString().equals(filterValue)) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } else { + if (!(getEventObjectVal(key).toString().equals(filterValue))) { + log.info(filterValue + "==" + key + "==" + getEventObjectVal(key) + COMP_FALSE); + return false; + } + } + } + return true; + } + + + public boolean isFilterMet(JSONObject jo) { + for (String key : jo.keySet()) { + if ("not".equals(key)) { + JSONObject njo = jo.getJSONObject(key); + for (String njoKey : njo.keySet()) { + if (!checkFilter(njo, njoKey, key)) + return false; + } + } else { + if (!checkFilter(jo, key, key)) + return false; + } + } + return true; + } + + /** + * returns a string or JSONObject or JSONArray + **/ + public Object getEventObjectVal(String keySeriesStr) { + keySeriesStr = keySeriesStr.replaceAll("\\[", "."); + keySeriesStr = keySeriesStr.replaceAll("\\]", "."); + if (keySeriesStr.contains("..")) { + keySeriesStr = keySeriesStr.replaceAll("\\.\\.", "."); + } + + if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1) + keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1); + String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length()); + Object keySeriesObj = event; + for (String aKeySet : keySet) { + if (keySeriesObj != null) { + if (keySeriesObj instanceof String) { + + log.info("STRING==" + keySeriesObj); + } else if (keySeriesObj instanceof JSONArray) { + keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(aKeySet)); + + } else if (keySeriesObj instanceof JSONObject) { + keySeriesObj = ((JSONObject) keySeriesObj).opt(aKeySet); + + } else { + log.info("unknown object==" + keySeriesObj); + } + } + } + + if (keySeriesObj == null) + return OBJECT_NOT_FOUND; + return keySeriesObj; + } + + public void setEventObjectVal(String keySeriesStr, Object value) { + setEventObjectVal(keySeriesStr, value, "string"); + } + + /** + * returns a string or JSONObject or JSONArray + **/ + public void setEventObjectVal(String keySeriesStr, Object value, String fieldType) { + keySeriesStr = keySeriesStr.replaceAll("\\[", "."); + keySeriesStr = keySeriesStr.replaceAll("\\]", "."); + if (keySeriesStr.contains("..")) { + keySeriesStr = keySeriesStr.replaceAll("\\.\\.", "."); + } + log.info("fieldType==" + fieldType); + + if (keySeriesStr.lastIndexOf(".") == keySeriesStr.length() - 1) + keySeriesStr = keySeriesStr.substring(0, keySeriesStr.length() - 1); + String[] keySet = keySeriesStr.split("\\.", keySeriesStr.length()); + Object keySeriesObj = event; + for (int i = 0; i < (keySet.length - 1); i++) { + + if (keySeriesObj instanceof JSONArray) { + + if (((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])) == null) // if + // the + // object + // is + // not + // there + // then + // add + // it + { + log.info("Object is null, must add it"); + if (keySet[i + 1].matches("[0-9]*")) // if index then array + ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONArray()); + else + ((JSONArray) keySeriesObj).put(Integer.parseInt(keySet[i]), new JSONObject()); + } + keySeriesObj = ((JSONArray) keySeriesObj).optJSONObject(Integer.parseInt(keySet[i])); + + } else if (keySeriesObj instanceof JSONObject) { + if (((JSONObject) keySeriesObj).opt(keySet[i]) == null) // if + // the + // object + // is + // not + // there + // then + // add + // it + { + if (keySet[i + 1].matches("[0-9]*")) // if index then array + ((JSONObject) keySeriesObj).put(keySet[i], new JSONArray()); + else + ((JSONObject) keySeriesObj).put(keySet[i], new JSONObject()); + log.info("Object is null, must add it"); + } + keySeriesObj = ((JSONObject) keySeriesObj).opt(keySet[i]); + } else { + log.info("unknown object==" + keySeriesObj); + } + } + if ("number".equals(fieldType)) { + DecimalFormat df = new DecimalFormat("#.0"); + if (value instanceof String) + ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], + Float.valueOf(df.format(Float.valueOf((String) value)))); + else + ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Float.valueOf(df.format(value))); + } else if ("integer".equals(fieldType) && value instanceof String) + ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], Integer.valueOf((String) value)); + else if ("JArray".equals(fieldType)) + ((JSONArray) keySeriesObj).put(value); + else + ((JSONObject) keySeriesObj).put(keySet[keySet.length - 1], value); + + } } diff --git a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java index a57ea3f0..7d27399d 100644 --- a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java +++ b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java @@ -25,7 +25,10 @@ import com.att.nsa.logging.LoggingContext; import com.att.nsa.logging.log4j.EcompFields;
import com.google.common.reflect.TypeToken;
import com.google.gson.Gson;
+import io.vavr.collection.Map;
import org.json.JSONObject;
+import org.onap.dcae.ApplicationSettings;
+import org.onap.dcae.VesApplication;
import org.onap.dcae.commonFunction.event.publishing.EventPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,37 +38,38 @@ import java.io.IOException; import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.text.SimpleDateFormat;
-import java.util.Arrays;
import java.util.Date;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-class EventProcessor implements Runnable {
+public class EventProcessor implements Runnable {
+ static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {
+ }.getType();
private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);
private static final String EVENT_LITERAL = "event";
private static final String COMMON_EVENT_HEADER = "commonEventHeader";
- static final Type EVENT_LIST_TYPE = new TypeToken<List<Event>>() {}.getType();
private final SimpleDateFormat dateFormat = new SimpleDateFormat("EEE, MM dd yyyy hh:mm:ss z");
- static Map<String, String[]> streamidHash = new HashMap<>();
public JSONObject event;
private EventPublisher eventPublisher;
+ private Map<String, String[]> streamidHash;
+ private ApplicationSettings properties;
- public EventProcessor(EventPublisher eventPublisher) {
+
+ public EventProcessor(EventPublisher eventPublisher, ApplicationSettings properties) {
this.eventPublisher = eventPublisher;
- streamidHash = CommonStartup.streamID.toJavaMap();
+ this.properties = properties;
+ this.streamidHash = properties.dMaaPStreamsMapping();
}
@Override
public void run() {
try {
while (true) {
- event = CommonStartup.fProcessingInputQueue.take();
+ event = VesApplication.fProcessingInputQueue.take();
// As long as the producer is running we remove elements from
// the queue.
- log.info("QueueSize:" + CommonStartup.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);
+ log.info("QueueSize:" + VesApplication.fProcessingInputQueue.size() + "\tEventProcessor\tRemoving element: " + event);
String uuid = event.get("VESuniqueId").toString();
LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);
@@ -73,14 +77,12 @@ class EventProcessor implements Runnable { String domain = event.getJSONObject(EVENT_LITERAL).getJSONObject(COMMON_EVENT_HEADER).getString("domain");
log.debug("event.VESuniqueId" + event.get("VESuniqueId") + "event.commonEventHeader.domain:" + domain);
- String[] streamIdList = streamidHash.get(domain);
- log.debug("streamIdList:" + Arrays.toString(streamIdList));
-
- if (streamIdList.length == 0) {
- log.error("No StreamID defined for publish - Message dropped" + event);
- } else {
- sendEventsToStreams(streamIdList);
- }
+ streamidHash.get(domain)
+ .onEmpty(() -> {
+ log.error("No StreamID defined for publish - Message dropped" + event);
+ }).forEach(streamIds -> {
+ sendEventsToStreams(streamIds);
+ });
log.debug("Message published" + event);
}
} catch (InterruptedException e) {
@@ -93,7 +95,7 @@ class EventProcessor implements Runnable { // Set collector timestamp in event payload before publish
addCurrentTimeToEvent(event);
- if (CommonStartup.eventTransformFlag) {
+ if (properties.eventTransformingEnabled()) {
// read the mapping json file
try (FileReader fr = new FileReader("./etc/eventTransform.json")) {
log.info("parse eventTransform.json");
@@ -168,5 +170,4 @@ class EventProcessor implements Runnable { method.invoke(configProcessors, parameter);
}
}
-}
-
+}
\ No newline at end of file diff --git a/src/main/java/org/onap/dcae/commonFunction/VESLogger.java b/src/main/java/org/onap/dcae/commonFunction/VESLogger.java index a967327e..2a392e81 100644 --- a/src/main/java/org/onap/dcae/commonFunction/VESLogger.java +++ b/src/main/java/org/onap/dcae/commonFunction/VESLogger.java @@ -8,9 +8,9 @@ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -35,128 +35,126 @@ import java.util.UUID; public class VESLogger { - public static final String VES_AGENT = "VES_AGENT"; - public static final String REQUEST_ID = "requestId"; - private static final String IP_ADDRESS = "127.0.0.1"; - private static final String HOST_NAME = "localhost"; - - public static Logger auditLog; - public static Logger metricsLog; - public static Logger errorLog; - public static Logger debugLog; - - // Common LoggingContext - private static LoggingContext commonLC; - // Thread-specific LoggingContext - private static LoggingContext threadLC; - public LoggingContext lc; - - /** - * Returns the common LoggingContext instance that is the base context for - * all subsequent instances. - * - * @return the common LoggingContext - */ - public static LoggingContext getCommonLoggingContext() { - if (commonLC == null) { - commonLC = new LoggingContextFactory.Builder().build(); - final UUID uuid = UUID.randomUUID(); - - commonLC.put(REQUEST_ID, uuid.toString()); - } - return commonLC; - } - - /** - * Get a logging context for the current thread that's based on the common - * logging context. Populate the context with context-specific values. - * - * @param aUuid - * uuid for request id - * @return a LoggingContext for the current thread - */ - public static LoggingContext getLoggingContextForThread(UUID aUuid) { - // note that this operation requires everything from the common context - // to be (re)copied into the target context. That seems slow, but it - // actually - // helps prevent the thread from overwriting supposedly common data. It - // also - // should be fairly quick compared with the overhead of handling the - // actual - // service call. - - threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build(); - // Establish the request-specific UUID, as long as we are here... - threadLC.put(REQUEST_ID, aUuid.toString()); - threadLC.put(EcompFields.kEndTimestamp, SaClock.now()); - - return threadLC; - } - - /** - * Get a logging context for the current thread that's based on the common - * logging context. Populate the context with context-specific values. - * - * @param aUuid - * uuid for request id - * @return a LoggingContext for the current thread - */ - public static LoggingContext getLoggingContextForThread(String aUuid) { - // note that this operation requires everything from the common context - // to be (re)copied into the target context. That seems slow, but it - // actually - // helps prevent the thread from overwriting supposedly common data. It - // also - // should be fairly quick compared with the overhead of handling the - // actual - // service call. - - threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build(); - // Establish the request-specific UUID, as long as we are here... - threadLC.put(REQUEST_ID, aUuid); - threadLC.put("statusCode", "COMPLETE"); - threadLC.put(EcompFields.kEndTimestamp, SaClock.now()); - return threadLC; - } - - public static void setUpEcompLogging() { - - // Create ECOMP Logger instances - auditLog = LoggerFactory.getLogger("com.att.ecomp.audit"); - metricsLog = LoggerFactory.getLogger("com.att.ecomp.metrics"); - debugLog = LoggerFactory.getLogger("com.att.ecomp.debug"); - errorLog = LoggerFactory.getLogger("com.att.ecomp.error"); - - final LoggingContext lc = getCommonLoggingContext(); - - String ipAddr = IP_ADDRESS; - String hostname = HOST_NAME; - try { - final InetAddress ip = InetAddress.getLocalHost(); - hostname = ip.getCanonicalHostName(); - ipAddr = ip.getHostAddress(); - } catch (UnknownHostException x) { - Log.debug(x.getMessage()); - } - - lc.put("serverName", hostname); - lc.put("serviceName", "VESCollecor"); - lc.put("statusCode", "RUNNING"); - lc.put("targetEntity", "NULL"); - lc.put("targetServiceName", "NULL"); - lc.put("server", hostname); - lc.put("serverIpAddress", ipAddr); - - // instance UUID is meaningless here, so we just create a new one each - // time the - // server starts. One could argue each new instantiation of the service - // should - // have a new instance ID. - lc.put("instanceUuid", ""); - lc.put("severity", ""); - lc.put(EcompFields.kEndTimestamp, SaClock.now()); - lc.put("EndTimestamp", SaClock.now()); - lc.put("partnerName", "NA"); - } + public static final String VES_AGENT = "VES_AGENT"; + public static final String REQUEST_ID = "requestId"; + private static final String IP_ADDRESS = "127.0.0.1"; + private static final String HOST_NAME = "localhost"; + + public static Logger auditLog; + public static Logger metricsLog; + public static Logger errorLog; + public static Logger debugLog; + + // Common LoggingContext + private static LoggingContext commonLC; + // Thread-specific LoggingContext + private static LoggingContext threadLC; + public LoggingContext lc; + + /** + * Returns the common LoggingContext instance that is the base context for + * all subsequent instances. + * + * @return the common LoggingContext + */ + public static LoggingContext getCommonLoggingContext() { + if (commonLC == null) { + commonLC = new LoggingContextFactory.Builder().build(); + final UUID uuid = UUID.randomUUID(); + + commonLC.put(REQUEST_ID, uuid.toString()); + } + return commonLC; + } + + /** + * Get a logging context for the current thread that's based on the common + * logging context. Populate the context with context-specific values. + * + * @param aUuid uuid for request id + * @return a LoggingContext for the current thread + */ + public static LoggingContext getLoggingContextForThread(UUID aUuid) { + // note that this operation requires everything from the common context + // to be (re)copied into the target context. That seems slow, but it + // actually + // helps prevent the thread from overwriting supposedly common data. It + // also + // should be fairly quick compared with the overhead of handling the + // actual + // service call. + + threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build(); + // Establish the request-specific UUID, as long as we are here... + threadLC.put(REQUEST_ID, aUuid.toString()); + threadLC.put(EcompFields.kEndTimestamp, SaClock.now()); + + return threadLC; + } + + /** + * Get a logging context for the current thread that's based on the common + * logging context. Populate the context with context-specific values. + * + * @param aUuid uuid for request id + * @return a LoggingContext for the current thread + */ + public static LoggingContext getLoggingContextForThread(String aUuid) { + // note that this operation requires everything from the common context + // to be (re)copied into the target context. That seems slow, but it + // actually + // helps prevent the thread from overwriting supposedly common data. It + // also + // should be fairly quick compared with the overhead of handling the + // actual + // service call. + + threadLC = new LoggingContextFactory.Builder().withBaseContext(getCommonLoggingContext()).build(); + // Establish the request-specific UUID, as long as we are here... + threadLC.put(REQUEST_ID, aUuid); + threadLC.put("statusCode", "COMPLETE"); + threadLC.put(EcompFields.kEndTimestamp, SaClock.now()); + return threadLC; + } + + public static void setUpEcompLogging() { + + // Create ECOMP Logger instances + auditLog = LoggerFactory.getLogger("com.att.ecomp.audit"); + metricsLog = LoggerFactory.getLogger("com.att.ecomp.metrics"); + debugLog = LoggerFactory.getLogger("com.att.ecomp.debug"); + errorLog = LoggerFactory.getLogger("com.att.ecomp.error"); + + final LoggingContext lc = getCommonLoggingContext(); + + String ipAddr = IP_ADDRESS; + String hostname = HOST_NAME; + try { + final InetAddress ip = InetAddress.getLocalHost(); + hostname = ip.getCanonicalHostName(); + ipAddr = ip.getHostAddress(); + } catch (UnknownHostException x) { + Log.debug(x.getMessage()); + } + + lc.put("serverName", hostname); + lc.put("serviceName", "VESCollecor"); + lc.put("statusCode", "RUNNING"); + lc.put("targetEntity", "NULL"); + lc.put("targetServiceName", "NULL"); + lc.put("server", hostname); + lc.put("serverIpAddress", ipAddr); + + // instance UUID is meaningless here, so we just create a new one each + // time the + // server starts. One could argue each new instantiation of the service + // should + // have a new instance ID. + lc.put("instanceUuid", ""); + lc.put("severity", ""); + lc.put(EcompFields.kEndTimestamp, SaClock.now()); + lc.put("EndTimestamp", SaClock.now()); + lc.put("partnerName", "NA"); + } } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java index 5865b12c..179e8826 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPConfigurationParser.java @@ -19,21 +19,19 @@ */ package org.onap.dcae.commonFunction.event.publishing; -import static io.vavr.API.List; -import static io.vavr.API.Try; -import static io.vavr.API.Tuple; -import static io.vavr.API.unchecked; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; - import io.vavr.collection.List; import io.vavr.collection.Map; import io.vavr.control.Option; import io.vavr.control.Try; +import org.onap.dcae.commonFunction.AnyNode; + import java.net.URL; import java.nio.file.Files; import java.nio.file.Path; -import org.onap.dcae.commonFunction.AnyNode; + +import static io.vavr.API.*; +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError; +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; /** * @author Pawel Szalapski (pawel.szalapski@nokia.com) @@ -43,23 +41,23 @@ public final class DMaaPConfigurationParser { public static Try<Map<String, PublisherConfig>> parseToDomainMapping(Path configLocation) { return readFromFile(configLocation) - .flatMap(DMaaPConfigurationParser::toJSON) - .flatMap(DMaaPConfigurationParser::toConfigMap); + .flatMap(DMaaPConfigurationParser::toJSON) + .flatMap(DMaaPConfigurationParser::toConfigMap); } private static Try<String> readFromFile(Path configLocation) { return Try(() -> new String(Files.readAllBytes(configLocation))) - .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation))); + .mapFailure(enhanceError(f("Could not read DMaaP configuration from location: '%s'", configLocation))); } private static Try<AnyNode> toJSON(String config) { return Try(() -> AnyNode.fromString(config)) - .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config))); + .mapFailure(enhanceError(f("DMaaP configuration '%s' is not a valid JSON document", config))); } private static Try<Map<String, PublisherConfig>> toConfigMap(AnyNode config) { return Try(() -> usesLegacyFormat(config) ? parseLegacyFormat(config) : parseNewFormat(config)) - .mapFailure(enhanceError(f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config))); + .mapFailure(enhanceError(f("Parsing DMaaP configuration: '%s' failed, probably it is in unexpected format", config))); } private static boolean usesLegacyFormat(AnyNode dMaaPConfig) { @@ -68,40 +66,40 @@ public final class DMaaPConfigurationParser { private static Map<String, PublisherConfig> parseLegacyFormat(AnyNode root) { return root.get("channels").toList().toMap( - channel -> channel.get("name").toString(), - channel -> { - String destinationsStr = channel.getAsOption("cambria.url") - .getOrElse(channel.getAsOption("cambria.hosts").get()) - .toString(); - String topic = channel.get("cambria.topic").toString(); - Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString); - Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString); - List<String> destinations = List(destinationsStr.split(",")); - return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); - }); + channel -> channel.get("name").toString(), + channel -> { + String destinationsStr = channel.getAsOption("cambria.url") + .getOrElse(channel.getAsOption("cambria.hosts").get()) + .toString(); + String topic = channel.get("cambria.topic").toString(); + Option<String> maybeUser = channel.getAsOption("basicAuthUsername").map(AnyNode::toString); + Option<String> maybePassword = channel.getAsOption("basicAuthPassword").map(AnyNode::toString); + List<String> destinations = List(destinationsStr.split(",")); + return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); + }); } private static Map<String, PublisherConfig> parseNewFormat(AnyNode root) { return root.keys().toMap( - channelName -> channelName, - channelName -> { - AnyNode channelConfig = root.get(channelName); - Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString); - Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString); - URL topicURL = unchecked( - () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply(); - String[] pathSegments = topicURL.getPath().substring(1).split("/"); - String topic = pathSegments[1]; - String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost(); - List<String> destinations = List(destination); - return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); - }); + channelName -> channelName, + channelName -> { + AnyNode channelConfig = root.get(channelName); + Option<String> maybeUser = channelConfig.getAsOption("aaf_username").map(AnyNode::toString); + Option<String> maybePassword = channelConfig.getAsOption("aaf_password").map(AnyNode::toString); + URL topicURL = unchecked( + () -> new URL(channelConfig.get("dmaap_info").get("topic_url").toString())).apply(); + String[] pathSegments = topicURL.getPath().substring(1).split("/"); + String topic = pathSegments[1]; + String destination = "events".equals(pathSegments[0]) ? topicURL.getAuthority() : topicURL.getHost(); + List<String> destinations = List(destination); + return buildBasedOnAuth(maybeUser, maybePassword, topic, destinations); + }); } private static PublisherConfig buildBasedOnAuth(Option<String> maybeUser, Option<String> maybePassword, String topic, List<String> destinations) { return maybeUser.flatMap(user -> maybePassword.map(password -> Tuple(user, password))) - .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2)) - .getOrElse(new PublisherConfig(destinations, topic)); + .map(credentials -> new PublisherConfig(destinations, topic, credentials._1, credentials._2)) + .getOrElse(new PublisherConfig(destinations, topic)); } } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java index fd9b3ae1..a0ee3bfb 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPEventPublisher.java @@ -21,20 +21,21 @@ package org.onap.dcae.commonFunction.event.publishing; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; - import com.att.nsa.cambria.client.CambriaBatchingPublisher; import com.att.nsa.clock.SaClock; import com.att.nsa.logging.LoggingContext; import com.att.nsa.logging.log4j.EcompFields; import io.vavr.collection.Map; import io.vavr.control.Try; -import java.io.IOException; import org.json.JSONObject; import org.onap.dcae.commonFunction.VESLogger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; + +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; + /** * @author Pawel Szalapski (pawel.szalapski@nokia.com) */ @@ -55,9 +56,9 @@ class DMaaPEventPublisher implements EventPublisher { public void sendEvent(JSONObject event, String domain) { clearVesUniqueIdFromEvent(event); publishersCache.getPublisher(domain) - .onEmpty(() -> - log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event))) - .forEach(publisher -> sendEvent(event, domain, publisher)); + .onEmpty(() -> + log.warn(f("Could not find event publisher for domain: '%s', dropping message: '%s'", domain, event))) + .forEach(publisher -> sendEvent(event, domain, publisher)); } @Override @@ -67,11 +68,11 @@ class DMaaPEventPublisher implements EventPublisher { private void sendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) { Try.run(() -> uncheckedSendEvent(event, domain, publisher)) - .onFailure(exc -> closePublisher(event, domain, exc)); + .onFailure(exc -> closePublisher(event, domain, exc)); } private void uncheckedSendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) - throws IOException { + throws IOException { int pendingMsgs = publisher.send("MyPartitionKey", event.toString()); if (pendingMsgs > PENDING_MESSAGE_LOG_THRESHOLD) { log.info("Pending messages count: " + pendingMsgs); @@ -83,7 +84,7 @@ class DMaaPEventPublisher implements EventPublisher { private void closePublisher(JSONObject event, String domain, Throwable e) { log.error(f("Unable to schedule event: '%s' on domain: '%s'. Closing publisher and dropping message.", - event, domain), e); + event, domain), e); publishersCache.closePublisherFor(domain); } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java index a7865a45..489fcbf0 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersBuilder.java @@ -19,15 +19,15 @@ */ package org.onap.dcae.commonFunction.event.publishing; -import static io.vavr.API.Try; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; - import com.att.nsa.cambria.client.CambriaBatchingPublisher; import com.att.nsa.cambria.client.CambriaClientBuilders; import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder; import io.vavr.control.Try; +import static io.vavr.API.Try; +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.enhanceError; +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; + /** * @author Pawel Szalapski (pawel.szalapski@nokia.com) */ @@ -36,7 +36,7 @@ final class DMaaPPublishersBuilder { @SuppressWarnings("mapFailure takes a generic varargs, unchecked because of Javas type system limitation, actually safe to do") static Try<CambriaBatchingPublisher> buildPublisher(PublisherConfig config) { return Try(() -> builder(config).build()) - .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config))); + .mapFailure(enhanceError(f("DMaaP client builder throws exception for this configuration: '%s'", config))); } private static PublisherBuilder builder(PublisherConfig config) { @@ -49,14 +49,14 @@ final class DMaaPPublishersBuilder { private static PublisherBuilder authenticatedBuilder(PublisherConfig config) { return unAuthenticatedBuilder(config) - .usingHttps() - .authenticatedByHttp(config.userName().get(), config.password().get()); + .usingHttps() + .authenticatedByHttp(config.userName().get(), config.password().get()); } private static PublisherBuilder unAuthenticatedBuilder(PublisherConfig config) { return new CambriaClientBuilders.PublisherBuilder() - .usingHosts(config.destinations().mkString(",")) - .onTopic(config.topic()) - .logSendFailuresAfter(5); + .usingHosts(config.destinations().mkString(",")) + .onTopic(config.topic()) + .logSendFailuresAfter(5); } } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java index 102d2774..4cdf92da 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/DMaaPPublishersCache.java @@ -20,24 +20,20 @@ */ package org.onap.dcae.commonFunction.event.publishing; -import static io.vavr.API.Option; -import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; - import com.att.nsa.cambria.client.CambriaBatchingPublisher; -import com.google.common.annotations.VisibleForTesting; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.google.common.cache.RemovalListener; -import com.google.common.cache.RemovalNotification; +import com.google.common.cache.*; import io.vavr.collection.Map; import io.vavr.control.Option; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; import java.io.IOException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import javax.annotation.Nonnull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; + +import static io.vavr.API.Option; +import static org.onap.dcae.commonFunction.event.publishing.VavrUtils.f; /** * @author Pawel Szalapski (pawel.szalapski@nokia.com) @@ -51,8 +47,8 @@ class DMaaPPublishersCache { DMaaPPublishersCache(Map<String, PublisherConfig> dMaaPConfiguration) { this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration); this.publishersCache = CacheBuilder.newBuilder() - .removalListener(new OnPublisherRemovalListener()) - .build(new CambriaPublishersCacheLoader()); + .removalListener(new OnPublisherRemovalListener()) + .build(new CambriaPublishersCacheLoader()); } DMaaPPublishersCache(CambriaPublishersCacheLoader dMaaPPublishersCacheLoader, @@ -60,8 +56,8 @@ class DMaaPPublishersCache { Map<String, PublisherConfig> dMaaPConfiguration) { this.dMaaPConfiguration = new AtomicReference<>(dMaaPConfiguration); this.publishersCache = CacheBuilder.newBuilder() - .removalListener(onPublisherRemovalListener) - .build(dMaaPPublishersCacheLoader); + .removalListener(onPublisherRemovalListener) + .build(dMaaPPublishersCacheLoader); } Option<CambriaBatchingPublisher> getPublisher(String streamID) { @@ -80,9 +76,9 @@ class DMaaPPublishersCache { synchronized void reconfigure(Map<String, PublisherConfig> newConfig) { Map<String, PublisherConfig> currentConfig = dMaaPConfiguration.get(); Map<String, PublisherConfig> removedConfigurations = currentConfig - .filterKeys(domain -> !newConfig.containsKey(domain)); + .filterKeys(domain -> !newConfig.containsKey(domain)); Map<String, PublisherConfig> changedConfigurations = newConfig - .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e))); + .filterKeys(e -> currentConfig.containsKey(e) && !currentConfig.get(e).equals(newConfig.get(e))); dMaaPConfiguration.set(newConfig); removedConfigurations.merge(changedConfigurations).forEach(e -> publishersCache.invalidate(e._1)); } @@ -99,7 +95,7 @@ class DMaaPPublishersCache { java.util.List<?> stuck = publisher.close(timeout, unit); if (!stuck.isEmpty()) { log.error(f("Publisher got stuck and did not manage to close in '%s' '%s', " - + "%s messages were dropped", stuck.size(), timeout, unit)); + + "%s messages were dropped", stuck.size(), timeout, unit)); } } catch (InterruptedException | IOException e) { log.error("Could not close Cambria publisher, some messages might have been dropped", e); @@ -113,11 +109,11 @@ class DMaaPPublishersCache { @Override public CambriaBatchingPublisher load(@Nonnull String domain) { return dMaaPConfiguration.get() - .get(domain) - .toTry(() -> new RuntimeException( - f("DMaaP configuration contains no configuration for domain: '%s'", domain))) - .flatMap(DMaaPPublishersBuilder::buildPublisher) - .get(); + .get(domain) + .toTry(() -> new RuntimeException( + f("DMaaP configuration contains no configuration for domain: '%s'", domain))) + .flatMap(DMaaPPublishersBuilder::buildPublisher) + .get(); } } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java index 4a056778..f1cbb8e5 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/PublisherConfig.java @@ -21,6 +21,7 @@ package org.onap.dcae.commonFunction.event.publishing; import io.vavr.collection.List; import io.vavr.control.Option; + import java.util.Objects; /** @@ -76,9 +77,9 @@ public final class PublisherConfig { } PublisherConfig that = (PublisherConfig) o; return Objects.equals(destinations, that.destinations) && - Objects.equals(topic, that.topic) && - Objects.equals(userName, that.userName) && - Objects.equals(password, that.password); + Objects.equals(topic, that.topic) && + Objects.equals(userName, that.userName) && + Objects.equals(password, that.password); } @Override @@ -89,10 +90,10 @@ public final class PublisherConfig { @Override public String toString() { return "PublisherConfig{" + - "destinations=" + destinations + - ", topic='" + topic + '\'' + - ", userName='" + userName + '\'' + - ", password='" + password + '\'' + - '}'; + "destinations=" + destinations + + ", topic='" + topic + '\'' + + ", userName='" + userName + '\'' + + ", password='" + password + '\'' + + '}'; } } diff --git a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java index 9bf3ef8c..78f34ff4 100644 --- a/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java +++ b/src/main/java/org/onap/dcae/commonFunction/event/publishing/VavrUtils.java @@ -19,11 +19,11 @@ */ package org.onap.dcae.commonFunction.event.publishing; -import static io.vavr.API.$; - import io.vavr.API; import io.vavr.API.Match.Case; +import static io.vavr.API.$; + /** * @author Pawel Szalapski (pawel.szalapski@nokia.com) */ |