swf-monitor/scripts/cric-declared-state.py

0001 #!/usr/bin/env python3
0002 """cric-declared-state.py — the declared record from CRIC.
0003 The doer of the ops agent's ``cric_declared_state`` (swf-epicprod
0004 docs/CONTINUOUS_PRODUCTION.md, Declared downtime): reads the PanDA queue
0005 status rules, the DDM endpoint status rules and the downtime objects of
0006 the CRIC ePIC's PanDA reads (datalake-cric.cern.ch) with the production
0007 proxy, keeps them as the declared record in the entry store
0008 (monitor_app/declared.py), and logs one ``declared_state_sync`` action
0009 naming what appeared, changed, expired and cleared. The scope is the
0010 EIC queues (vo eic), the endpoints those queues name, and the resource
0011 centres behind them.
0012 Credentials: ``X509_USER_PROXY`` (the ops agent's Rucio proxy, accepted
0013 by CRIC) and the grid CA directory (``X509_CERT_DIR``, default
0014 /etc/grid-security/certificates).
0015 Django-bootstrap standalone script, by hand::
0016     cd /data/wenauseic/github/swf-monitor/src
0017     ../../swf-testbed/.venv/bin/python ../scripts/cric-declared-state.py [--dry-run] [--created-by NAME]
0018 Prints one JSON line: the counts read and the changes. Exit status 1 when
0019 a CRIC read failed; nothing is written then.
0020 """
0021 import argparse
0022 import json
0023 import logging
0024 import os
0025 import sys
0026 import time
0027
0028 import requests
0029
0030 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0031 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0032 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0033 import django  # noqa: E402
0034 django.setup()
0035 from django.utils import timezone  # noqa: E402
0036 from monitor_app.epicprod_logging import log_epicprod_action  # noqa: E402
0037 from monitor_app import declared  # noqa: E402
0038
0039 log = logging.getLogger('cric-declared-state')
0040 TIMEOUT = 60
0041
0042
0043 def cric_get(url, proxy, ca_dir):
0044     """One CRIC read with the proxy as client certificate; raises on
0045     anything but a 200 with JSON."""
0046     r = requests.get(url, cert=(proxy, proxy), verify=ca_dir, timeout=TIMEOUT)
0047     if r.status_code != 200:
0048         raise RuntimeError(f'{url}: HTTP {r.status_code}: {r.text[:200]}')
0049     return r.json()
0050
0051
0052 def scope(queues_doc):
0053     """The EIC queues, the endpoints they name (astorages) and the
0054     resource centres behind them (each with its queues, so a site's
0055     window reaches the queue readers), from the pandaqueue document."""
0056     queues = set(queues_doc or {})
0057     endpoints, rcsites = set(), {}
0058     for name, q in (queues_doc or {}).items():
0059         for names in ((q or {}).get('astorages') or {}).values():
0060             endpoints.update(names or [])
0061         if (q or {}).get('rc_site'):
0062             rcsites.setdefault(q['rc_site'], set()).add(name)
0063     return queues, endpoints, rcsites
0064
0065
0066 def main():
0067     ap = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
0068     ap.add_argument('--dry-run', action='store_true',
0069                     help='read CRIC and print the records, write nothing')
0070     ap.add_argument('--created-by', default='cric_declared_state')
0071     args = ap.parse_args()
0072     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
0073                         format='%(asctime)s %(name)s %(levelname)s %(message)s')
0074     t0 = time.monotonic()
0075     proxy = os.environ.get('X509_USER_PROXY') or ''
0076     ca_dir = os.environ.get('X509_CERT_DIR') or '/etc/grid-security/certificates'
0077     if not proxy or not os.path.exists(proxy):
0078         msg = f'no proxy at X509_USER_PROXY={proxy!r}; CRIC refuses anonymous reads'
0079         log.error(msg)
0080         log_epicprod_action('prodops-agent', 'declared_state_sync', outcome='error',
0081                             username=args.created_by, sublevel='normal', live_default=True,
0082                             reason=msg,
0083                             level=logging.ERROR, message=f'declared state sync failed: {msg}')
0084         print(json.dumps({'error': msg}))
0085         return 1
0086     try:
0087         queues_doc = cric_get(declared.CRIC_QUEUES, proxy, ca_dir)
0088         queues, endpoints, rcsites = scope(queues_doc)
0089         found = []
0090         found += declared.rules_from_pandaqueuestatus(
0091             cric_get(declared.CRIC_QUEUE_STATUS, proxy, ca_dir), queues)
0092         found += declared.rules_from_ddmendpointstatus(
0093             cric_get(declared.CRIC_ENDPOINT_STATUS, proxy, ca_dir), endpoints)
0094         found += declared.windows_from_downtime(
0095             cric_get(declared.CRIC_DOWNTIME, proxy, ca_dir), rcsites)
0096     except Exception as e:                                    # noqa: BLE001
0097         log.exception('CRIC read failed')
0098         log_epicprod_action('prodops-agent', 'declared_state_sync', outcome='error',
0099                             username=args.created_by, sublevel='normal', live_default=True,
0100                             reason=str(e)[:300],
0101                             level=logging.ERROR,
0102                             duration_ms=int((time.monotonic() - t0) * 1000),
0103                             message=f'declared state sync failed: {str(e)[:300]}')
0104         print(json.dumps({'error': str(e)[:300]}))
0105         return 1
0106     now = timezone.now()
0107     counts = {'queues': len(queues), 'endpoints': len(endpoints), 'rcsites': len(rcsites),
0108               'records': len(found),
0109               'in_force': sum(1 for r in found if declared.standing_of(r, now) == 'active'),
0110               'future': sum(1 for r in found if declared.standing_of(r, now) == 'future')}
0111     if args.dry_run:
0112         print(json.dumps({'dry_run': True, **counts, 'records': found}, default=str))
0113         return 0
0114     changes = declared.sync(found, now, changed_by=args.created_by)
0115     moved = {k: v for k, v in changes.items() if k != 'unchanged' and v}
0116     # One incident per change for notice routing (swf-monitor
0117     # docs/NOTICE_ROUTING.md): a rule or window that appeared, changed,
0118     # expired or cleared, with its line as the summary and the target's
0119     # page as the url, so a subscriber sees a downtime when it is set.
0120     by_name = {r['name']: r for r in found}
0121     stored = {r['name']: r for r in declared.records()}
0122     for kind_of_change in ('appeared', 'changed', 'expired', 'cleared'):
0123         for name in changes.get(kind_of_change) or []:
0124             r = by_name.get(name) or stored.get(name) or {}
0125             target = r.get('target') or name
0126             kind = r.get('kind') or ''
0127             url = (f'/panda/epic-queues/{target}/#declared' if kind == 'queue'
0128                    else f'/rucio-endpoints/{target}/' if kind == 'endpoint'
0129                    else '/panda/epic-queues/')
0130             log_epicprod_action(
0131                 'prodops-agent', 'declared_state_changed',
0132                 subject_type={'queue': 'panda_queue', 'endpoint': 'ddm_endpoint'}.get(kind, 'rc_site'),
0133                 subject_key=target, username=args.created_by,
0134                 outcome=kind_of_change, sublevel='normal', live_default=True,
0135                 message=f"declared state {kind_of_change}: {target}: {declared.summary_line(r)}"[:300],
0136                 summary=declared.summary_line(r), url=url,
0137                 severity='warning' if kind_of_change in ('appeared', 'changed') else 'info',
0138                 kind=kind, record=name, declared_by=r.get('declared_by', ''),
0139                 end=r.get('end'), start=r.get('start'))
0140     log_epicprod_action(
0141         'prodops-agent', 'declared_state_sync',
0142         outcome='ok', username=args.created_by,
0143         sublevel='normal' if moved else 'low', live_default=bool(moved),
0144         duration_ms=int((time.monotonic() - t0) * 1000),
0145         message=('declared state from CRIC: '
0146                  + f"{counts['in_force']} in force, {counts['future']} future"
0147                  + ''.join(f"; {k} {', '.join(v)}" for k, v in moved.items())),
0148         **counts, **{k: len(v) for k, v in changes.items() if k != 'unchanged'},
0149         unchanged=changes['unchanged'])
0150     print(json.dumps({**counts, 'changes': changes}, default=str))
0151     return 0
0152
0153
0154 if __name__ == '__main__':
0155     sys.exit(main())