tasks/pipeline.py
def notify_failure(_, notification_type="merge", print_to_stdout=False):
    """
    Send failure notifications for the current pipeline. CI-only task.
    Use the --print-to-stdout option to test this locally, without sending
    real slack messages.
    """

    header = ""
    if notification_type == "merge":
        header = ":host-red: :merged: datadog-agent merge"
    elif notification_type == "deploy":
        header = ":host-red: :rocket: datadog-agent deploy"
    base = base_message(header)

    try:
        messages_to_send = generate_failure_messages(base)
    except Exception as e:
        buffer = io.StringIO()
        print(base, file=buffer)
        print("Found exception when generating notification:", file=buffer)
        traceback.print_exc(limit=-1, file=buffer)
        print("See the job log for the full exception traceback.", file=buffer)
        messages_to_send = {
            "@DataDog/agent-all": SlackMessage(buffer.getvalue()),
        }
        # Print traceback on job log
        print(e)
        traceback.print_exc()

    # Send messages
    for owner, message in messages_to_send.items():
        channel = GITHUB_SLACK_MAP.get(owner, "#datadog-agent-pipelines")
        if owner not in GITHUB_SLACK_MAP.keys():
            message.base_message += UNKNOWN_OWNER_TEMPLATE.format(owner=owner)
        message.coda = "If there is something wrong with the notification please contact #agent-platform"
        if print_to_stdout:
            print(f"Would send to {channel}:\n{str(message)}")
        else:
            send_slack_message(channel, str(message))

    Send failure notifications for the current pipeline. CI-only task.
    Use the --print-to-stdout option to test this locally, without sending
    real slack messages.
    
Send failure notifications for the current pipeline. CI-only task.
Use the --print-to-stdout option to test this locally, without sending
real slack messages.
Send failure notifications for the current pipeline . CI - only task . Use the -- print - to - stdout option to test this locally without sending real slack messages .
def notify_failure(_, notification_type="merge", print_to_stdout=False):
    header = ""
    if notification_type == "merge":
        header = ":host-red: :merged: datadog-agent merge"
    elif notification_type == "deploy":
        header = ":host-red: :rocket: datadog-agent deploy"
    base = base_message(header)
    try:
        messages_to_send = generate_failure_messages(base)
    except Exception as e:
        buffer = io.StringIO()
        print(base, file=buffer)
        print("Found exception when generating notification:", file=buffer)
        traceback.print_exc(limit=-1, file=buffer)
        print("See the job log for the full exception traceback.", file=buffer)
        messages_to_send = {
            "@DataDog/agent-all": SlackMessage(buffer.getvalue()),
        }
        print(e)
        traceback.print_exc()
    for owner, message in messages_to_send.items():
        channel = GITHUB_SLACK_MAP.get(owner, "#datadog-agent-pipelines")
        if owner not in GITHUB_SLACK_MAP.keys():
            message.base_message += UNKNOWN_OWNER_TEMPLATE.format(owner=owner)
        message.coda = "If there is something wrong with the notification please contact #agent-platform"
        if print_to_stdout:
            print(f"Would send to {channel}:\n{str(message)}")
        else:
            send_slack_message(channel, str(message))
def notify_failure ( _ , notification_type = "merge" , print_to_stdout = False ) : header = "" if notification_type == "merge" : header = ":host-red: :merged: datadog-agent merge" elif notification_type == "deploy" : header = ":host-red: :rocket: datadog-agent deploy" base = base_message ( header ) try : messages_to_send = generate_failure_messages ( base ) except Exception as e : buffer = io . StringIO ( ) print ( base , file = buffer ) print ( "Found exception when generating notification:" , file = buffer ) traceback . print_exc ( limit = - 1 , file = buffer ) print ( "See the job log for the full exception traceback." , file = buffer ) messages_to_send = { "@DataDog/agent-all" : SlackMessage ( buffer . getvalue ( ) ) , } print ( e ) traceback . print_exc ( ) for owner , message in messages_to_send . items ( ) : channel = GITHUB_SLACK_MAP . get ( owner , "#datadog-agent-pipelines" ) if owner not in GITHUB_SLACK_MAP . keys ( ) : message . base_message += UNKNOWN_OWNER_TEMPLATE . format ( owner = owner ) message . coda = "If there is something wrong with the notification please contact #agent-platform" if print_to_stdout : print ( f"Would send to {channel}:\n{str(message)}" ) else : send_slack_message ( channel , str ( message ) )
Send failure notifications for the current pipeline.
Send failure notifications for the current pipeline .
"""
    Send failure notifications for the current pipeline. CI-only task.
    Use the --print-to-stdout option to test this locally, without sending
    real slack messages.
    """ # Print traceback on job log # Send messages
_
notification_type
print_to_stdout