Luigi: Success emails - notifications

[pmajmudar]

Hi,

I was wondering if you'd considered adding a success email notification (as well as the error notification)?

Frequently with a long running job its useful to receive an email if the luigi.run() returns sucessfully (all tasks completed).

Is this something you'd consider?

Prash

[Tarrasch]

Hmm, I have no strong opinion on this. If you think it's useful for you real-world scenario. I guess I would be happy too see a PR. :)

[gpoulin]

Since you probably don't want an email for every tasks (correct me if I'm wrong), the simplest is to create a task that require the task that you want to receive an email for and simply send the email in the run method.

Personally, I have a daily task that require all needed tasks for the day and send email with general stats about the ran jobs.

[pmajmudar]

@gpoulin Yes, I could create a task for every Job flow that "requires" the previous task - but that felt a bit hacky. It would mean I'd have to have this extra task every time (we have basically done this so far to send the success email). We have a number of different jobs we want to run, at different frequencies (some daily, weekly, monthly). It'd be nice not to have to write this extra boilerplate.

@Tarrasch I was going to suggest adding some code to the luigi.run method and an extra option in the email config.

If the run method is successful and if the email success notification is set, it would send a "success" email.

How does that sound? I'll have a look at a PR

Prash

[dlstadther]

@pmajmudar There are two other options here for task notification:

• luigi-monitor
• hubot-luigi

Both of which integrate with Slack. I use luigi-monitor because it was super easy to setup. But hubot-luigi appears to contain more functionality.

If you're still trying to setup emails at the end of completed tasks, one of these projects may be a different solution.

[kgbaird]

👍 for this feature request. Our team currently has a fairly long running pipeline. We would like to be able to get notification when the pipeline has completed. We are currently proceeding with our own internal solution. It would be nice if this feature were available.

[kgbaird]

I just completed work to implement a solution for this internally at my company that I thought I would share. Our approach was to utilize the on_success trigger of Task with a method whose side effect was sending an e-mail. We too have several different pipelines running around, so we made a central function that could be applied to each task we wanted success e-mail sent when completed (typically the last task of a given pipeline) This approach only required and additional line of boilerplate to tasks that wanted success e-mail, which seemed reasonable enough. Here is a example of the code we came up with:

Using python 3.5.2 and luigi 2.3.3.

Our send e-mail function (pretty much just spews information/attributes about the completed task):

# Intentionally verbose and monolithic for demonstration purposes

def send_success_email(task: luigi.Task) -> None:
    """Notifies users of completed task. Intended to be added to luigi.Task.on_success"""
    local_domain_name = "mycompany.com"
    local_smtp = ".".join(("smtp", local_domain_name)) # Our company has its own SMTP server
    default_email_address = ["@".join((os.environ["USERNAME"], local_domain_name))] # E-mail addresses are [email protected] locally. Pick the user running it, if none provided
    luigi_config_singleton = luigi.configuration.LuigiConfigParser.instance()
    # Be sure to scrape the config file for information before providing a default
    to_addrs = luigi_config_singleton.get(
        "email",
        "receiver",
        default=default_email_address
        ).split(
            ","
            )
    name_task_class = task.task_family

    subject_prefix = luigi_config_singleton.get(
        "email",
        "prefix",
        default=None,
        )
    if subject_prefix is None:
        subject = list()
    else:
        subject = [subject_prefix + ":"]
    subject.extend([
        name_task_class,
        "ran successfully on",
        os.environ["COMPUTERNAME"],
        ])

    msg = EmailMessage()
    msg['Subject'] = " ".join(subject)
    msg['From'] = luigi_config_singleton.get(
        "email",
        "sender",
        default="@".join((os.environ["COMPUTERNAME"], local_domain_name))
        )
    msg["To"] = ",".join(to_addrs)
    # Prepare the body of the e-mail. Could consider more elegant HTML preparation tooling rather than somewhat messy string building
    # Start with a basic statement of what/when
    body = "<p>\n"
    body += " ".join((
        name_task_class,
        "task finished successfully at",
        datetime.datetime.now().strftime("%H:%M:%S on %Y-%m-%d"),
        ))
    body += "\n</p>\n"
    # Then start mining task attributes, pick some off, and dump them to a table
    body += "<table align=\"left\" border=1>\n"
    body += "\n".join([
        "<tr><td>{}</td><td>{}</td></tr>".format(key, value)
        for key, value in (
            ("Task ID", task.task_id),
            ("Task Class Name", name_task_class),
            ("Computer Name", os.environ["COMPUTERNAME"]),
            ("User Name", os.environ["USERNAME"]),
            ("Parameter(s)", task.get_params()),
            ("Requirement(s)", task.requires()),
            ("Input(s)", task.input()),
            ("Output(s)", task.output()),
            )
        ])
    body += "\n</table>\n"
    msg.set_content(
        body,
        subtype="html",
        )

    # Send the e-mail
    with smtplib.SMTP(
        luigi_config_singleton.get(
            "smtp",
            "host",
            default=local_smtp,
            ),
        luigi_config_singleton.get(
            "smtp",
            "port",
            default="25", # Local default SMTP port
            )
        ) as smtp:
        smtp.send_message(msg)
    return None

An example task:

class MyTask(luigi.Task):
    """An example task to demonstrate success e-mail integration design pattern"""
    def requires(self):
        """Task inputs"""
        pass # Define inputs here

    def run(self):
        """Do the actual task"""
        pass # Define the task here

    def output(self):
        """Task outputs"""
        pass # Define outputs here


MyTask.event_handler(luigi.Event.SUCCESS)(send_success_email) # Add our e-mail function to the on_success event

I hope this helps!

[dlstadther]

Closing as is a feature request and not an issue

[letroot]

@kgbaird Or you could just call
luigi.notifications.send_error_email(subject, message)
and customize the subject and message to be success-specific. Works for me!