gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge branch 'main' into fix/SIENTIAPDE-2013

Browse Source
This commit is contained in:
Eduardo Rios
2026-09-17 11:48:58 -03:00
committed by GitHub
parent d46bd55cb4 5cc8015fc7
commit 4ece5cf9e3
9 changed files with 560 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
orchestrator/activities/formatters.py
View File

@@ -824,6 +824,8 @@ class Formatters(SientiaMonitoring):
- group_name (str): Name of the receiver group
- contents (list): Content types to include (must contain "reports")
- ignore (list, optional): Notification IDs to exclude
- levels (list, optional): Notification levels to include
- ignore_models_id (list, optional): Model IDs to exclude
Returns:
dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
@@ -847,6 +849,8 @@ class Formatters(SientiaMonitoring):
already_added_keys = []
ignore_list = receiver_group.get('ignore', [])
levels_list = receiver_group.get('levels')
ignore_models_id_list = receiver_group.get('ignore_models_id') or []
for notification in notification_package:
alert_type = 'reports'
@@ -859,6 +863,11 @@ class Formatters(SientiaMonitoring):
alert_type in receiver_group['contents']
and notification_id not in ignore_list
and key not in already_added_keys
and (not levels_list or notification.get('level') in levels_list)
and (
not ignore_models_id_list
or notification.get('model_id') not in ignore_models_id_list
)
):
receiver_groups[group_name]['notifications'].append(notification)
already_added_keys.append(key)

142
orchestrator/activities/slot_manager.py
View File

@@ -396,6 +396,8 @@ class SlotManager(SientiaMonitoring):
- group_name (str): Name of the receiver group
- contents (list): Alert types to include (core_alerts, persistent_alerts)
- ignore (list, optional): Notification IDs to exclude
- levels (list, optional): Notification levels to include
- ignore_models_id (list, optional): Model IDs to exclude
- notification_ttl (int): Seconds before considering notification persistent
Returns:
@@ -415,48 +417,118 @@ class SlotManager(SientiaMonitoring):
for receiver_group in sending_configs:
group_name = receiver_group['group_name']
receiver_groups[group_name] = {**receiver_group, 'notifications': []}
receiver_groups[group_name]['notifications'] = []
already_added_keys = []
ignore_list = receiver_group.get('ignore', [])
for notification in notification_package:
alert_type = 'do_nothing'
notification_id = notification['notification_id']
# Check if notification was recently sent
key = f'{notification["trigger"]}:{notification_id}'
last_sent = self.redis_repository.get(key)
if last_sent is None:
alert_type = 'core_alerts'
else:
last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ)
# Check if "notification_ttl" seconds has passed since last sent
if (now() - last_sent) > timedelta(seconds=notification_ttl):
alert_type = 'persistent_alerts'
# Check if this group must be notified
if (
alert_type in receiver_group['contents']
and notification_id not in ignore_list
and key not in already_added_keys
):
receiver_groups[group_name]['notifications'].append(notification)
already_added_keys.append(key)
notifications = self._filter_group_notifications(
receiver_group, notification_package, notification_ttl
)
receiver_groups[group_name] = {**receiver_group, 'notifications': notifications}
# Remove groups with no notifications
receiver_groups = {
return {
group_name: group
for group_name, group in receiver_groups.items()
if group['notifications']
}
return receiver_groups
def _filter_group_notifications(
self,
receiver_group: dict[str, Any],
notification_package: list[dict[str, Any]],
notification_ttl: int,
) -> list[dict[str, Any]]:
"""
Select the notifications this receiver group should get.
Args:
receiver_group (dict[str, Any]): The receiver group configuration.
notification_package (list[dict[str, Any]]): Candidate notifications.
notification_ttl (int): Seconds before considering a notification persistent.
Returns:
list[dict[str, Any]]: Notifications matching this group's filters.
"""
already_added_keys: list[str] = []
ignore_list = receiver_group.get('ignore', [])
levels_list = receiver_group.get('levels')
ignore_models_id_list = receiver_group.get('ignore_models_id') or []
notifications = []
for notification in notification_package:
notification_id = notification['notification_id']
key = f'{notification["trigger"]}:{notification_id}'
alert_type = self._determine_alert_type(key, notification_ttl)
if (
alert_type in receiver_group['contents']
and self._check_keys_and_notif_id(
notification_id, ignore_list, key, already_added_keys
)
and self._check_levels_and_ignore_models_id(
notification, levels_list, ignore_models_id_list
)
):
notifications.append(notification)
already_added_keys.append(key)
return notifications
def _determine_alert_type(self, key: str, notification_ttl: int) -> str:
"""
Classify a notification as new, persistent, or not-yet-due.
Args:
key (str): The Redis cache key for this notification.
notification_ttl (int): Seconds before considering a notification persistent.
Returns:
str: 'core_alerts', 'persistent_alerts', or 'do_nothing'.
"""
last_sent = self.redis_repository.get(key)
if last_sent is None:
return 'core_alerts'
last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ)
if (now() - last_sent) > timedelta(seconds=notification_ttl):
return 'persistent_alerts'
return 'do_nothing'
@staticmethod
def _check_levels_and_ignore_models_id(notification, levels_list, ignore_models_id_list):
"""
Check if the notification's level is in the levels_list and its model_id is not in the ignore_models_id_list.
Args:
notification (dict): The notification to check.
levels_list (list, optional): List of levels to include. If None, all levels are included.
ignore_models_id_list (list, optional): List of model_ids to exclude. If None, no models are excluded.
Returns:
bool: True if the notification passes the checks, False otherwise.
"""
level_check = not levels_list or notification.get('level') in levels_list
model_id_check = (
not ignore_models_id_list or notification.get('model_id') not in ignore_models_id_list
)
return level_check and model_id_check
@staticmethod
def _check_keys_and_notif_id(
notification_id, ignore_list, key, already_added_keys: list[str]
) -> bool:
"""
Check if the key has already been added to the list of already added keys.
Args:
notification_id (str): The ID of the notification.
ignore_list (list): List of notification IDs to ignore.
key (str): The key to check.
already_added_keys (list[str]): List of keys that have already been added.
Returns:
bool: True if the key has not been added yet, False otherwise.
"""
return notification_id not in ignore_list and key not in already_added_keys
@activity.defn(name='store_notification_cache')
def store_notification_cache(self, input_data: dict[str, Any]) -> None: