src/monitor_app/error_corrections.py

0001 """The error-label correction root (docs/ERROR_ATTRIBUTION.md).
0002
0003 Label-reliability rules mark untrustworthy job error labels; every
0004 error-presentation reader applies corrections on the way out through
0005 this module, so a rule added once corrects every surface at the next
0006 read. Recorded history stays raw; the corrected reading leads in
0007 presentation with the original label preserved.
0008
0009 The corrected reading is refined from the matched jobs' payload exit
0010 codes (grade: pilot mechanical fields) — 128+N is termination by
0011 signal N, and the campaign payload's coded exits carry their own
0012 documented meanings.
0013 """
0014 import json
0015 import logging
0016 import re
0017 import time
0018
0019 logger = logging.getLogger(__name__)
0020
0021
0022 def exit_counts_of(value):
0023     """The exit-code histogram as a dict. The panda connection returns
0024     jsonb columns as JSON text; a malformed value reads as empty and
0025     is logged, never raised."""
0026     if isinstance(value, dict):
0027         return value
0028     if isinstance(value, str) and value:
0029         try:
0030             parsed = json.loads(value)
0031             return parsed if isinstance(parsed, dict) else {}
0032         except ValueError as e:
0033             logger.error('exit-counts JSON parse failed: %s', e)
0034     return {}
0035
0036 # Payload exit codes with documented meanings (the campaign run.sh
0037 # vocabulary plus the signal convention). Anything above 128 reads as
0038 # a signal termination even without an entry here.
0039 EXIT_READINGS = {
0040     139: 'payload segfault',
0041     134: 'payload abort',
0042     137: 'payload killed',
0043     143: 'payload stopped by SIGTERM (15); reason not established',
0044     78: 'payload Rucio output registration failure',
0045     65: 'payload validation failure',
0046 }
0047
0048 _RULES_TTL_SECONDS = 60.0
0049 _cache = {'rules': None, 'at': 0.0}
0050
0051
0052 def _rules():
0053     now = time.monotonic()
0054     if _cache['rules'] is None or now - _cache['at'] > _RULES_TTL_SECONDS:
0055         from .models import ErrorCorrectionRule
0056         try:
0057             _cache['rules'] = list(
0058                 ErrorCorrectionRule.objects.filter(active=True))
0059         except Exception as e:
0060             logger.error('error-correction rules load failed: %s', e)
0061             return _cache['rules'] or []
0062         _cache['at'] = now
0063     return _cache['rules']
0064
0065
0066 def _like(pattern, value):
0067     """SQL LIKE semantics: % any run, _ one character, whole value."""
0068     regex = ''.join('.*' if ch == '%' else '.' if ch == '_' else re.escape(ch)
0069                     for ch in pattern)
0070     return re.fullmatch(regex, str(value or '')) is not None
0071
0072
0073 def match(component, code, diag, sites=None):
0074     """The first active rule matching this label, else None. A rule with a
0075     queue_scope applies only when every site in ``sites`` (the job's
0076     computingsite, or a summary entry's sites) matches the scope."""
0077     for rule in _rules():
0078         try:
0079             if rule.component != str(component or ''):
0080                 continue
0081             if int(rule.code) != int(code):
0082                 continue
0083         except (TypeError, ValueError):
0084             continue
0085         if rule.diag_substring and rule.diag_substring not in str(diag or ''):
0086             continue
0087         if rule.queue_scope:
0088             in_scope = [s for s in (sites or []) if s]
0089             if not in_scope or not all(_like(rule.queue_scope, s) for s in in_scope):
0090                 continue
0091         return rule
0092     return None
0093
0094
0095 def exit_reading(exitcode):
0096     """The documented reading of one payload exit code, else None."""
0097     try:
0098         value = int(exitcode)
0099     except (TypeError, ValueError):
0100         return None
0101     if value in EXIT_READINGS:
0102         return EXIT_READINGS[value]
0103     if value > 128:
0104         return f'payload terminated by signal {value - 128}'
0105     if value > 0:
0106         return f'payload failure, exit code {value}'
0107     return None
0108
0109
0110 def correction(rule, exit_counts=None):
0111     """The corrected reading for one matched pattern.
0112
0113     ``exit_counts`` maps the matched jobs' transformation exit codes to
0114     their counts; the readings of those codes, largest first, are the
0115     corrected modes. With no readable exit profile the rule's fallback
0116     label stands.
0117     """
0118     profile = []
0119     for exitcode, value in exit_counts_of(exit_counts).items():
0120         if isinstance(value, dict):
0121             n, rep = int(value.get('n') or 0), value.get('rep')
0122             held = value.get('held')
0123         else:
0124             n, rep, held = int(value or 0), None, None
0125         profile.append((str(exitcode), n, rep, held))
0126     profile.sort(key=lambda t: (-t[1], t[0]))
0127     modes = []
0128     for exitcode, n, rep, held in profile:
0129         reading = exit_reading(exitcode)
0130         if reading:
0131             # held: the mode's core-seconds when the profile carries
0132             # them (the Snapper breakdown's patterns do), else None.
0133             modes.append({'reading': reading, 'exit_code': exitcode,
0134                           'count': n, 'rep_pandaid': rep,
0135                           'held': held})
0136     label = (modes[0]['reading'] if modes
0137              else (getattr(rule, 'corrected_label', '')
0138                    or 'unreliable label; payload failure of '
0139                       'undetermined mode'))
0140     return {
0141         'label': label,
0142         'modes': modes,
0143         'unreliable_label': True,
0144         'grade': 'pilot mechanical fields (payload exit codes)',
0145         'note': getattr(rule, 'note', '') or '',
0146         'evidence_url': getattr(rule, 'evidence_url', '') or '',
0147     }
0148
0149
0150 def reading(component, code, diag, exit_counts=None, sites=None):
0151     """Shared, evidence-bounded reading of a reported error.
0152
0153     Decode documented exits and explicit diagnostics. Unknown causes stay
0154     unknown; a warning copied into a pilot label is not a causal verdict.
0155     """
0156     rule = match(component, code, diag, sites=sites)
0157     if rule is not None:
0158         return correction(rule, exit_counts)
0159     try:
0160         code = int(code)
0161     except (TypeError, ValueError):
0162         return None
0163     diag = str(diag or '')
0164     lower = diag.lower()
0165     label = ''
0166     if component in ('executor', 'transformation'):
0167         label = exit_reading(code) or ''
0168         if code == 1:
0169             label = 'Payload failed (exit 1); exit code alone does not identify the cause'
0170     elif component == 'pilot' and code in (1305, 1315):
0171         modes = correction(None, exit_counts)
0172         if modes['modes']:
0173             modes['note'] = 'Payload exit codes take precedence over incidental wrapper warnings.'
0174             return modes
0175         if 'unknown groupid' in lower or 'bind mount' in lower:
0176             label = 'Execution failure; container warning does not establish the payload cause'
0177     elif component == 'ddm' and 'bytes mismatch' in lower:
0178         kind = 'Log file' if '.log.tgz' in lower or '.log.tar' in lower else 'File'
0179         label = kind + ' catalog registration refused: registered and submitted byte sizes differ'
0180     elif component == 'pilot' and code == 1212:
0181         label = ('Payload process killed by its memory limit' if 'out of memory' in lower
0182                  else 'Pilot reported payload memory exhaustion')
0183     elif component == 'pilot' and code == 1213:
0184         label = 'Job stopped at its runtime limit'
0185     elif component == 'pilot' and code == 1326:
0186         label = 'Input staging failed: no allowed input replica found'
0187         if 'allow_wan=false' in lower:
0188             label += '; remote (WAN) access was disabled'
0189     elif component == 'pilot' and code == 1152 and 'stage-out' in lower:
0190         kind = 'Log' if '.log.tgz' in lower or '.log.tar' in lower else 'Output'
0191         label = kind + ' transfer timed out during stage-out'
0192     elif component == 'dispatcher' and code == 100 and 'lost heartbeat' in lower:
0193         label = 'PanDA lost the job heartbeat; the reason contact stopped is unknown'
0194     elif component in ('supervisor', 'taskbuffer') and 'disappeared from remote schedd' in lower:
0195         label = 'Worker disappeared from the remote batch scheduler; Harvester cancelled it'
0196         if 'while the job was starting' in lower:
0197             label += ' before the payload started'
0198         elif 'while the job was running' in lower:
0199             label += ' while the payload was running'
0200         label += '; underlying batch cause not reported'
0201     elif component == 'taskbuffer' and code == 300 and re.search(r':\s*None\s*$', diag):
0202         label = 'Batch worker ended while the job was running; batch reason was not recorded'
0203     if not label:
0204         return None
0205     return {'label': label, 'modes': [], 'unreliable_label': False,
0206             'grade': 'reported diagnostic or documented exit code',
0207             'note': '', 'evidence_url': ''}
0208
0209
0210 def apply_to_summary(entries):
0211     """Attach ``correction`` to each error-summary entry whose label a
0212     rule marks unreliable. Entries carry error_source, error_code,
0213     error_diag, and optionally exit_counts. Decoration must never take
0214     a page down: a failure on one entry is logged and that entry keeps
0215     its raw label."""
0216     evidence = {}
0217     for entry in entries:
0218         try:
0219             interpreted = reading(entry.get('error_source'),
0220                                   entry.get('error_code'), entry.get('error_diag'),
0221                                   entry.get('exit_counts'), sites=entry.get('sites'))
0222             if interpreted is not None:
0223                 entry['correction'] = interpreted
0224             # Only generic payload labels receive representative-log evidence.
0225             # A witness explains its own job; it does not relabel every job
0226             # sharing an exit code or assign a payload cause to worker losses.
0227             if (entry.get('error_source'), entry.get('error_code')) in {('executor', 1), ('pilot', 1305)}:
0228                 from .epicprod_inventory import cached_payload_log_parts, diagnosis_from_log_texts
0229                 for sample in (entry.get('site_task_counts') or [])[:3]:
0230                     job = sample.get('representative_pandaid')
0231                     task = sample.get('taskid')
0232                     if not job or not task:
0233                         continue
0234                     if job not in evidence:
0235                         parts = cached_payload_log_parts(task, job)
0236                         evidence[job] = diagnosis_from_log_texts([p['text'] for p in parts]) if parts else {}
0237                     diagnosis = evidence[job]
0238                     if diagnosis.get('operation') == 'invalid_event_record':
0239                         entry['payload_evidence'] = {'pandaid': job, 'taskid': task,
0240                             'summary': diagnosis['failure_summary'], 'grade': 'payload log'}
0241                         break
0242         except Exception as e:
0243             logger.error('error-correction decoration failed for '
0244                          '%s:%s: %s', entry.get('error_source'),
0245                          entry.get('error_code'), e)
0246     return entries