Add cronjob for syncing heartbeats with heartbeat-missing incidents

changelog.d/1954.added.md

@@ -0,0 +1,4 @@
+Added management command suitable to run as a cronjob that looks for dead
+incidents (as per heartbeat freqeuncy and last seen heartbeat). Dead sources
+are reported as incidents, formerly dead sources get their respective
+incidents auto-closed.

docs/development/management-commands.rst

@@ -535,6 +535,21 @@ To also save the latest version in the database, use the `--save` flag:
 This will currently not do anything, but in the future you will be able to get
 a notification in the frontend if there is a new release registered in the database that you haven't seen yet.
 
+Handling source heartbeats
+==========================
+
+If there are sources capable of and configured to send heartbeats,
+there's a command suitable for creating and closing incidents to warn about
+missing heartbeats:
+
+    .. code:: console
+
+        $ python manage.py sync_heartbeat_incidents
+
+This command is suitable for running by cron. The smallest heartbeat frequency
+configured should be the same as or less often than the frequency of the cron-job.
+
+
 Deprecated/overlapping commands
 ===============================
 

docs/reference/special-incidents.rst

@@ -81,3 +81,38 @@ a source system.
 
 The ``object`` tag holds the integer value of the primary key of the generated
 incident.
+
+
+Missing heartbeat incidents
+===========================
+
+These stateful incidents may be generated when a source system has both the
+fields ``heartbeat_frequency`` (a duration) and ``last_seen`` (a timestamp)
+set.
+
+If it is more than ``last_seen`` + ``heartbeat_frequency`` since the source
+used the API, the incident is generated.
+
+If a previously marked source is back, th eincident is automatically closed.
+
+
+Description
+-----------
+
+"Missing heartbeat from source NAME, dead?"
+
+The NAME is the name of a source, without its type.
+
+Default level
+-------------
+
+4
+
+Tags
+----
+
+* problem_type=missing_heartbeat
+* source_system_id=INTEGER
+
+The ``source_system_id`` tag takes a value that is the integer primary key of
+a source system.

src/argus/incident/factories.py

@@ -1,5 +1,5 @@
 from zoneinfo import ZoneInfo
-from datetime import datetime
+from datetime import datetime, timedelta
 from random import randint, choice
 from typing import Optional, Any
 
@@ -47,6 +47,54 @@
     return c
 
 
+def create_fake_source(
+    name: Optional[str] = None,
+    type: Optional[str] = None,
+    user: Optional[str] = None,
+    last_seen: Optional[datetime] = None,
+    heartbeat_frequency: Optional[timedelta] = None,
+    commit=True,
+):
+    generated_name = factory.Sequence(lambda s: "source-%s" % s)
+    name = name or type or user or generated_name
+    if not type:
+        type = "type-" + name
+    if not user:
+        user = "user-" + name
+
+    source_user = SourceUserFactory(username=user)
+    source_type = SourceSystemTypeFactory(name=type)
+    source = SourceSystem(
+        name=name,
+        user=source_user,
+        type=source_type,
+        last_seen=last_seen,
+        heartbeat_frequency=heartbeat_frequency,
+    )
+    if commit:
+        source.save()
+    return source
+
+
+def create_dead_source(
+    name,
+    type: str = None,
+    user: str = None,
+    timestamp: Optional[None] = None,
+):
+    timestamp = timestamp if timestamp else timezone.now()
+    heartbeat_frequency = timedelta(minutes=5)
+    last_seen = timestamp - (2 * heartbeat_frequency)
+    source = create_fake_source(
+        name=name,
+        type=type,
+        user=user,
+        last_seen=last_seen,
+        heartbeat_frequency=heartbeat_frequency,
+    )
+    return source, timestamp
+
+
 def create_fake_incident(
     tags: Optional[list[str]] = None,
     description: Optional[str] = None,

src/argus/incident/heartbeat_utils.py

@@ -0,0 +1,112 @@
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+
+from django.utils import timezone
+
+from argus.incident.factories import create_stateful_incident
+from argus.incident.models import HEARTBEAT_TAG, SOURCE_TAG_KEY, Incident, SourceSystem
+
+
+LOG = logging.getLogger(__name__)
+DEFAULT_FUDGE_FACTOR = 10
+DEFAULT_LEVEL = 4
+_FUDGE = timedelta(seconds=DEFAULT_FUDGE_FACTOR)
+
+__all__ = [
+    "sync_heartbeats_with_heartbeat_incidents",
+]
+
+
+def sync_heartbeats_with_heartbeat_incidents():
+    timestamp = timezone.now()
+    alive_sources = _close_incidents_whose_sources_are_alive_again(timestamp)
+    new_incidents = _create_incidents_for_dead_sources(timestamp)
+    return alive_sources, new_incidents
+
+
+def _close_incidents_whose_sources_are_alive_again(timestamp):
+    argus = SourceSystem.objects.get(name="argus")
+    still_dead_sources = SourceSystem.objects.dead(timestamp - _FUDGE)
+    outdated_incidents = Incident.objects.heartbeat_incidents().open()
+    format_str = 'Source "{}" is again alive and well'
+    sources = []
+    for incident in outdated_incidents:
+        source_ids = incident.get_values_for_tag_key(SOURCE_TAG_KEY)
+        # We need exactly one source-tag.
+        # We don't need to check for multiple *identical* tags,
+        # that's a UniqueError, but we should also only have one source-tag
+        if not source_ids:
+            LOG.error('Heartbeat incident "%s" is missing source-tag!', incident)
+            continue
+        elif len(source_ids) > 1:
+            LOG.error('Heartbeat incident "%s" has multiple source tags!?', incident)
+            continue
+
+        source_id = source_ids[0]
+        if still_dead_sources.filter(pk=source_id).exists():
+            continue
+
+        try:
+            source = SourceSystem.objects.get(pk=source_id)
+        except SourceSystem.DoesNotExist:
+            LOG.error('Heartbeat source "%s" has disappeared!', source_id)
+            continue
+
+        # phew, *now* we can close the incident
+        incident.set_end(argus.user, timestamp, format_str.format(source))
+        sources.append(source)
+    return sources
+
+
+def _create_incidents_for_dead_sources(timestamp: Optional[datetime] = None):
+    # Create existing incidents whose sources have become dead
+    dead_sources = SourceSystem.objects.dead(timestamp - _FUDGE)
+    incident_owner = SourceSystem.objects.get(name="argus")
+    # prevent multiple incidents
+    heartbeat_incidents = Incident.objects.heartbeat_incidents().open()
+    source_ids = (incident.source_id for incident in heartbeat_incidents)
+    new_incidents = []
+    for source in dead_sources.exclude(id__in=source_ids):
+        incident = _get_or_create_incident_for_dead_source(source, incident_owner=incident_owner, timestamp=timestamp)
+        if incident:
+            new_incidents.append(incident)
+    return new_incidents
+
+
+def _get_or_create_incident_for_dead_source(
+    source: SourceSystem,
+    incident_owner: SourceSystem,
+    timestamp: Optional[datetime] = None,
+    level: int = DEFAULT_LEVEL,
+):
+    assert isinstance(source, SourceSystem), "source is not a SourceSystem"
+    if not timestamp:
+        timestamp = timezone.now()
+
+    source_tag = f"{SOURCE_TAG_KEY}={source.pk}"
+    tags = [HEARTBEAT_TAG, source_tag]
+
+    # One last check in case a heartbeat arrived in the meantime
+    source.refresh_from_db()
+    dead = source.is_dead(timestamp - _FUDGE)
+    if dead:
+        # prevent duplicates
+        existing_incidents = Incident.objects.heartbeat_incidents().open().from_tags(source_tag)
+        if existing_incidents.exists():
+            try:
+                return existing_incidents.get()
+            except Incident.MultipleObjectsReturned:
+                # TODO: fix, different function?
+                LOG.error('Source "%s" has multiple open heartbeat incidents', source)
+                return existing_incidents.first()
+        # save new incident
+        incident = create_stateful_incident(
+            f"Missing heartbeat from source {source}, dead?",
+            incident_owner,
+            level,
+            tags=tags,
+            start_time=timestamp,
+        )
+        return incident
+    return None

src/argus/incident/management/commands/sync_heartbeat_incidents.py

@@ -0,0 +1,22 @@
+from django.core.management.base import BaseCommand
+
+from argus.incident.heartbeat_utils import sync_heartbeats_with_heartbeat_incidents
+
+
+class Command(BaseCommand):
+    help = "Check that heartbeat-supporting sources are alive"
+
+    def handle(self, *args, **options):
+        alive_sources, new_incidents = sync_heartbeats_with_heartbeat_incidents()
+        if not (alive_sources or new_incidents):
+            return
+        if options["verbosity"] > 0:
+            for source in alive_sources:
+                self.stdout.write(f'Closed incident for source "{source.name}", it\'s back')
+            for incident in new_incidents:
+                self.stdout.write(f"Created incident {incident}")
+        else:
+            count_incidents = len(new_incidents)
+            self.stdout.write(f"Created {count_incidents} new incidents")
+            count_alive = len(alive_sources)
+            self.stdout.write(f"Closed {count_alive} existing incidents")