healthchecks/hc/api/transports.py

from django.conf import settings
from django.template.loader import render_to_string
from django.utils import timezone
import json
import requests
from six.moves.urllib.parse import quote

from hc.lib import emails


def tmpl(template_name, **ctx):
    template_path = "integrations/%s" % template_name
    return render_to_string(template_path, ctx).strip()


class Transport(object):
    def __init__(self, channel):
        self.channel = channel

    def notify(self, check):
        """ Send notification about current status of the check.

        This method returns None on success, and error message
        on error.

        """

        raise NotImplementedError()

    def is_noop(self, check):
        """ Return True if transport will ignore check's current status.

        This method is overriden in Webhook subclass where the user can
        configure webhook urls for "up" and "down" events, and both are
        optional.

        """

        return False

    def checks(self):
        return self.channel.user.check_set.order_by("created")


class Email(Transport):
    def notify(self, check, bounce_url):
        if not self.channel.email_verified:
            return "Email not verified"

        headers = {"X-Bounce-Url": bounce_url}

        ctx = {
            "check": check,
            "checks": self.checks(),
            "now": timezone.now(),
            "unsub_link": self.channel.get_unsub_link()
        }

        emails.alert(self.channel.value, ctx, headers)


class HttpTransport(Transport):

    def _request(self, method, url, **kwargs):
        try:
            options = dict(kwargs)
            if "headers" not in options:
                options["headers"] = {}

            options["timeout"] = 5
            options["headers"]["User-Agent"] = "healthchecks.io"

            r = requests.request(method, url, **options)
            if r.status_code not in (200, 201, 204):
                return "Received status code %d" % r.status_code
        except requests.exceptions.Timeout:
            # Well, we tried
            return "Connection timed out"
        except requests.exceptions.ConnectionError:
            return "Connection failed"

    def get(self, url):
        # Make 3 attempts--
        for x in range(0, 3):
            error = self._request("get", url)
            if error is None:
                break

        return error

    def post(self, url, **kwargs):
        # Make 3 attempts--
        for x in range(0, 3):
            error = self._request("post", url, **kwargs)
            if error is None:
                break

        return error


class Webhook(HttpTransport):
    def prepare(self, template, check, urlencode=False):
        """ Replace variables with actual values.

        There should be no bad translations if users use $ symbol in
        check's name or tags, because $ gets urlencoded to %24

        """

        def safe(s):
            return quote(s) if urlencode else s

        result = template
        if "$CODE" in result:
            result = result.replace("$CODE", str(check.code))

        if "$STATUS" in result:
            result = result.replace("$STATUS", check.status)

        if "$NOW" in result:
            s = timezone.now().replace(microsecond=0).isoformat()
            result = result.replace("$NOW", safe(s))

        if "$NAME" in result:
            result = result.replace("$NAME", safe(check.name))

        if "$TAG" in result:
            for i, tag in enumerate(check.tags_list()):
                placeholder = "$TAG%d" % (i + 1)
                result = result.replace(placeholder, safe(tag))

        return result

    def is_noop(self, check):
        if check.status == "down" and not self.channel.value_down:
            return True

        if check.status == "up" and not self.channel.value_up:
            return True

        return False

    def notify(self, check):
        url = self.channel.value_down
        if check.status == "up":
            url = self.channel.value_up

        assert url

        url = self.prepare(url, check, urlencode=True)
        if self.channel.post_data:
            payload = self.prepare(self.channel.post_data, check)
            return self.post(url, data=payload)
        else:
            return self.get(url)


class Slack(HttpTransport):
    def notify(self, check):
        text = tmpl("slack_message.json", check=check)
        payload = json.loads(text)
        return self.post(self.channel.slack_webhook_url, json=payload)


class HipChat(HttpTransport):
    def notify(self, check):
        text = tmpl("hipchat_message.html", check=check)
        payload = {
            "message": text,
            "color": "green" if check.status == "up" else "red",
        }
        return self.post(self.channel.value, json=payload)


class OpsGenie(HttpTransport):

    def notify(self, check):
        payload = {
            "apiKey": self.channel.value,
            "alias": str(check.code),
            "source": "healthchecks.io"
        }

        if check.status == "down":
            payload["tags"] = ",".join(check.tags_list())
            payload["message"] = tmpl("opsgenie_message.html", check=check)
            payload["note"] = tmpl("opsgenie_note.html", check=check)

        url = "https://api.opsgenie.com/v1/json/alert"
        if check.status == "up":
            url += "/close"

        return self.post(url, json=payload)


class PagerDuty(HttpTransport):
    URL = "https://events.pagerduty.com/generic/2010-04-15/create_event.json"

    def notify(self, check):
        description = tmpl("pd_description.html", check=check)
        payload = {
            "service_key": self.channel.value,
            "incident_key": str(check.code),
            "event_type": "trigger" if check.status == "down" else "resolve",
            "description": description,
            "client": "healthchecks.io",
            "client_url": settings.SITE_ROOT
        }

        return self.post(self.URL, json=payload)


class Pushbullet(HttpTransport):
    def notify(self, check):
        text = tmpl("pushbullet_message.html", check=check)
        url = "https://api.pushbullet.com/v2/pushes"
        headers = {
            "Access-Token": self.channel.value,
            "Conent-Type": "application/json"
        }
        payload = {
            "type": "note",
            "title": "healthchecks.io",
            "body": text
        }

        return self.post(url, json=payload, headers=headers)


class Pushover(HttpTransport):
    URL = "https://api.pushover.net/1/messages.json"

    def notify(self, check):
        others = self.checks().filter(status="down").exclude(code=check.code)
        ctx = {
            "check": check,
            "down_checks": others,
        }
        text = tmpl("pushover_message.html", **ctx)
        title = tmpl("pushover_title.html", **ctx)
        user_key, prio = self.channel.value.split("|")
        payload = {
            "token": settings.PUSHOVER_API_TOKEN,
            "user": user_key,
            "message": text,
            "title": title,
            "html": 1,
            "priority": int(prio),
        }

        # Emergency notification
        if prio == "2":
            payload["retry"] = settings.PUSHOVER_EMERGENCY_RETRY_DELAY
            payload["expire"] = settings.PUSHOVER_EMERGENCY_EXPIRATION

        return self.post(self.URL, data=payload)


class VictorOps(HttpTransport):
    def notify(self, check):
        description = tmpl("victorops_description.html", check=check)
        mtype = "CRITICAL" if check.status == "down" else "RECOVERY"
        payload = {
            "entity_id": str(check.code),
            "message_type": mtype,
            "entity_display_name": check.name_then_code(),
            "state_message": description,
            "monitoring_tool": "healthchecks.io",
        }

        return self.post(self.channel.value, json=payload)


class Discord(HttpTransport):
    def notify(self, check):
        text = tmpl("slack_message.json", check=check)
        payload = json.loads(text)
        url = self.channel.discord_webhook_url + "/slack"
        return self.post(url, json=payload)