-# Copyright (C) 2022-2024 Wind River Systems, Inc.
+# Copyright (C) 2022 Wind River Systems, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
class StxEventClient(BaseClient):
- def __init__(self, uow: unit_of_work.AbstractUnitOfWork, driver=None):
+ def __init__(self, driver=None):
super().__init__()
self.driver = driver if driver else StxFaultClientImp()
- self.uow = uow
def _get(self, id) -> alarmModel.FaultGenericModel:
return self.driver.getEventInfo(id)
def _set_stx_client(self):
self.driver.setFaultClient(self._pool_id)
- def suppression_list(self, alarm_id) -> List[alarmModel.FaultGenericModel]:
- return self.driver.getSuppressionList(alarm_id)
-
- def suppress(self, id) -> alarmModel.FaultGenericModel:
- return self.driver.suppressEvent(id)
-
# internal driver which implement client call to Stx Fault Management instance
class StxFaultClientImp(object):
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]
return alarmModel.FaultGenericModel(
alarmModel.EventTypeEnum.EVENT, self._eventconverter(event))
- def suppressEvent(self, id) -> alarmModel.FaultGenericModel:
- patch = [dict(path='/' + 'suppression_status', value='suppressed',
- op='replace')]
- event = self.fmclient.event_suppression.update(id, patch)
- logger.debug('suppressed event id ' + id + ':' + str(event.to_dict()))
- return alarmModel.FaultGenericModel(
- alarmModel.EventTypeEnum.EVENT, self._suppression_converter(event))
-
- def getSuppressionList(self, alarm_id) -> alarmModel.FaultGenericModel:
- suppression_list = []
- queryAsArray = []
- events = self.fmclient.event_suppression.list(q=queryAsArray)
- for event in events:
- if event.alarm_id == alarm_id:
- # logger.debug('suppression event:' + str(event.to_dict()))
- suppression_list.append(
- alarmModel.FaultGenericModel(
- alarmModel.EventTypeEnum.EVENT,
- self._suppression_converter(event)))
- return suppression_list
-
@ staticmethod
def _alarmconverter(alarm):
selected_keys = [
uuid.NAMESPACE_URL, event.probable_cause)))
return event
- @ staticmethod
- def _suppression_converter(event, clear=False):
- selected_keys = [
- 'alarm_id', 'description', 'suppression_status',
- 'links'
- ]
- content = event.to_dict()
- filtered = dict(
- filter(lambda item: item[0] in selected_keys, content.items()))
- setattr(event, 'filtered', filtered)
- setattr(event, 'uuid', event.uuid)
- setattr(event, 'alarm_id', event.alarm_id)
- setattr(event, 'description', event.description)
- setattr(event, 'suppression_status', event.suppression_status)
- setattr(event, 'alarm_type', None)
- setattr(event, 'alarm_def_id', None)
- setattr(event, 'probable_cause_id', None)
- setattr(event, 'state', None)
- setattr(event, 'timestamp', None)
- return event
-
@ staticmethod
def _alarmeventhasher(event, state=''):
# The event model and the alarm model have different parameter name