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.tca.utils;
\r
23 import com.fasterxml.jackson.core.JsonProcessingException;
\r
24 import com.fasterxml.jackson.databind.JsonNode;
\r
25 import com.google.common.base.Function;
\r
26 import com.google.common.base.Optional;
\r
27 import com.google.common.base.Predicate;
\r
28 import com.google.common.base.Predicates;
\r
29 import com.google.common.base.Supplier;
\r
30 import com.google.common.base.Suppliers;
\r
31 import com.google.common.collect.HashBasedTable;
\r
32 import com.google.common.collect.ImmutableList;
\r
33 import com.google.common.collect.ImmutableMap;
\r
34 import com.google.common.collect.Iterables;
\r
35 import com.google.common.collect.Lists;
\r
36 import com.google.common.collect.Maps;
\r
37 import com.google.common.collect.Table;
\r
38 import com.jayway.jsonpath.DocumentContext;
\r
39 import com.jayway.jsonpath.JsonPath;
\r
40 import com.jayway.jsonpath.TypeRef;
\r
41 import org.apache.commons.lang3.StringUtils;
\r
42 import org.apache.commons.lang3.tuple.ImmutablePair;
\r
43 import org.apache.commons.lang3.tuple.Pair;
\r
44 import org.openecomp.dcae.apod.analytics.aai.service.AAIEnrichmentClient;
\r
45 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
\r
46 import org.openecomp.dcae.apod.analytics.common.exception.MessageProcessingException;
\r
47 import org.openecomp.dcae.apod.analytics.common.service.processor.AbstractMessageProcessor;
\r
48 import org.openecomp.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;
\r
49 import org.openecomp.dcae.apod.analytics.model.domain.cef.AlertAction;
\r
50 import org.openecomp.dcae.apod.analytics.model.domain.cef.AlertType;
\r
51 import org.openecomp.dcae.apod.analytics.model.domain.cef.CommonEventHeader;
\r
52 import org.openecomp.dcae.apod.analytics.model.domain.cef.Criticality;
\r
53 import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
\r
54 import org.openecomp.dcae.apod.analytics.model.domain.cef.EventSeverity;
\r
55 import org.openecomp.dcae.apod.analytics.model.domain.cef.PerformanceCounter;
\r
56 import org.openecomp.dcae.apod.analytics.model.domain.cef.ThresholdCrossingAlertFields;
\r
57 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.ClosedLoopEventStatus;
\r
58 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.ControlLoopSchemaType;
\r
59 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Direction;
\r
60 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerEventName;
\r
61 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
\r
62 import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
\r
63 import org.openecomp.dcae.apod.analytics.model.facade.tca.AAI;
\r
64 import org.openecomp.dcae.apod.analytics.model.facade.tca.TCAVESResponse;
\r
65 import org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelIOUtils;
\r
66 import org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils;
\r
67 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFJsonProcessor;
\r
68 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyDomainFilter;
\r
69 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyEventNameFilter;
\r
70 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyThresholdsProcessor;
\r
71 import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFProcessorContext;
\r
72 import org.quartz.Job;
\r
73 import org.quartz.JobBuilder;
\r
74 import org.quartz.JobDataMap;
\r
75 import org.quartz.JobDetail;
\r
76 import org.quartz.Scheduler;
\r
77 import org.quartz.SchedulerException;
\r
78 import org.quartz.SimpleScheduleBuilder;
\r
79 import org.quartz.SimpleTrigger;
\r
80 import org.quartz.TriggerBuilder;
\r
81 import org.quartz.impl.StdSchedulerFactory;
\r
82 import org.slf4j.Logger;
\r
83 import org.slf4j.LoggerFactory;
\r
85 import java.io.IOException;
\r
86 import java.util.ArrayList;
\r
87 import java.util.Collections;
\r
88 import java.util.Comparator;
\r
89 import java.util.Date;
\r
90 import java.util.HashMap;
\r
91 import java.util.Iterator;
\r
92 import java.util.LinkedHashMap;
\r
93 import java.util.LinkedList;
\r
94 import java.util.List;
\r
95 import java.util.Map;
\r
96 import java.util.Properties;
\r
97 import java.util.Set;
\r
98 import java.util.SortedMap;
\r
99 import java.util.TreeMap;
\r
100 import java.util.UUID;
\r
102 import javax.annotation.Nonnull;
\r
103 import javax.annotation.Nullable;
\r
105 import static com.google.common.collect.Lists.newArrayList;
\r
106 import static org.apache.commons.lang3.time.DateFormatUtils.SMTP_DATETIME_FORMAT;
\r
109 * Utility Helper methods for TCA sub module only. Extends {@link AnalyticsModelJsonUtils} to get
\r
110 * pre configured Json Object Mapper understand serialization and deserialization of CEF Message
\r
113 * @author Rajiv Singla . Creation Date: 10/24/2016.
\r
115 public abstract class TCAUtils extends AnalyticsModelJsonUtils {
\r
117 private static final Logger LOG = LoggerFactory.getLogger(TCAUtils.class);
\r
120 * Threshold Comparator which is used to order thresholds based on their severity e.g. ( CRITICAL, MAJOR, MINOR,
\r
123 private static final Comparator<Threshold> THRESHOLD_COMPARATOR = new Comparator<Threshold>() {
\r
125 public int compare(Threshold threshold1, Threshold threshold2) {
\r
126 return threshold1.getSeverity().compareTo(threshold2.getSeverity());
\r
131 * {@link Function} that extracts {@link TCAPolicy#getMetricsPerEventName()} from {@link TCAPolicy}
\r
133 * @return TCA Policy Metrics Per Event Name list
\r
135 public static Function<TCAPolicy, List<MetricsPerEventName>> tcaPolicyMetricsExtractorFunction() {
\r
136 return new Function<TCAPolicy, List<MetricsPerEventName>>() {
\r
139 public List<MetricsPerEventName> apply(@Nonnull TCAPolicy tcaPolicy) {
\r
140 return tcaPolicy.getMetricsPerEventName();
\r
146 * {@link Function} that extracts {@link MetricsPerEventName#getEventName()} from
\r
147 * {@link MetricsPerEventName}
\r
149 * @return Event Names or a Metrics Per Event Name object
\r
151 public static Function<MetricsPerEventName, String> tcaEventNameExtractorFunction() {
\r
152 return new Function<MetricsPerEventName, String>() {
\r
154 public String apply(@Nonnull MetricsPerEventName metricsPerEventName) {
\r
155 return metricsPerEventName.getEventName();
\r
162 * Extracts {@link TCAPolicy} Event Names
\r
164 * @param tcaPolicy TCA Policy
\r
165 * @return List of event names in the TCA Policy
\r
167 public static List<String> getPolicyEventNames(@Nonnull final TCAPolicy tcaPolicy) {
\r
168 final List<MetricsPerEventName> metricsPerEventNames =
\r
169 tcaPolicyMetricsExtractorFunction().apply(tcaPolicy);
\r
171 return Lists.transform(metricsPerEventNames, tcaEventNameExtractorFunction());
\r
175 * A {@link Supplier} which caches {@link TCAPolicy} Event names as they are not expected to
\r
176 * change during runtime
\r
178 * @param tcaPolicy TCA Policy
\r
179 * @return a Supplier that memoize the TCA Policy event names
\r
181 public static Supplier<List<String>> getPolicyEventNamesSupplier(@Nonnull final TCAPolicy tcaPolicy) {
\r
182 return Suppliers.memoize(new Supplier<List<String>>() {
\r
184 public List<String> get() {
\r
185 return getPolicyEventNames(tcaPolicy);
\r
192 * Creates a Table to lookup thresholds of a {@link TCAPolicy} by its Event Name and Threshold Field path
\r
194 * @param tcaPolicy TCA Policy
\r
195 * @return A table with Keys of event name and field path containing List of threshold as values
\r
197 public static Table<String, String, List<Threshold>> getPolicyEventNameThresholdsTable(final TCAPolicy tcaPolicy) {
\r
198 final Table<String, String, List<Threshold>> domainFRTable = HashBasedTable.create();
\r
199 for (MetricsPerEventName metricsPerEventName : tcaPolicy.getMetricsPerEventName()) {
\r
200 final String eventName = metricsPerEventName.getEventName();
\r
201 final List<Threshold> thresholds = metricsPerEventName.getThresholds();
\r
202 for (Threshold threshold : thresholds) {
\r
203 final List<Threshold> existingThresholds = domainFRTable.get(eventName, threshold.getFieldPath());
\r
204 if (existingThresholds == null) {
\r
205 final LinkedList<Threshold> newThresholdList = new LinkedList<>();
\r
206 newThresholdList.add(threshold);
\r
207 domainFRTable.put(eventName, threshold.getFieldPath(), newThresholdList);
\r
209 domainFRTable.get(eventName, threshold.getFieldPath()).add(threshold);
\r
213 return domainFRTable;
\r
218 * A {@link Supplier} which caches Policy Event Name and Threshold Field Path Thresholds lookup table
\r
220 * @param tcaPolicy TCA Policy
\r
221 * @return Cached Supplier for table with Keys of event Name and field path containing thresholds as values
\r
223 public static Supplier<Table<String, String, List<Threshold>>> getPolicyEventNameThresholdsTableSupplier
\r
224 (final TCAPolicy tcaPolicy) {
\r
225 return Suppliers.memoize(new Supplier<Table<String, String, List<Threshold>>>() {
\r
227 public Table<String, String, List<Threshold>> get() {
\r
228 return getPolicyEventNameThresholdsTable(tcaPolicy);
\r
235 * Creates a {@link GenericMessageChainProcessor} of {@link TCACEFJsonProcessor},
\r
236 * {@link TCACEFPolicyDomainFilter} and {@link TCACEFPolicyEventNameFilter}s to
\r
237 * filter out messages which does not match policy domain or event Name
\r
239 * @param cefMessage CEF Message
\r
240 * @param tcaPolicy TCA Policy
\r
241 * @return Message Process Context after processing filter chain
\r
243 public static TCACEFProcessorContext filterCEFMessage(@Nullable final String cefMessage,
\r
244 @Nonnull final TCAPolicy tcaPolicy) {
\r
246 final TCACEFJsonProcessor jsonProcessor = new TCACEFJsonProcessor();
\r
247 final TCACEFPolicyDomainFilter domainFilter = new TCACEFPolicyDomainFilter();
\r
248 final TCACEFPolicyEventNameFilter eventNameFilter = new TCACEFPolicyEventNameFilter();
\r
249 // Create a list of message processors
\r
250 final ImmutableList<AbstractMessageProcessor<TCACEFProcessorContext>> messageProcessors =
\r
251 ImmutableList.of(jsonProcessor, domainFilter, eventNameFilter);
\r
252 final TCACEFProcessorContext processorContext = new TCACEFProcessorContext(cefMessage, tcaPolicy);
\r
253 // Create a message processors chain
\r
254 final GenericMessageChainProcessor<TCACEFProcessorContext> tcaProcessingChain =
\r
255 new GenericMessageChainProcessor<>(messageProcessors, processorContext);
\r
257 return tcaProcessingChain.processChain();
\r
262 * Extracts json path values for given json Field Paths from using Json path notation. Assumes
\r
263 * that values extracted are always long
\r
265 * @param message CEF Message
\r
266 * @param jsonFieldPaths Json Field Paths
\r
267 * @return Map containing key as json path and values as values associated with that json path
\r
269 public static Map<String, List<Long>> getJsonPathValue(@Nonnull String message, @Nonnull Set<String>
\r
272 final Map<String, List<Long>> jsonFieldPathMap = new HashMap<>();
\r
273 final DocumentContext documentContext = JsonPath.parse(message);
\r
275 for (String jsonFieldPath : jsonFieldPaths) {
\r
276 final List<Long> jsonFieldValues = documentContext.read(jsonFieldPath, new TypeRef<List<Long>>() {
\r
278 // If Json Field Values are not or empty
\r
279 if (jsonFieldValues != null && !jsonFieldValues.isEmpty()) {
\r
280 // Filter out all null values in the filed values list
\r
281 final List<Long> nonNullValues = Lists.newLinkedList(Iterables.filter(jsonFieldValues,
\r
282 Predicates.<Long>notNull()));
\r
283 // If there are non null values put them in the map
\r
284 if (!nonNullValues.isEmpty()) {
\r
285 jsonFieldPathMap.put(jsonFieldPath, nonNullValues);
\r
290 return jsonFieldPathMap;
\r
294 * Computes if any CEF Message Fields have violated any Policy Thresholds. For the same policy field path
\r
295 * it applies threshold in order of their severity and record the first threshold per message field path
\r
297 * @param messageFieldValues Field Path Values extracted from CEF Message
\r
298 * @param fieldThresholds Policy Thresholds for Field Path
\r
299 * @return Optional of violated threshold for a field path
\r
301 public static Optional<Threshold> thresholdCalculator(final List<Long> messageFieldValues, final List<Threshold>
\r
303 // order thresholds by severity
\r
304 Collections.sort(fieldThresholds, THRESHOLD_COMPARATOR);
\r
305 // Now apply each threshold to field values
\r
306 for (Threshold fieldThreshold : fieldThresholds) {
\r
307 for (Long messageFieldValue : messageFieldValues) {
\r
308 final Boolean isThresholdViolated =
\r
309 fieldThreshold.getDirection().operate(messageFieldValue, fieldThreshold.getThresholdValue());
\r
310 if (isThresholdViolated) {
\r
311 final Threshold violatedThreshold = Threshold.copy(fieldThreshold);
\r
312 violatedThreshold.setActualFieldValue(messageFieldValue);
\r
313 return Optional.of(violatedThreshold);
\r
317 return Optional.absent();
\r
321 * Prioritize Threshold to be reported in case there was multiple TCA violations in a single CEF message.
\r
322 * Grabs first highest priority violated threshold
\r
324 * @param violatedThresholdsMap Map containing field Path and associated violated Thresholds
\r
325 * @return First Highest priority violated threshold
\r
327 public static Threshold prioritizeThresholdViolations(final Map<String, Threshold> violatedThresholdsMap) {
\r
329 final List<Threshold> violatedThresholds = newArrayList(violatedThresholdsMap.values());
\r
331 if (violatedThresholds.size() == 1) {
\r
332 return violatedThresholds.get(0);
\r
334 Collections.sort(violatedThresholds, THRESHOLD_COMPARATOR);
\r
335 // Just grab the first violated threshold with highest priority
\r
336 return violatedThresholds.get(0);
\r
341 * Creates {@link MetricsPerEventName} object which contains violated thresholds
\r
343 * @param tcaPolicy TCA Policy
\r
344 * @param violatedThreshold Violated thresholds
\r
345 * @param eventName Event Name
\r
347 * @return MetricsPerEventName object containing one highest severity violated threshold
\r
349 public static MetricsPerEventName createViolatedMetrics(@Nonnull final TCAPolicy tcaPolicy,
\r
350 @Nonnull final Threshold violatedThreshold,
\r
351 @Nonnull final String eventName) {
\r
353 final ArrayList<MetricsPerEventName> metricsPerEventNames = newArrayList(
\r
354 Iterables.filter(tcaPolicy.getMetricsPerEventName(), new Predicate<MetricsPerEventName>() {
\r
356 public boolean apply(@Nonnull MetricsPerEventName metricsPerEventName) {
\r
357 return metricsPerEventName.getEventName().equals(eventName);
\r
360 // TCA policy must have only one metrics per event Name
\r
361 if (metricsPerEventNames.size() == 1) {
\r
362 final MetricsPerEventName violatedMetrics =
\r
363 MetricsPerEventName.copy(metricsPerEventNames.get(0));
\r
364 violatedMetrics.setThresholds(ImmutableList.of(Threshold.copy(violatedThreshold)));
\r
365 return violatedMetrics;
\r
367 final String errorMessage = String.format("TCA Policy must contain eventName: %s", eventName);
\r
368 throw new MessageProcessingException(errorMessage, LOG, new IllegalStateException(errorMessage));
\r
373 * Computes threshold violations
\r
375 * @param processorContext Filtered processor Context
\r
376 * @return processor context with any threshold violations
\r
378 public static TCACEFProcessorContext computeThresholdViolations(final TCACEFProcessorContext processorContext) {
\r
379 final TCACEFPolicyThresholdsProcessor policyThresholdsProcessor = new TCACEFPolicyThresholdsProcessor();
\r
380 return policyThresholdsProcessor.apply(processorContext);
\r
385 * Creates TCA Alert String - Alert String is created in both {@link EventListener} or {@link TCAVESResponse}
\r
388 * @param processorContextWithViolations processor context which has TCA violations
\r
389 * @param tcaAppName tca app name
\r
390 * @param isAlertInCEFFormat determines if output alert is in CEF format
\r
392 * @return TCA Alert String
\r
394 * @throws JsonProcessingException If alert cannot be parsed into JSON String
\r
396 public static String createTCAAlertString(final TCACEFProcessorContext processorContextWithViolations,
\r
397 final String tcaAppName,
\r
398 final Boolean isAlertInCEFFormat) throws JsonProcessingException {
\r
399 if (isAlertInCEFFormat != null && isAlertInCEFFormat) {
\r
400 final EventListener eventListenerWithViolations =
\r
401 addThresholdViolationFields(processorContextWithViolations);
\r
402 final String alertString = writeValueAsString(eventListenerWithViolations);
\r
403 LOG.debug("Created alert in CEF Format: {}", alertString);
\r
404 return alertString;
\r
406 final TCAVESResponse newTCAVESResponse =
\r
407 createNewTCAVESResponse(processorContextWithViolations, tcaAppName);
\r
408 final String alertString = writeValueAsString(newTCAVESResponse);
\r
409 LOG.debug("Created alert in Non CEF Format: {}", alertString);
\r
410 return alertString;
\r
415 * Adds threshold violation fields to {@link EventListener}
\r
417 * @param processorContextWithViolations processor context that contains violations
\r
418 * @return event listener with threshold crossing alert fields populated
\r
420 public static EventListener addThresholdViolationFields(
\r
421 final TCACEFProcessorContext processorContextWithViolations) {
\r
423 final MetricsPerEventName metricsPerEventName =
\r
424 processorContextWithViolations.getMetricsPerEventName();
\r
425 // confirm violations are indeed present
\r
426 if (metricsPerEventName == null) {
\r
427 final String errorMessage = "No violations metrics. Unable to add Threshold Violation Fields";
\r
428 throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
\r
431 // get violated threshold
\r
432 final Threshold violatedThreshold = metricsPerEventName.getThresholds().get(0);
\r
433 final EventListener eventListener = processorContextWithViolations.getCEFEventListener();
\r
434 final CommonEventHeader commonEventHeader = eventListener.getEvent().getCommonEventHeader();
\r
436 // create new threshold crossing alert fields
\r
437 final ThresholdCrossingAlertFields thresholdCrossingAlertFields = new ThresholdCrossingAlertFields();
\r
438 thresholdCrossingAlertFields.setEventStartTimestamp(commonEventHeader.getStartEpochMicrosec().toString());
\r
439 thresholdCrossingAlertFields.setEventSeverity(violatedThreshold.getSeverity());
\r
440 thresholdCrossingAlertFields.setCollectionTimestamp(SMTP_DATETIME_FORMAT.format(new Date()));
\r
441 thresholdCrossingAlertFields.setAlertAction(AlertAction.SET);
\r
442 thresholdCrossingAlertFields.setAlertType(AlertType.INTERFACE_ANOMALY);
\r
443 thresholdCrossingAlertFields.setAlertDescription(violatedThreshold.getDirection().toString());
\r
444 thresholdCrossingAlertFields.setInterfaceName(commonEventHeader.getReportingEntityName());
\r
445 thresholdCrossingAlertFields.setElementType(commonEventHeader.getEventName());
\r
447 // create new performance count
\r
448 final PerformanceCounter performanceCounter = new PerformanceCounter();
\r
449 performanceCounter.setCriticality(convertSeverityToCriticality(violatedThreshold.getSeverity()));
\r
450 performanceCounter.setName(violatedThreshold.getFieldPath());
\r
451 performanceCounter.setValue(violatedThreshold.getActualFieldValue().toString());
\r
452 performanceCounter.setThresholdCrossed(violatedThreshold.getThresholdValue().toString());
\r
454 // set additional parameters for threshold crossing alert fields
\r
455 thresholdCrossingAlertFields.setAdditionalParameters(ImmutableList.of(performanceCounter));
\r
457 // add threshold crossing fields to existing event listener
\r
458 eventListener.getEvent().setThresholdCrossingAlertFields(thresholdCrossingAlertFields);
\r
460 return eventListener;
\r
464 * Converts {@link EventSeverity} to {@link Criticality}
\r
466 * @param eventSeverity event severity
\r
468 * @return performance counter criticality
\r
470 private static Criticality convertSeverityToCriticality(final EventSeverity eventSeverity) {
\r
471 switch (eventSeverity) {
\r
473 return Criticality.CRIT;
\r
475 return Criticality.MAJ;
\r
477 return Criticality.UNKNOWN;
\r
482 * Creates {@link TCAVESResponse} object
\r
484 * @param processorContext processor Context with violations
\r
485 * @param tcaAppName TCA App Name
\r
487 * @return TCA VES Response Message
\r
489 public static TCAVESResponse createNewTCAVESResponse(final TCACEFProcessorContext processorContext,
\r
490 final String tcaAppName) {
\r
492 final MetricsPerEventName metricsPerEventName = processorContext.getMetricsPerEventName();
\r
493 // confirm violations are indeed present
\r
494 if (metricsPerEventName == null) {
\r
495 final String errorMessage = "No violations metrics. Unable to create VES Response";
\r
496 throw new MessageProcessingException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
\r
499 final Threshold violatedThreshold = metricsPerEventName.getThresholds().get(0);
\r
500 final EventListener eventListener = processorContext.getCEFEventListener();
\r
501 final CommonEventHeader commonEventHeader = eventListener.getEvent().getCommonEventHeader();
\r
503 final TCAVESResponse tcavesResponse = new TCAVESResponse();
\r
504 // ClosedLoopControlName included in the DCAE configuration Policy
\r
505 tcavesResponse.setClosedLoopControlName(violatedThreshold.getClosedLoopControlName());
\r
506 // version included in the DCAE configuration Policy
\r
507 tcavesResponse.setVersion(violatedThreshold.getVersion());
\r
508 // Generate a UUID for this output message
\r
509 tcavesResponse.setRequestID(UUID.randomUUID().toString());
\r
510 // commonEventHeader.startEpochMicrosec from the received VES message
\r
511 tcavesResponse.setClosedLoopAlarmStart(commonEventHeader.getStartEpochMicrosec());
\r
512 // commonEventHeader.lastEpochMicrosec from the received VES message for abated alerts
\r
513 if (violatedThreshold.getClosedLoopEventStatus() == ClosedLoopEventStatus.ABATED) {
\r
514 tcavesResponse.setClosedLoopAlarmEnd(commonEventHeader.getLastEpochMicrosec());
\r
516 // Concatenate name of this DCAE instance and name for this TCA instance, separated by dot
\r
517 tcavesResponse.setClosedLoopEventClient("DCAE_INSTANCE_ID." + tcaAppName);
\r
519 final AAI aai = new AAI();
\r
520 tcavesResponse.setAai(aai);
\r
522 // VM specific settings
\r
523 if (metricsPerEventName.getControlLoopSchemaType() == ControlLoopSchemaType.VM) {
\r
524 // Hard Coded - "VM"
\r
525 tcavesResponse.setTargetType(AnalyticsConstants.TCA_VES_RESPONSE_VM_TARGET_TYPE);
\r
526 // Hard Coded - "vserver.vserver-name"
\r
527 tcavesResponse.setTarget(AnalyticsConstants.TCA_VES_RESPONSE_VM_TARGET);
\r
528 // commonEventHeader.sourceName from the received VES message
\r
529 aai.setGenericServerId(commonEventHeader.getSourceName());
\r
531 // VNF specific settings
\r
532 // Hard Coded - "VNF"
\r
533 tcavesResponse.setTargetType(AnalyticsConstants.TCA_VES_RESPONSE_VNF_TARGET_TYPE);
\r
534 // Hard Coded - "generic-vnf.vnf-id"
\r
535 tcavesResponse.setTarget(AnalyticsConstants.TCA_VES_RESPONSE_VNF_TARGET);
\r
536 // commonEventHeader.sourceName from the received VES message
\r
537 aai.setGenericVNFId(commonEventHeader.getSourceName());
\r
540 // Hard Coded - "DCAE"
\r
541 tcavesResponse.setFrom(AnalyticsConstants.TCA_VES_RESPONSE_FROM);
\r
542 // policyScope included in the DCAE configuration Policy
\r
543 tcavesResponse.setPolicyScope(metricsPerEventName.getPolicyScope());
\r
544 // policyName included in the DCAE configuration Policy
\r
545 tcavesResponse.setPolicyName(metricsPerEventName.getPolicyName());
\r
546 // policyVersion included in the DCAE configuration Policy
\r
547 tcavesResponse.setPolicyVersion(metricsPerEventName.getPolicyVersion());
\r
548 // Extracted from violated threshold
\r
549 tcavesResponse.setClosedLoopEventStatus(violatedThreshold.getClosedLoopEventStatus().name());
\r
551 return tcavesResponse;
\r
556 * Determines {@link ControlLoopSchemaType} for given {@link TCAVESResponse} alert
\r
558 * @param tcavesResponse alert
\r
560 * @return control Loop Schema Type
\r
562 public static ControlLoopSchemaType determineControlLoopSchemaType(final TCAVESResponse tcavesResponse) {
\r
563 final AAI aai = tcavesResponse.getAai();
\r
564 if (aai.getGenericServerId() != null) {
\r
565 return ControlLoopSchemaType.VM;
\r
567 return ControlLoopSchemaType.VNF;
\r
572 * Determines {@link ControlLoopSchemaType} for given {@link TCAVESResponse} alert
\r
574 * @param tcavesResponse {@link TCAVESResponse} TCA alert
\r
576 * @return Source name
\r
578 public static String determineSourceName(final TCAVESResponse tcavesResponse) {
\r
579 final AAI aai = tcavesResponse.getAai();
\r
580 if (aai.getGenericServerId() != null) {
\r
581 return aai.getGenericServerId();
\r
583 return aai.getGenericVNFId();
\r
589 * Extract Domain and Event Name from processor context if present
\r
591 * @param processorContext processor context
\r
592 * @return Tuple of domain and event Name
\r
594 public static Pair<String, String> getDomainAndEventName(
\r
595 @Nullable final TCACEFProcessorContext processorContext) {
\r
597 String domain = null;
\r
598 String eventName = null;
\r
600 if (processorContext != null &&
\r
601 processorContext.getCEFEventListener() != null &&
\r
602 processorContext.getCEFEventListener().getEvent() != null &&
\r
603 processorContext.getCEFEventListener().getEvent().getCommonEventHeader() != null) {
\r
604 final CommonEventHeader commonEventHeader = processorContext.getCEFEventListener().getEvent()
\r
605 .getCommonEventHeader();
\r
607 if (commonEventHeader.getDomain() != null) {
\r
608 domain = commonEventHeader.getDomain().name();
\r
611 if (commonEventHeader.getEventName() != null) {
\r
612 eventName = commonEventHeader.getEventName();
\r
617 return new ImmutablePair<>(domain, eventName);
\r
622 * Creates {@link TCAPolicy} Metrics per Event Name list
\r
624 * @param eventNamesMap Map containing event Name as key and corresponding values
\r
626 * @return List of {@link MetricsPerEventName}
\r
628 public static List<MetricsPerEventName> createTCAPolicyMetricsPerEventNameList(
\r
629 final Map<String, Map<String, String>> eventNamesMap) {
\r
631 // create a new metrics per event Name list
\r
632 final List<MetricsPerEventName> metricsPerEventNames = new LinkedList<>();
\r
634 for (Map.Entry<String, Map<String, String>> eventNamesEntry : eventNamesMap.entrySet()) {
\r
636 // create new metrics per event Name instance
\r
637 final MetricsPerEventName newMetricsPerEventName =
\r
638 createNewMetricsPerEventName(eventNamesEntry);
\r
639 metricsPerEventNames.add(newMetricsPerEventName);
\r
641 // determine all threshold related values
\r
642 final Map<String, String> thresholdsValuesMaps =
\r
643 filterMapByKeyNamePrefix(eventNamesEntry.getValue(),
\r
644 AnalyticsConstants.TCA_POLICY_THRESHOLDS_PATH_POSTFIX);
\r
646 // create a map of all threshold values
\r
647 final Map<String, Map<String, String>> thresholdsMap =
\r
648 extractSubTree(thresholdsValuesMaps, 1, 2,
\r
649 AnalyticsConstants.TCA_POLICY_DELIMITER);
\r
651 // add thresholds to nmetrics per event Names threshold list
\r
652 for (Map<String, String> thresholdMap : thresholdsMap.values()) {
\r
653 newMetricsPerEventName.getThresholds().add(createNewThreshold(thresholdMap));
\r
658 return metricsPerEventNames;
\r
662 * Creates new instance of TCA Policy {@link Threshold} with values extracted from thresholdMap
\r
664 * @param thresholdMap threshold map with threshold values
\r
666 * @return new instance of TCA Policy Threshold
\r
668 public static Threshold createNewThreshold(final Map<String, String> thresholdMap) {
\r
669 final Threshold threshold = new Threshold();
\r
670 threshold.setClosedLoopControlName(thresholdMap.get("policy.closedLoopControlName"));
\r
671 threshold.setVersion(thresholdMap.get("policy.version"));
\r
672 threshold.setFieldPath(thresholdMap.get("policy.fieldPath"));
\r
673 threshold.setDirection(Direction.valueOf(thresholdMap.get("policy.direction")));
\r
674 threshold.setSeverity(EventSeverity.valueOf(thresholdMap.get("policy.severity")));
\r
675 threshold.setThresholdValue(Long.valueOf(thresholdMap.get("policy.thresholdValue")));
\r
676 threshold.setClosedLoopEventStatus(
\r
677 ClosedLoopEventStatus.valueOf(thresholdMap.get("policy.closedLoopEventStatus")));
\r
682 * Create new {@link MetricsPerEventName} instance with policy Name, policy Version and policy Scope
\r
683 * extracted from given eventNamesEntry
\r
685 * @param eventNamesEntry Event Names Entry
\r
687 * @return new instance of MetricsPerEventName
\r
689 public static MetricsPerEventName createNewMetricsPerEventName(
\r
690 final Map.Entry<String, Map<String, String>> eventNamesEntry) {
\r
691 // determine event Name
\r
692 final String eventName = eventNamesEntry.getKey();
\r
693 // determine event Name thresholds
\r
694 final Map<String, String> metricsPerEventNameThresholdsMap = eventNamesEntry.getValue();
\r
695 final MetricsPerEventName metricsPerEventName = new MetricsPerEventName();
\r
696 final List<Threshold> thresholds = new LinkedList<>();
\r
697 metricsPerEventName.setThresholds(thresholds);
\r
698 metricsPerEventName.setEventName(eventName);
\r
699 // bind policyName, policyVersion, policyScope and closedLoopControlName
\r
700 metricsPerEventName.setPolicyName(metricsPerEventNameThresholdsMap.get("policyName"));
\r
701 metricsPerEventName.setPolicyVersion(metricsPerEventNameThresholdsMap.get("policyVersion"));
\r
702 metricsPerEventName.setPolicyScope(metricsPerEventNameThresholdsMap.get("policyScope"));
\r
703 metricsPerEventName.setControlLoopSchemaType(ControlLoopSchemaType.valueOf(
\r
704 metricsPerEventNameThresholdsMap.get("controlLoopSchemaType")));
\r
705 return metricsPerEventName;
\r
709 * Converts a flattened key/value map which has keys delimited by a given delimiter.
\r
710 * The start Index and end index extract the sub-key value and returns a new map containing
\r
711 * sub-keys and values.
\r
713 * @param actualMap actual Map
\r
714 * @param startIndex start index
\r
715 * @param endIndex end index
\r
716 * @param delimiter delimiter
\r
718 * @return Map with new sub tree map
\r
720 public static Map<String, Map<String, String>> extractSubTree(
\r
721 final Map<String, String> actualMap, int startIndex, int endIndex, String delimiter) {
\r
723 final SortedMap<String, Map<String, String>> subTreeMap = new TreeMap<>();
\r
725 // iterate over actual map entries
\r
726 for (Map.Entry<String, String> actualMapEntry : actualMap.entrySet()) {
\r
727 final String actualMapKey = actualMapEntry.getKey();
\r
728 final String actualMapValue = actualMapEntry.getValue();
\r
730 // determine delimiter start and end index
\r
731 final int keyStartIndex = StringUtils.ordinalIndexOf(actualMapKey, delimiter, startIndex);
\r
732 final int keyEndIndex = StringUtils.ordinalIndexOf(actualMapKey, delimiter, endIndex);
\r
733 final int keyLength = actualMapKey.length();
\r
735 // extract sub-tree map
\r
736 if (keyStartIndex != -1 && keyEndIndex != -1 && keyEndIndex > keyStartIndex && keyLength > keyEndIndex) {
\r
737 final String thresholdKey = actualMapKey.substring(keyStartIndex + 1, keyEndIndex);
\r
738 final Map<String, String> existingThresholdMap = subTreeMap.get(thresholdKey);
\r
739 final String subMapKey = actualMapKey.substring(keyEndIndex + 1, keyLength);
\r
740 if (existingThresholdMap == null) {
\r
741 Map<String, String> newThresholdMap = new LinkedHashMap<>();
\r
742 newThresholdMap.put(subMapKey, actualMapValue);
\r
743 subTreeMap.put(thresholdKey, newThresholdMap);
\r
745 existingThresholdMap.put(subMapKey, actualMapValue);
\r
757 * Provides a view of underlying map that filters out entries with keys starting with give prefix
\r
759 * @param actualMap Target map that needs to be filtered
\r
760 * @param keyNamePrefix key prefix
\r
762 * @return a view of actual map which only show entries which have give prefix
\r
764 public static Map<String, String> filterMapByKeyNamePrefix(final Map<String, String> actualMap,
\r
765 final String keyNamePrefix) {
\r
766 return Maps.filterKeys(actualMap,
\r
767 new Predicate<String>() {
\r
769 public boolean apply(@Nullable String key) {
\r
770 return key != null && key.startsWith(keyNamePrefix);
\r
777 * Creates Quartz Scheduler
\r
779 * @param pollingIntervalMS polling interval
\r
780 * @param stdSchedulerFactory Quartz standard schedule factory instance
\r
781 * @param quartzPublisherPropertiesFileName quartz properties file name
\r
782 * @param jobDataMap job Data map
\r
783 * @param quartzJobClass Quartz Job Class
\r
784 * @param quartzJobName Quartz Job Name
\r
785 * @param quartzTriggerName Quartz Trigger name
\r
787 * @param <T> An implementation of Quartz {@link Job} interface
\r
788 * @return Configured Quartz Scheduler
\r
790 * @throws SchedulerException exception if unable to create to Quartz Scheduler
\r
792 public static <T extends Job> Scheduler createQuartzScheduler(final Integer pollingIntervalMS,
\r
793 final StdSchedulerFactory stdSchedulerFactory, final String quartzPublisherPropertiesFileName,
\r
794 final JobDataMap jobDataMap, final Class<T> quartzJobClass, final String quartzJobName,
\r
795 final String quartzTriggerName) throws SchedulerException {
\r
797 // Initialize a new Quartz Standard scheduler
\r
798 LOG.debug("Configuring quartz scheduler for Quartz Job: {} with properties file: {}",
\r
799 quartzJobClass.getSimpleName(), quartzPublisherPropertiesFileName);
\r
800 final Properties quartzProperties = AnalyticsModelIOUtils.loadPropertiesFile(
\r
801 quartzPublisherPropertiesFileName, new Properties());
\r
802 stdSchedulerFactory.initialize(quartzProperties);
\r
803 final Scheduler scheduler = stdSchedulerFactory.getScheduler();
\r
805 // Create a new job detail
\r
806 final JobDetail jobDetail = JobBuilder.newJob(quartzJobClass).withIdentity(quartzJobName,
\r
807 AnalyticsConstants.TCA_QUARTZ_GROUP_NAME).usingJobData(jobDataMap).build();
\r
809 // Create a new scheduling builder
\r
810 final SimpleScheduleBuilder simpleScheduleBuilder = SimpleScheduleBuilder.simpleSchedule()
\r
811 .withIntervalInMilliseconds(pollingIntervalMS) // job will use custom polling schedule
\r
812 .repeatForever(); // repeats while worker is running
\r
814 // Create a trigger for the TCA Publisher Job
\r
815 final SimpleTrigger simpleTrigger = TriggerBuilder.newTrigger()
\r
816 .withIdentity(quartzTriggerName, AnalyticsConstants.TCA_QUARTZ_GROUP_NAME)
\r
817 .startNow() // job starts right away
\r
818 .withSchedule(simpleScheduleBuilder).build();
\r
820 scheduler.scheduleJob(jobDetail, simpleTrigger);
\r
821 LOG.info("Scheduler Initialized successfully for JobName: {}", quartzJobClass.getSimpleName());
\r
827 * Does A&AI Enrichment for VM
\r
829 * @param tcavesResponse Outgoing alert object
\r
830 * @param aaiEnrichmentClient A&AI Enrichment client
\r
831 * @param aaiVMEnrichmentAPIPath A&AI VM Enrichment API Path
\r
832 * @param alertString alert String
\r
833 * @param vmSourceName vm source name
\r
835 public static void doAAIVMEnrichment(final TCAVESResponse tcavesResponse,
\r
836 final AAIEnrichmentClient aaiEnrichmentClient,
\r
837 final String aaiVMEnrichmentAPIPath,
\r
838 final String alertString,
\r
839 final String vmSourceName) {
\r
841 final String filterString = "vserver-name:EQUALS:" + vmSourceName;
\r
842 final ImmutableMap<String, String> queryParams = ImmutableMap.of(
\r
843 "search-node-type", "vserver", "filter", filterString);
\r
845 // fetch vm object resource Link from A&AI
\r
846 final String vmAAIResourceLinkDetails = aaiEnrichmentClient.getEnrichmentDetails(
\r
847 aaiVMEnrichmentAPIPath, queryParams, createAAIEnrichmentHeaders());
\r
848 final String vmObjectResourceLink = getVMObjectResourceLink(vmAAIResourceLinkDetails);
\r
850 if (vmObjectResourceLink == null) {
\r
851 LOG.warn("No A&AI Enrichment possible for alert message: {}.VM Object resource Link cannot be " +
\r
852 "determined for vmSourceName: {}.", alertString, vmSourceName);
\r
855 LOG.debug("Fetching VM A&AI Enrichment Details for VM Source Name: {}, Object resource Link: {}",
\r
856 vmSourceName, vmObjectResourceLink);
\r
858 // fetch vm A&AI Enrichment
\r
859 final String vmEnrichmentDetails = aaiEnrichmentClient.getEnrichmentDetails(
\r
860 vmObjectResourceLink, Collections.<String, String>emptyMap(), createAAIEnrichmentHeaders());
\r
863 enrichAAI(tcavesResponse.getAai(), vmEnrichmentDetails, alertString,
\r
864 AnalyticsConstants.AAI_VSERVER_KEY_PREFIX);
\r
872 * Does A&AI Enrichment for VNF
\r
874 * @param tcavesResponse Outgoing alert object
\r
875 * @param aaiEnrichmentClient A&AI Enrichment client
\r
876 * @param aaiVNFEnrichmentAPIPath A&AI VNF Enrichment API Path
\r
877 * @param alertString alert String
\r
878 * @param vnfSourceName vnf source name
\r
880 public static void doAAIVNFEnrichment(final TCAVESResponse tcavesResponse,
\r
881 final AAIEnrichmentClient aaiEnrichmentClient,
\r
882 final String aaiVNFEnrichmentAPIPath,
\r
883 final String alertString,
\r
884 final String vnfSourceName) {
\r
885 final ImmutableMap<String, String> queryParams = ImmutableMap.of("vnf-name", vnfSourceName);
\r
887 // fetch vnf A&AI Enrichment
\r
888 final String vnfEnrichmentDetails = aaiEnrichmentClient.getEnrichmentDetails(
\r
889 aaiVNFEnrichmentAPIPath, queryParams, createAAIEnrichmentHeaders());
\r
891 // enrich alert AAI
\r
892 enrichAAI(tcavesResponse.getAai(), vnfEnrichmentDetails, alertString, AnalyticsConstants.AAI_VNF_KEY_PREFIX);
\r
896 * Fetches VM Object Resource Link from A&AI Resource Link Json
\r
898 * @param vmAAIResourceLinkDetails VM Object Resource Link from A&AI Resource Link Json
\r
900 * @return object resource link String
\r
902 private static String getVMObjectResourceLink(final String vmAAIResourceLinkDetails) {
\r
903 if (StringUtils.isNotBlank(vmAAIResourceLinkDetails)) {
\r
905 final JsonNode jsonNode = ANALYTICS_MODEL_OBJECT_MAPPER.readTree(vmAAIResourceLinkDetails);
\r
906 final JsonNode resourceLinkJsonNode = jsonNode.findPath("resource-link");
\r
907 if (!resourceLinkJsonNode.isMissingNode()) {
\r
908 return resourceLinkJsonNode.asText();
\r
910 } catch (IOException e) {
\r
911 LOG.warn("Unable to determine VM Object link inside AAI Resource Link Response JSON: {}. Exception: {}",
\r
912 vmAAIResourceLinkDetails, e);
\r
919 * Creates Http Headers for A&AI Enrichment client
\r
921 * @return Http Headers Map for A&AI Enrichment client
\r
923 private static Map<String, String> createAAIEnrichmentHeaders() {
\r
924 final Map<String, String> aaiEnrichmentHeaders = new LinkedHashMap<>();
\r
925 final String transactionId = Long.toString(new Date().getTime());
\r
926 aaiEnrichmentHeaders.put("X-FromAppId", "dcae-analytics-tca");
\r
927 aaiEnrichmentHeaders.put("X-TransactionId", transactionId);
\r
928 aaiEnrichmentHeaders.put("Accept", "application/json");
\r
929 aaiEnrichmentHeaders.put("Real-Time", "true");
\r
930 aaiEnrichmentHeaders.put("Content-Type", "application/json");
\r
931 return aaiEnrichmentHeaders;
\r
936 * Populates A&AI details retrieved from A&AI Enrichment API into Alerts A&AI Object
\r
938 * @param preEnrichmentAAI A&AI Alert object which needs to be populated with A&AI Enrichment Details
\r
939 * @param aaiEnrichmentDetails A&AI Enrichment API fetched JSON String
\r
940 * @param alertString Alert String
\r
941 * @param keyPrefix Key prefix that needs to be added to each fetched A&AI Enrichment record
\r
943 private static void enrichAAI(final AAI preEnrichmentAAI, final String aaiEnrichmentDetails,
\r
944 final String alertString, final String keyPrefix) {
\r
946 if (aaiEnrichmentDetails == null) {
\r
947 LOG.warn("No A&AI Enrichment possible for AAI: {}. A&AI Enrichment details are absent." +
\r
948 "Skipping Enrichment for alert message:{}", preEnrichmentAAI, alertString);
\r
952 final AAI enrichmentDetailsAAI = getEnrichmentDetailsAAI(aaiEnrichmentDetails);
\r
954 if (enrichmentDetailsAAI != null) {
\r
955 final Set<Map.Entry<String, Object>> enrichedAAIEntrySet =
\r
956 enrichmentDetailsAAI.getDynamicProperties().entrySet();
\r
957 final Map<String, Object> preEnrichmentAAIDynamicProperties = preEnrichmentAAI.getDynamicProperties();
\r
959 // populate A&AI Enrichment details and add prefix to key
\r
960 for (Map.Entry<String, Object> enrichedAAIEntry : enrichedAAIEntrySet) {
\r
961 preEnrichmentAAIDynamicProperties.put(keyPrefix + enrichedAAIEntry.getKey(),
\r
962 enrichedAAIEntry.getValue());
\r
965 LOG.debug("A&AI Enrichment was completed successfully for alert message: {}. Enriched AAI: {}",
\r
966 alertString, preEnrichmentAAI);
\r
968 LOG.warn("No A&AI Enrichment possible for AAI: {}. Invalid A&AI Response: {}." +
\r
969 "Skipping Enrichment for alert message: {}",
\r
970 preEnrichmentAAI, aaiEnrichmentDetails, alertString);
\r
977 * Creates a new A&AI object with only top level A&AI Enrichment details
\r
979 * @param aaiEnrichmentDetails A&AI Enrichment details
\r
981 * @return new A&AI with only top level A&AI Enrichment details
\r
983 private static AAI getEnrichmentDetailsAAI(final String aaiEnrichmentDetails) {
\r
985 final JsonNode rootNode = ANALYTICS_MODEL_OBJECT_MAPPER.readTree(aaiEnrichmentDetails);
\r
986 final Iterator<Map.Entry<String, JsonNode>> fieldsIterator = rootNode.fields();
\r
987 while (fieldsIterator.hasNext()) {
\r
988 final Map.Entry<String, JsonNode> fieldEntry = fieldsIterator.next();
\r
989 final JsonNode jsonNode = fieldEntry.getValue();
\r
990 // remove all arrays, objects from A&AI Enrichment Json
\r
991 if (jsonNode.isPojo() || jsonNode.isObject() || jsonNode.isArray()) {
\r
992 fieldsIterator.remove();
\r
995 return ANALYTICS_MODEL_OBJECT_MAPPER.treeToValue(rootNode, AAI.class);
\r
996 } catch (IOException e) {
\r
997 LOG.error("Failed to Parse AAI Enrichment Details from JSON: {}, Exception: {}.", aaiEnrichmentDetails, e);
\r