Merge pull request #62 from google/mail_notifications

Mail notifications

Author: antoniolmm

cloud_config/generate_megalista_token.py

@@ -34,7 +34,8 @@
           'https://www.googleapis.com/auth/dfatrafficking',
           'https://www.googleapis.com/auth/ddmconversions',
           "https://www.googleapis.com/auth/analytics.edit",
-          'https://www.googleapis.com/auth/spreadsheets.readonly']
+          'https://www.googleapis.com/auth/spreadsheets.readonly',
+          'https://www.googleapis.com/auth/gmail.send']
 
 # The redirect URI set for the given Client ID. The redirect URI for Client ID
 # generated for an installed application will always have this value.

megalista_dataflow/error/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2022 Google LLC
+#
+# 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
+#
+#     https://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.

megalista_dataflow/error/error_handling.py

@@ -0,0 +1,185 @@
+# Copyright 2022 Google LLC
+#
+# 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
+#
+#     https://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 base64
+import logging
+from email.mime.text import MIMEText
+from typing import Iterable
+
+from apache_beam.options.value_provider import ValueProvider
+from google.oauth2.credentials import Credentials
+from googleapiclient.discovery import build
+
+from models.execution import DestinationType, Execution
+from models.oauth_credentials import OAuthCredentials
+
+
+class Error:
+  """
+    Holds errors executions and respective error messages
+  """
+
+  def __init__(self, execution: Execution, error_message):
+    self._execution = execution
+    self._error_message = error_message
+
+  @property
+  def execution(self):
+    return self._execution
+
+  @property
+  def error_message(self):
+    return self._error_message
+
+  def __str__(self):
+    return f'Execution: {self.execution}. Error message: {self.error_message}'
+
+  def __eq__(self, other):
+    return self.execution == other.execution and self.error_message == other.error_message
+
+  def __hash__(self):
+    return hash((self.execution, self.error_message))
+
+
+class ErrorNotifier:
+  """
+    Abstract class to notify errors. The mean is defined by the implementation.
+  """
+
+  def notify(self, destination_type: DestinationType, errors: Iterable[Error]):
+    raise NotImplementedError()
+
+
+class GmailNotifier(ErrorNotifier):
+  """
+    Notify errors sending emails through the gMail API. Uses the main application credentials.
+  """
+
+  def __init__(self, should_notify: ValueProvider, oauth_credentials: OAuthCredentials,
+               email_destinations: ValueProvider):
+    self._oauth_credentials = oauth_credentials
+    self._email_destinations = email_destinations
+    self._should_notify_param = should_notify
+    self._parsed_emails = None
+
+  def _get_gmail_service(self):
+    credentials = Credentials(
+      token=self._oauth_credentials.get_access_token(),
+      refresh_token=self._oauth_credentials.get_refresh_token(),
+      client_id=self._oauth_credentials.get_client_id(),
+      client_secret=self._oauth_credentials.get_client_secret(),
+      token_uri='https://accounts.google.com/o/oauth2/token',
+      scopes=[
+        'https://www.googleapis.com/auth/gmail.send'])
+
+    return build('gmail', 'v1', credentials=credentials)
+
+  def _should_notify(self):
+    if not self._should_notify_param.get():
+      return False
+    should_notify = self._should_notify_param.get()
+    return should_notify.lower() == 'true'
+
+  def notify(self, destination_type: DestinationType, errors: Iterable[Error]):
+    logger = logging.getLogger('megalista.GmailNotifier')
+    if not self._should_notify():
+      logger.info(f'Skipping sending emails notifying of errors: {", ".join(map(str, errors))}')
+      return
+
+    body = self._build_email_body(destination_type, errors)
+
+    gmail_service = self._get_gmail_service()
+
+    message = MIMEText(body, 'html')
+    message['to'] = ','.join(self.email_destinations)
+    message['from'] = 'me'
+    message['subject'] = f'[Action Required] Megalista error detected - {destination_type.name}'
+    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
+
+    logger.info(
+      f'Sending error email to {", ".join(self.email_destinations)} regarding the conector {destination_type.name}')
+
+    try:
+      gmail_service.users().messages().send(userId='me', body={'raw': raw}).execute()
+    except:
+      logger.error(
+        f'Error sending email to {", ".join(self.email_destinations)} regarding the conector {destination_type.name}')
+      raise
+
+  @property
+  def email_destinations(self) -> Iterable[str]:
+    if self._parsed_emails:
+      return self._parsed_emails
+
+    self._parsed_emails = list(map(lambda email: email.strip(), self._email_destinations.get().split(',')))
+    return self._parsed_emails
+
+  def _build_email_body(self, destination_type: DestinationType, errors: Iterable[Error]):
+    body = f'''<h3>Hello, Megalista user.</h3>
+           This is an error summary for the destination: <b>{destination_type.name}</b>.'''
+
+    body += '''<p>
+    <b>Errors list:</b>
+    <ul>'''
+
+    for error in errors:
+      body += f'''
+      <li>Error for source <b>"{error.execution.source.source_name}"</b> and destination 
+      <b>"{error.execution.destination.destination_name}"</b>: {error.error_message}</b>
+      </li>'''
+
+    body += '</ul>'
+
+    return body
+
+
+class ErrorHandler:
+  """
+    Accumulate errors and notify them.
+    Only record one message by Execution.
+    Notification details are decided by the ErrorNotifier received in the constructor.
+  """
+
+  def __init__(self, destination_type: DestinationType, error_notifier: ErrorNotifier):
+    self._destination_type = destination_type
+    self._error_notifier = error_notifier
+    self._errors = {}
+
+  def add_error(self, execution: Execution, error_message: str):
+    """
+      Add an error to be logged.
+      Only record one error per Execution, so the output message isn't too long.
+    """
+
+    if execution.destination.destination_type != self._destination_type:
+      raise ValueError(
+        f'Received a error of destination type: {execution.destination.destination_type}'
+        f' but this error handler is initialized with {self._destination_type} destination type')
+
+    error = Error(execution, error_message)
+
+    self._errors[error.execution] = error
+
+  @property
+  def errors(self):
+    return self._errors.copy()
+
+  def notify_errors(self):
+    """
+      Send the errors accumulated by email.
+      Does nothing if no errors were received.
+    """
+    if len(self.errors) == 0:
+      return
+
+    self._error_notifier.notify(self._destination_type, self.errors.values())

megalista_dataflow/error/error_handling_test.py

@@ -0,0 +1,132 @@
+# Copyright 2022 Google LLC
+#
+# 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
+#
+#     https://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.
+from typing import Iterable
+
+import pytest
+from apache_beam.options.value_provider import StaticValueProvider
+
+from error.error_handling import ErrorHandler, Error, GmailNotifier, ErrorNotifier
+from models.execution import DestinationType, Execution, AccountConfig, Source, SourceType, Destination
+from models.oauth_credentials import OAuthCredentials
+
+
+class MockErrorNotifier(ErrorNotifier):
+  def __init__(self):
+    self.were_errors_sent = False
+    self.errors_sent = {}
+    self.destination_type = None
+
+  def notify(self, destination_type: DestinationType, errors: Iterable[Error]):
+    self.were_errors_sent = True
+    self.errors_sent = {error.execution: error.error_message for error in errors}
+    self.destination_type = destination_type
+
+
+# ErrorHandler tests
+
+def create_execution(source_name, destination_name):
+  account_config = AccountConfig('', False, '', '', '')
+  source = Source(source_name, SourceType.BIG_QUERY, ['', ''])
+  destination = Destination(destination_name, DestinationType.ADS_OFFLINE_CONVERSION, [''])
+  return Execution(account_config, source, destination)
+
+
+def test_single_error_per_execution():
+  error_handler = ErrorHandler(DestinationType.ADS_OFFLINE_CONVERSION, MockErrorNotifier())
+
+  first_execution = create_execution('source1', 'destination1')
+  second_execution = create_execution('source1', 'destination2')
+
+  error_handler.add_error(first_execution, 'Error for first execution, fist input')
+  error_handler.add_error(first_execution, 'Error for first execution, second input')
+  error_handler.add_error(second_execution, 'Error for second execution, fist input')
+
+  returned_errors = error_handler.errors
+  assert len(returned_errors) == 2
+  assert returned_errors.keys() == {first_execution, second_execution}
+
+
+def test_destination_type_consistency():
+  error_handler = ErrorHandler(DestinationType.CM_OFFLINE_CONVERSION, MockErrorNotifier())
+  wrong_destination_type_execution = create_execution('source', 'destination')
+
+  with pytest.raises(ValueError):
+    error_handler.add_error(wrong_destination_type_execution, 'error message')
+
+
+def test_errors_sent_to_error_notifier():
+  mock_notifier = MockErrorNotifier()
+  error_handler = ErrorHandler(DestinationType.ADS_OFFLINE_CONVERSION, mock_notifier)
+
+  first_execution = create_execution('source1', 'destination1')
+  second_execution = create_execution('source1', 'destination2')
+
+  error_handler.add_error(first_execution, 'Error for first execution, fist input')
+  error_handler.add_error(second_execution, 'Error for second execution, fist input')
+
+  error_handler.notify_errors()
+
+  assert mock_notifier.were_errors_sent is True
+  assert mock_notifier.errors_sent == {first_execution: 'Error for first execution, fist input',
+                                       second_execution: 'Error for second execution, fist input'}
+  assert mock_notifier.destination_type == DestinationType.ADS_OFFLINE_CONVERSION
+
+
+def test_should_not_notify_when_empty_errors():
+  mock_notifier = MockErrorNotifier()
+  error_handler = ErrorHandler(DestinationType.ADS_OFFLINE_CONVERSION, mock_notifier)
+
+  error_handler.notify_errors()
+
+  assert mock_notifier.were_errors_sent is False
+
+
+# GmailNotifier tests
+
+def test_multiple_destinations_separated_by_comma():
+  first_email = '[email protected]'
+  second_email = '[email protected]'
+  third_email = '[email protected]'
+
+  credentials = OAuthCredentials('', '', '', '')
+  gmail_notifier = GmailNotifier(StaticValueProvider(str, 'true'), credentials,
+                                 StaticValueProvider(str, f'{first_email}, {second_email} ,{third_email}'))
+
+  emails = set(gmail_notifier.email_destinations)
+  assert len(emails) == 3
+  assert set(emails) == {first_email, third_email, second_email}
+
+
+def test_should_not_notify_when_param_is_false():
+  first_email = '[email protected]'
+  second_email = '[email protected]'
+  third_email = '[email protected]'
+
+  credentials = OAuthCredentials('', '', '', '')
+  gmail_notifier = GmailNotifier(StaticValueProvider(str, 'false'), credentials,
+                                 StaticValueProvider(str, f'{first_email}, {second_email} ,{third_email}'))
+
+  gmail_notifier.notify(DestinationType.ADS_OFFLINE_CONVERSION, [Error(create_execution('s', 'd'), 'error message')])
+
+
+def test_should_not_notify_when_param_is_empty():
+  first_email = '[email protected]'
+  second_email = '[email protected]'
+  third_email = '[email protected]'
+
+  credentials = OAuthCredentials('', '', '', '')
+  gmail_notifier = GmailNotifier(StaticValueProvider(str, None), credentials,
+                                 StaticValueProvider(str, f'{first_email}, {second_email} ,{third_email}'))
+
+  gmail_notifier.notify(DestinationType.ADS_OFFLINE_CONVERSION, [Error(create_execution('s', 'd'), 'error message')])