src/monitor_app/ready_history.py

0001 """Daily ready-task depth from the production front's recorded cycles."""
0002 from collections import defaultdict
0003 from datetime import datetime, timedelta, timezone
0004
0005 from .models import AppLog
0006
0007 DECISION_MAX_AGE = timedelta(minutes=75)  # Hourly decision heartbeat plus one cycle's grace.
0008 CYCLE_MAX_AGE = timedelta(minutes=15)
0009
0010
0011 def ready_samples(events):
0012     """Join each cycle to its current queues' most recent depth readings.
0013
0014     Decisions are stored on changes or at least hourly. The cycle records
0015     the number of regulated queues; its newest fresh queue readings avoid
0016     carrying retired queue state through later queue-registry changes.
0017     """
0018     latest = {}
0019     for stamp, action, queue, depth, queue_count, outcome in events:
0020         if action == 'front_decision':
0021             if queue and isinstance(depth, int) and depth >= 0:
0022                 latest[queue] = (stamp, depth)
0023             continue
0024         fresh = sorted((point for point in latest.values()
0025                         if stamp - point[0] <= DECISION_MAX_AGE), reverse=True)
0026         complete = (outcome == 'ok' and isinstance(queue_count, int)
0027                     and queue_count >= 0 and len(fresh) >= queue_count)
0028         yield stamp, sum(depth for _, depth in fresh[:queue_count]) if complete else None
0029
0030
0031 def daily_means(samples, start, end):
0032     """Time-weight samples over observed time, retaining gaps and coverage."""
0033     area, covered = defaultdict(float), defaultdict(float)
0034     for index, (stamp, count) in enumerate(samples):
0035         if count is None:
0036             continue
0037         begin = max(start, stamp)
0038         stop = min(end, stamp + CYCLE_MAX_AGE,
0039                    samples[index + 1][0] if index + 1 < len(samples) else end)
0040         while begin < stop:
0041             midnight = datetime.combine(begin.date() + timedelta(days=1), datetime.min.time(),
0042                                         tzinfo=timezone.utc)
0043             boundary = min(stop, midnight)
0044             seconds = (boundary - begin).total_seconds()
0045             key = str(begin.date())
0046             area[key] += count * seconds
0047             covered[key] += seconds
0048             begin = boundary
0049     return {day: {'ready_tasks': area[day] / seconds,
0050                   'ready_coverage_hours': seconds / 3600}
0051             for day, seconds in covered.items() if seconds > 0}
0052
0053
0054 def history(start, end):
0055     """One bounded action-stream read; no individual job or task queries."""
0056     begin = datetime.combine(start, datetime.min.time(), tzinfo=timezone.utc)
0057     stop = datetime.combine(end, datetime.min.time(), tzinfo=timezone.utc)
0058     rows = (AppLog.objects.filter(
0059         app_name='epicprod', timestamp__gte=begin - timedelta(hours=2), timestamp__lt=stop,
0060         extra_data__action__in=('front_decision', 'front_cycle'))
0061         .order_by('timestamp', 'id').values_list(
0062             'timestamp', 'extra_data__action', 'extra_data__subject_key',
0063             'extra_data__ready_total', 'extra_data__queues', 'extra_data__outcome'))
0064     return daily_means(list(ready_samples(rows.iterator())), begin, stop)