from cgtsclient.exc import EndpointException
from dcmanagerclient.api.client import client as get_dc_client
from fmclient.client import get_client as get_fm_client
-from fmclient.common.exceptions import HTTPNotFound
+from fmclient.common.exceptions import HTTPNotFound, HttpServerError
from o2app.adapter import unit_of_work
from o2common.config import config
logger.debug('alarm {} not in this resource pool {}'
.format(alarm, self._pool_id))
continue
+ except HttpServerError:
+ # TODO(jon): This exception needs to be removed when the
+ # INF-457 related FM client upgrade and issue fix occur.
+ logger.debug('alarm {} query failed'.format(alarm))
+ continue
ret.append(event)
return ret
try:
sub_is_https = False
os_client_args = config.get_stx_access_info(
- region_name=subcloud[0].name,
+ region_name=subcloud[0].region_name,
subcloud_hostname=subcloud[0].oam_floating_ip)
stx_client = get_stx_client(**os_client_args)
except EndpointException as e:
if CGTSCLIENT_ENDPOINT_ERROR_MSG in msg:
sub_is_https = True
os_client_args = config.get_stx_access_info(
- region_name=subcloud[0].name, sub_is_https=sub_is_https,
+ region_name=subcloud[0].region_name,
+ sub_is_https=sub_is_https,
subcloud_hostname=subcloud[0].oam_floating_ip)
stx_client = get_stx_client(**os_client_args)
else:
alarms = self.fmclient.alarm.list(expand=True)
if len(alarms) == 0:
return []
- logger.debug('alarm 1:' + str(alarms[0].to_dict()))
- # [print('alarm:' + str(alarm.to_dict())) for alarm in alarms if alarm]
+ [logger.debug(
+ 'alarm:' + str(alarm.to_dict())) for alarm in alarms if alarm]
return [alarmModel.FaultGenericModel(
alarmModel.EventTypeEnum.ALARM, self._alarmconverter(alarm))
for alarm in alarms if alarm]
def getAlarmInfo(self, id) -> alarmModel.FaultGenericModel:
try:
alarm = self.fmclient.alarm.get(id)
- logger.debug('get alarm id ' + id + ':' + str(alarm.to_dict()))
+ logger.debug(
+ 'get alarm id: ' + id + ', result:' + str(alarm.to_dict()))
except HTTPNotFound:
event = self.fmclient.event_log.get(id)
return alarmModel.FaultGenericModel(
def getEventList(self, **filters) -> List[alarmModel.FaultGenericModel]:
events = self.fmclient.event_log.list(alarms=True, expand=True)
- logger.debug('event 1:' + str(events[0].to_dict()))
- # [print('alarm:' + str(event.to_dict())) for event in events if event]
+ [logger.debug(
+ 'alarm:' + str(event.to_dict())) for event in events if event]
return [alarmModel.FaultGenericModel(
alarmModel.EventTypeEnum.EVENT, self._eventconverter(event))
for event in events if event]
@ staticmethod
def _alarmconverter(alarm):
+ selected_keys = [
+ 'alarm_id', 'alarm_state', 'entity_type_id', 'entity_instance_id',
+ 'reason_text', 'alarm_type', 'probable_cause',
+ 'proposed_repair_action', 'service_affecting', 'suppression',
+ 'suppression_status', 'mgmt_affecting', 'degrade_affecting'
+ ]
+ content = alarm.to_dict()
+ filtered = dict(
+ filter(lambda item: item[0] in selected_keys, content.items()))
+ setattr(alarm, 'filtered', filtered)
# setattr(alarm, 'alarm_def_id', uuid.uuid3(
# uuid.NAMESPACE_URL, alarm.alarm_id))
setattr(alarm, 'state', alarm.alarm_state)
@ staticmethod
def _eventconverter(event, clear=False):
+ selected_keys = [
+ 'event_log_id', 'state', 'entity_type_id',
+ 'entity_instance_id', 'reason_text', 'event_log_type',
+ 'probable_cause', 'proposed_repair_action',
+ 'service_affecting', 'suppression', 'suppression_status'
+ ]
+ content = event.to_dict()
+ filtered = dict(
+ filter(lambda item: item[0] in selected_keys, content.items()))
+ setattr(event, 'filtered', filtered)
setattr(event, 'alarm_id', event.event_log_id)
setattr(event, 'alarm_type', event.event_log_type)
if clear: