Truncate traceback for failure email (#3086)

Co-authored-by: Dillon Stadther <dlstadther+github@gmail.com>

doc/configuration.rst

@@ -437,6 +437,9 @@ sender
   User name in from field of error e-mails.
   Default value: luigi-client@<server_name>
 
+traceback_max_length
+  Maximum length for traceback included in error email. Default is 5000.
+
 
 [batch_notifier]
 ----------------

luigi/notifications.py

@@ -84,6 +84,9 @@ class email(luigi.Config):
         default='',
         config_path=dict(section='core', name='error-email'),
         description='Address to send error e-mails to')
+    traceback_max_length = luigi.parameter.IntParameter(
+        default=5000,
+        description='Max length for error traceback')
     sender = luigi.parameter.Parameter(
         default=DEFAULT_CLIENT_EMAIL,
         config_path=dict(section='core', name='email-sender'),
@@ -373,6 +376,11 @@ def format_task_error(headline, task, command, formatted_exception=None):
     :return: message body
     """
 
+    if formatted_exception:
+        if len(formatted_exception) > email().traceback_max_length:
+            truncated_exception = formatted_exception[:email().traceback_max_length]
+            formatted_exception = f"{truncated_exception}...Traceback exceeds max length and has been truncated."
+
     if formatted_exception:
         formatted_exception = wrap_traceback(formatted_exception)
     else:

test/worker_test.py

@@ -1324,6 +1324,17 @@ def run(self):
         self.assertEqual(1, len(emails))
         self.assertTrue(emails[0].find("Luigi: %s FAILED" % (a,)) != -1)
 
+    @email_patch
+    def test_run_error_long_traceback(self, emails):
+        class A(luigi.Task):
+            def run(self):
+                raise Exception("b0rk"*10500)
+
+        a = A()
+        luigi.build([a], workers=1, local_scheduler=True)
+        self.assertTrue(len(emails[0]) < 10000)
+        self.assertTrue(emails[0].find("Traceback exceeds max length and has been truncated"))
+
     @with_config({'batch_email': {'email_interval': '0'}, 'worker': {'send_failure_email': 'False'}})
     @email_patch
     def test_run_error_email_batch(self, emails):