dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/flowlet/TCAVESThresholdViolationCalculatorFlowlet.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.cdap.tca.flowlet;
  22
  23 import co.cask.cdap.api.annotation.HashPartition;
  24 import co.cask.cdap.api.annotation.Output;
  25 import co.cask.cdap.api.annotation.ProcessInput;
  26 import co.cask.cdap.api.annotation.Property;
  27 import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
  28 import co.cask.cdap.api.flow.flowlet.AbstractFlowlet;
  29 import co.cask.cdap.api.flow.flowlet.FlowletContext;
  30 import co.cask.cdap.api.flow.flowlet.OutputEmitter;
  31 import co.cask.cdap.api.metrics.Metrics;
  32 import com.fasterxml.jackson.core.JsonProcessingException;
  33 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
  34 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
  35 import org.openecomp.dcae.apod.analytics.cdap.common.domain.tca.ThresholdCalculatorOutput;
  36 import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCACalculatorMessageType;
  37 import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAMessageStatusEntity;
  38 import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;
  39 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;
  40 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
  41 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
  42 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFProcessorContext;
  43 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46
  47 import static org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAMessageStatusPersister.persist;
  48
  49 /**
  50  * TCA VES Message Filter filters out messages which are not applicable for TCA as per TCA Policy
  51  *
  52  * @author Rajiv Singla . Creation Date: 11/3/2016.
  53  */
  54 public class TCAVESThresholdViolationCalculatorFlowlet extends AbstractFlowlet {
  55
  56     private static final Logger LOG = LoggerFactory.getLogger(TCAVESThresholdViolationCalculatorFlowlet.class);
  57
  58     @Output(CDAPComponentsConstants.TCA_FIXED_VES_TCA_CALCULATOR_NAME_OUTPUT)
  59     protected OutputEmitter<ThresholdCalculatorOutput> tcaAlertOutputEmitter;
  60     protected Metrics metrics;
  61     private ObjectMappedTable<TCAMessageStatusEntity> vesMessageStatusTable;
  62
  63     @Property
  64     private final String messageStatusTableName;
  65     private Boolean enableAlertCEFFormat;
  66
  67     private TCAPolicy tcaPolicy;
  68
  69     /**
  70      * Creates an instance of TCA VES Threshold violation calculator flowlet with give message status table name
  71      *
  72      * @param messageStatusTableName message status table name
  73      */
  74     public TCAVESThresholdViolationCalculatorFlowlet(String messageStatusTableName) {
  75         this.messageStatusTableName = messageStatusTableName;
  76     }
  77
  78     @Override
  79     public void configure() {
  80         setName(CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_NAME_FLOWLET);
  81         setDescription(CDAPComponentsConstants.TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_DESCRIPTION_FLOWLET);
  82     }
  83
  84
  85     @Override
  86     public void initialize(FlowletContext flowletContext) throws Exception {
  87         super.initialize(flowletContext);
  88
  89         // parse Runtime Arguments to tca policy preferences
  90         tcaPolicy = CDAPTCAUtils.getValidatedTCAPolicyPreferences(flowletContext);
  91         // Parse runtime arguments
  92         final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(flowletContext);
  93         enableAlertCEFFormat = tcaAppPreferences.getEnableAlertCEFFormat();
  94         vesMessageStatusTable = getContext().getDataset(messageStatusTableName);
  95
  96     }
  97
  98     /**
  99      * Filters VES Messages that violates TCA Policy
 100      *
 101      * @param vesMessage VES Message
 102      * @throws JsonProcessingException if alert message cannot be parsed into JSON object
 103      */
 104     @ProcessInput(CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_ROUTER_OUTPUT)
 105     @HashPartition(AnalyticsConstants.TCA_VES_MESSAGE_ROUTER_PARTITION_KEY)
 106     public void filterVESMessages(String vesMessage) throws JsonProcessingException {
 107
 108         TCACalculatorMessageType calculatorMessageType = TCACalculatorMessageType.INAPPLICABLE;
 109         String alertMessage = null;
 110
 111         // Step 1: Filter incoming messages
 112         final TCACEFProcessorContext processorContext = TCAUtils.filterCEFMessage(vesMessage, tcaPolicy);
 113
 114         if (processorContext.canProcessingContinue()) {
 115
 116             // Step 2: Check if CEF Message violate any thresholds
 117             final TCACEFProcessorContext processorContextWithViolations =
 118                     TCAUtils.computeThresholdViolations(processorContext);
 119
 120             if (processorContextWithViolations.canProcessingContinue()) {
 121
 122                 // Step 3: Create Alert Message
 123                 final String tcaAppName = getContext().getApplicationSpecification().getName();
 124                 alertMessage =
 125                         TCAUtils.createTCAAlertString(processorContextWithViolations, tcaAppName, enableAlertCEFFormat);
 126                 calculatorMessageType = TCACalculatorMessageType.NON_COMPLIANT;
 127                 LOG.debug("VES Threshold Violation Detected. An alert message is be generated. {}", alertMessage);
 128
 129                 metrics.count(CDAPMetricsConstants.TCA_VES_NON_COMPLIANT_MESSAGES_METRIC, 1);
 130
 131                 // Step 4: Emit message to Alert Sink Flowlet
 132                 final ThresholdCalculatorOutput thresholdCalculatorOutput =
 133                         new ThresholdCalculatorOutput(processorContext.getMessage(),
 134                                 TCAUtils.writeValueAsString(processorContext.getTCAPolicy()),
 135                                 TCAUtils.writeValueAsString(processorContextWithViolations.getMetricsPerEventName()),
 136                                 alertMessage);
 137                 tcaAlertOutputEmitter.emit(thresholdCalculatorOutput);
 138
 139             } else {
 140
 141                 calculatorMessageType = TCACalculatorMessageType.COMPLIANT;
 142                 metrics.count(CDAPMetricsConstants.TCA_VES_COMPLIANT_MESSAGES_METRIC, 1);
 143             }
 144
 145         } else {
 146
 147             metrics.count(CDAPMetricsConstants.TCA_VES_INAPPLICABLE_MESSAGES_METRIC, 1);
 148         }
 149
 150         // save message to message status table
 151         final int instanceId = getContext().getInstanceId();
 152         persist(processorContext, instanceId, calculatorMessageType, vesMessageStatusTable, alertMessage);
 153     }
 154
 155
 156 }