swf-monitor/scripts/pool-reporter.py

0001 #!/usr/bin/env python3
0002 """pool-reporter.py — how full the batch pool our queues draw on is.
0003
0004 The E1 and BNL queues run in the SCDF shared pool, and a worker's wait
0005 there is decided by how much of the pool is claimed and how deep the
0006 queue ahead of it is. Neither figure is in any PanDA record: the pool's
0007 collector is the only source, and it answers the HTCondor protocol, not
0008 HTTP. This reporter asks it and posts the answer to swf-monitor's host
0009 report ingest (docs/POOL_REPORTER.md), so the web tier reads a stored
0010 record and never speaks to a collector in a request.
0011
0012 It runs on the monitor host, where the collector is reachable, in its
0013 own virtual environment: the HTCondor bindings are large and belong to
0014 this reporter rather than to the web tier's dependencies.
0015
0016 Usage::
0017
0018     pool-reporter.py                 # collect and post
0019     pool-reporter.py --print         # collect and print, post nothing
0020
0021 Configuration, from the environment or an environment file named by
0022 --env (mode 600): SWF_MONITOR_URL, SWF_REPORT_TOKEN. The pool is named
0023 by --pool and --collector, defaulting to the SCDF shared pool.
0024 """
0025 import argparse
0026 import json
0027 import os
0028 import ssl
0029 import sys
0030 import time
0031 from collections import Counter
0032 from datetime import datetime, timezone
0033
0034 DEFAULT_POOL = 'bnl-scdf'
0035 DEFAULT_COLLECTOR = 'condorspool01.sdcc.bnl.gov:9665'
0036 POST_TIMEOUT_S = 30
0037 # The pool holds tens of thousands of slots on hundreds of machines and
0038 # dozens of schedds. Only bounded summaries are delivered: a record is a
0039 # summary, not a payload, and the ingest refuses anything large.
0040 MAX_DOMAINS = 24
0041 MAX_SCHEDDS = 40
0042
0043
0044 def now_iso():
0045     return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
0046
0047
0048 def domain_of(machine):
0049     """The part of a machine name that identifies its pool.
0050
0051     Slot names arrive as ``slot1_14@spool1499.sdcc.bnl.gov`` in the job
0052     records and as bare machine names here; the domain is what the two
0053     have in common, and it is what says which pool a queue's workers ran
0054     in. A machine with no domain contributes its name's leading letters,
0055     which is how the OSG pool's bare ``compute34`` names are grouped.
0056     """
0057     name = (machine or '').split('@')[-1].strip().lower()
0058     if not name:
0059         return ''
0060     if '.' in name:
0061         return name.split('.', 1)[1]
0062     return name.rstrip('0123456789') or name
0063
0064
0065 def read_pool(collector):
0066     """One reading of the pool: what it holds, and the queue ahead."""
0067     import htcondor2 as htcondor
0068
0069     out = {'collector': collector}
0070     started = time.time()
0071     coll = htcondor.Collector(collector)
0072
0073     try:
0074         slots = coll.query(htcondor.AdType.Startd,
0075                            projection=['State', 'Machine', 'Cpus'])
0076     except Exception as exc:                                  # noqa: BLE001
0077         return {'collector': collector,
0078                 'error': 'startd query failed: {}: {}'.format(
0079                     exc.__class__.__name__, exc)}
0080
0081     states = Counter()
0082     cores = Counter()
0083     domains = Counter()
0084     for ad in slots:
0085         state = str(ad.get('State') or 'unknown').lower()
0086         states[state] += 1
0087         try:
0088             cpus = int(ad.get('Cpus') or 0)
0089         except (TypeError, ValueError):
0090             cpus = 0
0091         cores[state] += cpus
0092         domains[domain_of(ad.get('Machine'))] += 1
0093
0094     total = sum(states.values())
0095     claimed = states.get('claimed', 0)
0096     out['slots'] = {'total': total,
0097                     'claimed': claimed,
0098                     'unclaimed': states.get('unclaimed', 0),
0099                     'by_state': dict(states)}
0100     out['cores'] = {'total': sum(cores.values()),
0101                     'claimed': cores.get('claimed', 0)}
0102     out['claimed_fraction'] = round(float(claimed) / total, 4) if total else None
0103     kept = domains.most_common(MAX_DOMAINS)
0104     out['machine_domains'] = dict(kept)
0105     if len(domains) > len(kept):
0106         out['machine_domains_other'] = len(domains) - len(kept)
0107
0108     try:
0109         schedds = coll.query(htcondor.AdType.Schedd,
0110                              projection=['Name', 'TotalRunningJobs',
0111                                          'TotalIdleJobs', 'TotalHeldJobs'])
0112     except Exception as exc:                                  # noqa: BLE001
0113         out['schedd_error'] = '{}: {}'.format(exc.__class__.__name__, exc)
0114         out['collect_seconds'] = round(time.time() - started, 2)
0115         return out
0116
0117     def _int(ad, key):
0118         try:
0119             return int(ad.get(key) or 0)
0120         except (TypeError, ValueError):
0121             return 0
0122
0123     rows = [{'name': str(ad.get('Name') or ''),
0124              'running': _int(ad, 'TotalRunningJobs'),
0125              'idle': _int(ad, 'TotalIdleJobs'),
0126              'held': _int(ad, 'TotalHeldJobs')}
0127             for ad in schedds]
0128     rows.sort(key=lambda r: r['idle'] + r['running'], reverse=True)
0129     out['queue'] = {
0130         'schedds': len(rows),
0131         'running': sum(r['running'] for r in rows),
0132         'idle': sum(r['idle'] for r in rows),
0133         'held': sum(r['held'] for r in rows),
0134         'busiest': rows[:MAX_SCHEDDS],
0135     }
0136     out['collect_seconds'] = round(time.time() - started, 2)
0137     return out
0138
0139
0140 def collect(pool, collector):
0141     return {'pool': pool,
0142             'collected_at': now_iso(),
0143             'reporter_version': '1.0',
0144             'reading': read_pool(collector)}
0145
0146
0147 def load_env(path):
0148     """Read KEY=value lines from an environment file, if there is one."""
0149     if not path or not os.path.exists(path):
0150         return
0151     with open(path) as handle:
0152         for line in handle:
0153             line = line.strip()
0154             if not line or line.startswith('#') or '=' not in line:
0155                 continue
0156             key, value = line.split('=', 1)
0157             os.environ.setdefault(key.strip(), value.strip().strip('"\''))
0158
0159
0160 def post(record, url, token, key):
0161     """Deliver one record. Returns None on success, the reason on failure."""
0162     import urllib.request
0163     body = json.dumps(record).encode('utf-8')
0164     endpoint = '{}/api/host-reports/{}/'.format(url.rstrip('/'), key)
0165     request = urllib.request.Request(
0166         endpoint, data=body, method='POST',
0167         headers={'Content-Type': 'application/json',
0168                  'Authorization': 'Token {}'.format(token)})
0169     # The monitor serves a certificate this host's trust store carries
0170     # only through the combined bundle; the token is the authentication,
0171     # and the request never leaves the host.
0172     context = ssl._create_unverified_context()
0173     try:
0174         with urllib.request.urlopen(request, timeout=POST_TIMEOUT_S,
0175                                     context=context) as response:
0176             response.read()
0177         return None
0178     except Exception as exc:                                  # noqa: BLE001
0179         return '{}: {}'.format(exc.__class__.__name__, exc)
0180
0181
0182 def main():
0183     parser = argparse.ArgumentParser()
0184     parser.add_argument('--print', dest='show', action='store_true',
0185                         help='print the record, post nothing')
0186     parser.add_argument('--pool', default=DEFAULT_POOL,
0187                         help='the pool name, which is also the report key')
0188     parser.add_argument('--collector', default=DEFAULT_COLLECTOR)
0189     parser.add_argument('--env', default=os.path.expanduser(
0190         '~/.swf-pool-reporter.env'))
0191     args = parser.parse_args()
0192
0193     record = collect(args.pool, args.collector)
0194     if args.show:
0195         print(json.dumps(record, indent=2, sort_keys=True))
0196         return 0
0197
0198     load_env(args.env)
0199     url = os.environ.get('SWF_MONITOR_URL')
0200     token = os.environ.get('SWF_REPORT_TOKEN')
0201     if not url or not token:
0202         print('SWF_MONITOR_URL and SWF_REPORT_TOKEN are required '
0203               '(environment or --env file)', file=sys.stderr)
0204         return 2
0205
0206     failure = post(record, url, token, args.pool)
0207     if failure:
0208         print('post failed: {}'.format(failure), file=sys.stderr)
0209         return 1
0210     reading = record.get('reading') or {}
0211     if reading.get('error'):
0212         print('reported, with an error in the reading: {}'
0213               .format(reading['error']))
0214         return 0
0215     slots = reading.get('slots') or {}
0216     queue = reading.get('queue') or {}
0217     print('reported: {} of {} slots claimed, {} jobs waiting ahead'
0218           .format(slots.get('claimed'), slots.get('total'),
0219                   queue.get('idle')))
0220     return 0
0221
0222
0223 if __name__ == '__main__':
0224     sys.exit(main())