swf-testbed/episodes/backfill.py

0001 """Backfill workflow episodes from recorded messages.
0002
0003 Replays a past execution's recorded bus messages through the same
0004 definitions and engine the live builder runs, then drives the
0005 completion pass to closure. The episode carries the recorded times,
0006 not the replay time. Bounded by message-log retention.
0007
0008 Usage (testbed venv, SWF_MONITOR_HTTP_URL and SWF_API_TOKEN in the
0009 environment):
0010
0011     python -m episodes.backfill <execution_id> [<execution_id> ...]
0012 """
0013
0014 import os
0015 import sys
0016 import time
0017
0018 import requests
0019
0020 from swf_common_lib.episodes import EpisodeBuilder, MonitorEpisodeIngest
0021
0022 from . import ALL_DEFINITIONS
0023
0024 COMPLETION_ATTEMPTS = 60
0025 COMPLETION_INTERVAL_SECONDS = 5
0026
0027
0028 def backfill(execution_id: str) -> bool:
0029     base = (os.environ.get('SWF_MONITOR_HTTP_URL') or '').rstrip('/')
0030     token = os.environ.get('SWF_API_TOKEN') or ''
0031     if not base:
0032         print('SWF_MONITOR_HTTP_URL is not set')
0033         return False
0034
0035     session = requests.Session()
0036     if token:
0037         session.headers['Authorization'] = f'Token {token}'
0038     response = session.get(f'{base}/api/workflow-messages/',
0039                            params={'execution_id': execution_id},
0040                            timeout=60)
0041     response.raise_for_status()
0042     rows = response.json()
0043     if isinstance(rows, dict):
0044         rows = rows.get('results') or []
0045     rows.sort(key=lambda row: row.get('sent_at') or '')
0046     messages = []
0047     for row in rows:
0048         content = row.get('message_content') or {}
0049         if content.get('execution_id') != execution_id:
0050             continue
0051         # The recorded sent time backs any message whose writer
0052         # stamped no timestamp of its own.
0053         if row.get('sent_at'):
0054             content.setdefault('sent_at', row['sent_at'])
0055         messages.append(content)
0056     if not messages:
0057         print(f'{execution_id}: no recorded messages')
0058         return False
0059
0060     ingest = MonitorEpisodeIngest(base_url=base, token=token,
0061                                   builder_identity='episode-backfill')
0062     builder = EpisodeBuilder(
0063         [definition() for definition in ALL_DEFINITIONS], ingest)
0064     handled = sum(1 for m in messages if builder.handle_message(m))
0065     print(f'{execution_id}: {handled}/{len(messages)} recorded messages '
0066           f'replayed')
0067
0068     for _ in range(COMPLETION_ATTEMPTS):
0069         builder.tick()
0070         if not builder.active:
0071             print(f'{execution_id}: episode closed')
0072             return True
0073         time.sleep(COMPLETION_INTERVAL_SECONDS)
0074     print(f'{execution_id}: completion did not converge')
0075     return False
0076
0077
0078 if __name__ == '__main__':
0079     if len(sys.argv) < 2:
0080         print(__doc__)
0081         sys.exit(2)
0082     results = [backfill(execution_id) for execution_id in sys.argv[1:]]
0083     sys.exit(0 if all(results) else 1)