diff --git a/changelog.d/1954.added.md b/changelog.d/1954.added.md new file mode 100644 index 000000000..c24f04910 --- /dev/null +++ b/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. diff --git a/docs/development/management-commands.rst b/docs/development/management-commands.rst index 567159de4..df8e5952a 100644 --- a/docs/development/management-commands.rst +++ b/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 =============================== diff --git a/docs/reference/special-incidents.rst b/docs/reference/special-incidents.rst index 487435e14..47de0fc67 100644 --- a/docs/reference/special-incidents.rst +++ b/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, the incident is automatically closed. + + +Description +----------- + +"Missing heartbeat from source NAME (TYPE), dead?" + +The NAME is the name of a source, while TYPE is the type of that source. + +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. diff --git a/src/argus/incident/factories.py b/src/argus/incident/factories.py index 282ed1461..8ec0a7c51 100644 --- a/src/argus/incident/factories.py +++ b/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 @@ def update_tags(incident: Incident, *tags): 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, diff --git a/src/argus/incident/heartbeat_utils.py b/src/argus/incident/heartbeat_utils.py new file mode 100644 index 000000000..84a793b06 --- /dev/null +++ b/src/argus/incident/heartbeat_utils.py @@ -0,0 +1,179 @@ +import logging +from datetime import datetime, timedelta +from typing import List, Tuple, 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 +HEARTBEAT_FUDGE = timedelta(seconds=DEFAULT_FUDGE_FACTOR) +INCIDENT_DESCRIPTION_TEMPLATE = "Missing heartbeat from source {source}, dead?" + +__all__ = [ + "sync_heartbeats_with_heartbeat_incidents", +] + + +def sync_heartbeats_with_heartbeat_incidents() -> Tuple[List[SourceSystem], List[Incident], List[Incident]]: + """Sync heartbeat-configured sources with heartbeat incidents + + Returns a list of no longer dead sources, freshly created incidents, and + incidents for sources that are still dead. + """ + timestamp = timezone.now() + alive_sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(timestamp) + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + remaining_incidents = list(set(remaining_incidents + existing_incidents)) + return alive_sources, new_incidents, remaining_incidents + + +def _close_heartbeat_incidents(timestamp: datetime) -> Tuple[List[SourceSystem], List[Incident], List[Incident]]: + """Close all existing heartbeat incidents that can be closed + + Returns a list of reanimated sources foll: owed by a list of freshly + created incidents and ifinally a list of incidents that could not be closed. + """ + outdated_incidents = Incident.objects.heartbeat_incidents().open() + sources = [] + closed_incidents = [] + remaining_incidents = [] + for incident in outdated_incidents: + source, incident = _attempt_closing_heartbeat_incident(incident, timestamp) + if source: + sources.append(source) + if incident.open: + remaining_incidents.append(incident) + else: + closed_incidents.append(incident) + return sources, closed_incidents, remaining_incidents + + +def _attempt_closing_heartbeat_incident( + incident: Incident, timestamp: datetime +) -> Tuple[Optional[SourceSystem], Incident]: + """Try to close a heartbeat incident + + - If the source is alive again, return both the source and the closed + incident. + - If the source is still dead, return None and the still open incident. + + If something was wrong with the source or incident, an error is logged. + + - If the source has disappeared return None and the force-closed incident. + - If the incident is missing its source tag return None and the + force-closed incident. + - If the incident has multiple source tags, return None and the still open + incident.[*] + + [*] This should never happen but is included for completeness. + """ + argus = SourceSystem.objects.get(name="argus") + alive_description = 'Source "{}" is again alive and well' + missing_tag_description = "The source tag of incident {} is missing, closing" + missing_source_description = "The source of incident {} is gone, closing" + + 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.id) + incident.set_end(argus.user, timestamp, missing_tag_description.format(incident.id)) + return None, incident + elif len(source_ids) > 1: # WAT, inconceivable + LOG.error('Heartbeat incident "%s" has multiple source tags!?', incident.id) + return None, incident + + source_id = source_ids[0] + try: + source = SourceSystem.objects.get(pk=source_id) + except SourceSystem.DoesNotExist: + LOG.error('Heartbeat source "%s" has disappeared!', source_id) + incident.set_end(argus.user, timestamp, missing_source_description.format(incident.id)) + return None, incident + + if source.is_dead(timestamp - HEARTBEAT_FUDGE): + return None, incident + + # phew, *now* we can close the incident + incident.set_end(argus.user, timestamp, alive_description.format(source)) + return source, incident + + +def _create_incidents_for_dead_sources(timestamp: Optional[datetime] = None) -> tuple[list[Incident], list[Incident]]: + """ + Creates heartbeat incidents for dead sources as of TIMESTAMP + + Keeps track of freshly created incidents versus already existing incidents. + """ + # Create existing incidents whose sources have become dead + dead_sources = SourceSystem.objects.dead(timestamp - HEARTBEAT_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 = [] + existing_incidents = [] + for source in dead_sources.exclude(id__in=source_ids): + incident, new = _get_or_create_incident_for_dead_source( + source, incident_owner=incident_owner, timestamp=timestamp + ) + if incident: + if new: + new_incidents.append(incident) + else: + existing_incidents.append(incident) + return new_incidents, existing_incidents + + +def _get_or_create_incident_for_dead_source( + source: SourceSystem, + incident_owner: SourceSystem, + timestamp: Optional[datetime] = None, + level: int = DEFAULT_LEVEL, +) -> Tuple[Optional[Incident], Optional[bool]]: + """ + Create a heartbeat incident for one specific dead source + + If the source is actually dead, returns the incident and whether it was + freshly created (True) or not (False). + + Logs if there are multiple incidents for the source, and returns one of the + sources and that it was not feshly created (False). + + If the source is actually alive, returns (None, None) + """ + 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 - HEARTBEAT_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(), False + except Incident.MultipleObjectsReturned: + LOG.error('Source "%s" has multiple open heartbeat incidents', source) + return existing_incidents.first(), False + # save new incident + incident = create_stateful_incident( + INCIDENT_DESCRIPTION_TEMPLATE.format(source=str(source)), + incident_owner, + level, + tags=tags, + start_time=timestamp, + ) + return incident, True + return None, None diff --git a/src/argus/incident/management/commands/sync_heartbeat_incidents.py b/src/argus/incident/management/commands/sync_heartbeat_incidents.py new file mode 100644 index 000000000..4f988ead8 --- /dev/null +++ b/src/argus/incident/management/commands/sync_heartbeat_incidents.py @@ -0,0 +1,39 @@ +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, remaining_incidents = sync_heartbeats_with_heartbeat_incidents() + if not (alive_sources or new_incidents or remaining_incidents): + return + count_closed = len(alive_sources) + count_incidents = len(new_incidents) + count_dead = len(remaining_incidents) + + if options["verbosity"] > 0: + self.stdout.write( + f"Heartbeat incidents: Created: {count_incidents}, Closed: {count_closed}, Remaining: {count_dead}" + ) + + if options["verbosity"] > 1: + if count_incidents: + self.stdout.write() + self.stdout.write("Created incidents:") + for incident in new_incidents: + self.stdout.write(f"- {incident}") + + if count_closed: + self.stdout.write() + self.stdout.write("Closed incidents:") + for source in alive_sources: + self.stdout.write(f"- {source.name}") + + if count_dead: + self.stdout.write() + self.stdout.write("Remaining incidents:") + for incident in remaining_incidents: + self.stdout.write(f"- {incident}") diff --git a/src/argus/incident/models.py b/src/argus/incident/models.py index 020599a66..b159c7608 100644 --- a/src/argus/incident/models.py +++ b/src/argus/incident/models.py @@ -20,6 +20,8 @@ from .validators import validate_key, validate_lowercase +SOURCE_TAG_KEY = "source_system_id" +HEARTBEAT_TAG = "problem_type=missing_heartbeat" MINIMUM_DURATION = timedelta(seconds=60) MAXIMUM_DURATION = timedelta(days=1) LOG = logging.getLogger(__name__) @@ -48,6 +50,28 @@ def save(self, *args, **kwargs): super().save(*args, **kwargs) +class SourceSystemQuerySet(models.QuerySet): + def has_heartbeat(self): + """Find active sources configured to send heartbeats""" + return self.filter(heartbeat_frequency__isnull=False, last_seen__isnull=False) + + def with_next_expected_heartbeat(self, timestamp: Optional[datetime] = None): + """Annotate heartbeat sources with the timestamp of the next predicted heartbeat""" + timestamp = timestamp if timestamp else timezone.now() + qs = self.has_heartbeat() + qs = qs.annotate(next_heartbeat=F("last_seen") + F("heartbeat_frequency")) + return qs + + def dead(self, timestamp: Optional[datetime] = None): + """Find sources that have missed heartbeats + + The calculation is done in the database. + """ + timestamp = timestamp if timestamp else timezone.now() + qs = self.with_next_expected_heartbeat(timestamp) + return qs.filter(next_heartbeat__lt=timestamp) + + class SourceSystem(models.Model): name = models.TextField() type = models.ForeignKey(to=SourceSystemType, on_delete=models.PROTECT, related_name="instances") @@ -67,6 +91,8 @@ class SourceSystem(models.Model): ], ) + objects = SourceSystemQuerySet.as_manager() + class Meta: constraints = [ models.UniqueConstraint(fields=["name", "type"], name="%(class)s_unique_name_per_type"), @@ -79,9 +105,16 @@ def update_last_seen(self, timestamp: Optional[datetime] = None): self.last_seen = timestamp if timestamp else timezone.now() self.save() + def is_dead(self, timestamp) -> None | bool: + """Check if an expected heartbeat of a source is missing""" + if not self.heartbeat_frequency: + return None + dead = self.last_seen + self.heartbeat_frequency < timestamp + return dead + class TagQuerySet(models.QuerySet): - def parse(self, *tags): + def parse(self, *tags: str): "Return a list of querysets that match `tags`" set_dict = defaultdict(set) for k, v in (Tag.split(tag) for tag in tags): @@ -94,6 +127,17 @@ def create_from_tag(self, tag): tag, _ = self.get_or_create(key=key, value=value) return tag + def from_tags(self, *tags: str): + qss = self.parse(*tags) + tagobjs = [] + for qs in qss: + tagobjs.extend(list(qs)) + return set(tagobjs) + + def from_tag_keys(self, *keys: str): + qs = self.filter(key__in=keys) + return qs + class Tag(models.Model): TAG_DELIMITER = "=" @@ -249,6 +293,13 @@ def from_tags(self, *tags): qs = reduce(and_, qs) return qs.distinct() + def from_tag_keys(self, *keys): + qs = self.filter(incident_tag_relations__tag__key__in=keys) + return qs + + def heartbeat_incidents(self): + return self.from_tags(HEARTBEAT_TAG) + def is_longer_than_minutes(self, minutes): min_duration = timedelta(minutes=minutes) open = self.open().annotate(duration=timezone.now() - F("start_time")) @@ -469,6 +520,27 @@ def acked(self): return self.events.filter((acks_query & acks_not_expired_query) | ack_is_just_being_created).exists() + def has_tags(self, *tags: str): + tags = Tag.objects.from_tags(*tags) + if not tags: + return False + return bool(set(self.deprecated_tags).issuperset(tags)) + + def has_tag_keys(self, *keys: str): + return self.incident_tag_relations.only("tag").filter(tag__key__in=keys).exists() + + def is_heartbeat_incident(self): + heartbeat_tag = self.has_tags(HEARTBEAT_TAG) + source_tag = self.has_tag_keys(SOURCE_TAG_KEY) + return heartbeat_tag and source_tag + + def get_values_for_tag_key(self, key): + values = [] + for tag in self.deprecated_tags: + if tag.key == key: + values.append(tag.value) + return values + def is_acked_by(self, group: str) -> bool: return group in self.acks.active().group_names() @@ -492,6 +564,7 @@ def create_first_event(self): # @transaction.atomic def set_open(self, actor: User, timestamp: datetime = None, description=""): + "Incident reopened by user" if not self.stateful: raise ValidationError("Cannot set a stateless incident as open") if self.open: @@ -509,6 +582,7 @@ def set_open(self, actor: User, timestamp: datetime = None, description=""): # @transaction.atomic def set_closed(self, actor: User, timestamp: datetime = None, description=""): + "Incident closed by human user" if not self.stateful: raise ValidationError("Cannot set a stateless incident as closed") if not self.open: @@ -525,15 +599,22 @@ def set_closed(self, actor: User, timestamp: datetime = None, description=""): ) # @transaction.atomic - def set_end(self, actor: User): + def set_end(self, actor: User, timestamp: datetime = None, description: str = ""): + "Incident ended by machine source" if not self.stateful: raise ValidationError("Cannot set a stateless incident as ended") if not self.open: return - self.end_time = timezone.now() + self.end_time = timestamp or timezone.now() self.save(update_fields=["end_time"]) - Event.objects.create(incident=self, actor=actor, timestamp=self.end_time, type=Event.Type.INCIDENT_END) + Event.objects.create( + incident=self, + actor=actor, + timestamp=self.end_time, + type=Event.Type.INCIDENT_END, + description=description, + ) # @transaction.atomic def create_ack(self, actor: User, timestamp=None, description="", expiration=None): diff --git a/tests/incident/management_commands/test_sync_heartbeat_incidents.py b/tests/incident/management_commands/test_sync_heartbeat_incidents.py new file mode 100644 index 000000000..cfcbe4108 --- /dev/null +++ b/tests/incident/management_commands/test_sync_heartbeat_incidents.py @@ -0,0 +1,131 @@ +from unittest.mock import patch +from io import StringIO +import contextlib + +from django.core.management import call_command +from django.test import TestCase + +from argus.incident.models import get_or_create_default_instances + + +class TestSyncHeartbeatIncidents(TestCase): + def setUp(self): + get_or_create_default_instances() + + def test_when_nothing_to_report_and_not_verbose_then_print_nothing(self): + # No verbosity flag + F = StringIO() + + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=1) + + output = F.getvalue().strip() + self.assertFalse(output) + + def test_when_nothing_to_report_and_verbose_then_print_nothing(self): + # Flag "-v 2" + F = StringIO() + + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=2) + + output = F.getvalue().strip() + self.assertFalse(output) + + def test_when_something_to_report_but_quiet_print_nothing(self): + # Flag "-v 0" + F = StringIO() + + alive_sources = [] + new_incidents = ["golgamfrincham"] + remaining_incidents = [] + with patch( + "argus.incident.management.commands.sync_heartbeat_incidents.sync_heartbeats_with_heartbeat_incidents", + return_value=(alive_sources, new_incidents, remaining_incidents), + ): + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=0) + + output = F.getvalue() + self.assertFalse(output) + + def test_when_new_dead_source_incidents_then_print_overview_line(self): + # No verbosity flag + F = StringIO() + + alive_sources = [] + new_incidents = ["golgamfrincham"] + remaining_incidents = [] + with patch( + "argus.incident.management.commands.sync_heartbeat_incidents.sync_heartbeats_with_heartbeat_incidents", + return_value=(alive_sources, new_incidents, remaining_incidents), + ): + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=1) + + output = F.getvalue().strip() + self.assertEqual( + output, + "Heartbeat incidents: Created: 1, Closed: 0, Remaining: 0", + ) + + def test_when_new_dead_source_incidents_and_verbose_then_print_incident_in_addition_to_overview(self): + # Flag "-v 2" + F = StringIO() + + alive_sources = [] + new_incidents = ["pollywog"] + remaining_incidents = [] + with patch( + "argus.incident.management.commands.sync_heartbeat_incidents.sync_heartbeats_with_heartbeat_incidents", + return_value=(alive_sources, new_incidents, remaining_incidents), + ): + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=2) + + output = F.getvalue() + self.assertEqual( + output, ("Heartbeat incidents: Created: 1, Closed: 0, Remaining: 0\n\nCreated incidents:\n- pollywog\n") + ) + + def test_when_reanimated_source_then_print_overview_line(self): + # No verbosity flag + F = StringIO() + + class FakeSource: + name = "wui" + + alive_sources = [FakeSource] + new_incidents = [] + remaining_incidents = [] + with patch( + "argus.incident.management.commands.sync_heartbeat_incidents.sync_heartbeats_with_heartbeat_incidents", + return_value=(alive_sources, new_incidents, remaining_incidents), + ): + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=1) + + output = F.getvalue().strip() + self.assertEqual(output, "Heartbeat incidents: Created: 0, Closed: 1, Remaining: 0") + + def test_when_reanimated_source_and_verbose_then_print_source_name_in_addition_to_overview(self): + # Flag "-v 2" + F = StringIO() + + class FakeSource: + name = "oopsy" + + alive_sources = [FakeSource] + new_incidents = [] + remaining_incidents = [] + with patch( + "argus.incident.management.commands.sync_heartbeat_incidents.sync_heartbeats_with_heartbeat_incidents", + return_value=(alive_sources, new_incidents, remaining_incidents), + ): + with contextlib.redirect_stdout(F): + call_command("sync_heartbeat_incidents", verbosity=2) + + output = F.getvalue() + self.assertEqual( + output, ("Heartbeat incidents: Created: 0, Closed: 1, Remaining: 0\n\nClosed incidents:\n- oopsy\n") + ) diff --git a/tests/incident/test_factories.py b/tests/incident/test_factories.py index be2638554..cf2279535 100644 --- a/tests/incident/test_factories.py +++ b/tests/incident/test_factories.py @@ -106,7 +106,7 @@ def test_create_fake_incident_does_not_raise_error_on_extra_args(self): def TestCreateStatelessIncident(TestCase): - def test_create_stateless_incident(self): + def test_it_should_always_return_a_stateless_incident(self): source_name = "source_a" sst = SourceSystemTypeFactory(name=source_name) user = SourceUserFactory(username=source_name) diff --git a/tests/incident/test_heartbeat_utils.py b/tests/incident/test_heartbeat_utils.py new file mode 100644 index 000000000..66ba5db4e --- /dev/null +++ b/tests/incident/test_heartbeat_utils.py @@ -0,0 +1,224 @@ +from datetime import timedelta + +from django.test import TestCase, tag +from django.utils.timezone import now as tznow + +from argus.incident.factories import SourceSystemFactory, create_dead_source, create_fake_incident +from argus.incident.heartbeat_utils import ( + SOURCE_TAG_KEY, + HEARTBEAT_TAG, + _close_heartbeat_incidents, + _create_incidents_for_dead_sources, + _get_or_create_incident_for_dead_source, + sync_heartbeats_with_heartbeat_incidents, +) +from argus.incident.models import Incident, Tag, get_or_create_default_instances + + +class MakeImmutableFixtures: + def setUp(self): + _, sst, self.owner_source = get_or_create_default_instances() + self.alive_source = SourceSystemFactory( + name="spring_blossom", last_seen=tznow(), heartbeat_frequency=timedelta(days=1) + ) + self.irrelevant_incident = create_fake_incident(source=self.owner_source.name, tags=["test=test"]) + + +class TestGetOrCreateIncidentForDeadSource(MakeImmutableFixtures, TestCase): + def test_when_source_is_not_dead_we_create_nothing(self): + incident, new = _get_or_create_incident_for_dead_source( + self.alive_source, incident_owner=self.owner_source, timestamp=self.alive_source.last_seen + ) + self.assertIsNone(incident) + self.assertIsNone(new) + + def test_when_source_is_dead_then_created_incident_has_correct_description_and_tags(self): + zombie_source, timestamp = create_dead_source("zombie_walking") + incident, new = _get_or_create_incident_for_dead_source( + zombie_source, incident_owner=self.owner_source, timestamp=timestamp + ) + self.assertTrue(new) + self.assertEqual(incident.description, f"Missing heartbeat from source {zombie_source}, dead?") + self.assertEqual(timestamp, incident.start_time) + tags = [tag.representation for tag in incident.deprecated_tags] + expected_tags = [HEARTBEAT_TAG, f"{SOURCE_TAG_KEY}={zombie_source.pk}"] + self.assertEqual(set(tags), set(expected_tags)) + + def test_when_source_is_dead_and_timestamp_not_set_incident_generates_a_timestamp(self): + in_timestamp = tznow() - timedelta(seconds=60) + zombie_source, timestamp = create_dead_source("zombie_walking", timestamp=in_timestamp) + self.assertEqual(in_timestamp, timestamp) + incident, _ = _get_or_create_incident_for_dead_source(zombie_source, incident_owner=self.owner_source) + self.assertNotEqual(in_timestamp, incident.start_time) + + def test_when_dead_sources_exist_we_only_make_one_incident_per_source(self): + self.assertFalse(Incident.objects.heartbeat_incidents().exists()) + + in_timestamp = tznow() - timedelta(seconds=60) + zombie_source, timestamp = create_dead_source("zombie_walking", timestamp=in_timestamp) + incident1, new1 = _get_or_create_incident_for_dead_source(zombie_source, incident_owner=self.owner_source) + incident2, new2 = _get_or_create_incident_for_dead_source(zombie_source, incident_owner=self.owner_source) + self.assertTrue(new1) + self.assertFalse(new2) + self.assertEqual(incident1.pk, incident2.pk) + + +class TestCreateIncidentsForDeadSources(MakeImmutableFixtures, TestCase): + def test_when_no_dead_sources_returns_tuple_of_two_empty_lists(self): + new_incidents, existing_incidents = _create_incidents_for_dead_sources(tznow()) + self.assertFalse(new_incidents) + self.assertFalse(existing_incidents) + + def test_when_dead_sources_exist_return_list_of_created_incidents_and_an_empty_list(self): + zombie_source, timestamp = create_dead_source("zombie_walking") + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + self.assertTrue(new_incidents) + self.assertFalse(existing_incidents) + tags = [tag.representation for tag in new_incidents[0].deprecated_tags] + self.assertIn(HEARTBEAT_TAG, tags) + self.assertIn(f"{SOURCE_TAG_KEY}={zombie_source.pk}", tags) + + +@tag("db") +class TestCloseHeartbeatIncidents(MakeImmutableFixtures, TestCase): + def test_when_no_reanimated_sources_return_empty_lists(self): + sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(tznow()) + self.assertFalse(sources) + self.assertFalse(closed_incidents) + self.assertFalse(remaining_incidents) + + def test_when_a_reanimated_source_with_heartbeat_incident_exist_close_it(self): + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + zombie_source, timestamp = create_dead_source("zombie_walking") + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + self.assertTrue(Incident.objects.heartbeat_incidents().open().exists()) + self.assertFalse(existing_incidents) + + # reawaken source + zombie_source.last_seen = tznow() + zombie_source.save() + + incident = new_incidents[0] + sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(tznow()) + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + self.assertIn(zombie_source, sources) + self.assertIn(incident, closed_incidents) + self.assertFalse(remaining_incidents) + + def test_when_a_heartbeat_incident_lacks_the_sourcetag_close_it(self): + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + zombie_source, timestamp = create_dead_source("zombie_walking") + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + + # remove source tag + tags = Tag.objects.filter(key=SOURCE_TAG_KEY) + incident = new_incidents[0] + incident.incident_tag_relations.filter(tag__in=tags).delete() + incident.refresh_from_db() + self.assertFalse(Incident.objects.heartbeat_incidents().from_tag_keys(tags[0].representation).open().exists()) + + # reawaken source + zombie_source.last_seen = tznow() + zombie_source.save() + + sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(tznow()) + + # incidents without source tags are just closed regardless + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + self.assertNotIn(zombie_source, sources) + self.assertIn(incident, closed_incidents) + self.assertEqual(remaining_incidents, []) + + def test_when_a_heartbeat_incident_has_multiple_sourcetags_leave_it_alone(self): + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + zombie_source, timestamp = create_dead_source("zombie_walking") + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + incident = new_incidents[0] + self.assertEqual(len(incident.deprecated_tags), 2) + + additional_tag = Tag.objects.create(key=SOURCE_TAG_KEY, value=str(2**32 - 1)) + incident.incident_tag_relations.create(added_by=zombie_source.user, tag=additional_tag) + incident.refresh_from_db() + self.assertEqual(len(incident.deprecated_tags), 3) + + # reawaken source + zombie_source.last_seen = tznow() + zombie_source.save() + + sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(tznow()) + # not found, so not closed! + self.assertTrue(Incident.objects.heartbeat_incidents().open().exists()) + self.assertNotIn(zombie_source, sources) + self.assertEqual(closed_incidents, []) + self.assertIn(incident, remaining_incidents) + + def test_close_all_incidents_pointing_to_a_nonexistent_source(self): + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + zombie_source, timestamp = create_dead_source("zombie_walking") + new_incidents, existing_incidents = _create_incidents_for_dead_sources(timestamp) + incident = new_incidents[0] + self.assertEqual(len(incident.deprecated_tags), 2) + + zombie_source.delete() + + sources, closed_incidents, remaining_incidents = _close_heartbeat_incidents(tznow()) + # source not found, so closed! + self.assertFalse(Incident.objects.heartbeat_incidents().open().exists()) + self.assertNotIn(zombie_source, sources) + self.assertIn(incident, closed_incidents) + self.assertEqual(remaining_incidents, []) + + +class TestSyncHeartbeatsWithHeartbeatIncidents(MakeImmutableFixtures, TestCase): + def test_when_no_relevant_incidents_or_sources_return_two_empty_lists(self): + sources, new_incidents, remaining_incidents = sync_heartbeats_with_heartbeat_incidents() + self.assertEqual(sources, []) + self.assertEqual(new_incidents, []) + self.assertEqual(remaining_incidents, []) + + def test_when_relevant_incidents_exist_return_them(self): + self.assertFalse(Incident.objects.heartbeat_incidents().exists()) + in_timestamp = tznow() - timedelta(seconds=60) + zombie_source, timestamp = create_dead_source("zombie_walking", timestamp=in_timestamp) + + # Get or create heartbeat incident inside the sync + sources, new_incidents, remaining_incidents = sync_heartbeats_with_heartbeat_incidents() + self.assertEqual(Incident.objects.heartbeat_incidents().open().count(), 1) + self.assertEqual(sources, []) + self.assertTrue(new_incidents) + self.assertEqual(remaining_incidents, []) + + def test_when_relevant_incidents_exist_return_incident_without_duplications(self): + self.assertFalse(Incident.objects.heartbeat_incidents().exists()) + in_timestamp = tznow() - timedelta(seconds=60) + zombie_source, timestamp = create_dead_source("zombie_walking", timestamp=in_timestamp) + + # Get or create heartbeat incident outside of the sync + incident, _ = _get_or_create_incident_for_dead_source(zombie_source, incident_owner=self.owner_source) + self.assertEqual(Incident.objects.heartbeat_incidents().open().count(), 1) + + # Get or create heartbeat incident inside the sync, without dupliaction + sources, new_incidents, remaining_incidents = sync_heartbeats_with_heartbeat_incidents() + self.assertEqual(Incident.objects.heartbeat_incidents().open().count(), 1) + self.assertEqual(sources, []) + self.assertEqual(new_incidents, []) + self.assertTrue(remaining_incidents) + + # should not duplicate incident + result_incident = remaining_incidents[0] + self.assertEqual(incident.pk, result_incident.pk) + self.assertEqual(incident.description, result_incident.description) + + def test_when_relevant_incidents_exist_and_source_is_alive_again_return_reanimated_sources(self): + self.assertFalse(Incident.objects.heartbeat_incidents().exists()) + in_timestamp = tznow() - timedelta(seconds=60) + zombie_source, timestamp = create_dead_source("zombie_walking", timestamp=in_timestamp) + _get_or_create_incident_for_dead_source(zombie_source, incident_owner=self.owner_source) + # reawaken source + zombie_source.last_seen = tznow() + zombie_source.save() + + sources, new_incidents, remaining_incidents = sync_heartbeats_with_heartbeat_incidents() + self.assertIn(zombie_source, sources) + self.assertEqual(new_incidents, []) + self.assertEqual(remaining_incidents, []) diff --git a/tests/incident/test_incident.py b/tests/incident/test_incident.py index 36d845fff..15303540e 100644 --- a/tests/incident/test_incident.py +++ b/tests/incident/test_incident.py @@ -8,8 +8,10 @@ EventFactory, StatefulIncidentFactory, StatelessIncidentFactory, + create_fake_incident, + create_fake_source, ) -from argus.incident.models import Event +from argus.incident.models import HEARTBEAT_TAG, SOURCE_TAG_KEY, Event from argus.incident.factories import SourceSystemFactory, SourceUserFactory from argus.util.testing import disconnect_signals, connect_signals @@ -121,3 +123,41 @@ def tearDown(self): def test_given_stateless_incident_then_returns_empty_string(self): self.assertEqual(StatelessIncidentFactory().humanize_age, "") + + +class TestIncidentTagMethods(TestCase): + def test_when_incident_lacks_tag_return_False(self): + incident = create_fake_incident(tags=["existing=badtag"]) + result = incident.has_tags("existing=goodtag") + self.assertFalse(result) + + def test_when_incident_has_tag_return_True(self): + incident = create_fake_incident(tags=["existing=goodtag"]) + result = incident.has_tags("existing=goodtag") + self.assertTrue(result) + + def test_when_incident_lacks_tag_key_return_False(self): + incident = create_fake_incident(tags=["existing=goodtag"]) + result = incident.has_tag_keys("nonexisting") + self.assertFalse(result) + + def test_when_incident_has_tag_key_return_True(self): + incident = create_fake_incident(tags=["existing=goodtag"]) + result = incident.has_tag_keys("existing") + self.assertTrue(result) + + def test_when_incident_lacks_magical_heartbeat_tags_return_False(self): + incident = create_fake_incident(tags=["existing=goodtag"]) + result = incident.is_heartbeat_incident() + self.assertFalse(result) + + def test_when_incident_has_both_magical_heartbeat_tags_return_True(self): + source_name = "hearteat-source" + source = create_fake_source(name=source_name) + tags = [ + HEARTBEAT_TAG, + f"{SOURCE_TAG_KEY}={source.pk}", + ] + incident = create_fake_incident(source=source_name, tags=tags) + result = incident.is_heartbeat_incident() + self.assertTrue(result) diff --git a/tests/incident/test_queryset.py b/tests/incident/test_queryset.py index 9507b9b58..06275e6c7 100644 --- a/tests/incident/test_queryset.py +++ b/tests/incident/test_queryset.py @@ -8,8 +8,12 @@ SourceUserFactory, StatefulIncidentFactory, StatelessIncidentFactory, + TagFactory, + create_dead_source, + create_fake_incident, ) -from argus.incident.models import Incident, Event +from argus.incident.models import Incident, Event, SourceSystem, Tag, get_or_create_default_instances +from argus.incident.heartbeat_utils import _get_or_create_incident_for_dead_source class IncidentQuerySetTestCase(TestCase): @@ -124,3 +128,70 @@ def test_reopen_incidents(self): qs.reopen(self.user, description="Bar") result = set(e.incident for e in Event.objects.filter(type=Event.Type.REOPEN, description="Bar")) self.assertEqual(result, set(qs.all())) + + +class TestTagQuerySetMethods(TestCase): + def test_from_tags_converts_tagstrings_to_existing_tags_and_returns_a_set_of_tags(self): + tag1 = TagFactory() + tag2 = TagFactory() + tag3 = TagFactory() + result = Tag.objects.from_tags(tag1.representation, tag2.representation) + self.assertNotIn(tag3, result) + self.assertEqual(set((tag1, tag2)), result) + + def test_given_existing_tags_from_tag_keys_returns_a_queryset_of_tags_the_given_tag_key(self): + tag1 = TagFactory(key="foo", value="foo") + tag2 = TagFactory(key="foo", value="bar") + tag3 = TagFactory(key="xux", value="foo") + result = Tag.objects.from_tag_keys(tag1.key) + self.assertNotIn(tag3, result) + self.assertEqual(set((tag1, tag2)), set(result)) + + +class TestSourceSystemQuerySet(TestCase): + def setUp(self): + disconnect_signals() + + def tearDown(self): + connect_signals() + + def test_when_no_dead_sources_return_empty_queryset(self): + result = SourceSystem.objects.dead(timezone.now()) + self.assertFalse(result.exists()) + + def test_when_a_dead_source_return_annotated_queryset(self): + zombie_source, timestamp = create_dead_source("zombie_walking") + result = SourceSystem.objects.dead(timestamp) + self.assertEqual(result.count(), 1) + self.assertEqual(result[0].name, "zombie_walking") + self.assertTrue(result[0].next_heartbeat) + + +@tag("db") +class TestIncidentQuerySetTagMethods(TestCase): + def test_when_no_incident_has_source_with_key_return_empty_queryset(self): + self.assertFalse(Incident.objects.from_tag_keys("blbl", "nonexistent").exists()) + + def test_when_incident_has_source_with_key_return_queryset_with_incident(self): + incident = create_fake_incident(tags=["blbl=foo"]) + result = Incident.objects.from_tag_keys("blbl") + self.assertIn(incident, result) + + +@tag("db") +class TestIncidentQuerySetHeartbeatIncidents(TestCase): + def setUp(self): + _, sst, self.owner_source = get_or_create_default_instances() + self.zombie_source, self.timestamp = create_dead_source("zombie_walking") + self.irrelevant_incident = create_fake_incident(source=self.owner_source.name, tags=["test=test"]) + + def test_when_no_heartbeat_incidents_returns_empty_queryset(self): + self.assertFalse(Incident.objects.heartbeat_incidents().exists()) + + def test_when_heartbeat_incidents_exist_return_them(self): + incident, _ = _get_or_create_incident_for_dead_source( + self.zombie_source, incident_owner=self.owner_source, timestamp=self.timestamp + ) + result = Incident.objects.heartbeat_incidents() + self.assertIn(incident, result) + self.assertNotIn(self.irrelevant_incident, result) diff --git a/tests/incident/test_source_system.py b/tests/incident/test_source_system.py index c90c82f1c..512b27c17 100644 --- a/tests/incident/test_source_system.py +++ b/tests/incident/test_source_system.py @@ -50,6 +50,14 @@ def test_updates_last_seen_field_to_current_timestamp(self): self.assertEqual(source.last_seen, testtime) +class SourceSystemIsDeadTests(TestCase): + def test_when_heartbeat_frequency_not_set_always_returns_None(self): + source = SourceSystemFactory(heartbeat_frequency=None) + self.assertIsNone(source.heartbeat_frequency) + # timestamp does not matter + self.assertIsNone(source.is_dead(datetime.max)) + + @tag("api", "integration") class SourceSystemPostingTests(APITestCase): def setUp(self):