swf-monitor/scripts/storage-sweep.py

0001 #!/usr/bin/env python3
0002 """storage-sweep.py — the storage pass: production output on the JLab
0003 RSEs from registration until it settles, kept in the storage store and
0004 published as the epicprod ``storage`` Snapper component.
0005
0006 The prod-ops agent's doer for the ``storage_sweep`` message, enqueued
0007 by cron every four hours. Logic in ``swf_epicprod/analytics/storage.py``,
0008 publication in ``monitor_app/snapper_storage.py``; design in
0009 ``swf-epicprod/docs/STORAGE.md``. Django-bootstrap standalone script —
0010 also usable by hand.
0011
0012 Usage::
0013
0014     cd /data/wenauseic/github/swf-monitor/src
0015     source ../../swf-testbed/.venv/bin/activate && source ~/.env
0016     python ../scripts/storage-sweep.py [--campaigns 26.07]
0017                                        [--limit-files N] [--limit-datasets N]
0018                                        [--resume PASS_ID] [--publish-only]
0019                                        [--no-publish] [--dump projection.json]
0020
0021 A run with a limit works on a copy of the store and never publishes.
0022 ``--publish-only`` skips the crawl and publishes the projection of the
0023 store's last completed pass, for a pass whose publish step failed.
0024 """
0025 import argparse
0026 import json
0027 import os
0028 import signal
0029 import sys
0030
0031 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0032 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0033 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0034
0035 import django  # noqa: E402
0036 django.setup()
0037
0038 from swf_epicprod.analytics.storage import (  # noqa: E402
0039     PassInProgress, log, mark_pass_interrupted, project_store, run_pass)
0040
0041
0042 def _on_signal(signum, frame):
0043     raise KeyboardInterrupt(signal.Signals(signum).name)
0044
0045
0046 def main():
0047     # A signal to the doer (a stop by hand, or systemd's stop timeout)
0048     # ends the pass through the interrupted path below instead of killing
0049     # the process with its pass row left as if still running.
0050     signal.signal(signal.SIGTERM, _on_signal)
0051     parser = argparse.ArgumentParser()
0052     parser.add_argument('--publish-only', action='store_true',
0053                         help='no catalog reads: publish the projection of '
0054                              'the store\'s last completed pass (after a '
0055                              'failed publish)')
0056     parser.add_argument('--campaigns', default='',
0057                         help='comma-separated campaign families '
0058                              '(default: the delivery record\'s targets)')
0059     parser.add_argument('--limit-files', type=int, default=0,
0060                         help='cap the file tier for a validation run '
0061                              '(works on a copy of the store, no publish)')
0062     parser.add_argument('--limit-datasets', type=int, default=0,
0063                         help='cap the dataset tier for a validation run')
0064     parser.add_argument('--resume', type=int, default=0, metavar='PASS_ID',
0065                         help='continue an interrupted pass by id, skipping '
0066                              'the locations it already stamped')
0067     parser.add_argument('--no-publish', action='store_true',
0068                         help='run the pass without publishing the component')
0069     parser.add_argument('--dump', default='',
0070                         help='write the projection JSON to this path')
0071     parser.add_argument('--created-by', default='prodops_agent')
0072     args = parser.parse_args()
0073     mode_name = 'incremental'
0074     campaigns = tuple(c.strip() for c in args.campaigns.split(',')
0075                       if c.strip()) or None
0076     validation = bool(args.limit_files or args.limit_datasets)
0077
0078     if args.publish_only:
0079         summary, data = project_store()
0080     else:
0081         try:
0082             summary, data = run_pass(campaigns=campaigns,
0083                                      limit_files=args.limit_files,
0084                                      limit_datasets=args.limit_datasets,
0085                                      resume_pass=args.resume or None)
0086         except PassInProgress as exc:
0087             # Another pass holds the store when the cron enqueue lands:
0088             # skipped, not failed.
0089             print('SUMMARY ' + json.dumps({'mode': mode_name,
0090                                            'skipped': str(exc)}))
0091             return 4
0092         except KeyboardInterrupt as exc:
0093             # The pass row records the interruption and stays unfinished:
0094             # the next pass redoes the interval and is not blocked once
0095             # this process is gone (storage.py, run_pass).
0096             pass_id = mark_pass_interrupted(str(exc) or 'interrupt')
0097             log(f'pass {pass_id} interrupted: {exc}')
0098             print('SUMMARY ' + json.dumps({'mode': mode_name,
0099                                            'pass_id': pass_id,
0100                                            'interrupted': str(exc)}))
0101             return 143
0102     if args.dump:
0103         with open(args.dump, 'w') as handle:
0104             json.dump(data, handle, indent=1, sort_keys=True)
0105         print(f'projection written to {args.dump}')
0106
0107     if validation or args.no_publish:
0108         summary['published'] = False
0109     else:
0110         from monitor_app.snapper_storage import publish_storage
0111         publication = publish_storage(data, since=data['interval']['start'])
0112         summary['published'] = True
0113         summary['revision'] = publication.update.revision
0114         summary['content_changed'] = publication.update.content_changed
0115         # The ghost product is rebuilt as the pass's last step, so the
0116         # Storage exceptions page, the epicprod_storage tool and the REST
0117         # listing serve this pass's ghosts at once (STORAGE.md, Retrieval;
0118         # swf-monitor docs/CACHED_PRODUCTS.md). A failure here is logged
0119         # and never fails the sweep: the product's TTL rebuilds it behind
0120         # the next page view.
0121         try:
0122             from swf_epicprod.analytics.storage_listings import (
0123                 refresh_ghost_product)
0124             summary['ghost_product'] = refresh_ghost_product()
0125         except Exception as exc:                                  # noqa: BLE001
0126             log(f'ERROR ghost product refresh: {exc}')
0127             summary['ghost_product'] = {'error': str(exc)}
0128     print('SUMMARY ' + json.dumps(summary))
0129     return 0 if not summary.get('errors') else 3
0130
0131
0132 if __name__ == '__main__':
0133     sys.exit(main())