swf-monitor/tests/test_es_snapshot_timeline.py

0001 """A running snapshot must never be presented as a preemption or extended in time."""
0002 import os
0003 import unittest
0004
0005 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0006 import django
0007 django.setup()
0008
0009 from monitor_app.panda.queries import es_record_report, es_slot_timeline, _es_range_query
0010 from monitor_app.es_plot import slot_plot_svg
0011
0012
0013 class SnapshotTimelineTest(unittest.TestCase):
0014     def setUp(self):
0015         self.record = {
0016             'started_at': 1000, 'written_at': 1300, 'slots': 2,
0017             'done': [{'unit_id': 'saved', 'slot': 0, 'started_at': 1010,
0018                       'ended_at': 1100, 'events': 10, 'close': 1}],
0019             'in_flight': [{'unit_id': 'working', 'slot': 0, 'started_at': 1200, 'events': 10}],
0020             'awaiting_close': [{'unit_id': 'pending', 'slot': 1, 'started_at': 1010,
0021                                 'ended_at': 1250, 'events': 10}],
0022             'closes': [{'index': 1, 'ok': True, 'events': 10,
0023                         'started_at': 1100, 'ended_at': 1120}],
0024             'closing': [{'index': 2, 'units': ['pending'], 'started_at': 1280}],
0025         }
0026         self.job = {'starttime': '1970-01-01T00:16:40Z', 'endtime': None}
0027
0028     def test_running_snapshot_has_no_cut_and_stops_at_observation(self):
0029         es = es_record_report(self.record, 'running')
0030         timeline = es_slot_timeline(self.job, es)
0031         self.assertIsNone(es['preempted'])
0032         self.assertEqual(es['ranges_interrupted'], [])
0033         self.assertEqual(es['ranges_unshipped'], [])
0034         self.assertEqual(timeline['wall_s'], 300)
0035         self.assertIsNone(timeline['cut_s'])
0036         self.assertTrue(timeline['snapshot_stale'])
0037         self.assertIn('running', {u['status'] for row in timeline['rows'] for u in row['units']})
0038         svg = slot_plot_svg(timeline)
0039         self.assertIn('close 2: running', svg)
0040         self.assertNotIn('preempted at', svg)
0041
0042     def test_lost_job_preserves_saved_work_and_marks_only_unsaved_work(self):
0043         es = es_record_report(self.record, 'finished', loss_confirmed=True)
0044         job = dict(self.job, endtime='1970-01-01T00:30:00Z')
0045         timeline = es_slot_timeline(job, es)
0046         self.assertEqual(len(es['ranges_done']), 1)
0047         self.assertEqual(es['preempted']['unshipped_events'], 10)
0048         self.assertEqual(es['preempted']['interrupted_events'], 10)
0049         self.assertEqual(timeline['cut_s'], 300)
0050         self.assertEqual(timeline['wall_s'], 300)
0051         self.assertFalse(timeline['live'])
0052
0053     def test_normal_harness_end_is_not_a_lost_node(self):
0054         es = es_record_report(dict(self.record, ended_at=1300), 'finished')
0055         self.assertIsNone(es['preempted'])
0056         self.assertFalse(es['live'])
0057         timeline = es_slot_timeline(dict(self.job, endtime='1970-01-01T00:21:40Z'), es)
0058         self.assertTrue(timeline['normal_end'])
0059         self.assertFalse(timeline['snapshot_stale'])
0060
0061     def test_lost_worker_observation_outlives_stale_panda_endtime(self):
0062         es = es_record_report(self.record, 'finished', loss_confirmed=True)
0063         job = dict(self.job, endtime='1970-01-01T00:16:55Z')
0064         timeline = es_slot_timeline(job, es)
0065         self.assertEqual(timeline['wall_s'], 300)
0066         self.assertGreaterEqual(timeline['tail_idle_s'], 0)
0067         self.assertLessEqual(timeline['busy_fraction'], 1)
0068         self.assertTrue(all(u['end_s'] <= timeline['wall_s']
0069                             for row in timeline['rows'] for u in row['units']))
0070
0071     def test_missing_final_report_is_not_evidence_of_preemption(self):
0072         es = es_record_report(self.record, 'finished')
0073         timeline = es_slot_timeline(dict(self.job, endtime='1970-01-01T00:16:55Z'), es)
0074         self.assertTrue(timeline['incomplete'])
0075         self.assertIsNone(timeline['cut_s'])
0076         self.assertEqual(timeline['wall_s'], 300)
0077         self.assertEqual(es['ranges_interrupted'], [])
0078         self.assertEqual(es['ranges_unshipped'], [])
0079         self.assertTrue(all(u['status'] not in ('interrupted', 'unshipped')
0080                             for row in timeline['rows'] for u in row['units']))
0081
0082
0083
0084     def test_completed_successor_supersedes_cancelled_bookkeeping_only(self):
0085         # Range1 was recovered by consumer102; range2 really remains cancelled.
0086         # Range3 remains ready under the current jobset.
0087         data = [
0088             (1, 101, 10, 1, 4, 1, 1, None, 1, None, None),
0089             (1, 101, 10, 2, 4, 2, 2, None, 1, None, None),
0090             (1, 102, 10, 1, 3, 1, 1, None, 1, None, None),
0091             (1, 103, 10, 3, 0, 3, 3, None, 1, None, None),
0092         ]
0093         class Cursor:
0094             def __enter__(self): return self
0095             def __exit__(self, *args): pass
0096             def execute(self, *args): pass
0097             def fetchall(self): return data
0098         class Connection:
0099             def cursor(self): return Cursor()
0100         result = _es_range_query(Connection(), 'test', [], 0, consumer_ids=[102])
0101         self.assertEqual(result['events_requested'], 3)
0102         self.assertEqual(result['events_done'], 1)
0103         self.assertEqual({c['status']: c['n'] for c in result['range_counts']}, {0: 1, 4: 1})
0104         self.assertEqual(result['attempt_counts'][0]['n'], 1)
0105
0106 if __name__ == '__main__':
0107     unittest.main()