-# 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.
import redis
import json
+
from o2app import bootstrap
-from o2common.config import config
from o2common.adapter.notifications import SmoNotifications
+from o2common.config import config
+from o2common.helper import o2logging
from o2dms.domain import commands
from o2ims.domain import commands as imscmd
from o2ims.domain.subscription_obj import Message2SMO, RegistrationMessage
from o2ims.domain.alarm_obj import AlarmEvent2SMO
-from o2common.helper import o2logging
logger = o2logging.get_logger(__name__)
r = redis.Redis(**config.get_redis_host_and_port())
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))
+ cmd = imscmd.PurgeAlarmEvent(data=AlarmEvent2SMO(
+ id=data['id'], 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]
-# 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.
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.
def _set_stx_client(self):
self.driver.setFaultClient(self._pool_id)
+ def delete(self, id) -> alarmModel.FaultGenericModel:
+ return self.driver.deleteAlarm(id)
+
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.ALARM, self._alarmconverter(alarm))
+ def deleteAlarm(self, id) -> alarmModel.FaultGenericModel:
+ alarm = self.fmclient.alarm.delete(id)
+ logger.debug('delete alarm id ' + id + ':' + str(alarm.to_dict()))
+ return alarmModel.FaultGenericModel(
+ alarmModel.EventTypeEnum.ALARM, self._alarmconverter(alarm))
+
def getEventList(self, **filters) -> List[alarmModel.FaultGenericModel]:
events = self.fmclient.event_log.list(alarms=True, expand=True)
[logger.debug(
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 = []
+ query_as_array = []
+ events = self.fmclient.event_suppression.list(q=query_as_array)
+ 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
MANUAL = 'MANUAL'
+class AlarmEventRecordModifications(AgRoot):
+ def __init__(self, ack: bool = None,
+ clear: PerceivedSeverityEnum = None) -> None:
+ super().__init__()
+ self.alarmAcknowledged = ack
+ self.perceivedSeverity = clear
+
+
class AlarmDefinition(AgRoot, Serializer):
def __init__(self, id: str, name: str, change_type: AlarmChangeTypeEnum,
desc: str, prop_action: str, clearing_type: ClearingTypeEnum,
class AlarmEvent2SMO(Serializer):
def __init__(self, eventtype: AlarmNotificationEventEnum,
- id: str, ref: str, updatetime: str) -> None:
+ id: str, updatetime: str) -> None:
self.notificationEventType = eventtype
- self.objectRef = ref
self.id = id
self.updatetime = updatetime
-# 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.adapter.notifications import AbstractNotifications
+from o2common.config import conf
+from o2common.domain.filter import gen_orm_filter
+from o2common.helper import o2logging
+from o2common.service.unit_of_work import AbstractUnitOfWork
+from o2ims.adapter.clients.fault_client import StxAlarmClient
+from o2ims.domain import commands, alarm_obj
+logger = o2logging.get_logger(__name__)
+
+
+def purge_alarm_event(
+ cmd: commands.PubAlarm2SMO,
+ uow: AbstractUnitOfWork,
+ notifications: AbstractNotifications,
+):
+ """
+ Purges an alarm event and notifies relevant subscribers.
+
+ This method performs the following steps:
+ 1. Retrieves data from the command object and initializes the fault client.
+ 2. Uses the Unit of Work pattern to find and delete the corresponding
+ alarm event record.
+ 3. Updates the alarm event record's hash, extensions, changed time,
+ and perceived severity.
+ 4. Commits the changes to the database.
+ 5. Finds and processes all alarm subscriptions, deciding whether to
+ send notifications based on subscription filters.
+
+ Parameters:
+ - cmd (commands.PubAlarm2SMO): Command object containing the alarm
+ event data.
+ - uow (AbstractUnitOfWork): Unit of Work object for managing
+ database transactions.
+ - notifications (AbstractNotifications): Abstract notifications
+ object for sending notifications.
+
+ Exceptions:
+ - Any exceptions that might occur during database operations or
+ notification sending.
+ """
+ fault_client = StxAlarmClient(uow)
+ data = cmd.data
+ with uow:
+ alarm_event_record = uow.alarm_event_records.get(data.id)
+ alarm = fault_client.delete(alarm_event_record.alarmEventRecordId)
+ alarm_event_record.hash = alarm.hash
+ alarm_event_record.extensions = json.dumps(alarm.filtered)
+ alarm_event_record.alarmChangedTime = alarm.updatetime.\
+ strftime("%Y-%m-%dT%H:%M:%S")
+ alarm_event_record.perceivedSeverity = \
+ alarm_obj.PerceivedSeverityEnum.CLEARED
+
+ uow.alarm_event_records.update(alarm_event_record)
+
+ uow.commit()
+
+ alarm = uow.alarm_event_records.get(data.id)
+ subs = uow.alarm_subscriptions.list()
+ for sub in subs:
+ sub_data = sub.serialize()
+ logger.debug('Alarm Subscription: {}'.format(
+ sub_data['alarmSubscriptionId']))
+
+ if not sub_data.get('filter', None):
+ callback_smo(notifications, sub, data, alarm)
+ continue
+ try:
+ args = gen_orm_filter(alarm_obj.AlarmEventRecord,
+ sub_data['filter'])
+ except KeyError:
+ logger.warning(
+ 'Alarm Subscription {} filter {} has wrong attribute '
+ 'name or value. Ignore the filter'.format(
+ sub_data['alarmSubscriptionId'],
+ sub_data['filter']))
+ callback_smo(notifications, sub, data, alarm)
+ continue
+ args.append(alarm_obj.AlarmEventRecord.
+ alarmEventRecordId == data.id)
+ count, _ = uow.alarm_event_records.list_with_count(*args)
+ if count != 0:
+ logger.debug(
+ 'Alarm Event {} skip for subscription {} because of '
+ 'the filter.'
+ .format(data.id, sub_data['alarmSubscriptionId']))
+ continue
+ callback_smo(notifications, sub, data, alarm)
+
+
+def callback_smo(notifications: AbstractNotifications,
+ sub: alarm_obj.AlarmSubscription,
+ msg: alarm_obj.AlarmEvent2SMO,
+ alarm: alarm_obj.AlarmEventRecord):
+ sub_data = sub.serialize()
+ alarm_data = alarm.serialize()
+ callback = {
+ 'globalCloudID': conf.DEFAULT.ocloud_global_id,
+ 'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
+ 'notificationEventType': msg.notificationEventType,
+ 'objectRef': msg.objectRef,
+ 'alarmEventRecordId': alarm_data['alarmEventRecordId'],
+ 'resourceTypeID': alarm_data['resourceTypeId'],
+ 'resourceID': alarm_data['resourceId'],
+ 'alarmDefinitionID': alarm_data['alarmDefinitionId'],
+ 'probableCauseID': alarm_data['probableCauseId'],
+ 'alarmRaisedTime': alarm_data['alarmRaisedTime'],
+ 'alarmChangedTime': alarm_data['alarmChangedTime'],
+ 'alarmAcknowledgeTime': alarm_data['alarmAcknowledgeTime'],
+ 'alarmAcknowledged': alarm_data['alarmAcknowledged'],
+ 'perceivedSeverity': alarm_data['perceivedSeverity'],
+ 'extensions': json.loads(alarm_data['extensions'])
+ }
+ logger.info('callback URL: {}'.format(sub_data['callback']))
+ logger.debug('callback data: {}'.format(json.dumps(callback)))
+
+ return notifications.send(sub_data['callback'], callback)
-# 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.
event: events.AlarmEventChanged,
publish: Callable,
):
- logger.debug('In notify_alarm_event_change')
publish("AlarmEventChanged", event)
logger.debug("published Alarm Event Changed: {}".format(
event.id))
+
+
+def notify_alarm_event_purge(
+ event: events.AlarmEventPurged,
+ publish: Callable,
+):
+ publish("AlarmEventPurged", event)
+ logger.debug("published Alarm Event Purged: {}".format(
+ event.id))
# 'alarmAcknowledgeTime,alarmAcknowledged,extensions}'
)
+ alarm_event_record_patch = api_monitoring_v1.model(
+ "AlarmPatchDto",
+ {
+ 'alarmAcknowledged': fields.Boolean(
+ example=True,
+ description='Boolean value indication of a management ' +
+ 'system has acknowledged the alarm.'),
+ 'perceivedSeverity': fields.String(
+ example='5',
+ description='indicate that the alarm record is requested ' +
+ 'to be cleared. Only the value "5" for "CLEARED" is ' +
+ 'permitted in a request message content. ')
+ },
+ mask='{alarmAcknowledged,}'
+ )
+
class SubscriptionDTO:
-# 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.
from o2common.views.pagination_route import link_header, PAGE_PARAM
from o2common.views.route_exception import NotFoundException, \
BadRequestException
+from o2ims.domain.alarm_obj import PerceivedSeverityEnum
from o2ims.views import alarm_view
from o2ims.views.api_ns import api_ims_monitoring as api_monitoring_v1
from o2ims.views.alarm_dto import AlarmDTO, SubscriptionDTO, \
class AlarmGetRouter(Resource):
model = AlarmDTO.alarm_event_record_get
+ patch = AlarmDTO.alarm_event_record_patch
@api_monitoring_v1.doc('Get Alarm Event Record Information')
@api_monitoring_v1.marshal_with(model)
raise NotFoundException(
"Alarm Event Record {} doesn't exist".format(alarmEventRecordId))
+ @api_monitoring_v1.doc('Patch Alarm Event Record Information')
+ @api_monitoring_v1.expect(patch)
+ @api_monitoring_v1.marshal_with(patch)
+ def patch(self, alarmEventRecordId):
+ data = api_monitoring_v1.payload
+ ack_action = data.get('alarmAcknowledged', None)
+ clear_action = data.get('perceivedSeverity', None)
+
+ ack_is_none = ack_action is None
+ clear_is_none = clear_action is None
+ if (ack_is_none and clear_is_none) or (not ack_is_none and
+ not clear_is_none):
+ raise BadRequestException('Either "alarmAcknowledged" or '
+ '"perceivedSeverity" shall be included '
+ 'in a request, but not both.')
+ if ack_action:
+ result = alarm_view.alarm_event_record_ack(alarmEventRecordId,
+ bus.uow)
+ if result is not None:
+ return result
+ elif clear_action:
+ if clear_action != PerceivedSeverityEnum.CLEARED:
+ raise BadRequestException(
+ 'Only the value "5" for "CLEARED" is permitted of '
+ '"perceivedSeverity".')
+
+ result = alarm_view.alarm_event_record_clear(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, AlarmEventRecordModifications, \
+ PerceivedSeverityEnum
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)
+ # Check the record does not exist, return None. Otherwise, the
+ # acknowledge request will update the record even if it is
+ # acknowledged.
+ if alarm_event_record is None:
+ return None
+ alarm_event_record.alarmAcknowledged = True
+ alarm_event_record.alarmAcknowledgeTime = datetime.\
+ now().strftime("%Y-%m-%dT%H:%M:%S")
+ uow.alarm_event_records.update(alarm_event_record)
+ uow.commit()
+
+ result = AlarmEventRecordModifications(True)
+ return result
+
+
+def alarm_event_record_clear(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.perceivedSeverity == \
+ PerceivedSeverityEnum.CLEARED:
+ raise BadRequestException(
+ "Alarm Event Record {} has already been marked as CLEARED."
+ .format(alarmEventRecordId))
+ alarm_event_record.events.append(events.AlarmEventPurged(
+ id=alarm_event_record.alarmEventRecordId,
+ notificationEventType=AlarmNotificationEventEnum.CLEAR,
+ updatetime=alarm_event_record.alarmAcknowledgeTime))
+
+ uow.alarm_event_records.update(alarm_event_record)
+ uow.commit()
+
+ result = AlarmEventRecordModifications(
+ clear=PerceivedSeverityEnum.CLEARED)
+ _handle_events(messagebus.MessageBus.get_instance())
+ 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