ems/boco/src/main/java/org/onap/vfc/nfvo/emsdriver/northbound/client/NorthMessageMgr.java

   1 /*
   2  * Copyright 2017 BOCO Corporation.  CMCC Technologies Co., Ltd
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.onap.vfc.nfvo.emsdriver.northbound.client;
  17
  18 import com.alibaba.fastjson.JSONObject;
  19 import evel_javalibrary.att.com.AgentMain;
  20 import evel_javalibrary.att.com.AgentMain.EVEL_ERR_CODES;
  21 import evel_javalibrary.att.com.EvelFault;
  22 import evel_javalibrary.att.com.EvelFault.EVEL_SEVERITIES;
  23 import evel_javalibrary.att.com.EvelFault.EVEL_SOURCE_TYPES;
  24 import evel_javalibrary.att.com.EvelFault.EVEL_VF_STATUSES;
  25 import evel_javalibrary.att.com.EvelHeader;
  26 import evel_javalibrary.att.com.EvelScalingMeasurement;
  27
  28 import org.apache.commons.logging.Log;
  29 import org.apache.commons.logging.LogFactory;
  30 import org.apache.log4j.Level;
  31 import org.onap.vfc.nfvo.emsdriver.collector.alarm.AlarmManager;
  32 import org.onap.vfc.nfvo.emsdriver.commons.constant.Constant;
  33 import org.onap.vfc.nfvo.emsdriver.commons.utils.DriverThread;
  34 import org.onap.vfc.nfvo.emsdriver.configmgr.ConfigurationInterface;
  35 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannel;
  36 import org.onap.vfc.nfvo.emsdriver.messagemgr.MessageChannelFactory;
  37
  38 import java.text.ParseException;
  39 import java.text.SimpleDateFormat;
  40 import java.util.Date;
  41 import java.util.Map;
  42 import java.util.Properties;
  43
  44 public class NorthMessageMgr extends DriverThread {
  45         protected static final Log logger = LogFactory.getLog(NorthMessageMgr.class);
  46     private MessageChannel alarmChannel = MessageChannelFactory.getMessageChannel(Constant.RESULT_CHANNEL_KEY);
  47     private MessageChannel collectResultPMChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_PM_CHANNEL_KEY);
  48     private MessageChannel collectResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_CHANNEL_KEY);
  49     private ConfigurationInterface configurationInterface;
  50
  51     private boolean threadStop = false;
  52     private Level level = Level.DEBUG;
  53
  54     @Override
  55     public void dispose() {
  56         logger.info("NorthMessageMgr Thread start threadStop=" + threadStop);
  57         try {
  58             Properties properties = configurationInterface.getProperties();
  59             String eventApiUrl = properties.getProperty("event_api_url");
  60             String port = properties.getProperty("port");
  61             String path = properties.getProperty("path");
  62             String topic = properties.getProperty("topic");
  63             String username = properties.getProperty("username");
  64             String password = properties.getProperty("password");
  65             String levelStr = properties.getProperty("level");
  66             if ("debug".equals(levelStr)) {
  67                 level = Level.DEBUG;
  68             } else {
  69                 level = Level.INFO;
  70             }
  71
  72             //login north
  73             eventApiUrl = "http://" + eventApiUrl;
  74             logger.info("AgentMain.evel_initialize start event_api_url=[" + eventApiUrl + "]port=[" + port + "]path=[" + path + "]"
  75                     + "topic=[" + topic + "]username=[" + username + /*"]password=[" + password +*/ "]level=[" + level + "]");
  76             try {
  77                 EVEL_ERR_CODES evecode = AgentMain.evel_initialize(eventApiUrl, Integer.parseInt(port),
  78                         path, topic,
  79                         username,
  80                         password,
  81                         level);
  82                 logger.info("AgentMain.evel_initialize sucess EVEL_ERR_CODES=" + evecode);
  83             } catch (Exception e) {
  84                 logger.error("AgentMain.evel_initialize fail ", e);
  85             }
  86         } catch (Exception e2) {
  87                 logger.error("NorthMessageMgr start fail ", e2);
  88         }
  89
  90         new HeatBeatTread().start();
  91         //
  92         new AlarmMessageRecv().start();
  93
  94         new ResultMessageRecv().start();
  95
  96         new CollectMessageRecv().start();
  97
  98         logger.info("NorthMessageMgr start sucess ");
  99     }
 100
 101     /**
 102      * @return the configurationInterface
 103      */
 104     public ConfigurationInterface getConfigurationInterface() {
 105         return configurationInterface;
 106     }
 107
 108     /**
 109      * @param configurationInterface the configurationInterface to set
 110      */
 111     public void setConfigurationInterface(
 112             ConfigurationInterface configurationInterface) {
 113         this.configurationInterface = configurationInterface;
 114     }
 115
 116     class HeatBeatTread extends Thread {
 117         @Override
 118         public void run() {
 119
 120             while (!threadStop) {
 121
 122                 try {
 123                     EvelHeader header = EvelHeader.evel_new_heartbeat("Hearbeat_EMS", "EMS-driver");
 124                     header.evel_nfnamingcode_set("EMS-driver");
 125                     header.evel_nfcnamingcode_set("EMS-driver");
 126                     AgentMain.evel_post_event(header);
 127                     logger.info("HeatBeat send!");
 128                     try {
 129                         Thread.sleep(60 * 1000L);//60 secs
 130                     } catch (Exception e) {
 131                         logger.error("Unable to sleep the HB thread ", e);
 132                     }
 133                 } catch (Exception e) {
 134                         logger.error("HeatBeatTread exception", e);
 135                 }
 136             }
 137         }
 138     }
 139
 140     class AlarmMessageRecv extends Thread {
 141         long timeStamp = System.currentTimeMillis();
 142
 143         @Override
 144         public void run() {
 145             while (!threadStop) {
 146
 147                 try {
 148                     if (System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE) {
 149                         timeStamp = System.currentTimeMillis();
 150
 151                         logger.info("ALARM_CHANNEL Msg size :" + alarmChannel.size());
 152                     }
 153
 154                     Object obj = alarmChannel.poll();
 155                     if (obj == null) {
 156                         continue;
 157                     }
 158                     if (obj instanceof String) {
 159                         String result = (String) obj;
 160                         JSONObject reagobj = JSONObject.parseObject(result);
 161
 162                         EvelFault evelFault = this.resultEvelFault(reagobj);
 163
 164                         //send
 165                         logger.info("AgentMain.evel_post_event alarm start");
 166                         AgentMain.evel_post_event(evelFault);
 167                         logger.info("AgentMain.evel_post_event alarm sucess");
 168                     } else {
 169                         logger.error("AlarmMessageRecv receive Object = " + obj);
 170                     }
 171
 172                 } catch (Exception e) {
 173                         logger.error("AlarmMessageRecv exception", e);
 174                 }
 175             }
 176         }
 177
 178         private EvelFault resultEvelFault(JSONObject reagobj) {
 179
 180             String eventName = null;
 181             EvelHeader.PRIORITIES pri = null;
 182             EVEL_SEVERITIES severity = null;
 183             EVEL_VF_STATUSES status = null;
 184             String alarmStatus = reagobj.getString("alarmStatus");
 185             String origSeverity = reagobj.getString("origSeverity");
 186             if ("0".equals(alarmStatus)) {
 187                 status = EVEL_VF_STATUSES.EVEL_VF_STATUS_IDLE;
 188                 eventName = "Fault_" + reagobj.getString("neType") + "_" + reagobj.getString("alarmTitle") + "Cleared";
 189
 190                 if ("1".equals(origSeverity)) {
 191                     severity = EVEL_SEVERITIES.EVEL_SEVERITY_CRITICAL;
 192                     pri = EvelHeader.PRIORITIES.EVEL_PRIORITY_HIGH;
 193                 } else if ("2".equals(origSeverity)) {
 194                     severity = EVEL_SEVERITIES.EVEL_SEVERITY_MAJOR;
 195                     pri = EvelHeader.PRIORITIES.EVEL_PRIORITY_MEDIUM;
 196                 } else if ("3".equals(origSeverity)) {
 197                     severity = EVEL_SEVERITIES.EVEL_SEVERITY_MINOR;
 198                     pri = EvelHeader.PRIORITIES.EVEL_PRIORITY_NORMAL;
 199                 } else if ("4".equals(origSeverity)) {
 200                     severity = EVEL_SEVERITIES.EVEL_SEVERITY_WARNING;
 201                     pri = EvelHeader.PRIORITIES.EVEL_PRIORITY_LOW;
 202
 203                 }
 204             } else {
 205                 status = EVEL_VF_STATUSES.EVEL_VF_STATUS_ACTIVE;
 206                 eventName = "Fault_" + reagobj.getString("neType") + "_" + reagobj.getString("alarmTitle");
 207                 pri = EvelHeader.PRIORITIES.EVEL_PRIORITY_NORMAL;
 208                 severity = EVEL_SEVERITIES.EVEL_SEVERITY_NORMAL;
 209             }
 210
 211             String evnId = reagobj.getString("alarmId");
 212             String alarmCondition = reagobj.getString("specificProblem");
 213
 214             String specificProblem = reagobj.getString("specificProblem");
 215
 216             EvelFault flt = new EvelFault(eventName, evnId, alarmCondition,
 217                     specificProblem, pri, severity,
 218                     EVEL_SOURCE_TYPES.EVEL_SOURCE_VIRTUAL_NETWORK_FUNCTION,
 219                     status);
 220             flt.evel_nfcnamingcode_set("");
 221             flt.evel_nfnamingcode_set("");
 222             flt.evel_header_type_set("applicationVnf");
 223             String eventTime = reagobj.getString("eventTime");
 224             SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 225             Date eventTimeD = new Date();
 226             try {
 227                 eventTimeD = format.parse(eventTime);
 228             } catch (ParseException e) {
 229                 logger.error("ParseException ", e);
 230             }
 231             flt.evel_start_epoch_set(eventTimeD.getTime());
 232             flt.evel_last_epoch_set(eventTimeD.getTime());
 233
 234             flt.evel_fault_category_set(reagobj.getString("alarmType"));
 235             flt.evel_fault_interface_set(reagobj.getString("objectName"));
 236             String neUID = reagobj.getString("neUID");
 237             flt.evel_reporting_entity_id_set(neUID.substring(0, 9));//
 238             flt.evel_reporting_entity_name_set(neUID.substring(0, 9));
 239             flt.evel_header_set_sourceid(true, reagobj.getString("neName"));
 240             flt.evel_header_set_source_name(reagobj.getString("objectName"));
 241
 242             flt.evel_header_set_priority(pri);
 243             for (String key : reagobj.keySet()) {
 244                 flt.evel_fault_addl_info_add(key, reagobj.getString(key));
 245             }
 246
 247             return flt;
 248         }
 249     }
 250
 251     class ResultMessageRecv extends Thread {
 252         long timeStamp = System.currentTimeMillis();
 253
 254         @Override
 255         public void run() {
 256             while (!threadStop) {
 257
 258                 try {
 259                     if (System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE) {
 260                         timeStamp = System.currentTimeMillis();
 261
 262                         logger.debug("COLLECT_RESULT_CHANNEL Msg size :" + collectResultChannel.size());
 263                     }
 264
 265                     Object obj = collectResultChannel.poll();
 266                     if (obj == null) {
 267                         continue;
 268                     }
 269                     if (obj instanceof String) {
 270                         //http
 271                         Properties properties = configurationInterface.getProperties();
 272                         String msbAddress = properties.getProperty("msbAddress");
 273                         String url = properties.getProperty("dataNotifyUrl");
 274                         String postUrl = "http://" + msbAddress + url;
 275                         HttpClientUtil.doPost(postUrl, (String) obj, Constant.ENCODING_UTF8);
 276                     }
 277
 278                 } catch (Exception e) {
 279                         logger.error("ResultMessageRecv exception", e);
 280                 }
 281             }
 282         }
 283     }
 284
 285     class CollectMessageRecv extends Thread {
 286         long timeStamp = System.currentTimeMillis();
 287
 288         @Override
 289         public void run() {
 290                         logger.info("CollectMessageRecv Thread is start threadStop=" + threadStop);
 291             while (!threadStop) {
 292
 293                 try {
 294                     if (System.currentTimeMillis() - timeStamp > Constant.ONEMINUTE) {
 295                         timeStamp = System.currentTimeMillis();
 296
 297                         logger.debug(Constant.COLLECT_RESULT_PM_CHANNEL_KEY + " Msg size :" + collectResultPMChannel.size());
 298                     }
 299
 300                     Object obj = collectResultPMChannel.poll();
 301                     if (obj == null) {
 302                         continue;
 303                     }
 304                     if (obj instanceof Map) {
 305                         @SuppressWarnings("unchecked")
 306                         Map<String, String> reMap = (Map<String, String>) obj;
 307                         logger.debug("reMap =" + reMap);
 308                         EvelScalingMeasurement evelScalingMeasurement = this.resultEvelScalingMeasurement(reMap);
 309                         logger.debug("evelScalingMeasurement=" + evelScalingMeasurement);
 310                         //send
 311                         logger.info("AgentMain.evel_post_event start");
 312                         AgentMain.evel_post_event(evelScalingMeasurement);
 313                         logger.info("AgentMain.evel_post_event sucess");
 314
 315                     } else {
 316                         logger.error("CollectMessageRecv receive Object = " + obj);
 317                     }
 318
 319                 } catch (Exception e) {
 320                         logger.error("CollectMessageRecv exception", e);
 321                 }
 322              }
 323         }
 324
 325         private EvelScalingMeasurement resultEvelScalingMeasurement(Map<String, String> reMap) {
 326             String evname = "Mfvs_" + reMap.get("ElementType") + reMap.get("ObjectType");
 327             String evid = reMap.get("StartTime") + reMap.get("ObjectType") + reMap.get("rmUID");
 328             int period = Integer.parseInt(reMap.get("Period") != null ? reMap.get("Period") : "15");
 329             EvelScalingMeasurement sm = new EvelScalingMeasurement(period, evname, evid);
 330
 331             for (String key : reMap.keySet()) {
 332                 sm.evel_measurement_custom_measurement_add(reMap.get("ElementType"), key, reMap.get(key));
 333
 334             }
 335
 336             sm.evel_nfcnamingcode_set("");
 337             sm.evel_nfnamingcode_set("");
 338             sm.evel_header_type_set("applicationVnf");
 339             String rmUID = reMap.get("rmUID");
 340             sm.evel_reporting_entity_id_set(rmUID.substring(0, 9));//
 341             String dn = reMap.get("Dn");
 342             if (dn != null)
 343                 sm.evel_reporting_entity_name_set(dn.substring(0, dn.indexOf(";") > -1 ? dn.indexOf(";") : dn.length()));//0 is valid index
 344             else {
 345                 // decide  the flow if Dn is null
 346
 347             }
 348             sm.evel_header_set_sourceid(true, reMap.get("rmUID"));
 349             sm.evel_header_set_source_name(reMap.get("rmUID"));
 350
 351             sm.evel_header_set_priority(EvelHeader.PRIORITIES.EVEL_PRIORITY_NORMAL);
 352             return sm;
 353         }
 354     }
 355
 356 }