add code to post silences to Alertmanager (team#41827)

    "fabric_tpa.libvirt",

    "fabric_tpa.reboot",

    "fabric_tpa.retire",

    "fabric_tpa.silence",

    "fabric_tpa.user",

]:

    try:

from datetime import datetime, timedelta

import logging

import os

from invoke import Context

import invoke.exceptions

from fabric import task

import requests

import requests.exceptions

@task(iterable=["matchers"])

def create(

    con: Context,

    matchers: str,

    comment: str,

    created_by: str | None = None,

    starts_at: str | None = None,

    ends_at: str | None = None,

    http_user: str = "tor-guest",

    http_password: str = "",

    url: str = "https://alertmanager.torproject.org",

):

    """create a silence in Alertmanager

    Matchers are a series of key=value strings.

    starts_at default to "now" and ends_at default to "an hour from

    now", otherwise should be specified as RFC 3339 (AKA ISO 8601)

    strings, e.g. 2024-10-16T15:10:11.484Z.

    """

    # set sane defaults

    if not starts_at or starts_at == "now":

        # start is "now" by default

        starts_at = datetime.utcnow().isoformat()

    if not ends_at:

        # ends in one hour by default

        ends_at = (datetime.utcnow() + timedelta(hours=1)).isoformat()

    if not created_by:

        # use the local username as a default for the creator

        created_by = os.getlogin()

    # parse a series of key=value pairs and add them to the matcher_objs object

    matcher_objs = []

    error = False

    for matcher in matchers:

        if "=" not in matcher:

            logging.error("badly specified match, missing equal (=) sign: %s", matcher)

            error = True

            continue

        name, value = matcher.split("=", maxsplit=1)

        matcher_objs.append(

            {

                "name": name,

                "value": value,

                "isRegex": True,

                "isEqual": True,

            }

        )

    if error:

        raise invoke.exceptions.Exit("errors while parsing matchers, aborting")

    # post the silence

    # documentation of this API:

    # https://petstore.swagger.io/?url=https://raw.githubusercontent.com/prometheus/alertmanager/main/api/v2/openapi.yaml#/silence/postSilences

    silence_obj = {

        # id is optional, it's used to update an existing one. this

        # code could be therefore refactored to support modifying

        # existing alerts as well.

        "matchers": matcher_objs,

        "startsAt": starts_at,

        "endsAt": ends_at,

        "createdBy": created_by,

        "comment": comment,

    }

    logging.info("sending silence %r", silence_obj)

    result = requests.post(

        url + "/api/v2/silences", json=silence_obj, auth=(http_user, http_password)

    )

    # error handling

    logging.debug("response code: %s", result)

    try:

        response_obj = result.json()

    except requests.exceptions.JSONDecodeError as e:

        raise invoke.exceptions.Exit(

            "cannot parse JSON response from Alertmanager: %s" % e

        )

    if result.status_code != requests.codes.ok:

        raise invoke.exceptions.Exit(

            "failed to post silence to Alertmanager, responded with status code: %s, JSON: %s"

            % (

                result.status_code,

                response_obj,

            )

        )

    # show confirmation to the user

    logging.debug("response object: %s", response_obj)

    silence_id = response_obj.get("silenceID")

    if not silence_id:

        raise invoke.exceptions.Exit(

            "silence posted, but missing silenceID from Alertmanager, status code: %s, response object: %s"

            % (result.status_code, response_obj)

        )

    logging.info("posted silence %s: %s/#/silences/%s", silence_id, url, silence_id)