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