swf-monitor/scripts/content-validate.py

0001 #!/usr/bin/env python3
0002 """content-validate.py — reconcile a sample's dataset against the record.
0003
0004 The production-ops agent's doer for content validation (swf-epicprod
0005 docs/EPICPROD_VALIDATION.md, Content validation). The record states which
0006 file is the delivered output of each work unit and how many events it
0007 carries; the Rucio dataset states what it holds. This reads both and
0008 names what would make an availability signal false.
0009
0010 It proposes and never acts: nothing is detached, nothing is written to
0011 Rucio. Acceptance is an operator's single action, and acceptance never
0012 deletes.
0013
0014 Reading JLab Rucio is a remote call, so it belongs here rather than in a
0015 page. The finding is stored as a cached product, `content_validation:<task>`,
0016 which is what pages read.
0017
0018 Django-bootstrap standalone script — also usable by hand::
0019
0020     cd /data/wenauseic/github/swf-monitor/src
0021     ../../swf-testbed/.venv/bin/python ../scripts/content-validate.py \
0022         [--task <composed name>] [--hours 168] [--json]
0023
0024 The last stdout line is a JSON summary; findings go to stderr readably.
0025 """
0026 import argparse
0027 import json
0028 import os
0029 import sys
0030 from datetime import datetime, timedelta, timezone as dt_timezone
0031
0032 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0033 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0034 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0035
0036 import django  # noqa: E402
0037 django.setup()
0038
0039 DEFAULT_HOURS = 168
0040
0041
0042 def _log(msg):
0043     print(msg, file=sys.stderr, flush=True)
0044
0045
0046 def tasks_to_check(since, name=''):
0047     """Tasks whose record has moved: those are the ones worth reconciling."""
0048     from pcs.models import DeliveredOutput, ProdTask
0049     if name:
0050         task = ProdTask.objects.filter(name=name).first() or \
0051             ProdTask.objects.filter(dataset__composed_name=name).first()
0052         return [task] if task else []
0053     ids = (DeliveredOutput.objects
0054            .filter(updated_at__gte=since)
0055            .values_list('prod_task_id', flat=True).distinct())
0056     return list(ProdTask.objects.filter(id__in=list(ids)))
0057
0058
0059 def store(task, findings):
0060     """Keep the finding where pages can read it without a remote call.
0061
0062     The house cached-product store, written by a forced rebuild: this doer
0063     is the builder, and a page is only ever a reader (docs/CACHED_PRODUCTS.md).
0064     """
0065     from monitor_app.cached_product import get_product
0066     key = f'content_validation:{task.composed_name or task.name}'
0067     payload = {'task': task.composed_name or task.name, 'findings': findings,
0068                'built_at': datetime.now(dt_timezone.utc).isoformat()}
0069     get_product(key, lambda: payload, ttl_seconds=24 * 3600, refresh=True)
0070     return key
0071
0072
0073 def main():
0074     parser = argparse.ArgumentParser()
0075     parser.add_argument('--task', default='', help='one task by composed name')
0076     parser.add_argument('--hours', type=float, default=DEFAULT_HOURS,
0077                         help='reconcile tasks whose record moved in this window')
0078     parser.add_argument('--json', action='store_true',
0079                         help='print every finding as JSON')
0080     args = parser.parse_args()
0081
0082     from pcs import content_validation
0083
0084     since = datetime.now(dt_timezone.utc) - timedelta(hours=args.hours)
0085     summary = {'tasks': 0, 'datasets': 0, 'sound': 0, 'unsound': 0,
0086                'unreadable': 0, 'delivered_events': 0}
0087
0088     for task in tasks_to_check(since, name=args.task):
0089         findings = content_validation.reconcile_all(task)
0090         if not findings:
0091             continue
0092         summary['tasks'] += 1
0093         for finding in findings:
0094             summary['datasets'] += 1
0095             if finding['error']:
0096                 summary['unreadable'] += 1
0097             elif finding['sound']:
0098                 summary['sound'] += 1
0099             else:
0100                 summary['unsound'] += 1
0101             summary['delivered_events'] += finding['delivered_events']
0102             _log(f"{finding['task']} {finding['dataset']}: "
0103                  f"units={finding['units']} matched={len(finding['matched'])} "
0104                  f"events={finding['delivered_events']} "
0105                  f"orphans={len(finding['orphan_files'])} "
0106                  f"missing={len(finding['units_without_file'])} "
0107                  f"several={len(finding['units_with_several'])} "
0108                  f"no_count={len(finding['files_without_events'])} "
0109                  f"{'sound' if finding['sound'] else 'NOT SOUND'}"
0110                  f"{' — ' + finding['error'] if finding['error'] else ''}")
0111             if args.json:
0112                 _log(json.dumps(finding, indent=2, default=str))
0113         try:
0114             store(task, findings)
0115         except Exception as e:                                # noqa: BLE001
0116             _log(f'ERROR: finding not stored for {task}: {e}')
0117
0118     print(json.dumps(summary))
0119     return 0
0120
0121
0122 if __name__ == '__main__':
0123     sys.exit(main())