sdnr/wt/mountpoint-registrar/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointregistrar/vesdomain/stnddefined/StrimziKafkaStndDefinedFaultVESMsgConsumer.java

   1 /*
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt mountpoint-registrar
   4  * =================================================================================================
   5  * Copyright (C) 2022 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  */
  18
  19 package org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.vesdomain.stnddefined;
  20
  21 import com.fasterxml.jackson.core.JsonProcessingException;
  22 import com.fasterxml.jackson.databind.JsonNode;
  23 import java.time.Instant;
  24 import java.time.ZoneId;
  25 import java.util.Map;
  26 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.config.GeneralConfig;
  27 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.InvalidMessageException;
  28 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.impl.StrimziKafkaVESMsgConsumerImpl;
  29 import org.onap.ccsdk.features.sdnr.wt.mountpointregistrar.vesdomain.fault.FaultNotificationClient;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.data.provider.rev201110.SeverityType;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 public class StrimziKafkaStndDefinedFaultVESMsgConsumer extends StrimziKafkaVESMsgConsumerImpl {
  35
  36     private static final Logger LOG = LoggerFactory.getLogger(StrimziKafkaStndDefinedFaultVESMsgConsumer.class);
  37     Map<String, String> payloadMapMessage = null;
  38     String faultNodeId;
  39     String notificationType;
  40
  41     public StrimziKafkaStndDefinedFaultVESMsgConsumer(GeneralConfig generalConfig) {
  42         super(generalConfig);
  43         LOG.info("StrimziKafkaStndDefinedFaultVESMsgConsumer started successfully");
  44     }
  45
  46     /*
  47      * Supports processing of notifyNewAlarm and notifyClearedAlarm messages ONLY
  48      */
  49     @Override
  50     public void processMsg(String msg) throws InvalidMessageException, JsonProcessingException {
  51         LOG.debug("Processing StndDefined Fault message {}", msg);
  52         JsonNode rootNode = convertMessageToJsonNode(msg);
  53         try {
  54
  55             faultNodeId = rootNode.at("/event/commonEventHeader/sourceName").textValue();
  56             notificationType = rootNode.at("/event/stndDefinedFields/data/notificationType").textValue();
  57
  58             if (notificationType.equalsIgnoreCase("notifyNewAlarm")) {
  59                 LOG.info("Read stndDefined Fault message of type - {} with id {} from Kafka topic", notificationType,
  60                         faultNodeId);
  61                 processNewAlarm(rootNode);
  62             } else if (notificationType.equalsIgnoreCase("notifyClearedAlarm")) {
  63                 LOG.info("Read stdnDefined Fault message of type - {} with id {} from Kafka topic", notificationType,
  64                         faultNodeId);
  65                 processClearedAlarm(rootNode);
  66             } else {
  67                 LOG.warn(
  68                         "Read stdnDefined Fault message of type - {} with id {} from Kafka topic. No suitable implementation for processing this message",
  69                         notificationType, faultNodeId);
  70                 throw new InvalidMessageException();
  71             }
  72             // Send Fault Notification
  73             String baseUrl = getBaseUrl();
  74             String sdnrUser = getSDNRUser();
  75             String sdnrPasswd = getSDNRPasswd();
  76
  77             FaultNotificationClient faultClient = new FaultNotificationClient(baseUrl);
  78             LOG.debug("Setting RESTConf Authorization values - {} : {}", sdnrUser, sdnrPasswd);
  79             faultClient.setAuthorization(sdnrUser, sdnrPasswd);
  80             String message = faultClient.prepareMessageFromPayloadMap(payloadMapMessage);
  81             faultClient.sendNotification(message);
  82         } catch (NullPointerException e) {
  83             LOG.warn("Message is invalid, sending aborted, processing stopped because one of fields is missing");
  84             throw new InvalidMessageException("Missing field");
  85         }
  86
  87     }
  88
  89     private void processClearedAlarm(JsonNode rootNode) {
  90         String faultOccurrenceTime =
  91                 Instant.ofEpochMilli(rootNode.at("/event/commonEventHeader/startEpochMicrosec").longValue() / 1000)
  92                         .atZone(ZoneId.of("Z")).toString();
  93         int faultSequence = rootNode.at("/event/commonEventHeader/sequence").intValue();
  94         String faultObjectId = rootNode.at("/event/stndDefinedFields/data/alarmId").textValue();
  95         String faultReason = rootNode.at("/event/stndDefinedFields/data/probableCause").textValue();
  96         String faultSeverity = SeverityType.NonAlarmed.toString();
  97
  98         payloadMapMessage = FaultNotificationClient.createFaultNotificationPayloadMap(faultNodeId,
  99                 Integer.toString(faultSequence), faultOccurrenceTime, faultObjectId, faultReason, faultSeverity);
 100
 101     }
 102
 103     private void processNewAlarm(JsonNode rootNode) {
 104         String faultOccurrenceTime =
 105                 Instant.ofEpochMilli(rootNode.at("/event/commonEventHeader/startEpochMicrosec").longValue() / 1000)
 106                         .atZone(ZoneId.of("Z")).toString();
 107         int faultSequence = rootNode.at("/event/commonEventHeader/sequence").intValue();
 108         String faultObjectId = rootNode.at("/event/stndDefinedFields/data/alarmId").textValue();
 109         String faultReason = rootNode.at("/event/stndDefinedFields/data/probableCause").textValue();
 110         String faultSeverity =
 111                 getSDNRSeverityType(rootNode.at("/event/stndDefinedFields/data/perceivedSeverity").textValue());
 112
 113         payloadMapMessage = FaultNotificationClient.createFaultNotificationPayloadMap(faultNodeId,
 114                 Integer.toString(faultSequence), faultOccurrenceTime, faultObjectId, faultReason, faultSeverity);
 115
 116     }
 117
 118     /*
 119      * 3GPP Definition PerceivedSeverity: type: string enum: - INDETERMINATE -
 120      * CRITICAL - MAJOR - MINOR - WARNING - CLEARED
 121      *
 122      */
 123     private String getSDNRSeverityType(String faultSeverity) {
 124         if (faultSeverity.equalsIgnoreCase("critical")) {
 125             faultSeverity = SeverityType.Critical.toString();
 126         } else if (faultSeverity.equalsIgnoreCase("major")) {
 127             faultSeverity = SeverityType.Major.toString();
 128         } else if (faultSeverity.equalsIgnoreCase("minor")) {
 129             faultSeverity = SeverityType.Minor.toString();
 130         } else if (faultSeverity.equalsIgnoreCase("warning") || faultSeverity.equalsIgnoreCase("indeterminate")) {
 131             faultSeverity = SeverityType.Warning.toString();
 132         } else if (faultSeverity.equalsIgnoreCase("cleared")) {
 133             faultSeverity = SeverityType.NonAlarmed.toString();
 134         } else {
 135             faultSeverity = SeverityType.NonAlarmed.toString();
 136         }
 137         return faultSeverity;
 138     }
 139
 140 }