File indexing completed on 2026-11-09 10:47:23
0001
0002 """osgsub-reporter.py — what only the OSG submit host can see.
0003
0004 Runs on osgsub01 and posts one record per run to swf-monitor
0005 (docs/OSG_SUBMIT_REPORTER.md): the exclusions actually in force in the
0006 submit descriptions, the queue-to-file mapping harvester uses, what the
0007 pool currently offers, and what the local schedd is doing. The monitor
0008 cannot read any of it — ssh to this host is refused except through the
0009 facility gateway with agent forwarding — so the host pushes.
0010
0011 Written for this host as it is: python 3.6, standard library only, no
0012 virtual environment. Every source that cannot be read is delivered as
0013 an error field rather than dropped, because a missing field and an
0014 unreadable source are otherwise the same thing to a reader.
0015
0016 Usage::
0017
0018 osgsub-reporter.py # collect and post
0019 osgsub-reporter.py --print # collect and print, post nothing
0020 osgsub-reporter.py --no-pool # skip the condor queries
0021
0022 Configuration, from the environment or an environment file named by
0023 --env (mode 600): SWF_MONITOR_URL, SWF_REPORT_TOKEN.
0024 """
0025 from __future__ import print_function
0026
0027 import argparse
0028 import json
0029 import os
0030 import re
0031 import ssl
0032 import subprocess
0033 import sys
0034 import time
0035 from datetime import datetime
0036
0037 HOST = 'osgsub01'
0038 SDF_DIR = '/var/data/atlpan/harvester_common'
0039 QUEUE_CONFIG = '/opt/harvester/etc/panda/panda_queueconfig.json'
0040 DEFAULT_STATE = os.path.expanduser('~/.swf-osgsub-reporter')
0041 POST_TIMEOUT_S = 30
0042 CONDOR_TIMEOUT_S = 120
0043 BUFFER_KEEP = 48
0044
0045
0046
0047
0048 PAIR_RE = re.compile(
0049 r'GLIDEIN_Site\s*=\?=\s*"([^"]+)"\s*&&\s*'
0050 r'regexp\(\s*"\^\(([^)]*)\)\(\[\.\]\|\$\)"',
0051 re.IGNORECASE)
0052
0053
0054 def now_iso():
0055 return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
0056
0057
0058 def run(cmd, timeout=CONDOR_TIMEOUT_S):
0059 """A command's stdout, or None with the reason on the record."""
0060 try:
0061 done = subprocess.run(cmd, stdout=subprocess.PIPE,
0062 stderr=subprocess.PIPE,
0063 universal_newlines=True, timeout=timeout)
0064 except Exception as exc:
0065 return None, '{}: {}'.format(exc.__class__.__name__, exc)
0066 if done.returncode != 0:
0067 return None, 'exit {}: {}'.format(done.returncode,
0068 (done.stderr or '').strip()[:400])
0069 return done.stdout, None
0070
0071
0072 def parse_sdf(path):
0073 """The exclusions and submission shape of one submit description."""
0074 out = {'file': os.path.basename(path)}
0075 try:
0076 with open(path) as handle:
0077 text = handle.read()
0078 out['modified'] = datetime.utcfromtimestamp(
0079 os.path.getmtime(path)).strftime('%Y-%m-%dT%H:%M:%SZ')
0080 except OSError as exc:
0081 out['error'] = 'unreadable: {}'.format(exc)
0082 return out
0083
0084 sites = re.search(r'^\+UNDESIRED_Sites\s*=\s*"([^"]*)"', text, re.M)
0085 out['excluded_sites'] = (
0086 [s.strip() for s in sites.group(1).split(',') if s.strip()]
0087 if sites else [])
0088
0089 req = re.search(r'^Requirements\s*=\s*(.+)$', text, re.M)
0090 out['requirements'] = req.group(1).strip() if req else None
0091 pairs = []
0092 if req:
0093 for site, names in PAIR_RE.findall(req.group(1)):
0094 for node in names.split('|'):
0095 if node.strip():
0096 pairs.append({'site': site, 'node': node.strip()})
0097 out['excluded_site_nodes'] = pairs
0098
0099 for field, pattern in (
0100 ('executable', r'^executable\s*=\s*(.+)$'),
0101 ('job_duration_category', r'^\+JobDurationCategory\s*=\s*"([^"]*)"'),
0102 ('request_cpus', r'^request_cpus\s*=\s*(.+)$'),
0103 ('request_memory', r'^request_memory\s*=\s*(.+)$'),
0104 ('request_disk', r'^request_disk\s*=\s*(.+)$')):
0105 found = re.search(pattern, text, re.M)
0106 out[field] = found.group(1).strip() if found else None
0107 return out
0108
0109
0110 def submit_descriptions():
0111 """Every submit description on the host, parsed."""
0112 try:
0113 names = sorted(n for n in os.listdir(SDF_DIR) if n.endswith('.sdf'))
0114 except OSError as exc:
0115 return {'error': 'cannot list {}: {}'.format(SDF_DIR, exc)}
0116 return {n: parse_sdf(os.path.join(SDF_DIR, n)) for n in names}
0117
0118
0119 def queue_map():
0120 """Which queue uses which submit description, and its worker limits."""
0121 try:
0122 with open(QUEUE_CONFIG) as handle:
0123 config = json.load(handle)
0124 except (OSError, ValueError) as exc:
0125 return {'error': 'cannot read {}: {}'.format(QUEUE_CONFIG, exc)}
0126 out = {}
0127 for queue, block in config.items():
0128 blob = json.dumps(block)
0129 found = re.findall(r'submit_pilot2[\w\-.]*\.sdf', blob)
0130 if not found:
0131 continue
0132 submitter = block.get('submitter') or {}
0133 out[queue] = {
0134 'submit_description': sorted(set(found))[0],
0135 'max_workers': submitter.get('maxWorkers'),
0136 'max_new_workers_per_cycle':
0137 submitter.get('maxNewWorkersPerCycle'),
0138 'workflow': block.get('workflow'),
0139 }
0140 return out
0141
0142
0143 def pool(requirements, pairs):
0144 """What the pool offers, what the queue admits, what we exclude."""
0145 out = {}
0146 total, err = run(['condor_status', '-af', 'Name'])
0147 if err:
0148 return {'error': 'condor_status unavailable: {}'.format(err)}
0149 out['slots_total'] = len(total.strip().splitlines()) if total.strip() else 0
0150
0151 if requirements:
0152 admitted, err = run(['condor_status', '-constraint', requirements,
0153 '-af', 'Name'])
0154 out['slots_admitted'] = (
0155 len(admitted.strip().splitlines()) if admitted and admitted.strip()
0156 else (None if err else 0))
0157 if err:
0158 out['slots_admitted_error'] = err
0159
0160 if pairs:
0161 terms = []
0162 for site in sorted(set(p['site'] for p in pairs)):
0163 names = '|'.join(p['node'] for p in pairs if p['site'] == site)
0164 terms.append('(GLIDEIN_Site =?= "{}" && regexp("^({})([.]|$)", '
0165 'Machine) =?= True)'.format(site, names))
0166 banned = ' || '.join(terms)
0167 hit, err = run(['condor_status', '-constraint', banned,
0168 '-af', 'Machine', 'GLIDEIN_Site'])
0169 if err:
0170 out['slots_excluded_error'] = err
0171 else:
0172 lines = [ln.strip() for ln in (hit or '').splitlines() if ln.strip()]
0173 out['slots_excluded_now'] = len(lines)
0174 out['excluded_present'] = sorted(set(lines))
0175
0176 sites, err = run(['condor_status', '-af', 'GLIDEIN_Site'])
0177 if err:
0178 out['sites_error'] = err
0179 else:
0180 counts = {}
0181 for line in (sites or '').splitlines():
0182 name = line.strip() or 'unknown'
0183 counts[name] = counts.get(name, 0) + 1
0184 out['slots_by_site'] = counts
0185 return out
0186
0187
0188 def schedd():
0189 """What the local schedd is holding, by queue where it is named."""
0190 out, err = run(['condor_q', '-all', '-af', 'JobStatus', 'HoldReason',
0191 'MATCH_EXP_JOB_GLIDEIN_ResourceName'])
0192 if err:
0193 return {'error': 'condor_q unavailable: {}'.format(err)}
0194 states = {1: 'idle', 2: 'running', 5: 'held'}
0195 counts = {'idle': 0, 'running': 0, 'held': 0, 'other': 0}
0196 held_reasons = {}
0197 for line in (out or '').splitlines():
0198 parts = line.split(None, 1)
0199 if not parts:
0200 continue
0201 try:
0202 code = int(parts[0])
0203 except ValueError:
0204 continue
0205 counts[states.get(code, 'other')] = counts.get(
0206 states.get(code, 'other'), 0) + 1
0207 if code == 5 and len(parts) > 1:
0208 reason = parts[1].strip()[:300]
0209 held_reasons[reason] = held_reasons.get(reason, 0) + 1
0210 out_rec = {'workers': counts}
0211 if held_reasons:
0212 out_rec['held_reasons'] = held_reasons
0213 return out_rec
0214
0215
0216 IDLE_ANALYZE_MIN = 100
0217 ANALYSIS_LOG = 'match-analysis.jsonl'
0218 ANALYSIS_KEEP_S = 7 * 24 * 3600
0219 QUEUE_ARG_RE = re.compile(r'-q (\S+)')
0220 ANALYSIS_PATTERNS = (
0221 ('slots', re.compile(r'Of (\d+) slots on (\d+) machines')),
0222 ('rejected_by_job', re.compile(r'(\d+) slots are rejected by your job')),
0223 ('reject_job', re.compile(r'(\d+) slots reject your job')),
0224 ('willing', re.compile(r'(\d+) slots match and are willing')),
0225 ('if_drained', re.compile(r'(\d+) slots would match if drained')),
0226 ('last_match', re.compile(r'Last successful match: (.+)$')),
0227 ('last_failed_match', re.compile(r'Last failed match: (.+)$')),
0228 ('failed_reason', re.compile(r'^\s*Reason: (.+)$')),
0229 )
0230
0231
0232 def _analysis(job_id):
0233 """condor's own account of why one idle pilot is not running: the
0234 pool's slots split by who rejects whom, and the last match times."""
0235 out, err = run(['condor_q', '-better-analyze', job_id])
0236 if err:
0237 return {'error': err}
0238 found = {}
0239 for line in (out or '').splitlines():
0240 for key, pattern in ANALYSIS_PATTERNS:
0241 m = pattern.search(line)
0242 if m and key not in found:
0243 found[key] = (int(m.group(1)) if m.group(1).isdigit()
0244 else m.group(1).strip())
0245 return found
0246
0247
0248 def idle_analysis(state_dir):
0249 """Per OSG queue with at least IDLE_ANALYZE_MIN idle pilots: the idle
0250 count, the oldest idle pilot's age, and condor's match analysis of
0251 that pilot. Each run's result is also appended to a seven-day log in
0252 the state directory, since the schedd keeps about an hour and a half
0253 of history and a stall outlasts it (BNL_OSG_EPIC_PROD_1, 10/6 22:00 to
0254 10/7 03:00 UTC, unexplained once the history had rotated)."""
0255 out, err = run(['condor_q', '-all', '-constraint', 'JobStatus == 1',
0256 '-af', 'ClusterId', 'ProcId', 'QDate', 'Args'])
0257 if err:
0258 return {'error': 'condor_q unavailable: {}'.format(err)}
0259 queues = {}
0260 for line in (out or '').splitlines():
0261 parts = line.split(None, 3)
0262 if len(parts) < 4:
0263 continue
0264 m = QUEUE_ARG_RE.search(parts[3])
0265 if not m:
0266 continue
0267 q = queues.setdefault(m.group(1), {'idle': 0, 'oldest_qdate': None})
0268 q['idle'] += 1
0269 try:
0270 qdate = int(parts[2])
0271 except ValueError:
0272 continue
0273 if q['oldest_qdate'] is None or qdate < q['oldest_qdate']:
0274 q['oldest_qdate'] = qdate
0275 q['job'] = '{}.{}'.format(parts[0], parts[1])
0276 now = time.time()
0277 result = {}
0278 for name, q in sorted(queues.items()):
0279 entry = {'idle': q['idle']}
0280 if q['oldest_qdate'] is not None:
0281 entry['oldest_idle_minutes'] = int((now - q['oldest_qdate']) / 60)
0282 if q['idle'] >= IDLE_ANALYZE_MIN and q.get('job'):
0283 entry['job'] = q['job']
0284 entry['analysis'] = _analysis(q['job'])
0285 result[name] = entry
0286 _append_analysis(state_dir, {'at': now_iso(), 'queues': result})
0287 return result
0288
0289
0290 def _append_analysis(state_dir, line):
0291 """Append one run's analysis and keep seven days of them."""
0292 path = os.path.join(state_dir, ANALYSIS_LOG)
0293 cutoff = datetime.utcfromtimestamp(time.time() - ANALYSIS_KEEP_S).strftime(
0294 '%Y-%m-%dT%H:%M:%SZ')
0295 try:
0296 os.makedirs(state_dir, exist_ok=True)
0297 kept = []
0298 if os.path.exists(path):
0299 with open(path) as handle:
0300 for raw in handle:
0301 try:
0302 if json.loads(raw).get('at', '') >= cutoff:
0303 kept.append(raw if raw.endswith('\n') else raw + '\n')
0304 except ValueError:
0305 continue
0306 kept.append(json.dumps(line, sort_keys=True) + '\n')
0307 tmp = path + '.tmp'
0308 with open(tmp, 'w') as handle:
0309 handle.writelines(kept)
0310 os.replace(tmp, path)
0311 except OSError as exc:
0312 print('could not write the match analysis log: {}'.format(exc),
0313 file=sys.stderr)
0314
0315
0316 NEGOTIATOR_TIMEOUT_S = 30
0317
0318
0319 def _config_value(name):
0320 """One condor configuration value, or None when it is not defined."""
0321 out, err = run(['condor_config_val', name], timeout=NEGOTIATOR_TIMEOUT_S)
0322 return None if err else (out or '').strip() or None
0323
0324
0325 def negotiators():
0326 """Each pool negotiator's last cycle, and this schedd's submitters it
0327 held at their fair-share limit in that cycle. A held submitter gets
0328 no new slots in that pool, whatever its free slots would accept:
0329 on 10/24 the OSPool negotiator held atlpan@osgsub01 while free
0330 UChicago slots matched our pilot both ways. One small ad per
0331 negotiator, about 0.2 s a pool; the pools are the schedd's own
0332 collector and the pools it flocks to."""
0333 host = _config_value('FULL_HOSTNAME')
0334 pools = []
0335 for name in ('COLLECTOR_HOST', 'FLOCK_TO'):
0336 for pool in (_config_value(name) or '').replace(',', ' ').split():
0337 if pool not in pools:
0338 pools.append(pool)
0339 result = {'pools': pools, 'submitter_host': host, 'negotiators': [],
0340 'errors': {}}
0341 if not host:
0342 result['errors']['FULL_HOSTNAME'] = 'not defined'
0343 return result
0344 seen = set()
0345 for pool in pools:
0346 out, err = run(['condor_status', '-pool', pool, '-negotiator', '-af',
0347 'Name', 'LastNegotiationCycleTime0',
0348 'LastNegotiationCycleSubmittersShareLimit0'],
0349 timeout=NEGOTIATOR_TIMEOUT_S)
0350 if err:
0351 result['errors'][pool] = err
0352 continue
0353 for line in (out or '').splitlines():
0354 parts = line.split(None, 2)
0355 if len(parts) < 2 or parts[0] in seen:
0356 continue
0357 seen.add(parts[0])
0358 limited = [] if len(parts) < 3 or parts[2] == 'undefined' else [
0359 s.strip() for s in parts[2].split(',')]
0360 try:
0361 cycle_at = datetime.utcfromtimestamp(int(parts[1])).strftime(
0362 '%Y-%m-%dT%H:%M:%SZ')
0363 except ValueError:
0364 cycle_at = None
0365 result['negotiators'].append({
0366 'name': parts[0], 'pool': pool, 'cycle_at': cycle_at,
0367 'share_limited': [s for s in limited
0368 if s.endswith('@' + host)]})
0369 return result
0370
0371
0372 HARVESTER_LOG_DIR = '/var/log/harvester'
0373
0374
0375 def _proc_present(pattern):
0376 """Whether any process matches, without requiring pgrep to exist."""
0377 out, err = run(['pgrep', '-f', pattern], timeout=15)
0378 if err and 'exit 1' in err:
0379 return False
0380 if err:
0381 return None
0382 return bool((out or '').strip())
0383
0384
0385 def health():
0386 """The submit host's own state, for the platform component.
0387
0388 Daemon liveness and log freshness on the idiom the PanDA server host
0389 reporter uses (docs/SNAPPER_PLATFORM.md, Server host): a daemon is
0390 named by its log, and the age of that log's last write is what says
0391 whether it is working. Held workers with their reasons are the
0392 signal this host alone carries: a site refusing every submission
0393 shows here and nowhere else.
0394 """
0395 out = {}
0396
0397 daemons = {}
0398 try:
0399 names = [n for n in os.listdir(HARVESTER_LOG_DIR)
0400 if n.startswith('panda-') and n.endswith('.log')]
0401 except OSError as exc:
0402 daemons = {'error': 'cannot list {}: {}'.format(HARVESTER_LOG_DIR, exc)}
0403 names = []
0404 now = time.time()
0405 for name in sorted(names):
0406 label = name[len('panda-'):-len('.log')]
0407 path = os.path.join(HARVESTER_LOG_DIR, name)
0408 try:
0409 daemons[label] = {
0410 'log_age_seconds': int(now - os.path.getmtime(path)),
0411 'log_bytes': os.path.getsize(path),
0412 }
0413 except OSError as exc:
0414 daemons[label] = {'error': str(exc)}
0415 out['harvester_daemons'] = daemons
0416 out['harvester_process'] = _proc_present('panda_harvester-uwsgi')
0417 out['schedd_process'] = _proc_present('condor_schedd')
0418
0419 try:
0420 with open('/proc/loadavg') as handle:
0421 parts = handle.read().split()
0422 out['load'] = {'1m': float(parts[0]), '5m': float(parts[1]),
0423 '15m': float(parts[2])}
0424 except (OSError, ValueError, IndexError) as exc:
0425 out['load_error'] = str(exc)
0426
0427 mem = {}
0428 try:
0429 with open('/proc/meminfo') as handle:
0430 for line in handle:
0431 key, _, rest = line.partition(':')
0432 if key in ('MemTotal', 'MemAvailable', 'SwapTotal',
0433 'SwapFree'):
0434 mem[key] = int(rest.split()[0])
0435 out['memory_kb'] = mem
0436 except (OSError, ValueError, IndexError) as exc:
0437 out['memory_error'] = str(exc)
0438
0439 volumes = {}
0440 for mount in ('/', '/var'):
0441 try:
0442 st = os.statvfs(mount)
0443 total = st.f_blocks * st.f_frsize
0444 free = st.f_bavail * st.f_frsize
0445 volumes[mount] = {
0446 'total_bytes': total, 'free_bytes': free,
0447 'used_percent': round(100.0 * (total - free) / total, 1)
0448 if total else None}
0449 except OSError as exc:
0450 volumes[mount] = {'error': str(exc)}
0451 out['volumes'] = volumes
0452 return out
0453
0454
0455 def collect(with_pool=True, state_dir=DEFAULT_STATE):
0456 """The record this run delivers."""
0457 record = {'host': HOST, 'collected_at': now_iso(),
0458 'reporter_version': '1.2'}
0459 started = time.time()
0460 record['submit_descriptions'] = submit_descriptions()
0461 record['queues'] = queue_map()
0462
0463
0464
0465 prod = (record['submit_descriptions'] or {}).get(
0466 'submit_pilot2_push_bnl_osg.sdf') or {}
0467 record['exclusions'] = {
0468 'submit_description': prod.get('file'),
0469 'modified': prod.get('modified'),
0470 'sites': prod.get('excluded_sites'),
0471 'site_nodes': prod.get('excluded_site_nodes'),
0472 'queues': sorted(q for q, v in (record['queues'] or {}).items()
0473 if isinstance(v, dict)
0474 and v.get('submit_description')
0475 == 'submit_pilot2_push_bnl_osg.sdf'),
0476 }
0477 record['health'] = health()
0478 if with_pool:
0479 record['pool'] = pool(prod.get('requirements'),
0480 prod.get('excluded_site_nodes') or [])
0481 record['schedd'] = schedd()
0482 record['idle_analysis'] = idle_analysis(state_dir)
0483 record['negotiators'] = negotiators()
0484 record['collect_seconds'] = round(time.time() - started, 2)
0485 return record
0486
0487
0488 def load_env(path):
0489 """Read KEY=value lines from an environment file, if there is one."""
0490 if not path or not os.path.exists(path):
0491 return
0492 with open(path) as handle:
0493 for line in handle:
0494 line = line.strip()
0495 if not line or line.startswith('#') or '=' not in line:
0496 continue
0497 key, value = line.split('=', 1)
0498 os.environ.setdefault(key.strip(), value.strip().strip('"\''))
0499
0500
0501 def post(record, url, token):
0502 """Deliver one record. Returns None on success, the reason on failure."""
0503 import urllib.request
0504 body = json.dumps(record).encode('utf-8')
0505 endpoint = '{}/api/host-reports/{}/'.format(url.rstrip('/'), HOST)
0506 request = urllib.request.Request(
0507 endpoint, data=body, method='POST',
0508 headers={'Content-Type': 'application/json',
0509 'Authorization': 'Token {}'.format(token)})
0510
0511
0512 context = ssl._create_unverified_context()
0513 try:
0514 with urllib.request.urlopen(request, timeout=POST_TIMEOUT_S,
0515 context=context) as response:
0516 response.read()
0517 return None
0518 except Exception as exc:
0519 return '{}: {}'.format(exc.__class__.__name__, exc)
0520
0521
0522 def buffered(state_dir):
0523 """Records held from earlier runs the monitor could not take."""
0524 try:
0525 names = sorted(os.listdir(state_dir))
0526 except OSError:
0527 return []
0528 return [os.path.join(state_dir, n) for n in names
0529 if n.startswith('record-') and n.endswith('.json')]
0530
0531
0532 def buffer_record(state_dir, record):
0533 """Hold a record for the next run, keeping the buffer bounded."""
0534 try:
0535 os.makedirs(state_dir, exist_ok=True)
0536 path = os.path.join(state_dir, 'record-{}.json'.format(
0537 record['collected_at'].replace(':', '')))
0538 with open(path, 'w') as handle:
0539 json.dump(record, handle)
0540 held = buffered(state_dir)
0541 for stale in held[:-BUFFER_KEEP]:
0542 os.unlink(stale)
0543 except OSError as exc:
0544 print('could not buffer the record: {}'.format(exc), file=sys.stderr)
0545
0546
0547 def main():
0548 parser = argparse.ArgumentParser()
0549 parser.add_argument('--print', dest='show', action='store_true',
0550 help='print the record, post nothing')
0551 parser.add_argument('--no-pool', dest='pool', action='store_false',
0552 help='skip the condor queries')
0553 parser.add_argument('--env', default=os.path.expanduser(
0554 '~/.swf-osgsub-reporter.env'))
0555 parser.add_argument('--state', default=DEFAULT_STATE)
0556 args = parser.parse_args()
0557
0558 record = collect(with_pool=args.pool, state_dir=args.state)
0559 if args.show:
0560 print(json.dumps(record, indent=2, sort_keys=True))
0561 return 0
0562
0563 load_env(args.env)
0564 url = os.environ.get('SWF_MONITOR_URL')
0565 token = os.environ.get('SWF_REPORT_TOKEN')
0566 if not url or not token:
0567 print('SWF_MONITOR_URL and SWF_REPORT_TOKEN are required '
0568 '(environment or --env file)', file=sys.stderr)
0569 buffer_record(args.state, record)
0570 return 2
0571
0572
0573 for path in buffered(args.state):
0574 try:
0575 with open(path) as handle:
0576 held = json.load(handle)
0577 except (OSError, ValueError):
0578 os.unlink(path)
0579 continue
0580 if post(held, url, token) is None:
0581 os.unlink(path)
0582 else:
0583 break
0584
0585 failure = post(record, url, token)
0586 if failure:
0587 print('post failed, record buffered: {}'.format(failure),
0588 file=sys.stderr)
0589 buffer_record(args.state, record)
0590 return 1
0591 excl = record.get('exclusions') or {}
0592 print('reported: {} site(s), {} node(s) excluded, {} slot(s) removed now'
0593 .format(len(excl.get('sites') or []),
0594 len(excl.get('site_nodes') or []),
0595 (record.get('pool') or {}).get('slots_excluded_now', '?')))
0596 return 0
0597
0598
0599 if __name__ == '__main__':
0600 sys.exit(main())
