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