GlitchTip/glitchtip/uptime/models.py

import uuid
from datetime import timedelta

from django.core.validators import MaxValueValidator
from django.db import models
from django.db.models import OuterRef, Subquery

from glitchtip.base_models import CreatedModel

from .constants import MonitorCheckReason, MonitorType


class MonitorManager(models.Manager):
    def with_check_annotations(self, on_update=False):
        """
        Adds MonitorCheck annotations:
        latest_is_up - Most recent check is_up result
        last_change - Most recent check where is_up state changed
        last_change_on_update: Same idea as last_change, but handled differently when monitor has not yet been updated (for alerts).
        Example: Monitor state: { latest_is_up } since { last_change }
        """
        initial_annotation = self.annotate(
            latest_is_up=Subquery(
                MonitorCheck.objects.filter(monitor_id=OuterRef("id"))
                .order_by("-start_check")
                .values("is_up")[:1]
            )
        )
        if on_update:
            return initial_annotation.annotate(
                last_change_on_update=Subquery(
                    MonitorCheck.objects.filter(monitor_id=OuterRef("id"))
                    .filter(is_up=OuterRef("latest_is_up"))
                    .order_by("-start_check")
                    .values("start_check")[:1]
                )
            )
        else:
            return initial_annotation.annotate(
                last_change=Subquery(
                    MonitorCheck.objects.filter(monitor_id=OuterRef("id"))
                    .exclude(is_up=OuterRef("latest_is_up"))
                    .order_by("-start_check")
                    .values("start_check")[:1]
                )
            )


class Monitor(CreatedModel):
    monitor_type = models.CharField(
        max_length=12, choices=MonitorType.choices, default=MonitorType.PING
    )
    endpoint_id = models.UUIDField(
        blank=True,
        null=True,
        editable=False,
        help_text="Used for referencing heartbeat endpoint",
    )
    name = models.CharField(max_length=200)
    url = models.URLField(max_length=2000, blank=True)
    expected_status = models.PositiveSmallIntegerField(default=200)
    expected_body = models.CharField(max_length=2000, blank=True)
    environment = models.ForeignKey(
        "environments.Environment", on_delete=models.SET_NULL, null=True, blank=True,
    )
    project = models.ForeignKey(
        "projects.Project", on_delete=models.SET_NULL, null=True, blank=True,
    )
    organization = models.ForeignKey(
        "organizations_ext.Organization", on_delete=models.CASCADE
    )
    interval = models.DurationField(
        default=timedelta(minutes=1),
        validators=[MaxValueValidator(timedelta(hours=23, minutes=59, seconds=59))],
    )

    objects = MonitorManager()

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        if self.monitor_type == MonitorType.HEARTBEAT and not self.endpoint_id:
            self.endpoint_id = uuid.uuid4()
        super().save(*args, **kwargs)
        # pylint: disable=import-outside-toplevel
        from glitchtip.uptime.tasks import perform_checks

        if self.monitor_type != MonitorType.HEARTBEAT:
            perform_checks.apply_async(args=([self.pk],), countdown=1)


class MonitorCheck(CreatedModel):
    monitor = models.ForeignKey(
        Monitor, on_delete=models.CASCADE, related_name="checks"
    )
    is_up = models.BooleanField()
    start_check = models.DateTimeField(
        help_text="Time when the start of this check was performed"
    )
    reason = models.PositiveSmallIntegerField(
        choices=MonitorCheckReason.choices, default=0, null=True, blank=True
    )
    response_time = models.DurationField(blank=True, null=True)
    data = models.JSONField(null=True, blank=True)

    class Meta:
        indexes = [
            models.Index(fields=["start_check", "monitor"]),
        ]
        ordering = ("-created",)

    def __str__(self):
        return self.up_or_down

    @property
    def up_or_down(self):
        if self.is_up:
            return "Up"
        return "Down"