dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/sparkcompute/tca/SimpleTCAPlugin.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.plugins.sparkcompute.tca;
  22
  23 import co.cask.cdap.api.annotation.Description;
  24 import co.cask.cdap.api.annotation.Name;
  25 import co.cask.cdap.api.annotation.Plugin;
  26 import co.cask.cdap.api.data.format.StructuredRecord;
  27 import co.cask.cdap.api.data.format.StructuredRecord.Builder;
  28 import co.cask.cdap.api.data.schema.Schema;
  29 import co.cask.cdap.etl.api.PipelineConfigurer;
  30 import co.cask.cdap.etl.api.StageMetrics;
  31 import co.cask.cdap.etl.api.batch.SparkCompute;
  32 import co.cask.cdap.etl.api.batch.SparkExecutionPluginContext;
  33 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  34 import org.apache.spark.api.java.JavaRDD;
  35 import org.apache.spark.api.java.function.Function;
  36 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
  37 import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCACalculatorMessageType;
  38 import org.openecomp.dcae.apod.analytics.cdap.common.utils.ValidationUtils;
  39 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.tca.SimpleTCAPluginConfig;
  40 import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;
  41 import org.openecomp.dcae.apod.analytics.cdap.plugins.validator.SimpleTCAPluginConfigValidator;
  42 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerFunctionalRole;
  43 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
  44 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
  45 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFJsonProcessor;
  46 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFProcessorContext;
  47 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 /**
  52  * @author Rajiv Singla . Creation Date: 2/13/2017.
  53  */
  54
  55 @Plugin(type = SparkCompute.PLUGIN_TYPE)
  56 @Name("SimpleTCAPlugin")
  57 @Description("Used to create TCA (Threshold Crossing Alert) based on given Policy")
  58 @SuppressFBWarnings("SE_INNER_CLASS")
  59 public class SimpleTCAPlugin extends SparkCompute<StructuredRecord, StructuredRecord> {
  60
  61     private static final Logger LOG = LoggerFactory.getLogger(SimpleTCAPlugin.class);
  62     private static final long serialVersionUID = 1L;
  63
  64     private final SimpleTCAPluginConfig pluginConfig;
  65
  66     /**
  67      * Create an instance of Simple TCA Plugin with give Simple TCA Plugin Config
  68      *
  69      * @param pluginConfig Simple TCA Plugin Config
  70      */
  71     public SimpleTCAPlugin(SimpleTCAPluginConfig pluginConfig) {
  72         this.pluginConfig = pluginConfig;
  73         LOG.info("Creating instance of Simple TCA Plugin with plugin config: {}", pluginConfig);
  74     }
  75
  76     @Override
  77     public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  78         super.configurePipeline(pipelineConfigurer);
  79         ValidationUtils.validateSettings(pluginConfig, new SimpleTCAPluginConfigValidator());
  80         final Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema();
  81         CDAPPluginUtils.validateSchemaContainsFields(inputSchema, pluginConfig.getVesMessageFieldName());
  82         CDAPPluginUtils.setOutputSchema(pipelineConfigurer, pluginConfig.getSchema());
  83     }
  84
  85     @Override
  86     public JavaRDD<StructuredRecord> transform(final SparkExecutionPluginContext context,
  87                                                final JavaRDD<StructuredRecord> input) throws Exception {
  88         final StageMetrics metrics = context.getMetrics();
  89
  90         LOG.debug("Invoking Spark Transform for Simple TCA Plugin");
  91         return input.map(new Function<StructuredRecord, StructuredRecord>() {
  92
  93             @Override
  94             public StructuredRecord call(StructuredRecord inputStructuredRecord) throws Exception {
  95                 TCACalculatorMessageType calculatorMessageType;
  96                 String alertMessage = null;
  97
  98                 // Get input structured record
  99                 final String cefMessage = inputStructuredRecord.get(pluginConfig.getVesMessageFieldName());
 100
 101                 // Get TCA Policy
 102                 final TCAPolicy tcaPolicy = CDAPPluginUtils.readValue(pluginConfig.getPolicyJson(), TCAPolicy.class);
 103
 104                 // create initial processor context
 105                 final TCACEFProcessorContext initialProcessorContext =
 106                         new TCACEFProcessorContext(cefMessage, tcaPolicy);
 107
 108                 final TCACEFJsonProcessor jsonProcessor = new TCACEFJsonProcessor();
 109                 final TCACEFProcessorContext jsonProcessorContext =
 110                         jsonProcessor.processMessage(initialProcessorContext);
 111
 112                 if (jsonProcessorContext.getCEFEventListener() != null) {
 113
 114                     LOG.debug("Json to CEF parsing successful. Parsed object {}",
 115                             jsonProcessorContext.getCEFEventListener());
 116
 117                     // compute violations
 118                     final TCACEFProcessorContext processorContextWithViolations =
 119                             TCAUtils.computeThresholdViolations(jsonProcessorContext);
 120
 121                     // if violation are found then create alert message
 122                     if (processorContextWithViolations.canProcessingContinue()) {
 123
 124                         alertMessage = TCAUtils.createTCAAlertString(processorContextWithViolations,
 125                                 pluginConfig.getReferenceName(), pluginConfig.getEnableAlertCEFFormat());
 126                         calculatorMessageType = TCACalculatorMessageType.NON_COMPLIANT;
 127
 128                         LOG.debug("VES Threshold Violation Detected.An alert message is be generated: {}",
 129                                 alertMessage);
 130
 131                         final MetricsPerFunctionalRole metricsPerFunctionalRole =
 132                                 processorContextWithViolations.getMetricsPerFunctionalRole();
 133                         if (metricsPerFunctionalRole != null
 134                                 && metricsPerFunctionalRole.getThresholds() != null
 135                                 && metricsPerFunctionalRole.getThresholds().get(0) != null) {
 136                             final Threshold violatedThreshold = metricsPerFunctionalRole.getThresholds().get(0);
 137                             LOG.debug("CEF Message: {}, Violated Threshold: {}", cefMessage, violatedThreshold);
 138                         }
 139
 140                         metrics.count(CDAPMetricsConstants.TCA_VES_NON_COMPLIANT_MESSAGES_METRIC, 1);
 141
 142                     } else {
 143                         LOG.debug("No Threshold Violation Detected. No alert will be generated.");
 144                         calculatorMessageType = TCACalculatorMessageType.COMPLIANT;
 145                         metrics.count(CDAPMetricsConstants.TCA_VES_COMPLIANT_MESSAGES_METRIC, 1);
 146                     }
 147
 148                 } else {
 149                     LOG.info("Unable to parse provided json message to CEF format. Invalid message: {}", cefMessage);
 150                     calculatorMessageType = TCACalculatorMessageType.INAPPLICABLE;
 151                 }
 152
 153                 LOG.debug("Calculator message type: {} for message: {}", calculatorMessageType, cefMessage);
 154
 155                 final Schema outputSchema = Schema.parseJson(pluginConfig.getSchema());
 156
 157                 // create new output record builder and copy any input record values to output record builder
 158                 final Builder outputRecordBuilder =
 159                         CDAPPluginUtils.createOutputStructuredRecordBuilder(outputSchema, inputStructuredRecord);
 160
 161                 // add alert field
 162                 final Builder outputRecordBuilderWithAlertField =
 163                         CDAPPluginUtils.addFieldValueToStructuredRecordBuilder(outputRecordBuilder,
 164                                 outputSchema, pluginConfig.getAlertFieldName(), alertMessage);
 165
 166                 // add message field type
 167                 final Builder outRecordBuilderWithMessageTypeField =
 168                         CDAPPluginUtils.addFieldValueToStructuredRecordBuilder(outputRecordBuilderWithAlertField,
 169                                 outputSchema, pluginConfig.getMessageTypeFieldName(), calculatorMessageType.toString());
 170
 171                 return outRecordBuilderWithMessageTypeField.build();
 172             }
 173         });
 174     }
 175 }