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