[GH-1420] Notification Improvements Stage 1

backend/dataall/base/feature_toggle_checker.py

@@ -3,11 +3,14 @@
 """
 
 import functools
+import logging
 from typing import List, Any, Optional, Callable
 
 from dataall.base.config import config
 from dataall.base.utils.decorator_utls import process_func
 
+log = logging.getLogger(__name__)
+
 
 def is_feature_enabled(config_property: str):
     def decorator(f):
@@ -56,3 +59,20 @@ def decorated(*args, **kwargs):
         return fn_decorator(decorated)
 
     return decorator
+
+
+def is_config_active(config_property: str, default_value: Any):
+    def decorator(f):
+        fn, fn_decorator = process_func(f)
+
+        @functools.wraps(fn)
+        def decorated(*args, **kwargs):
+            value = config.get_property(config_property, default_value)
+            if not value:
+                log.info(f'Config - {config_property} is inactive')
+                return
+            return fn(*args, **kwargs)
+
+        return fn_decorator(decorated)
+
+    return decorator

backend/dataall/core/groups/db/constants.py

@@ -0,0 +1,2 @@
+class DataallGroups:
+    admin = 'DAAdministrators'

backend/dataall/modules/catalog/tasks/catalog_indexer_task.py

@@ -8,6 +8,7 @@
 from dataall.base.db import get_engine
 from dataall.base.loader import load_modules, ImportMode
 from dataall.base.utils.alarm_service import AlarmService
+from dataall.modules.notifications.services.admin_notifications import AdminNotificationService
 
 log = logging.getLogger(__name__)
 
@@ -31,7 +32,14 @@ def index_objects(cls, engine, with_deletes='False'):
                     CatalogIndexerTask._delete_old_objects(indexed_object_uris)
                 return len(indexed_object_uris)
         except Exception as e:
+            error_log = f'Error occurred while indexing objects during the cataloging task. Exception: {e}'
+            log.error(error_log)
             AlarmService().trigger_catalog_indexing_failure_alarm(error=str(e))
+            AdminNotificationService().notify_admins_with_error_log(
+                process_error='Exception occurred during cataloging task',
+                error_logs=[error_log],
+                process_name=cls.__name__,
+            )
             raise e
 
     @classmethod

backend/dataall/modules/notifications/services/admin_notifications.py

@@ -0,0 +1,35 @@
+import logging
+from typing import List
+
+from dataall.base.feature_toggle_checker import is_config_active
+from dataall.core.groups.db.constants import DataallGroups
+from dataall.modules.notifications.services.ses_email_notification_service import SESEmailNotificationService
+
+log = logging.getLogger(__name__)
+
+
+class AdminNotificationService:
+    """
+    Send email notifications to Admin Group i.e. DAAdministrators in data.all
+    Args -
+        1. process_error - string describing in short the error / exception details
+        2. error_logs - List of all the exception error logs
+        3. process_name - Code where the exception occurred. Example, inside an ECS task like cataloging task, etc or inside a graphql service
+    """
+
+    @staticmethod
+    @is_config_active(
+        config_property='modules.datasets_base.features.share_notifications.email.parameters.admin_notifications',
+        default_value=False,
+    )
+    def notify_admins_with_error_log(process_error: str, error_logs: List[str], process_name: str = ''):
+        subject = f'Data.all alert | Attention Required | Failure in : {process_name}'
+        email_message = f"""
+            Following error occurred - <br><br> {process_error} 
+        """
+        email_message += '<br><br>'.join(error_logs)
+        email_message += '<br><br> Please check the logs in cloudwatch for more details'
+
+        SESEmailNotificationService.create_and_send_email_notifications(
+            subject=subject, msg=email_message, recipient_groups_list=[DataallGroups.admin]
+        )

backend/dataall/modules/notifications/services/ses_email_notification_service.py

@@ -1,9 +1,10 @@
 # Email Notification Provider implements the email notification service abstract method
 import logging
 
-from dataall.base.aws.cognito import Cognito
 from dataall.base.aws.ses import Ses
+from dataall.base.config import config
 from dataall.base.services.service_provider_factory import ServiceProviderFactory
+from dataall.core.groups.db.constants import DataallGroups
 from dataall.modules.notifications.services.base_email_notification_service import BaseEmailNotificationService
 
 log = logging.getLogger(__name__)
@@ -37,19 +38,29 @@ def send_email_task(subject, message, recipient_groups_list, recipient_email_lis
         email_provider = SESEmailNotificationService.get_email_provider_instance(
             recipient_groups_list, recipient_email_list
         )
+        identityProvider = ServiceProviderFactory.get_service_provider_instance()
         try:
-            identityProvider = ServiceProviderFactory.get_service_provider_instance()
-
-            email_ids_to_send_emails = email_provider.get_email_ids_from_groupList(
-                email_provider.recipient_group_list, identityProvider
-            )
+            email_ids_to_send_emails = set()
+            if len(recipient_groups_list) > 0:
+                email_ids_to_send_emails = email_provider.get_email_ids_from_groupList(
+                    email_provider.recipient_group_list, identityProvider
+                )
 
             if len(recipient_email_list) > 0:
                 email_ids_to_send_emails.update(recipient_email_list)
 
             SESEmailNotificationService.send_email_to_users(email_ids_to_send_emails, email_provider, message, subject)
 
         except Exception as e:
+            email_ids_to_send_emails = email_provider.get_email_ids_from_groupList(
+                [DataallGroups.admin], identityProvider
+            )
+            SESEmailNotificationService.send_email_to_users(
+                email_ids_to_send_emails,
+                email_provider,
+                f'Error sending email due to: {e}',
+                'Data.all alert | Attention Required | Failure in: Email Notification Service',
+            )
             raise e
         else:
             return True
@@ -60,3 +71,29 @@ def send_email_to_users(email_list, email_provider, message, subject):
         # https://aws.amazon.com/blogs/messaging-and-targeting/how-to-send-messages-to-multiple-recipients-with-amazon-simple-email-service-ses/
         for emailId in email_list:
             email_provider.send_email([emailId], message, subject)
+
+    @staticmethod
+    def create_and_send_email_notifications(subject, msg, recipient_groups_list=None, recipient_email_ids=None):
+        """
+        Method to directly send email notification instead of creating an SQS Task
+        This approach is used while sending email notifications in an ECS task ( e.g. persistent email reminder task, share expiration task, etc )
+        Emails send to groups mentioned in recipient_groups_list and / or emails mentioned in recipient_email_ids
+        """
+        if recipient_groups_list is None:
+            recipient_groups_list = []
+        if recipient_email_ids is None:
+            recipient_email_ids = []
+
+        if share_notification_config := config.get_property(
+            'modules.datasets_base.features.share_notifications', default=None
+        ):
+            for share_notification_config_type, n_config in share_notification_config.items():
+                if n_config.get('active', False):
+                    if share_notification_config_type == 'email':
+                        SESEmailNotificationService.send_email_task(
+                            subject, msg, recipient_groups_list, recipient_email_ids
+                        )
+                else:
+                    log.info(f'Notification type : {share_notification_config_type} is not active')
+        else:
+            log.info('Notifications are not active')