Skip to content

Send email notifications

Send email notifications for all tasks that have failed.

When validation tasks have failed, at most two separate emails will be sent. The first will contain all general failures of subscription validation workflows. The second email contains an overview of the prefix list validations that have failed.

gather_failed_tasks()

Gather all tasks that have failed.

Source code in gso/workflows/tasks/send_email_notifications.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
@step("Gather all tasks that have failed")
def gather_failed_tasks() -> State:
    """Gather all tasks that have failed."""
    failed_prefix_list_tasks = get_suspended_tasks_by_workflow_name("validate_prefix_list")
    all_other_tasks = list(set(get_failed_tasks()) - set(failed_prefix_list_tasks))

    return {
        "failed_tasks": [
            {"process_id": failure.process_id, "last_step": failure.last_step, "failed_reason": failure.failed_reason}
            for failure in all_other_tasks
        ],
        "failed_prefix_list_checks": [{"process_id": failure.process_id} for failure in failed_prefix_list_tasks],
    }

send_email_notification(failed_tasks)

Send out an email notification for all tasks that have failed.

Source code in gso/workflows/tasks/send_email_notifications.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@step("Send notification email for failed tasks")
def send_email_notification(failed_tasks: list[dict]) -> None:
    """Send out an email notification for all tasks that have failed."""
    general_settings = load_oss_params().GENERAL
    all_alerts = ""
    for failure in failed_tasks:
        failed_task_url = f"{general_settings.public_hostname}/workflows/{failure["process_id"]}"
        failed_subscription = get_subscription_by_process_id(failure["process_id"])
        all_alerts = f"{all_alerts}------\n\n"
        if failed_subscription:
            all_alerts = (
                f"{all_alerts}Product name: {failed_subscription.product.name}\n"
                f"Description: {failed_subscription.description}\n"
            )
        all_alerts = (
            f'{all_alerts}The step "{failure["last_step"]}" failed for the following reason: '
            f'"{failure["failed_reason"]}".\n\nPlease inspect the full workflow at the following link: '
            f'{failed_task_url}.\n\n'
        )

    send_mail(
        f"GAP {general_settings.environment} environment - One or more tasks have failed!",
        (
            f"Please check the following tasks in GAP which have failed.\n\n{all_alerts}------"
            f"\n\nRegards, the GÉANT Automation Platform.\n\n"
        ),
    )

send_prefix_list_email_notification(failed_prefix_list_checks)

Send out an email notification for all prefix list validation tasks that have failed.

Source code in gso/workflows/tasks/send_email_notifications.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@step("Send notification emails for failed prefix list tasks")
def send_prefix_list_email_notification(failed_prefix_list_checks: list[dict]) -> None:
    """Send out an email notification for all prefix list validation tasks that have failed."""
    general_settings = load_oss_params().GENERAL
    all_alerts = ""
    for failure in failed_prefix_list_checks:
        failed_task_url = f"{general_settings.public_hostname}/workflows/{failure["process_id"]}"
        failed_subscription = get_subscription_by_process_id(failure["process_id"])
        all_alerts = f"{all_alerts}------\n\n"
        if failed_subscription:
            all_alerts = (
                f"{all_alerts}Product name: {failed_subscription.product.name}\n"
                f"Description: {failed_subscription.description}\n"
            )
        all_alerts = (
            f"{all_alerts}This prefix list has drifted!\n\n"
            f"Please redeploy this prefix list at the following link: {failed_task_url}.\n\n"
        )

    send_mail(
        f"GAP {general_settings.environment} environment - One or more prefix lists have diverged!",
        (
            f"Please check the following prefix lists in GAP which have diverged.\n\n{all_alerts}------"
            f"\n\nRegards, the GÉANT Automation Platform.\n\n"
        ),
    )

task_send_email_notifications()

Gather all failed tasks, and send an email notification if needed.

Source code in gso/workflows/tasks/send_email_notifications.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@workflow("Send email notifications for all failed tasks", target=Target.SYSTEM)
def task_send_email_notifications() -> StepList:
    """Gather all failed tasks, and send an email notification if needed."""
    tasks_have_failed = conditional(lambda state: len(state["failed_tasks"]) > 0)
    prefix_tasks_have_failed = conditional(lambda state: len(state["failed_prefix_list_checks"]) > 0)

    return (
        init
        >> gather_failed_tasks
        >> tasks_have_failed(send_email_notification)
        >> prefix_tasks_have_failed(send_prefix_list_email_notification)
        >> done
    )