import copy
import datetime
import logging
from django.conf import settings
from django.utils import timezone
from .forms import DateTimeForm
from .status import STATUS
logger = logging.getLogger("Happyface4")
[docs]
class DummyInstanceStatus(object):
"""
Helper object that can be used to propagate a single instance status to the template.
This is generally only used for the statuses info(-1) and technical issue(-2).
"""
def __init__(self, time, status):
"""Initialize the status.
:param time: The timestamp for this status. Normally just the time of the request.
:type time: datetime.datetime
:param status: The status in form of a number between -2 and 2.
:type status: int
"""
self.time = time
self.status = status
self.itered = False
def __len__(self):
# Only one status
return 1
def __iter__(self):
# little hack to make this object iterable so that it returns itself
# in only one iteration
return [self].__iter__()
[docs]
def latest(self, _):
# To imitate the latest method of an QuerySet. As this object only represent one
# status it returns itself.
return self
[docs]
def last(self):
# To imitate the `last` method of a QuerySet. As this object only represent one
# status it returns itself.
return self
[docs]
class DummyInstanceStatusList(list):
"""Sequence of :class:`DummyInstanceStatus` objects that mimics the QuerySet
interface consumed by the templates and the nav aggregation (iteration,
``len``, ``latest`` and ``last``).
Like the stored-status QuerySets (see
``InstanceStatusManager.get_instance_status``) the sequence is ordered
chronologically ascending, so ``last()`` returns the newest entry.
"""
[docs]
def latest(self, field):
# imitate QuerySet.latest: the newest status by time
return max(self, key=lambda s: s.time)
[docs]
def last(self):
# imitate QuerySet.last: newest status in chronological ordering
return self[-1]
def _latest_status_time(instance_name, category, time):
"""Timestamp of the newest stored status at or before ``time``.
Looks up the instance's own last status first (its real check grid),
then the globally newest stored status (the getData cycle grid), and
returns ``None`` when nothing is stored at all.
"""
from .models import InstanceStatus
own = (
InstanceStatus.objects.filter(
time__lte=time, instance__name=instance_name, category__name=category
)
.order_by("-time")
.values_list("time", flat=True)
.first()
)
if own is not None:
return own
return (
InstanceStatus.objects.filter(time__lte=time)
.order_by("-time")
.values_list("time", flat=True)
.first()
)
[docs]
def expand_dummy_status(dummy, time, time_range, instance, anchor=None):
"""Spread a single :class:`DummyInstanceStatus` over the requested time range.
Info and technical-issue instances (and instances without any stored
status in the range) previously rendered a *single* overview bar, which
broke the timestep alignment on the overview page (HFC #23). This helper
builds a chronological series on the same grid as real instances: one
step per ``PULL_INTERVAL * get_data_every`` going back from ``anchor``.
The series is anchored to the **data grid**, not to the request time, so
overlapping requests for the same time-slot always render the same bar
timestamps and the bars line up with the stored-status rows (the newest
bar sits at the instance's last real check before ``time``). Without any
stored status at all the series falls back to anchoring at ``time``.
:param dummy: the single placeholder status to expand
:param time: the requested time
:param time_range: the range to cover; may be ``None``
:param instance: the instance configuration dict (for ``get_data_every``)
:param anchor: newest timestep of the series (defaults to ``time``);
callers pass the instance's last real check time for grid alignment
:returns: a :class:`DummyInstanceStatusList` covering the range
"""
step = settings.PULL_INTERVAL * instance.get("get_data_every", 1)
if not time_range or time_range <= step:
steps = 1
else:
steps = max(int(time_range / step), 1)
newest = anchor or time
return DummyInstanceStatusList(
DummyInstanceStatus(time=newest - (steps - 1 - i) * step, status=dummy.status)
for i in range(steps)
)
[docs]
class CategoryNav(object):
def __init__(self, name, instance):
self.name = name
self.instances = [instance]
[docs]
def latest_status(self):
return max([i.latest_status() for i in self.instances])
[docs]
def statuses_list(self):
cat_status_list = []
# find instance with most status entries
most_status_entries = copy.deepcopy(
max([inst.status for inst in self.instances], key=len)
)
# find the worst status at each time
for timepoint in most_status_entries:
instance_status_list = []
for inst in self.instances:
if type(inst.status) is not DummyInstanceStatus:
timediffs = [
abs(timepoint.time - status.time) for status in inst.status
]
nearest_status_index = timediffs.index(min(timediffs))
nearest_status = inst.status[nearest_status_index]
if type(nearest_status) is int:
instance_status_list.append(nearest_status)
else:
instance_status_list.append(nearest_status.status)
else:
instance_status_list.append(inst.status.status)
cat_status_list.append(
{"time": timepoint.time, "status": max(instance_status_list)}
)
return cat_status_list
[docs]
class InstanceNav(object):
def __init__(self, name, verbose_name, status, order=-1, carry_over=None):
self.status = status
self.verbose_name = verbose_name
self.name = name
self.order = order
# optional per-module status carry-over for the overview grid
# (HFC #2/#23): how long the last known status stays valid for
# display, e.g. timedelta(hours=2) for modules that report sparsely.
# None keeps the default behaviour (only the nearest slot is filled).
self.carry_over = carry_over
[docs]
def latest_status(self):
# returns the newest status
return self.status.latest("time").status
[docs]
def get_Analyses_from_apps(apps):
filtered_apps = []
for app in apps.get_app_configs():
if app.path.startswith(settings.ANALYSES_DIR):
filtered_apps.append(app)
return filtered_apps
[docs]
def get_apps_with_instances_in_category(app_config_list, category):
"""
Creates a list of all apps with their respective instances that are in the category.
This is another filtering step that is used to determine which analyses to
call with what arguments to get all the divs in the category.
:param app_config_list: list of configuration objects of the Analyses.
:type app_config_list: list of Subclasses of the ``django.apps.AppConfig`` class
:param category: the name of the category in which the analyses have registered an instance
:type category: str
:returns: all apps that have an instance in the category
:rtype: list of tuples of the ``AppConfig`` subclass of an analysis with the
instances of the corresponding class
.. note::
The returned tuples contain the ``AppConfig`` subclass as well as the
entire configuration dict of that particular instance.
"""
apps_in_category = []
for app_config in app_config_list:
instances_in_category = []
for instance in app_config.instances:
if instance["category"] == category:
instances_in_category.append(instance)
if len(instances_in_category) != 0:
apps_in_category.append((app_config, instances_in_category))
return apps_in_category
[docs]
def render_instances(analyses_list, category, time, request):
divs = []
instance_orders = []
ananlyses_to_render = get_apps_with_instances_in_category(analyses_list, category)
# catch case where the category string doesn't match the category of any analysis
if not ananlyses_to_render:
return []
for app_config, instances_list in ananlyses_to_render:
# the instance is a tuple where the first item is the app config and the second is a list of all
# instances of the app that have to be rendered
for instance in instances_list:
anaDiv = app_config.builddiv(time, instance, request)
divs.append(anaDiv)
instance_orders.append(instance.get("order", -1))
# show analyses with negative/unspecified ordering specifications last
instance_orders = [len(instance_orders) if o < 0 else o for o in instance_orders]
# order analyses according to ordering specifications
sorted_orders, sorted_divs = zip(*sorted(zip(instance_orders, divs)))
return sorted_divs
[docs]
def get_statuses_of_instances(app_list, time, time_range):
"""
generates a list of nav_elements from a list of analyses
:param app_list: list of the analysis configurations.
:type app_list: list of ``django.apps.AppConfig`` subclasses
:param time: the requested time of the statuses.
:type time: ~datetime.datetime
:param time_range: the time range in which to look for the statuses.
:type time_range: ~datetime.timedelta
:returns: the sorted list of :class:`CategoryNav` elements.
:rtype: list
"""
from .models import InstanceStatus
categories_list = []
dummy_anchor_cache = {}
for app in app_list:
for instance in app.instances:
name = instance["name"]
verbose_name = instance["verbose_name"]
cat = instance["category"]
order = instance.get("order", -1)
if instance.get("is_info", False):
instance_status = DummyInstanceStatus(time=time, status=STATUS.INFO)
else:
try:
instance_status = app.get_instance_status(
instance=instance, time=time, time_range=time_range
)
except:
logger.error(
f"No 'get_instance_status' method implemented or 'is_info' flag set for instance {name}."
)
instance_status = DummyInstanceStatus(
time=time, status=STATUS.TECHNICAL_ISSUE
)
if type(instance_status) is DummyInstanceStatus:
# HFC #23: info / technical-issue placeholders (and instances
# without stored statuses) are spread over the requested
# range so their overview rows align with the timestep grid
# of the other instances instead of rendering a single bar.
# The series is anchored to the instance's last real check
# (or the global getData cycle grid) so overlapping requests
# for the same time-slot always render the same timestamps.
key = (name, cat)
if key not in dummy_anchor_cache:
dummy_anchor_cache[key] = _latest_status_time(name, cat, time)
instance_status = expand_dummy_status(
instance_status,
time,
time_range,
instance,
anchor=dummy_anchor_cache[key],
)
instance_object = InstanceNav(
name=name,
verbose_name=verbose_name,
status=instance_status,
order=order,
carry_over=instance.get("carry_over"),
)
# check if category of current instance already exists in the categories list
if cat not in [n.name for n in categories_list]:
# if not create a new Category object and pass it the current instance
# object
categories_list.append(CategoryNav(name=cat, instance=instance_object))
else:
# if the category object already exists append the instance object to the
# instances list
for c in categories_list:
if c.name == cat:
c.instances.append(instance_object)
# sort the array
categories_list.sort(key=lambda cat: cat.name.lower())
for cat in categories_list:
# order instances as specified in config
cat.instances.sort(
key=lambda ins: len(cat.instances) if ins.order < 0 else ins.order
)
return categories_list
# severity order for status-based overview sorting (HFC #2): problems first,
# ok after them, info last
_STATUS_SEVERITY_RANK = {
STATUS.CRITICAL: 0,
STATUS.WARNING: 1,
STATUS.TECHNICAL_ISSUE: 2,
STATUS.OK: 3,
STATUS.INFO: 4,
}
[docs]
def sort_categories(categories_list, sort_key="name"):
"""Sort nav categories for the overview page (HFC #2).
:param categories_list: the list of :class:`CategoryNav` objects as
produced by :func:`get_statuses_of_instances`
:param sort_key: ``name`` (default, alphabetical, unchanged behavior) or
``status`` (most severe category first: critical > warning >
technical issue > ok > info, ties broken alphabetically)
:returns: a newly sorted list; the input list is not modified
"""
if sort_key == "status":
return sorted(
categories_list,
key=lambda c: (
_STATUS_SEVERITY_RANK.get(
c.latest_status(), len(_STATUS_SEVERITY_RANK)
),
c.name.lower(),
),
)
return sorted(categories_list, key=lambda c: c.name.lower())
def _global_latest_status_time(time):
"""Timestamp of the globally newest stored status at or before ``time``
(the getData cycle grid anchor), or ``None`` when nothing is stored."""
from .models import InstanceStatus
return (
InstanceStatus.objects.filter(time__lte=time)
.order_by("-time")
.values_list("time", flat=True)
.first()
)
def _nearest_times(times, slots):
"""For each slot, the nearest entry of ``times`` (ties resolved to the
earlier time); ``None`` entries when ``times`` is empty."""
result = []
for slot in slots:
if times:
result.append(
min(times, key=lambda t: (abs((slot - t).total_seconds()), t))
)
else:
result.append(None)
return result
[docs]
def build_overview_grid(categories, time, slot_count=None, anchor=None):
"""Map the nav categories onto a fixed overview slot grid.
The grid holds ``slot_count`` equidistant slots of ``PULL_INTERVAL``
covering the overview range, anchored at the globally newest stored
status at or before ``time``, snapped onto the ``PULL_INTERVAL`` lattice
(fallback: ``time``, snapped likewise). Every category
and every instance is mapped onto exactly these slots, so all rows of
the overview page are horizontally aligned regardless of how much data
each instance has.
Mapping rules:
* stored statuses are assigned to their nearest slot; a slot without a
status renders as "no data" (``status = None``),
* per-module carry-over (``InstanceNav.carry_over``, from the instance
config key ``carry_over``): the last known status stays valid for the
configured window — every slot whose latest stored status is at most
``carry_over`` old shows that status instead of "no data" (for modules
that report sparsely, e.g. SAM tests every hour or SITECONF once a
day). Without the key only the nearest slot is filled (unchanged
default),
* the link target of a slot is the **nearest stored status time** of
that instance (ties resolved to the earlier time) — for carried slots
it is the carried status's own time, so the shown page matches the
displayed status; slots of placeholder series (info / technical
issue, no stored rows) link to the slot time itself,
* placeholder series fill every slot with their constant status,
* a category slot shows the worst status of its instances at that slot
and links to the nearest stored time across the category.
:param categories: list of :class:`CategoryNav` as produced by
:func:`get_statuses_of_instances`
:param time: the requested time
:param slot_count: number of slots (default ``OVERVIEW_RANGE /
PULL_INTERVAL``)
:param anchor: newest slot time (default: the newest stored status time
at or before ``time``, snapped onto the ``PULL_INTERVAL`` lattice;
fallback: ``time`` snapped likewise). An explicitly passed anchor is
snapped onto the lattice too (a no-op for on-lattice times)
:returns: template-friendly category dicts
``{"name", "slots": [{"time", "status", "link_time"}...],
"instances": [{"name", "verbose_name", "slots": [...]}]}``
"""
if slot_count is None:
slot_count = max(int(settings.OVERVIEW_RANGE / settings.PULL_INTERVAL), 1)
step = settings.PULL_INTERVAL
if anchor is None:
anchor = _global_latest_status_time(time) or time
# snap the anchor onto the PULL_INTERVAL lattice so slot times (and
# the tick labels) stay on stable :00/:15/:30/:45 marks instead of
# inheriting the newest stored status's minute offset; a no-op for
# anchors already on the lattice. The minute-of-hour floor assumes
# PULL_INTERVAL is a whole-minute divisor of 60 (the shipped 15 min;
# the overview range and tick cadence couple to it anyway) — other
# values would drift the labels across hours.
minutes = int(step.total_seconds() // 60)
anchor = anchor.replace(
minute=(anchor.minute // minutes) * minutes,
second=0,
microsecond=0,
)
slots = [anchor - (slot_count - 1 - i) * step for i in range(slot_count)]
result = []
for cat in categories:
cat_status = [None] * slot_count
all_times = []
instances_data = []
for inst in cat.instances:
status_obj = inst.status
if type(status_obj) is DummyInstanceStatusList:
constant = status_obj[0].status
slots_data = [
{"time": t, "status": constant, "link_time": t} for t in slots
]
else:
rows = sorted(status_obj, key=lambda r: r.time)
times = [r.time for r in rows]
links = _nearest_times(times, slots)
carry = getattr(inst, "carry_over", None)
# base fill: every row maps to its NEAREST slot (clamped to
# the grid, later rows win). Rejecting rows farther than
# half a slot from their nearest slot made the fill depend
# on how well the module's row cadence aligns with the slot
# lattice (which follows the anchor's cycle drift) — a few
# minutes of lattice mismatch greyed out whole ribbons.
slots_data = [
{"time": t, "status": None, "link_time": lt}
for t, lt in zip(slots, links)
]
for row in rows:
idx = int(
round(
(row.time - slots[0]).total_seconds() / step.total_seconds()
)
)
idx = max(0, min(idx, slot_count - 1))
slots_data[idx]["status"] = row.status
slots_data[idx]["link_time"] = row.time
if carry is not None:
# carry-over: slots without their own row show the last
# known status within the configured window
row_idx = 0
for i, sd in enumerate(slots_data):
if sd["status"] is not None:
continue
while (
row_idx + 1 < len(rows)
and rows[row_idx + 1].time <= slots[i]
):
row_idx += 1
if (
rows
and rows[row_idx].time <= slots[i]
and slots[i] - rows[row_idx].time <= carry
):
sd["status"] = rows[row_idx].status
sd["link_time"] = rows[row_idx].time
all_times.extend(times)
instances_data.append(
{
"name": inst.name,
"verbose_name": inst.verbose_name,
"slots": slots_data,
}
)
for i, sd in enumerate(slots_data):
if sd["status"] is not None:
cat_status[i] = (
max(cat_status[i], sd["status"])
if cat_status[i] is not None
else sd["status"]
)
cat_links = _nearest_times(all_times, slots)
cat_slots = [
{"time": t, "status": s, "link_time": l if s is not None else None}
for t, s, l in zip(slots, cat_status, cat_links)
]
result.append(
{
"name": cat.name,
"slots": cat_slots,
"instances": instances_data,
}
)
return result