X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=o2ims%2Fadapter%2Fclients%2Ffault_client.py;h=989edfc693237e05627fb3b4715dcdc8faba6c98;hb=874ff786f1180878d74de7c4d85034247703112b;hp=953beb9216a48b358d8ea8125e39560817e12f48;hpb=f19e210d02bfe198549730c496f5642236ed3700;p=pti%2Fo2.git diff --git a/o2ims/adapter/clients/fault_client.py b/o2ims/adapter/clients/fault_client.py index 953beb9..989edfc 100644 --- a/o2ims/adapter/clients/fault_client.py +++ b/o2ims/adapter/clients/fault_client.py @@ -1,4 +1,4 @@ -# 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. @@ -99,10 +99,9 @@ class StxAlarmClient(BaseClient): 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) @@ -113,12 +112,6 @@ class StxEventClient(BaseClient): 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): @@ -157,7 +150,7 @@ 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: @@ -165,7 +158,8 @@ class StxFaultClientImp(object): 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: @@ -199,8 +193,8 @@ class StxFaultClientImp(object): 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] @@ -208,7 +202,8 @@ class StxFaultClientImp(object): 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( @@ -219,8 +214,8 @@ class StxFaultClientImp(object): 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] @@ -231,27 +226,6 @@ class StxFaultClientImp(object): 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 = [ @@ -297,27 +271,6 @@ class StxFaultClientImp(object): 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