dcae-analytics-tca/src/main/java/org/openecomp/dcae/apod/analytics/tca/processor/TCACEFPolicyThresholdsProcessor.java

   1 /*
   2  * ===============================LICENSE_START======================================
   3  *  dcae-analytics
   4  * ================================================================================
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *   You may obtain a copy of the License at
  10  *
  11  *          http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *  ============================LICENSE_END===========================================
  19  */
  20
  21 package org.openecomp.dcae.apod.analytics.tca.processor;
  22
  23 import com.google.common.base.Optional;
  24 import com.google.common.collect.Table;
  25 import org.openecomp.dcae.apod.analytics.common.exception.MessageProcessingException;
  26 import org.openecomp.dcae.apod.analytics.model.domain.cef.Domain;
  27 import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
  28 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName;
  29 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
  30 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
  31 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 import java.util.HashMap;
  36 import java.util.List;
  37 import java.util.Map;
  38 import java.util.Set;
  39
  40 import javax.annotation.Nonnull;
  41
  42 /**
  43  *<p>
  44  *     TCA CEF Policy Threshold processor
  45  *     <br>
  46  *     Pre Conditions: Domain and Functional Role must be present in CEF Event Listener Object
  47  *</p>
  48  *
  49  * @author Rajiv Singla . Creation Date: 11/9/2016.
  50  */
  51 public class TCACEFPolicyThresholdsProcessor extends AbstractTCAECEFPolicyProcessor {
  52
  53     private static final long serialVersionUID = 1L;
  54
  55     private static final Logger LOG = LoggerFactory.getLogger(TCACEFPolicyThresholdsProcessor.class);
  56
  57     @Override
  58     public TCACEFProcessorContext preProcessor(@Nonnull TCACEFProcessorContext processorContext) {
  59         // validates Domain and Functional Role are present
  60         final EventListener eventListener = processorContext.getCEFEventListener();
  61         final Domain domain = eventListener.getEvent().getCommonEventHeader().getDomain();
  62         final String eventName = eventListener.getEvent().getCommonEventHeader().getEventName();
  63         if (domain == null || eventName == null) {
  64             final String errorMessage = "CEF Event Listener domain or eventName not Present. " +
  65                     "Invalid use of this Processor";
  66             throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
  67         }
  68         return super.preProcessor(processorContext);
  69     }
  70
  71     @Override
  72     public String getProcessorDescription() {
  73         return "Applies TCA Policy rules to incoming CEF message. If any thresholds are violated attaches max " +
  74                 "Severity violated threshold to TCA Processor Context";
  75     }
  76
  77     @Override
  78     public TCACEFProcessorContext processMessage(TCACEFProcessorContext processorContext) {
  79
  80         final String cefMessage = processorContext.getMessage();
  81
  82         // Determine domain and eventName
  83         final EventListener eventListener = processorContext.getCEFEventListener();
  84         final String eventName = eventListener.getEvent().getCommonEventHeader().getEventName();
  85
  86         // Get Table containing event Name and Thresholds Field Path
  87         final TCAPolicy tcaPolicy = processorContext.getTCAPolicy();
  88         final Table<String, String, List<Threshold>> eventNameFieldPathsTable =
  89                 TCAUtils.getPolicyEventNameThresholdsTableSupplier(tcaPolicy).get();
  90
  91         // Get Policy Field Paths for that event Name
  92         final Map<String, List<Threshold>> policyFieldPathsMap = eventNameFieldPathsTable.row(eventName);
  93         final Set<String> policyFieldPaths = policyFieldPathsMap.keySet();
  94
  95         // Get Json Values for Policy Fields
  96         final Map<String, List<Long>> messageFieldValuesMap = TCAUtils.getJsonPathValue(cefMessage, policyFieldPaths);
  97
  98         // Determine all violated thresholds per message field Path
  99         final Map<String, Threshold> violatedThresholdsMap = new HashMap<>();
 100         for (Map.Entry<String, List<Long>> messageFieldValuesMapEntry : messageFieldValuesMap.entrySet()) {
 101             final String messageFieldPath = messageFieldValuesMapEntry.getKey();
 102             final List<Threshold> messageFieldAssociatedPolicyThresholds = policyFieldPathsMap.get(messageFieldPath);
 103             if (messageFieldAssociatedPolicyThresholds != null) {
 104                 final Optional<Threshold> thresholdOptional = TCAUtils.thresholdCalculator(
 105                         messageFieldValuesMapEntry.getValue(), messageFieldAssociatedPolicyThresholds);
 106                 if (thresholdOptional.isPresent()) {
 107                     violatedThresholdsMap.put(messageFieldPath, thresholdOptional.get());
 108                 }
 109             }
 110         }
 111
 112         // No threshold were violated
 113         if (violatedThresholdsMap.isEmpty()) {
 114
 115             final String terminationMessage = "No Policy Threshold violated by the VES CEF Message.";
 116             setTerminatingProcessingMessage(terminationMessage, processorContext);
 117
 118         } else {
 119
 120             // If there are policy violations then determine max priority violation
 121             final Threshold maxSeverityThresholdViolation =
 122                     TCAUtils.prioritizeThresholdViolations(violatedThresholdsMap);
 123             final MetricsPerEventName violatedMetrics = TCAUtils.createViolatedMetrics(tcaPolicy,
 124                     maxSeverityThresholdViolation, eventName);
 125             // attach policy violation to processor Context
 126             processorContext.setMetricsPerEventName(violatedMetrics);
 127
 128             final String finishMessage = String.format("Policy Threshold violation detected for threshold: %s",
 129                     maxSeverityThresholdViolation);
 130             setFinishedProcessingMessage(finishMessage, processorContext);
 131
 132         }
 133
 134         return processorContext;
 135     }
 136 }