-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
pubsub.subscribe('DmsChanged')
pubsub.subscribe('ResourceChanged')
pubsub.subscribe('AlarmEventChanged')
+ pubsub.subscribe('AlarmEventPurged')
for m in pubsub.listen():
try:
eventtype=data['notificationEventType'],
updatetime=data['updatetime']))
bus.handle(cmd)
+ elif channel == 'AlarmEventPurged':
+ datastr = m['data']
+ data = json.loads(datastr)
+ logger.info('AlarmEventPurged with cmd:{}'.format(data))
+ ref = api_monitoring_base + \
+ monitor_api_version + '/alarms/' + data['id']
+ cmd = imscmd.PurgeAlarmEvent(data=AlarmEvent2SMO(
+ id=data['id'], ref=ref,
+ eventtype=data['notificationEventType'],
+ updatetime=data['updatetime']))
+ bus.handle(cmd)
else:
logger.info("unhandled:{}".format(channel))
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
pserver_dev_handler, agg_compute_handler, agg_network_handler,\
agg_storage_handler, agg_undefined_handler
from o2ims.service.command import notify_handler, registration_handler,\
- notify_alarm_handler
+ notify_alarm_handler, purge_alarm_handler
from o2ims.service.event import ocloud_event, resource_event, \
resource_pool_event, alarm_event, dms_event, resource_type_event
notify_resourcepool_change],
events.AlarmEventChanged: [alarm_event.\
notify_alarm_event_change],
+ events.AlarmEventPurged: [alarm_event.\
+ notify_alarm_event_purge],
} # type: Dict[Type[events.Event], Callable]
commands.PubMessage2SMO: notify_handler.notify_change_to_smo,
commands.PubAlarm2SMO: notify_alarm_handler.notify_alarm_to_smo,
commands.Register2SMO: registration_handler.registry_to_smo,
+ commands.PurgeAlarmEvent: purge_alarm_handler.purge_alarm_event,
} # type: Dict[Type[commands.Command], Callable]
return (count, result)
def _update(self, alarm_event_record: alarm_obj.AlarmEventRecord):
- self.session.add(alarm_event_record)
+ self.session.merge(alarm_event_record)
def _delete(self, alarm_event_record_id):
self.session.query(alarm_obj.AlarmEventRecord).filter_by(
-# Copyright (C) 2022 Wind River Systems, Inc.
+# Copyright (C) 2022-2024 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, driver=None):
+ def __init__(self, uow: unit_of_work.AbstractUnitOfWork, 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):
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
-# Copyright (C) 2022 Wind River Systems, Inc.
+# Copyright (C) 2022-2024 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.
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
@dataclass
class UpdateAlarm(UpdateFaultObject):
parentid: str
+
+
+@dataclass
+class PurgeAlarmEvent(UpdateFaultObject):
+ data: AlarmEvent2SMO
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
id: str
notificationEventType: AlarmNotificationEventEnum
updatetime: datetime.now()
+
+
+@dataclass
+class AlarmEventPurged(Event):
+ id: str
+ notificationEventType: AlarmNotificationEventEnum
+ updatetime: datetime.now()
--- /dev/null
+# Copyright (C) 2024 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.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+
+from o2common.service.unit_of_work import AbstractUnitOfWork
+from o2common.adapter.notifications import AbstractNotifications
+
+from o2ims.adapter.clients.fault_client import StxEventClient
+from o2ims.domain import commands, alarm_obj
+
+from o2common.helper import o2logging
+logger = o2logging.get_logger(__name__)
+
+
+def purge_alarm_event(
+ cmd: commands.PubAlarm2SMO,
+ uow: AbstractUnitOfWork,
+ notifications: AbstractNotifications,
+):
+ logger.debug('In purge_alarm_event')
+ fault_client = StxEventClient(uow)
+ data = cmd.data
+ with uow:
+ alarm_event_record = uow.alarm_event_records.get(data.id)
+ alarm_id = json.loads(alarm_event_record.extensions).get('alarm_id')
+
+ events = fault_client.suppression_list(alarm_id)
+ for event_id in events:
+ event = fault_client.suppress(event_id.id)
+ alarm_event_record.hash = event.hash
+ alarm_event_record.extensions = json.dumps(event.filtered)
+ alarm_event_record.alarmChangedTime = event.updatetime.\
+ strftime("%Y-%m-%dT%H:%M:%S")
+ alarm_event_record.perceivedSeverity = \
+ alarm_obj.PerceivedSeverityEnum.CLEARED
+
+ uow.alarm_event_records.update(alarm_event_record)
+ break
+ uow.commit()
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
publish("AlarmEventChanged", event)
logger.debug("published Alarm Event Changed: {}".format(
event.id))
+
+
+def notify_alarm_event_purge(
+ event: events.AlarmEventPurged,
+ publish: Callable,
+):
+ logger.debug('In notify_alarm_event_purge')
+ publish("AlarmEventPurged", event)
+ logger.debug("published Alarm Event Purged: {}".format(
+ event.id))
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
raise NotFoundException(
"Alarm Event Record {} doesn't exist".format(alarmEventRecordId))
+ @api_monitoring_v1.doc('Patch Alarm Event Record Information')
+ @api_monitoring_v1.marshal_with(model)
+ def patch(self, alarmEventRecordId):
+ result = alarm_view.alarm_event_record_ack(alarmEventRecordId, bus.uow)
+ if result is not None:
+ return result
+ raise NotFoundException(
+ "Alarm Event Record {} doesn't exist".format(alarmEventRecordId))
+
# ---------- Alarm Subscriptions ---------- #
@api_monitoring_v1.route("/v1/alarmSubscriptions")
-# Copyright (C) 2021 Wind River Systems, Inc.
+# Copyright (C) 2021-2024 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.
# See the License for the specific language governing permissions and
# limitations under the License.
+from datetime import datetime
import uuid as uuid
-from o2common.service import unit_of_work
+from o2common.service import unit_of_work, messagebus
from o2common.views.view import gen_filter, check_filter
from o2common.views.pagination_view import Pagination
from o2common.views.route_exception import BadRequestException, \
NotFoundException
+from o2ims.domain import events
from o2ims.views.alarm_dto import SubscriptionDTO
-from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
+from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord, \
+ AlarmNotificationEventEnum
from o2common.helper import o2logging
# from o2common.config import config
return first.serialize() if first is not None else None
+def alarm_event_record_ack(alarmEventRecordId: str,
+ uow: unit_of_work.AbstractUnitOfWork):
+ with uow:
+ alarm_event_record = uow.alarm_event_records.get(alarmEventRecordId)
+ if alarm_event_record is None:
+ return None
+ elif alarm_event_record.alarmAcknowledged == 'true':
+ raise BadRequestException(
+ "Alarm Event Record {} has already been acknowledged."
+ .format(alarmEventRecordId))
+ alarm_event_record.alarmAcknowledged = True
+ alarm_event_record.alarmAcknowledgeTime = datetime.\
+ now().strftime("%Y-%m-%dT%H:%M:%S")
+ bus = messagebus.MessageBus.get_instance()
+ alarm_event_record.events.append(events.AlarmEventPurged(
+ id=alarm_event_record.alarmEventRecordId,
+ notificationEventType=AlarmNotificationEventEnum.ACKNOWLEDGE,
+ updatetime=alarm_event_record.alarmAcknowledgeTime))
+
+ uow.alarm_event_records.update(alarm_event_record)
+ uow.commit()
+
+ result = alarm_event_record.serialize()
+ _handle_events(bus)
+ return result
+
+
+def _handle_events(bus: messagebus.MessageBus):
+ # handle events
+ events = bus.uow.collect_new_events()
+ for event in events:
+ bus.handle(event)
+ return True
+
+
def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
pagination = Pagination(**kwargs)
query_kwargs = pagination.get_pagination()
-# Copyright (C) 2022 Wind River Systems, Inc.
+# Copyright (C) 2022-2024 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.
# pylint: disable=redefined-outer-name
import shutil
import subprocess
+import sys
import time
from pathlib import Path
from unittest.mock import MagicMock
from mock_alchemy.mocking import UnifiedAlchemyMagicMock
+# Mock cgtsclient, dcmanagerclient, fmclient
+modules_to_mock = [
+ 'cgtsclient',
+ 'cgtsclient.client',
+ 'cgtsclient.exc',
+ 'dcmanagerclient',
+ 'dcmanagerclient.api',
+ 'dcmanagerclient.api.client',
+ 'fmclient',
+ 'fmclient.client',
+ 'fmclient.common',
+ 'fmclient.common.exceptions'
+]
+
+for module_name in modules_to_mock:
+ sys.modules[module_name] = MagicMock()
+
from o2app.bootstrap import bootstrap
from o2ims.views import configure_namespace
from o2app.adapter import unit_of_work
Code Review / pti / o2.git / commitdiff