diff options
Diffstat (limited to 'dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor')
6 files changed, 575 insertions, 0 deletions
diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/AbstractTCAECEFPolicyProcessor.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/AbstractTCAECEFPolicyProcessor.java new file mode 100644 index 0000000..b2da5a4 --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/AbstractTCAECEFPolicyProcessor.java @@ -0,0 +1,61 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import org.onap.dcae.apod.analytics.common.exception.MessageProcessingException; +import org.onap.dcae.apod.analytics.common.service.processor.AbstractMessageProcessor; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; + +/** + * <p> + * Encapsulates common functionality for all TCA CEF Policy Processors + * </p> + * + * @author Rajiv Singla . Creation Date: 11/9/2016. + */ +public abstract class AbstractTCAECEFPolicyProcessor extends AbstractMessageProcessor<TCACEFProcessorContext> { + + private static final Logger LOG = LoggerFactory.getLogger(AbstractTCAECEFPolicyProcessor.class); + + /** + * For all TCA Policy Processor the pre processor ensures that {@link EventListener} object is + * present + * + * @param processorContext incoming Processor Context + * @return Pre processed Processor Context + */ + @Override + public TCACEFProcessorContext preProcessor(@Nonnull TCACEFProcessorContext processorContext) { + // validates CEF Event Listener is Present + final EventListener cefEventListener = processorContext.getCEFEventListener(); + if (cefEventListener == null) { + final String errorMessage = String.format( + "CEF Event Listener is not Present.Invalid use of Processor: %s. CEF Message: %s", + getProcessorInfo().getProcessorName(), processorContext.getMessage()); + throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage)); + } + return super.preProcessor(processorContext); + } +} diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFJsonProcessor.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFJsonProcessor.java new file mode 100644 index 0000000..fdeeb3d --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFJsonProcessor.java @@ -0,0 +1,98 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import org.apache.commons.lang3.StringUtils; +import org.onap.dcae.apod.analytics.common.exception.MessageProcessingException; +import org.onap.dcae.apod.analytics.common.service.processor.AbstractMessageProcessor; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.onap.dcae.apod.analytics.tca.utils.TCAUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** + *<p> + * Processor that converts incoming presumed JSON string CEF message to {@link EventListener} object + * <br> + * Pre Conditions: None + *</p> + * + * @author Rajiv Singla . Creation Date: 11/5/2016. + */ +public class TCACEFJsonProcessor extends AbstractMessageProcessor<TCACEFProcessorContext> { + + + private static final long serialVersionUID = 1L; + + private static final Logger LOG = LoggerFactory.getLogger(TCACEFJsonProcessor.class); + + + @Override + public String getProcessorDescription() { + return "Converts incoming TCA CEF Message to Event Listener object"; + } + + @Override + public TCACEFProcessorContext processMessage(TCACEFProcessorContext processorContext) { + + final String cefMessage = processorContext.getMessage(); + + // If CEF Message is null then processor should stop processing + if (cefMessage == null) { + String errorMessage = "Null CEF message cannot be converted to CEF Event Listener Object"; + throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage)); + } + + // If CEF Message is blank then processor stop processing + if (StringUtils.isBlank(cefMessage)) { + setTerminatingProcessingMessage("Blank CEF message cannot be converted to CEF Event Listener Object", + processorContext); + return processorContext; + } + + // trim cef message + final String trimmedCEFMessage = cefMessage.trim(); + + // if message does not start with curly brace and ends with curly brace, it is not a valid cef message + // processor will stop processing + if (!(trimmedCEFMessage.startsWith("{") && trimmedCEFMessage.endsWith("}"))) { + setTerminatingProcessingMessage("CEF Message must start with curly brace and must end with curly brace", + processorContext); + return processorContext; + } + + // try parsing the cef message + try { + final EventListener eventListener = TCAUtils.readValue(trimmedCEFMessage, EventListener.class); + setFinishedProcessingMessage("CEF JSON to Event Listener Conversion Successful", processorContext); + // set new Event Listener in the Processor Context + processorContext.setCEFEventListener(eventListener); + return processorContext; + } catch (IOException e) { + final String errorMessage = String.format("Parsing Failed for CEF Message: %s, Error: %s", cefMessage, e); + // If parsing fails throw an exception + throw new MessageProcessingException(errorMessage, LOG, e); + } + + } +} diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyDomainFilter.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyDomainFilter.java new file mode 100644 index 0000000..e5db327 --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyDomainFilter.java @@ -0,0 +1,84 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import org.onap.dcae.apod.analytics.model.domain.cef.Domain; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; + +/** + * <p> + * TCA Processor which acts like a filter to filter out messages which does not belong to TCA Policy Domain + * <br> + * Pre Conditions: CEF Event Listener must be present + * </p> + * + * @author Rajiv Singla . Creation Date: 11/7/2016. + */ +public class TCACEFPolicyDomainFilter extends AbstractTCAECEFPolicyProcessor { + + + private static final long serialVersionUID = 1L; + + @Override + public String getProcessorDescription() { + return "Filters out CEF Messages which does not match TCAPolicy Domain"; + } + + @Override + public TCACEFProcessorContext processMessage(TCACEFProcessorContext processorContext) { + + // Safe to get event Listener here without null check as pre processor will validate if + // event listener is indeed present + final EventListener eventListener = processorContext.getCEFEventListener(); + + Domain cefMessageDomain; + + // Extract CEF domain as it is must be present as per CEF Schema + if (eventListener.getEvent() != null && + eventListener.getEvent().getCommonEventHeader() != null && + eventListener.getEvent().getCommonEventHeader().getDomain() != null) { + cefMessageDomain = eventListener.getEvent().getCommonEventHeader().getDomain(); + + } else { + final String terminatingMessage = "Invalid CEF Message.Common Event Header Domain not present."; + setTerminatingProcessingMessage(terminatingMessage, processorContext); + return processorContext; + } + + // Get Policy Domain. TCA Policy Validation must ensure that Domain is indeed present + // no null check will be required here + final String policyDomain = processorContext.getTCAPolicy().getDomain(); + + // If Policy domain matches CEF message domain then continue processing + if (cefMessageDomain.toString().equalsIgnoreCase(policyDomain)) { + final String finishMessage = String.format("Policy Domain and CEF Message Domain match successful." + + " Message Domain: %s, Policy Domain: %s", cefMessageDomain, policyDomain); + setFinishedProcessingMessage(finishMessage, processorContext); + } else { + // If policy domain does not match with CEF message terminate processing chain + final String terminatingMessage = String.format("Policy Domain and CEF Message Domain match unsuccessful." + + " Message Domain: %s, Policy Domain: %s", cefMessageDomain, policyDomain); + setTerminatingProcessingMessage(terminatingMessage, processorContext); + } + + return processorContext; + } +} diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyEventNameFilter.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyEventNameFilter.java new file mode 100644 index 0000000..73cbe78 --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyEventNameFilter.java @@ -0,0 +1,91 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import com.google.common.base.Joiner; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy; + +import java.util.List; + +import static org.onap.dcae.apod.analytics.tca.utils.TCAUtils.getPolicyEventNamesSupplier; + +/** + * <p> + * TCA Processor that acts like a filter to filter out messages which does not belong to TCA Policy Event Name + * <br> + * Pre Conditions: CEF Event Listener must be present + * </p> + * + * @author Rajiv Singla . Creation Date: 11/9/2016. + */ +public class TCACEFPolicyEventNameFilter extends AbstractTCAECEFPolicyProcessor { + + private static final long serialVersionUID = 1L; + + @Override + public String getProcessorDescription() { + return "Filters out CEF Messages which does not match Policy Functional Roles"; + } + + @Override + public TCACEFProcessorContext processMessage(TCACEFProcessorContext processorContext) { + + // Safe to get event Listener here without null check as pre processor will validate if + // event listener is indeed present + final EventListener eventListener = processorContext.getCEFEventListener(); + + String cefMessageEventName; + + if (eventListener.getEvent() != null && + eventListener.getEvent().getCommonEventHeader() != null && + eventListener.getEvent().getCommonEventHeader().getEventName() != null) { + cefMessageEventName = eventListener.getEvent().getCommonEventHeader().getEventName(); + } else { + String terminationMessage = "Invalid CEF Message.Common Event Header Event Name not present."; + setTerminatingProcessingMessage(terminationMessage, processorContext); + return processorContext; + } + + // Determine Policy Functional Roles + final TCAPolicy tcaPolicy = processorContext.getTCAPolicy(); + final List<String> policyEventNames = getPolicyEventNamesSupplier(tcaPolicy).get(); + final String policyEventNamesString = Joiner.on(",").join(policyEventNames); + + // If Policy event names contains CEF message event names then continue processing + if (policyEventNames.contains(cefMessageEventName)) { + final String finishMessage = String.format( + "Policy Event Name and CEF Message Event Name match successful." + + "Message EventName: %s, Policy Event Names: %s", + cefMessageEventName, policyEventNamesString); + setFinishedProcessingMessage(finishMessage, processorContext); + } else { + // If Policy event names does not contain CEF message event names then terminate processing + final String terminatingMessage = String.format( + "Policy Event name and CEF Message Event name match unsuccessful." + + "Message EventName: %s, Policy Event Names: %s", + cefMessageEventName, policyEventNamesString); + setTerminatingProcessingMessage(terminatingMessage, processorContext); + } + + return processorContext; + } +} diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyThresholdsProcessor.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyThresholdsProcessor.java new file mode 100644 index 0000000..c7d780d --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFPolicyThresholdsProcessor.java @@ -0,0 +1,138 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import com.google.common.base.Optional; +import com.google.common.collect.Table; +import org.onap.dcae.apod.analytics.common.exception.MessageProcessingException; +import org.onap.dcae.apod.analytics.model.domain.cef.Domain; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.Threshold; +import org.onap.dcae.apod.analytics.tca.utils.TCAUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.math.BigDecimal; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import javax.annotation.Nonnull; + +/** + *<p> + * TCA CEF Policy Threshold processor + * <br> + * Pre Conditions: Domain and Functional Role must be present in CEF Event Listener Object + *</p> + * + * @author Rajiv Singla . Creation Date: 11/9/2016. + */ +public class TCACEFPolicyThresholdsProcessor extends AbstractTCAECEFPolicyProcessor { + + private static final long serialVersionUID = 1L; + + private static final Logger LOG = LoggerFactory.getLogger(TCACEFPolicyThresholdsProcessor.class); + + @Override + public TCACEFProcessorContext preProcessor(@Nonnull TCACEFProcessorContext processorContext) { + // validates Domain and Functional Role are present + final EventListener eventListener = processorContext.getCEFEventListener(); + final Domain domain = eventListener.getEvent().getCommonEventHeader().getDomain(); + final String eventName = eventListener.getEvent().getCommonEventHeader().getEventName(); + if (domain == null || eventName == null) { + final String errorMessage = "CEF Event Listener domain or eventName not Present. " + + "Invalid use of this Processor"; + throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage)); + } + return super.preProcessor(processorContext); + } + + @Override + public String getProcessorDescription() { + return "Applies TCA Policy rules to incoming CEF message. If any thresholds are violated attaches max " + + "Severity violated threshold to TCA Processor Context"; + } + + @Override + public TCACEFProcessorContext processMessage(TCACEFProcessorContext processorContext) { + + final String cefMessage = processorContext.getMessage(); + + // Determine domain and eventName + final EventListener eventListener = processorContext.getCEFEventListener(); + final String eventName = eventListener.getEvent().getCommonEventHeader().getEventName(); + + // Get Table containing event Name and Thresholds Field Path + final TCAPolicy tcaPolicy = processorContext.getTCAPolicy(); + final Table<String, String, List<Threshold>> eventNameFieldPathsTable = + TCAUtils.getPolicyEventNameThresholdsTableSupplier(tcaPolicy).get(); + + // Get Policy Field Paths for that event Name + final Map<String, List<Threshold>> policyFieldPathsMap = eventNameFieldPathsTable.row(eventName); + final Set<String> policyFieldPaths = policyFieldPathsMap.keySet(); + + // Get Json Values for Policy Fields + final Map<String, List<BigDecimal>> messageFieldValuesMap = + TCAUtils.getJsonPathValue(cefMessage, policyFieldPaths); + + // Determine all violated thresholds per message field Path + final Map<String, Threshold> violatedThresholdsMap = new HashMap<>(); + for (Map.Entry<String, List<BigDecimal>> messageFieldValuesMapEntry : messageFieldValuesMap.entrySet()) { + final String messageFieldPath = messageFieldValuesMapEntry.getKey(); + final List<Threshold> messageFieldAssociatedPolicyThresholds = policyFieldPathsMap.get(messageFieldPath); + if (messageFieldAssociatedPolicyThresholds != null) { + final Optional<Threshold> thresholdOptional = TCAUtils.thresholdCalculator( + messageFieldValuesMapEntry.getValue(), messageFieldAssociatedPolicyThresholds); + if (thresholdOptional.isPresent()) { + violatedThresholdsMap.put(messageFieldPath, thresholdOptional.get()); + } + } + } + + // No threshold were violated + if (violatedThresholdsMap.isEmpty()) { + + final String terminationMessage = "No Policy Threshold violated by the VES CEF Message."; + setTerminatingProcessingMessage(terminationMessage, processorContext); + + } else { + + // If there are policy violations then determine max priority violation + final Threshold maxSeverityThresholdViolation = + TCAUtils.prioritizeThresholdViolations(violatedThresholdsMap); + final MetricsPerEventName violatedMetrics = TCAUtils.createViolatedMetrics(tcaPolicy, + maxSeverityThresholdViolation, eventName); + // attach policy violation to processor Context + processorContext.setMetricsPerEventName(violatedMetrics); + + final String finishMessage = String.format("Policy Threshold violation detected for threshold: %s", + maxSeverityThresholdViolation); + setFinishedProcessingMessage(finishMessage, processorContext); + + } + + return processorContext; + } +} diff --git a/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFProcessorContext.java b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFProcessorContext.java new file mode 100644 index 0000000..4448b6b --- /dev/null +++ b/dcae-analytics-tca/src/main/java/org/onap/dcae/apod/analytics/tca/processor/TCACEFProcessorContext.java @@ -0,0 +1,103 @@ +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 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.apod.analytics.tca.processor; + +import org.onap.dcae.apod.analytics.common.service.processor.AbstractProcessorContext; +import org.onap.dcae.apod.analytics.model.domain.cef.EventListener; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName; +import org.onap.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy; + +/** + * TCA CEF Policy Processor Context + * + * @author Rajiv Singla . Creation Date: 11/7/2016. + */ +public class TCACEFProcessorContext extends AbstractProcessorContext { + + private static final long serialVersionUID = 1L; + + private final TCAPolicy tcaPolicy; + private EventListener eventListener; + private MetricsPerEventName metricsPerEventName; + + public TCACEFProcessorContext(final String message, boolean canProcessingContinue, final TCAPolicy tcaPolicy) { + super(message, canProcessingContinue); + this.tcaPolicy = tcaPolicy; + // present only if cef incoming message can be parsed successfully to Event Listener Object + this.eventListener = null; + // present only if there are any threshold violations are detected + this.metricsPerEventName = null; + } + + // Auxiliary Constructor which default canProcessingContinue Flag to true + public TCACEFProcessorContext(final String message, final TCAPolicy tcaPolicy) { + this(message, true, tcaPolicy); + } + + /** + * Returns {@link TCAPolicy} Object + * + * @return TCA Policy + */ + public TCAPolicy getTCAPolicy() { + return tcaPolicy; + } + + /** + * Returns Common Event Format {@link EventListener} if present else null + * + * @return CEF Event Listener + */ + public EventListener getCEFEventListener() { + return eventListener; + } + + + /** + * Sets new {@link EventListener} + * + * @param eventListener set new value for CEF event listener + */ + public void setCEFEventListener(final EventListener eventListener) { + this.eventListener = eventListener; + } + + + /** + * Returns TCA Policy {@link MetricsPerEventName} which was has violated Threshold for the CEF Message if + * present else null + * + * @return Violated Threshold + */ + public MetricsPerEventName getMetricsPerEventName() { + return metricsPerEventName; + } + + /** + * Assign new TCA Policy {@link MetricsPerEventName} which was has violated Threshold for the CEF Message + * + * @param metricsPerEventName new value for Metrics Per Functional Role with violated threshold + */ + public void setMetricsPerEventName(MetricsPerEventName metricsPerEventName) { + this.metricsPerEventName = metricsPerEventName; + } + +} |