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