dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESThresholdViolationCalculatorFlowlet.java

   1 /*\r
   2  * ===============================LICENSE_START======================================\r
   3  *  dcae-analytics\r
   4  * ================================================================================\r
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * ================================================================================\r
   7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   8  *  you may not use this file except in compliance with the License.\r
   9  *   You may obtain a copy of the License at\r
  10  *\r
  11  *          http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  *  Unless required by applicable law or agreed to in writing, software\r
  14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  *  See the License for the specific language governing permissions and\r
  17  *  limitations under the License.\r
  18  *  ============================LICENSE_END===========================================\r
  19  */\r
  20 \r
  21 package org.openecomp.dcae.apod.analytics.cdap.tca.flowlet;\r
  22 \r
  23 import co.cask.cdap.api.annotation.HashPartition;\r
  24 import co.cask.cdap.api.annotation.Output;\r
  25 import co.cask.cdap.api.annotation.ProcessInput;\r
  26 import co.cask.cdap.api.annotation.Property;\r
  27 import co.cask.cdap.api.dataset.lib.ObjectMappedTable;\r
  28 import co.cask.cdap.api.flow.flowlet.AbstractFlowlet;\r
  29 import co.cask.cdap.api.flow.flowlet.FlowletContext;\r
  30 import co.cask.cdap.api.flow.flowlet.OutputEmitter;\r
  31 import co.cask.cdap.api.metrics.Metrics;\r
  32 import com.fasterxml.jackson.core.JsonProcessingException;\r
  33 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;\r
  34 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;\r
  35 import org.openecomp.dcae.apod.analytics.cdap.common.domain.tca.ThresholdCalculatorOutput;\r
  36 import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCACalculatorMessageType;\r
  37 import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAMessageStatusEntity;\r
  38 import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;\r
  39 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\r
  40 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;\r
  41 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;\r
  42 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFProcessorContext;\r
  43 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;\r
  44 import org.slf4j.Logger;\r
  45 import org.slf4j.LoggerFactory;\r
  46 \r
  47 import static org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAMessageStatusPersister.persist;\r
  48 \r
  49 /**\r
  50  * TCA VES Message Filter filters out messages which are not applicable for TCA as per TCA Policy\r
  51  *\r
  52  * @author Rajiv Singla . Creation Date: 11/3/2016.\r
  53  */\r
  54 public class TCAVESThresholdViolationCalculatorFlowlet extends AbstractFlowlet {\r
  55 \r
  56     private static final Logger LOG = LoggerFactory.getLogger(TCAVESThresholdViolationCalculatorFlowlet.class);\r
  57 \r
  58     @Output(CDAPComponentsConstants.TCA_FIXED_VES_TCA_CALCULATOR_NAME_OUTPUT)\r
  59     protected OutputEmitter<ThresholdCalculatorOutput> tcaAlertOutputEmitter;\r
  60     protected Metrics metrics;\r
  61     private ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable;\r
  62 \r
  63     @Property\r
  64     private final String messageStatusTableName;\r
  65     private Boolean enableAlertCEFFormat;\r
  66 \r
  67     private TCAPolicy tcaPolicy;\r
  68 \r
  69     /**\r
  70      * Creates an instance of TCA VES Threshold violation calculator flowlet with give message status table name\r
  71      *\r
  72      * @param messageStatusTableName message status table name\r
  73      */\r
  74     public TCAVESThresholdViolationCalculatorFlowlet(String messageStatusTableName) {\r
  75         this.messageStatusTableName = messageStatusTableName;\r
  76     }\r
  77 \r
  78     @Override\r
  79     public void configure() {\r
  80         setName(CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_NAME_FLOWLET);\r
  81         setDescription(CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_DESCRIPTION_FLOWLET);\r
  82     }\r
  83 \r
  84 \r
  85     @Override\r
  86     public void initialize(FlowletContext flowletContext) throws Exception {\r
  87         super.initialize(flowletContext);\r
  88 \r
  89         // parse Runtime Arguments to tca policy preferences\r
  90         tcaPolicy = CDAPTCAUtils.getValidatedTCAPolicyPreferences(flowletContext);\r
  91         // Parse runtime arguments\r
  92         final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(flowletContext);\r
  93         enableAlertCEFFormat = tcaAppPreferences.getEnableAlertCEFFormat();\r
  94         vesMessageStatusTable = getContext().getDataset(messageStatusTableName);\r
  95 \r
  96     }\r
  97 \r
  98     /**\r
  99      * Filters VES Messages that violates TCA Policy\r
 100      *\r
 101      * @param vesMessage VES Message\r
 102      * @throws JsonProcessingException if alert message cannot be parsed into JSON object\r
 103      */\r
 104     @ProcessInput(CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_ROUTER_OUTPUT)\r
 105     @HashPartition(AnalyticsConstants.TCA_VES_MESSAGE_ROUTER_PARTITION_KEY)\r
 106     public void filterVESMessages(String vesMessage) throws JsonProcessingException {\r
 107 \r
 108         TCACalculatorMessageType calculatorMessageType = TCACalculatorMessageType.INAPPLICABLE;\r
 109         String alertMessage = null;\r
 110 \r
 111         // Step 1: Filter incoming messages\r
 112         final TCACEFProcessorContext processorContext = TCAUtils.filterCEFMessage(vesMessage, tcaPolicy);\r
 113 \r
 114         if (processorContext.canProcessingContinue()) {\r
 115 \r
 116             // Step 2: Check if CEF Message violate any thresholds\r
 117             final TCACEFProcessorContext processorContextWithViolations =\r
 118                     TCAUtils.computeThresholdViolations(processorContext);\r
 119 \r
 120             if (processorContextWithViolations.canProcessingContinue()) {\r
 121 \r
 122                 // Step 3: Create Alert Message\r
 123                 final String tcaAppName = getContext().getApplicationSpecification().getName();\r
 124                 alertMessage =\r
 125                         TCAUtils.createTCAAlertString(processorContextWithViolations, tcaAppName, enableAlertCEFFormat);\r
 126                 calculatorMessageType = TCACalculatorMessageType.NON_COMPLIANT;\r
 127                 LOG.debug("VES Threshold Violation Detected. An alert message is be generated. {}", alertMessage);\r
 128 \r
 129                 metrics.count(CDAPMetricsConstants.TCA_VES_NON_COMPLIANT_MESSAGES_METRIC, 1);\r
 130 \r
 131                 // Step 4: Emit message to Alert Sink Flowlet\r
 132                 final ThresholdCalculatorOutput thresholdCalculatorOutput =\r
 133                         new ThresholdCalculatorOutput(processorContext.getMessage(),\r
 134                                 TCAUtils.writeValueAsString(processorContext.getTCAPolicy()),\r
 135                                 TCAUtils.writeValueAsString(processorContextWithViolations.getMetricsPerEventName()),\r
 136                                 alertMessage);\r
 137                 tcaAlertOutputEmitter.emit(thresholdCalculatorOutput);\r
 138 \r
 139             } else {\r
 140 \r
 141                 calculatorMessageType = TCACalculatorMessageType.COMPLIANT;\r
 142                 metrics.count(CDAPMetricsConstants.TCA_VES_COMPLIANT_MESSAGES_METRIC, 1);\r
 143             }\r
 144 \r
 145         } else {\r
 146 \r
 147             metrics.count(CDAPMetricsConstants.TCA_VES_INAPPLICABLE_MESSAGES_METRIC, 1);\r
 148         }\r
 149 \r
 150         // save message to message status table\r
 151         final int instanceId = getContext().getInstanceId();\r
 152         persist(processorContext, instanceId, calculatorMessageType, vesMessageStatusTable, alertMessage);\r
 153     }\r
 154 \r
 155 \r
 156 }\r