#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MVS JobLog Doctor V2 Standalone analyzer for JES2/JESMSGLG/JESYSMSG/JESJCL/SYSOUT text exports. Design goals: - Parse job identity, steps, condition codes, abnormal terminations, message families, and dataset mentions. - Classify issues with severity, confidence, root-cause candidates, and remediation guidance. - Export HTML, JSON, CSV steps, CSV issues, and CSV messages. No external dependency is required. """ from __future__ import annotations import argparse import csv import html import json import os import re import sys from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Dict, Iterable, List, Optional, Tuple VERSION = "2.0.0" SEVERITY_RANK = { "INFO": 0, "WARNING": 1, "ERROR": 2, "CRITICAL": 3, } SEVERITY_SCORE = { "INFO": 1, "WARNING": 5, "ERROR": 15, "CRITICAL": 30, } ABEND_KB: Dict[str, Dict[str, object]] = { "S0C1": { "title": "Operation exception", "area": "Program execution", "description": "The program attempted to execute an invalid operation code.", "recommendations": [ "Verify that the correct load module was executed.", "Check for corrupted load libraries or invalid branch addresses.", "Review the compile/link-edit chain and program entry point.", ], }, "S0C4": { "title": "Protection exception", "area": "Memory addressing", "description": "A program attempted to access storage that was not addressable or not permitted.", "recommendations": [ "Check pointer/address handling and table bounds.", "Review recent copybook or linkage section changes.", "Inspect dump registers and failing instruction if available.", ], }, "S0C7": { "title": "Data exception", "area": "Data quality / COBOL numeric fields", "description": "Invalid numeric data was used, often packed decimal or zoned decimal content that does not match the program definition.", "recommendations": [ "Validate input file layout against the COBOL copybook.", "Check packed decimal and zoned decimal fields in the failing record.", "Review upstream feeds and recent file format changes.", ], }, "S222": { "title": "Operator or system cancel", "area": "Operations control", "description": "The job or started task was cancelled by an operator or automation.", "recommendations": [ "Check console automation, scheduler actions, and operator commands.", "Verify whether the cancel was intentional or timeout-driven.", ], }, "S322": { "title": "CPU time limit exceeded", "area": "Runtime limit", "description": "The step exceeded its CPU time limit.", "recommendations": [ "Review TIME parameters and scheduler limits.", "Check for loops, data volume spikes, or missing indexes in database access.", ], }, "S522": { "title": "Wait time limit exceeded", "area": "Runtime wait", "description": "The step exceeded its allowed wait time.", "recommendations": [ "Check external resources, enqueue waits, database locks, and tape or dataset availability.", "Review scheduler and system wait limits.", ], }, "S806": { "title": "Load module not found", "area": "Program/library resolution", "description": "The requested program could not be loaded, commonly due to missing or incorrect STEPLIB/JOBLIB/LNKLST definitions.", "recommendations": [ "Verify the EXEC PGM name.", "Check STEPLIB, JOBLIB, LNKLST, and APF library definitions.", "Confirm that the load module exists in the expected library.", ], }, "S837": { "title": "End-of-volume / space management problem", "area": "Dataset allocation", "description": "A dataset could not extend as required, often linked to volume or allocation constraints.", "recommendations": [ "Check SPACE parameters, volume availability, and dataset limits.", "Review SMS storage class and allocation policies.", ], }, "S878": { "title": "Insufficient virtual storage", "area": "Region/storage", "description": "The program could not obtain enough virtual storage.", "recommendations": [ "Review REGION or MEMLIMIT settings.", "Look for storage leaks or excessive in-memory tables.", ], }, "S80A": { "title": "Insufficient virtual storage", "area": "Region/storage", "description": "The address space could not obtain required storage.", "recommendations": [ "Increase REGION or MEMLIMIT if appropriate.", "Investigate storage fragmentation or abnormal memory consumption.", ], }, "S913": { "title": "Security authorization failure", "area": "RACF/security", "description": "The job was not authorized to access a protected resource, commonly a dataset.", "recommendations": [ "Check RACF/ACF2/Top Secret access for the job user ID.", "Review ICH408I or related security messages near the failure.", "Validate dataset profile, access level, and group membership.", ], }, "SB37": { "title": "Dataset space exhausted", "area": "Dataset allocation", "description": "The dataset ran out of space and could not be extended.", "recommendations": [ "Increase primary/secondary SPACE allocation.", "Check SMS storage class, volume free space, and extent limits.", "Consider larger allocation units or a different storage class.", ], }, "SD37": { "title": "Dataset secondary allocation missing or exhausted", "area": "Dataset allocation", "description": "The dataset needed more space but no suitable secondary allocation was available.", "recommendations": [ "Add or increase secondary SPACE allocation.", "Check whether the dataset is sequential and volume constrained.", ], }, "SE37": { "title": "Volume or extent limit reached", "area": "Dataset allocation", "description": "The dataset could not obtain additional extents or volumes.", "recommendations": [ "Review extent count, volume count, and SMS rules.", "Allocate a new dataset with more appropriate space parameters.", ], }, "U4038": { "title": "Language Environment user abend", "area": "Application runtime", "description": "The program ended under Language Environment with an application-level failure.", "recommendations": [ "Review CEEDUMP, application messages, and LE runtime output.", "Find the original application error before the U4038 wrapper abend.", ], }, "U4088": { "title": "Language Environment severe error", "area": "Application runtime", "description": "Language Environment detected a severe condition.", "recommendations": [ "Review LE messages, CEEDUMP, and runtime options.", "Look for preceding COBOL, PL/I, or C runtime messages.", ], }, "U0999": { "title": "Application-defined user abend", "area": "Application logic", "description": "The application explicitly issued a user abend.", "recommendations": [ "Review application-specific messages before the abend.", "Check documentation for the user abend code in the application or package.", ], }, } MESSAGE_FAMILIES: List[Tuple[str, str, str]] = [ ("$HASP", "JES2", "JES2 job scheduling and spool messages"), ("IEF", "Job management", "JCL execution, allocation, and step termination"), ("IEA", "System", "z/OS system action or status"), ("IEC", "Dataset I/O", "DFSMS dataset, catalog, or I/O condition"), ("IGD", "SMS allocation", "Storage Management Subsystem allocation"), ("ICH", "Security", "RACF security and authorization"), ("IRR", "Security", "RACF security and auditing"), ("IDC", "IDCAMS/VSAM", "IDCAMS, VSAM, and catalog utility output"), ("IGY", "COBOL compiler", "Enterprise COBOL compiler messages"), ("IEW", "Binder", "Binder/link-edit messages"), ("CEE", "Language Environment", "Language Environment runtime messages"), ("DFH", "CICS", "CICS messages"), ("DSN", "Db2", "Db2 for z/OS messages"), ("DFS", "IMS", "IMS messages"), ("IKJ", "TSO/REXX", "TSO, CLIST, or REXX messages"), ] KNOWN_MESSAGE_GUIDANCE: Dict[str, Dict[str, str]] = { "IEF142I": { "severity": "INFO", "title": "Step completed with condition code", "recommendation": "Use the condition code to determine whether the step ended normally, with warnings, or with errors.", }, "IEF450I": { "severity": "CRITICAL", "title": "Step abnormally ended", "recommendation": "Focus on this step and inspect preceding application, dataset, security, or runtime messages.", }, "IEF472I": { "severity": "CRITICAL", "title": "Completion code issued", "recommendation": "Decode the system/user completion code and correlate with the failing step.", }, "IEF453I": { "severity": "CRITICAL", "title": "JCL error", "recommendation": "Review JESJCL and converter/interpreter messages before execution.", }, "ICH408I": { "severity": "CRITICAL", "title": "Security access denied", "recommendation": "Check RACF profile, access level, job user ID, and dataset/resource class.", }, } DATASET_PATTERN = re.compile( r"\b(?:DSN=|DATA\s+SET\s+|DATASET\s+|DSNAME=|CLUSTER\s+NAME\s+IS\s+)?" r"([A-Z0-9#$@][A-Z0-9#$@-]{0,7}(?:\.[A-Z0-9#$@][A-Z0-9#$@-]{0,7}){1,21})\b", re.IGNORECASE, ) MESSAGE_ID_PATTERN = re.compile(r"(?[A-Z0-9#$@]+)\s+(?P.*?)\s+-\s+STEP\s+WAS\s+EXECUTED\s+-\s+COND\s+CODE\s+(?P\d{1,4})", re.IGNORECASE, ) IEF450_PATTERN = re.compile( r"\bIEF450I\s+(?P[A-Z0-9#$@]+)\s+(?P.*?)\s+-\s+ABEND=(?P[A-Z0-9]+)(?:\s+(?PU\d{3,4}))?(?:\s+REASON=(?P[A-F0-9]+))?", re.IGNORECASE, ) IEF472_PATTERN = re.compile( r"\bIEF472I\s+(?P[A-Z0-9#$@]+)\s+(?P.*?)\s+-\s+COMPLETION\s+CODE\s+-\s+SYSTEM=(?P[A-Z0-9]+)(?:\s+USER=(?P[A-Z0-9]+))?(?:\s+REASON=(?P[A-F0-9]+))?", re.IGNORECASE, ) COND_CODE_ANY_PATTERN = re.compile(r"\b(?:COND\s+CODE|RETURN\s+CODE|MAXCC|RC)\s*[=: ]\s*(\d{1,4})\b", re.IGNORECASE) ABEND_ANY_PATTERN = re.compile(r"\b(?:ABEND[= -]*)?([SU][0-9A-F]{3,4}|S[0-9A-F]{3}|SB37|SD37|SE37|U\d{4})\b", re.IGNORECASE) SPACE_ABEND_PATTERN = re.compile(r"\b(?:SB37|SD37|SE37|B37-|D37-|E37-|IEC030I|IEC031I|IEC032I)\b", re.IGNORECASE) SECURITY_PATTERN = re.compile(r"\b(?:ICH408I|S913|NOT\s+AUTHORIZED|ACCESS\s+DENIED|RACF)\b", re.IGNORECASE) JCL_ERROR_PATTERN = re.compile(r"\b(?:JCL\s+ERROR|IEF453I|IEFC\d{3,5}[A-Z]|IEFC\w+)\b", re.IGNORECASE) DATASET_NOT_FOUND_PATTERN = re.compile(r"\b(?:NOT\s+CATALOGED|NOT\s+CATALOGUED|DATA\s+SET\s+NOT\s+FOUND|DATASET\s+NOT\s+FOUND|DSN\s+NOT\s+FOUND)\b", re.IGNORECASE) COMPILER_ERROR_PATTERN = re.compile(r"\b(?:IGY[A-Z0-9]*\d+[ES]|IEW\d+[ES]|CEE\d+[ES]|DSN\d+[ES]|DFH\d+[ES]|IDC\d+[ES])\b", re.IGNORECASE) APP_ERROR_PATTERN = re.compile(r"\b[A-Z][A-Z0-9]{2,10}\d{2,6}E\b") @dataclass class JobInfo: job_name: Optional[str] = None job_id: Optional[str] = None owner: Optional[str] = None started_line: Optional[int] = None ended_line: Optional[int] = None status: str = "UNKNOWN" raw_start: Optional[str] = None raw_end: Optional[str] = None @dataclass class StepResult: sequence: int step_name: str proc_step: Optional[str] = None program: Optional[str] = None return_code: Optional[int] = None abend_code: Optional[str] = None system_code: Optional[str] = None user_code: Optional[str] = None reason_code: Optional[str] = None status: str = "UNKNOWN" line_number: Optional[int] = None evidence: List[str] = field(default_factory=list) @dataclass class MessageHit: line_number: int message_id: str family: str family_description: str severity_hint: str text: str @dataclass class DatasetMention: line_number: int dataset_name: str dd_name: Optional[str] text: str @dataclass class Issue: severity: str code: str title: str description: str recommendation: str confidence: str line_number: Optional[int] = None step_name: Optional[str] = None message_id: Optional[str] = None abend_code: Optional[str] = None dataset_name: Optional[str] = None evidence: Optional[str] = None @dataclass class RootCauseCandidate: rank: int category: str title: str confidence: str evidence: List[str] recommendations: List[str] @dataclass class AnalysisSummary: version: str analyzed_at_utc: str input_name: str profile: str status: str highest_severity: str risk_score: int max_return_code: Optional[int] abend_codes: List[str] step_count: int issue_count: int critical_count: int error_count: int warning_count: int info_count: int dataset_count: int message_count: int @dataclass class AnalysisResult: summary: AnalysisSummary job: JobInfo steps: List[StepResult] issues: List[Issue] messages: List[MessageHit] datasets: List[DatasetMention] root_causes: List[RootCauseCandidate] timeline: List[Dict[str, object]] @dataclass class Thresholds: rc_warning: int = 4 rc_error: int = 8 rc_critical: int = 12 max_risk_score: int = 100 def normalize_abend(code: Optional[str]) -> Optional[str]: if not code: return None c = code.strip().upper().replace("=", "") if c in {"0000", "U0000", "S0000"}: return None if c in {"B37", "D37", "E37"}: return "S" + c if re.fullmatch(r"[0-9A-F]{3}", c): return "S" + c if re.fullmatch(r"U\d{3}", c): return "U0" + c[1:] return c def parse_step_body(body: str) -> Tuple[Optional[str], str]: tokens = [t for t in body.strip().split() if t] if not tokens: return None, "UNKNOWN" if len(tokens) == 1: return None, tokens[0].upper() return tokens[-2].upper(), tokens[-1].upper() def classify_message(message_id: str) -> Tuple[str, str]: mid = message_id.upper() for prefix, family, description in MESSAGE_FAMILIES: if mid.startswith(prefix): return family, description return "Other", "Unclassified z/OS or application message" def severity_from_message(message_id: str, line: str) -> str: mid = message_id.upper() if mid in KNOWN_MESSAGE_GUIDANCE: return KNOWN_MESSAGE_GUIDANCE[mid]["severity"] if mid.endswith("E"): return "ERROR" if mid.endswith("W"): return "WARNING" if any(x in line.upper() for x in ["ABEND", "FAILED", "ERROR", "NOT AUTHORIZED", "JCL ERROR"]): return "ERROR" return "INFO" def issue_exists(issues: List[Issue], code: str, line_number: Optional[int], step_name: Optional[str]) -> bool: for issue in issues: if issue.code == code and issue.line_number == line_number and issue.step_name == step_name: return True return False def safe_int(value: Optional[str]) -> Optional[int]: if value is None: return None try: return int(value) except (TypeError, ValueError): return None def issue_for_abend(abend_code: str, line_number: Optional[int], step_name: Optional[str], evidence: str) -> Issue: normalized = normalize_abend(abend_code) or abend_code.upper() kb = ABEND_KB.get(normalized, {}) title = str(kb.get("title", "Abnormal termination")) description = str(kb.get("description", f"The job or step ended with abend code {normalized}.")) recommendations = kb.get("recommendations", ["Decode the completion code and inspect messages immediately before the abend."]) recommendation = " ".join(str(x) for x in recommendations) return Issue( severity="CRITICAL", code="ABEND_DETECTED", title=f"{normalized} - {title}", description=description, recommendation=recommendation, confidence="HIGH" if normalized in ABEND_KB else "MEDIUM", line_number=line_number, step_name=step_name, message_id=None, abend_code=normalized, evidence=evidence.strip(), ) class MVSJobLogDoctor: def __init__(self, thresholds: Thresholds, profile: str = "production", redact: bool = False): self.thresholds = thresholds self.profile = profile self.redact = redact self.job = JobInfo() self.steps: Dict[str, StepResult] = {} self.step_sequence = 0 self.messages: List[MessageHit] = [] self.datasets: List[DatasetMention] = [] self.issues: List[Issue] = [] self.timeline: List[Dict[str, object]] = [] self.current_step_from_jcl: Optional[str] = None self.current_dd_from_jcl: Optional[str] = None def analyze_text(self, text: str, input_name: str = "") -> AnalysisResult: for idx, line in enumerate(text.splitlines(), start=1): self._parse_line(idx, line.rstrip("\n")) self._finalize_status() root_causes = self._build_root_causes() summary = self._build_summary(input_name) return AnalysisResult( summary=summary, job=self.job, steps=list(sorted(self.steps.values(), key=lambda s: s.sequence)), issues=self._sort_issues(self.issues), messages=self.messages, datasets=self.datasets, root_causes=root_causes, timeline=sorted(self.timeline, key=lambda x: (int(x.get("line_number") or 0), str(x.get("type") or ""))), ) def _parse_line(self, line_number: int, line: str) -> None: if not line.strip(): return redacted_line = self._redact_line(line) self._parse_job_identity(line_number, redacted_line) self._parse_jcl_context(line_number, redacted_line) self._parse_messages(line_number, redacted_line) self._parse_datasets(line_number, redacted_line) self._parse_step_condition(line_number, redacted_line) self._parse_abend(line_number, redacted_line) self._parse_completion_code(line_number, redacted_line) self._parse_generic_patterns(line_number, redacted_line) def _parse_job_identity(self, line_number: int, line: str) -> None: match = JOB_CARD_PATTERN.search(line) if match and not self.job.job_name: self.job.job_name = match.group(1).upper() self.timeline.append({"line_number": line_number, "type": "JOB_CARD", "text": line.strip()}) match = HASP_START_PATTERN.search(line) if match: self.job.job_name = self.job.job_name or match.group(1).upper() self.job.started_line = line_number self.job.raw_start = line.strip() self.timeline.append({"line_number": line_number, "type": "JOB_START", "text": line.strip()}) match = HASP_END_PATTERN.search(line) if match: self.job.job_name = self.job.job_name or match.group(1).upper() self.job.ended_line = line_number self.job.raw_end = line.strip() self.timeline.append({"line_number": line_number, "type": "JOB_END", "text": line.strip()}) match = IEF403_PATTERN.search(line) if match: self.job.job_name = self.job.job_name or match.group(1).upper() self.job.started_line = self.job.started_line or line_number self.timeline.append({"line_number": line_number, "type": "JOB_START", "text": line.strip()}) match = IEF404_PATTERN.search(line) if match: self.job.job_name = self.job.job_name or match.group(1).upper() self.job.ended_line = self.job.ended_line or line_number self.timeline.append({"line_number": line_number, "type": "JOB_END", "text": line.strip()}) jobid_match = re.search(r"\b(JOB\d{5,8}|STC\d{5,8}|TSU\d{5,8})\b", line, re.IGNORECASE) if jobid_match and not self.job.job_id: self.job.job_id = jobid_match.group(1).upper() def _parse_jcl_context(self, line_number: int, line: str) -> None: exec_match = EXEC_PATTERN.search(line) if exec_match: step_name = exec_match.group(1).upper() program = exec_match.group(2).upper() if exec_match.group(2) else None step = self._get_or_create_step(step_name, line_number=line_number) if program: step.program = program step.evidence.append(line.strip()) self.current_step_from_jcl = step_name self.timeline.append({"line_number": line_number, "type": "JCL_EXEC", "step": step_name, "text": line.strip()}) return dd_match = DD_PATTERN.search(line) if dd_match: self.current_dd_from_jcl = dd_match.group(1).upper() def _parse_messages(self, line_number: int, line: str) -> None: for match in MESSAGE_ID_PATTERN.finditer(line): message_id = match.group(1).upper() family, description = classify_message(message_id) severity = severity_from_message(message_id, line) hit = MessageHit( line_number=line_number, message_id=message_id, family=family, family_description=description, severity_hint=severity, text=line.strip(), ) self.messages.append(hit) if severity in {"ERROR", "CRITICAL"} or message_id in KNOWN_MESSAGE_GUIDANCE: self.timeline.append({ "line_number": line_number, "type": "MESSAGE", "message_id": message_id, "severity": severity, "text": line.strip(), }) if message_id in KNOWN_MESSAGE_GUIDANCE and message_id not in {"IEF142I"}: guidance = KNOWN_MESSAGE_GUIDANCE[message_id] step_name = self._step_from_known_context(line) if not issue_exists(self.issues, f"MSG_{message_id}", line_number, step_name): self.issues.append(Issue( severity=guidance["severity"], code=f"MSG_{message_id}", title=guidance["title"], description=f"Message {message_id} detected in the job log.", recommendation=guidance["recommendation"], confidence="HIGH", line_number=line_number, step_name=step_name, message_id=message_id, evidence=line.strip(), )) elif severity in {"ERROR", "CRITICAL"} and not message_id.startswith(("IEF142",)): if APP_ERROR_PATTERN.search(message_id) or message_id.startswith(("IGY", "IEW", "CEE", "IDC", "DSN", "DFH")): step_name = self._step_from_known_context(line) or self.current_step_from_jcl self.issues.append(Issue( severity=severity, code="MESSAGE_ERROR", title=f"Error message {message_id} detected", description=f"The log contains an error-class message from family {family}.", recommendation="Inspect the full message text and the few lines before it to identify the failing program, utility, or resource.", confidence="MEDIUM", line_number=line_number, step_name=step_name, message_id=message_id, evidence=line.strip(), )) def _parse_datasets(self, line_number: int, line: str) -> None: for match in DATASET_PATTERN.finditer(line): dataset = match.group(1).upper() if dataset.startswith(("HTTP.", "HTTPS.", "WWW.")): continue self.datasets.append(DatasetMention( line_number=line_number, dataset_name=dataset, dd_name=self.current_dd_from_jcl, text=line.strip(), )) def _parse_step_condition(self, line_number: int, line: str) -> None: match = IEF142_PATTERN.search(line) if not match: return proc_step, step_name = parse_step_body(match.group("body")) rc = safe_int(match.group("cc")) step = self._get_or_create_step(step_name, proc_step=proc_step, line_number=line_number) step.return_code = rc step.status = self._status_from_rc(rc) step.evidence.append(line.strip()) self.timeline.append({"line_number": line_number, "type": "STEP_RC", "step": step_name, "rc": rc, "text": line.strip()}) if rc is not None and rc >= self.thresholds.rc_warning: severity = "WARNING" if rc >= self.thresholds.rc_error: severity = "ERROR" if rc >= self.thresholds.rc_critical: severity = "CRITICAL" self.issues.append(Issue( severity=severity, code="HIGH_RETURN_CODE", title=f"Step {step_name} ended with RC={rc:04d}", description="The step completed normally from a system perspective but returned a non-zero condition code.", recommendation="Inspect the step output, utility messages, compiler messages, or application messages to determine whether this RC is acceptable.", confidence="HIGH", line_number=line_number, step_name=step_name, message_id="IEF142I", evidence=line.strip(), )) def _parse_abend(self, line_number: int, line: str) -> None: match = IEF450_PATTERN.search(line) if match: proc_step, step_name = parse_step_body(match.group("body")) abend = normalize_abend(match.group("abend")) user = normalize_abend(match.group("user")) if match.group("user") else None step = self._get_or_create_step(step_name, proc_step=proc_step, line_number=line_number) step.abend_code = abend step.user_code = user step.reason_code = match.group("reason") step.status = "ABENDED" step.evidence.append(line.strip()) self.timeline.append({"line_number": line_number, "type": "ABEND", "step": step_name, "abend": abend, "text": line.strip()}) if abend: self.issues.append(issue_for_abend(abend, line_number, step_name, line)) if user and user != abend: self.issues.append(issue_for_abend(user, line_number, step_name, line)) return # Generic abend detection fallback. It avoids IEF472I because a dedicated parser handles it. if "IEF472I" not in line.upper() and "ABEND" in line.upper(): for abend_match in ABEND_ANY_PATTERN.finditer(line): abend = normalize_abend(abend_match.group(1)) if not abend: continue step_name = self._step_from_known_context(line) or self.current_step_from_jcl if not issue_exists(self.issues, "ABEND_DETECTED", line_number, step_name): self.issues.append(issue_for_abend(abend, line_number, step_name, line)) def _parse_completion_code(self, line_number: int, line: str) -> None: match = IEF472_PATTERN.search(line) if not match: return proc_step, step_name = parse_step_body(match.group("body")) system_code = normalize_abend(match.group("system")) user_code = normalize_abend(match.group("user")) if match.group("user") else None step = self._get_or_create_step(step_name, proc_step=proc_step, line_number=line_number) step.system_code = system_code step.user_code = user_code step.reason_code = match.group("reason") step.status = "ABENDED" step.evidence.append(line.strip()) self.timeline.append({"line_number": line_number, "type": "COMPLETION_CODE", "step": step_name, "system": system_code, "user": user_code, "text": line.strip()}) if system_code and not issue_exists(self.issues, "ABEND_DETECTED", line_number, step_name): self.issues.append(issue_for_abend(system_code, line_number, step_name, line)) if user_code and user_code not in {system_code, "U0000"}: self.issues.append(issue_for_abend(user_code, line_number, step_name, line)) def _parse_generic_patterns(self, line_number: int, line: str) -> None: upper = line.upper() step_name = self._step_from_known_context(line) or self.current_step_from_jcl if SPACE_ABEND_PATTERN.search(line): dataset = self._first_dataset_in_line(line) self.issues.append(Issue( severity="CRITICAL", code="DATASET_SPACE_PROBLEM", title="Dataset space or extent problem detected", description="The log contains a B37/D37/E37 or IEC space-related symptom.", recommendation="Check SPACE parameters, secondary allocation, SMS storage class, target volumes, and extent limits.", confidence="HIGH", line_number=line_number, step_name=step_name, dataset_name=dataset, evidence=line.strip(), )) if SECURITY_PATTERN.search(line): dataset = self._first_dataset_in_line(line) self.issues.append(Issue( severity="CRITICAL", code="SECURITY_ACCESS_PROBLEM", title="Security or authorization problem detected", description="The log contains a RACF/security access symptom.", recommendation="Validate the job user ID, RACF profile, access level, dataset or resource class, and group membership.", confidence="HIGH", line_number=line_number, step_name=step_name, dataset_name=dataset, evidence=line.strip(), )) if JCL_ERROR_PATTERN.search(line): self.issues.append(Issue( severity="CRITICAL", code="JCL_ERROR", title="JCL error detected", description="The job contains a converter/interpreter or execution-time JCL error symptom.", recommendation="Inspect JESJCL and the messages immediately around the JCL error. Check DD statements, PROC parameters, and symbolic substitutions.", confidence="HIGH", line_number=line_number, step_name=step_name, evidence=line.strip(), )) if DATASET_NOT_FOUND_PATTERN.search(line): dataset = self._first_dataset_in_line(line) self.issues.append(Issue( severity="ERROR", code="DATASET_NOT_FOUND", title="Dataset not found or not cataloged", description="The log indicates that a referenced dataset could not be located or cataloged.", recommendation="Check DSN spelling, catalog availability, GDG generation, DISP settings, and scheduling dependencies.", confidence="HIGH", line_number=line_number, step_name=step_name, dataset_name=dataset, evidence=line.strip(), )) if COMPILER_ERROR_PATTERN.search(line): message = COMPILER_ERROR_PATTERN.search(line).group(0).upper() self.issues.append(Issue( severity="ERROR", code="TOOLCHAIN_ERROR", title=f"Toolchain error message {message} detected", description="The log contains a compiler, binder, utility, Db2, CICS, or runtime error message.", recommendation="Inspect the specific tool output section and correlate with the step return code.", confidence="MEDIUM", line_number=line_number, step_name=step_name, message_id=message, evidence=line.strip(), )) # MAXCC fallback, often from IDCAMS or utilities. if "MAXCC" in upper: rc_match = COND_CODE_ANY_PATTERN.search(line) rc = safe_int(rc_match.group(1)) if rc_match else None if rc is not None and rc >= self.thresholds.rc_warning: severity = "WARNING" if rc >= self.thresholds.rc_error: severity = "ERROR" if rc >= self.thresholds.rc_critical: severity = "CRITICAL" self.issues.append(Issue( severity=severity, code="UTILITY_MAXCC", title=f"Utility MAXCC={rc:04d}", description="A utility reported a non-zero maximum condition code.", recommendation="Inspect utility-specific messages before the MAXCC line.", confidence="MEDIUM", line_number=line_number, step_name=step_name, evidence=line.strip(), )) def _get_or_create_step(self, step_name: str, proc_step: Optional[str] = None, line_number: Optional[int] = None) -> StepResult: name = (step_name or "UNKNOWN").upper() if name not in self.steps: self.step_sequence += 1 self.steps[name] = StepResult( sequence=self.step_sequence, step_name=name, proc_step=proc_step, line_number=line_number, ) else: if proc_step and not self.steps[name].proc_step: self.steps[name].proc_step = proc_step if line_number and not self.steps[name].line_number: self.steps[name].line_number = line_number return self.steps[name] def _status_from_rc(self, rc: Optional[int]) -> str: if rc is None: return "UNKNOWN" if rc == 0: return "OK" if rc < self.thresholds.rc_error: return "WARNING" if rc < self.thresholds.rc_critical: return "ERROR" return "CRITICAL" def _step_from_known_context(self, line: str) -> Optional[str]: for pattern in [IEF142_PATTERN, IEF450_PATTERN, IEF472_PATTERN]: match = pattern.search(line) if match: _, step_name = parse_step_body(match.group("body")) return step_name return None def _first_dataset_in_line(self, line: str) -> Optional[str]: match = DATASET_PATTERN.search(line) if not match: return None return match.group(1).upper() def _redact_line(self, line: str) -> str: if not self.redact: return line def repl(match: re.Match[str]) -> str: value = match.group(1).upper() parts = value.split(".") if len(parts) <= 2: return "HLQ.***" return parts[0] + ".***." + parts[-1] return DATASET_PATTERN.sub(lambda m: repl(m), line) def _finalize_status(self) -> None: has_critical = any(issue.severity == "CRITICAL" for issue in self.issues) has_error = any(issue.severity == "ERROR" for issue in self.issues) any_abend = any(step.status == "ABENDED" or step.abend_code or step.system_code for step in self.steps.values()) max_rc = self._max_return_code() if any_abend or has_critical: self.job.status = "FAILED" elif has_error or (max_rc is not None and max_rc >= self.thresholds.rc_error): self.job.status = "ERROR" elif max_rc is not None and max_rc >= self.thresholds.rc_warning: self.job.status = "WARNING" elif self.steps: self.job.status = "OK" else: self.job.status = "UNKNOWN" def _max_return_code(self) -> Optional[int]: values = [step.return_code for step in self.steps.values() if step.return_code is not None] if not values: return None return max(values) def _build_root_causes(self) -> List[RootCauseCandidate]: candidates: List[RootCauseCandidate] = [] seen = set() def add(category: str, title: str, confidence: str, evidence: Iterable[str], recommendations: Iterable[str]) -> None: key = (category, title) if key in seen: return seen.add(key) candidates.append(RootCauseCandidate( rank=len(candidates) + 1, category=category, title=title, confidence=confidence, evidence=list(evidence)[:6], recommendations=list(recommendations)[:6], )) for issue in self._sort_issues(self.issues): if issue.abend_code: kb = ABEND_KB.get(issue.abend_code, {}) recs = kb.get("recommendations", [issue.recommendation]) add( category=str(kb.get("area", "Abend")), title=issue.title, confidence=issue.confidence, evidence=[issue.evidence or "", f"Step: {issue.step_name or 'UNKNOWN'}"], recommendations=[str(x) for x in recs], ) elif issue.code in {"DATASET_SPACE_PROBLEM", "DATASET_NOT_FOUND", "SECURITY_ACCESS_PROBLEM", "JCL_ERROR"}: add( category=issue.code.replace("_", " ").title(), title=issue.title, confidence=issue.confidence, evidence=[issue.evidence or "", f"Dataset: {issue.dataset_name or 'UNKNOWN'}", f"Step: {issue.step_name or 'UNKNOWN'}"], recommendations=[issue.recommendation], ) if len(candidates) >= 5: break if not candidates: max_rc = self._max_return_code() if max_rc is not None and max_rc > 0: add( category="Return code", title=f"Non-zero maximum return code {max_rc:04d}", confidence="MEDIUM", evidence=[f"MAX RC: {max_rc:04d}"], recommendations=["Review step outputs and utility messages associated with the highest return code."], ) else: add( category="No critical root cause", title="No obvious failure signature detected", confidence="LOW", evidence=["No ABEND, high RC, or critical message was detected by the current rules."], recommendations=["Review the full job output manually if business validation still failed."], ) for idx, candidate in enumerate(candidates, start=1): candidate.rank = idx return candidates def _build_summary(self, input_name: str) -> AnalysisSummary: counts = {severity: 0 for severity in SEVERITY_RANK} for issue in self.issues: counts[issue.severity] = counts.get(issue.severity, 0) + 1 highest = "INFO" for issue in self.issues: if SEVERITY_RANK[issue.severity] > SEVERITY_RANK[highest]: highest = issue.severity if not self.issues: highest = "INFO" raw_score = 0 for issue in self.issues: raw_score += SEVERITY_SCORE.get(issue.severity, 0) for step in self.steps.values(): if step.return_code is not None: raw_score += min(step.return_code, 16) if step.abend_code or step.system_code: raw_score += 25 risk_score = min(self.thresholds.max_risk_score, raw_score) abends = sorted({ code for code in ( [step.abend_code for step in self.steps.values()] + [step.system_code for step in self.steps.values()] + [step.user_code for step in self.steps.values()] + [issue.abend_code for issue in self.issues] ) if code and code != "U0000" }) return AnalysisSummary( version=VERSION, analyzed_at_utc=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), input_name=os.path.basename(input_name), profile=self.profile, status=self.job.status, highest_severity=highest, risk_score=risk_score, max_return_code=self._max_return_code(), abend_codes=abends, step_count=len(self.steps), issue_count=len(self.issues), critical_count=counts.get("CRITICAL", 0), error_count=counts.get("ERROR", 0), warning_count=counts.get("WARNING", 0), info_count=counts.get("INFO", 0), dataset_count=len(self.datasets), message_count=len(self.messages), ) def _sort_issues(self, issues: List[Issue]) -> List[Issue]: return sorted(issues, key=lambda x: (-SEVERITY_RANK.get(x.severity, 0), x.line_number or 0, x.code)) DEMO_JOBLOG = """//PAYROLL1 JOB (ACCT),'PAYROLL TEST',CLASS=A,MSGCLASS=X,NOTIFY=&SYSUID $HASP373 PAYROLL1 STARTED - INIT 3 - CLASS A - SYS SYS1 IEF403I PAYROLL1 - STARTED - TIME=21.15.01 //STEP010 EXEC PGM=IEFBR14 IEF142I PAYROLL1 STEP010 - STEP WAS EXECUTED - COND CODE 0000 //STEP020 EXEC PGM=SORT //SORTIN DD DSN=PROD.PAYROLL.TRANS.G0007V00,DISP=SHR //SORTOUT DD DSN=PROD.PAYROLL.SORTED.G0007V00,DISP=(NEW,CATLG,DELETE) ICE000I 1 - CONTROL STATEMENTS FOR 5694-A01, Z/OS DFSORT ICE201I 0 RECORD TYPE IS F - DATA STARTS IN POSITION 1 IEF142I PAYROLL1 STEP020 - STEP WAS EXECUTED - COND CODE 0004 //STEP030 EXEC PGM=PAYCALC //PAYIN DD DSN=PROD.PAYROLL.SORTED.G0007V00,DISP=SHR PAYERR01E INVALID PACKED DECIMAL IN EMPLOYEE GROSS PAY FIELD CEE3207S THE SYSTEM DETECTED A DATA EXCEPTION (SYSTEM COMPLETION CODE=0C7). IEF450I PAYROLL1 STEP030 - ABEND=S0C7 U0000 REASON=00000000 TIME=21.16.33 IEF472I PAYROLL1 STEP030 - COMPLETION CODE - SYSTEM=S0C7 USER=0000 REASON=00000000 IEF404I PAYROLL1 - ENDED - TIME=21.16.34 $HASP395 PAYROLL1 ENDED - ABEND=S0C7 """ SPACE_JOBLOG = """//ARCHIVE1 JOB (ACCT),'ARCHIVE TEST',CLASS=B,MSGCLASS=X $HASP373 ARCHIVE1 STARTED - INIT 2 - CLASS B - SYS SYS1 IEF403I ARCHIVE1 - STARTED - TIME=02.10.00 //STEP010 EXEC PGM=IDCAMS //OUTDD DD DSN=PROD.ARCHIVE.MONTHLY.BACKUP,DISP=(NEW,CATLG,DELETE),SPACE=(CYL,(50,0)) IEC030I B37-04,IFG0554A,ARCHIVE1,STEP010,OUTDD,3390,VOL001,PROD.ARCHIVE.MONTHLY.BACKUP IDC3009I ** VSAM CATALOG RETURN CODE IS 8 - REASON CODE IS IGG0CLEH-42 IDC0005I NUMBER OF RECORDS PROCESSED WAS 00000000 IDC0001I FUNCTION COMPLETED, HIGHEST CONDITION CODE WAS 12 IEF450I ARCHIVE1 STEP010 - ABEND=SB37 U0000 REASON=00000004 TIME=02.12.16 IEF472I ARCHIVE1 STEP010 - COMPLETION CODE - SYSTEM=SB37 USER=0000 REASON=00000004 IEF404I ARCHIVE1 - ENDED - TIME=02.12.17 $HASP395 ARCHIVE1 ENDED - ABEND=SB37 """ SECURITY_JOBLOG = """//SECFAIL JOB (ACCT),'SECURITY TEST',CLASS=A,MSGCLASS=X $HASP373 SECFAIL STARTED - INIT 4 - CLASS A - SYS SYS1 IEF403I SECFAIL - STARTED - TIME=10.00.01 //STEP010 EXEC PGM=IEBGENER //SYSUT1 DD DSN=PROD.SECRET.CUSTOMER.FILE,DISP=SHR ICH408I USER(BATCH01 ) GROUP(PAYROLL ) NAME(BATCH USER) PROD.SECRET.CUSTOMER.FILE CL(DATASET ) INSUFFICIENT ACCESS AUTHORITY ICH408I ACCESS INTENT(READ ) ACCESS ALLOWED(NONE ) IEF450I SECFAIL STEP010 - ABEND=S913 U0000 REASON=00000000 TIME=10.00.03 IEF472I SECFAIL STEP010 - COMPLETION CODE - SYSTEM=S913 USER=0000 REASON=00000000 IEF404I SECFAIL - ENDED - TIME=10.00.04 $HASP395 SECFAIL ENDED - ABEND=S913 """ def result_to_dict(result: AnalysisResult) -> Dict[str, object]: return asdict(result) def write_json(result: AnalysisResult, path: str) -> None: with open(path, "w", encoding="utf-8") as fh: json.dump(result_to_dict(result), fh, indent=2, ensure_ascii=False) def write_csv_issues(result: AnalysisResult, path: str) -> None: fields = list(asdict(result.issues[0]).keys()) if result.issues else [ "severity", "code", "title", "description", "recommendation", "confidence", "line_number", "step_name", "message_id", "abend_code", "dataset_name", "evidence" ] with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for issue in result.issues: writer.writerow(asdict(issue)) def write_csv_steps(result: AnalysisResult, path: str) -> None: fields = list(asdict(result.steps[0]).keys()) if result.steps else [ "sequence", "step_name", "proc_step", "program", "return_code", "abend_code", "system_code", "user_code", "reason_code", "status", "line_number", "evidence" ] with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for step in result.steps: row = asdict(step) row["evidence"] = " | ".join(step.evidence) writer.writerow(row) def write_csv_messages(result: AnalysisResult, path: str) -> None: fields = list(asdict(result.messages[0]).keys()) if result.messages else [ "line_number", "message_id", "family", "family_description", "severity_hint", "text" ] with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for message in result.messages: writer.writerow(asdict(message)) def html_escape(value: object) -> str: if value is None: return "" return html.escape(str(value)) def severity_badge(severity: str) -> str: cls = { "CRITICAL": "critical", "ERROR": "error", "WARNING": "warning", "INFO": "info", }.get(severity, "info") return f'{html_escape(severity)}' def render_html_report(result: AnalysisResult) -> str: summary = result.summary root_rows = [] for candidate in result.root_causes: evidence = "
".join(html_escape(x) for x in candidate.evidence if x) recommendations = "
".join(html_escape(x) for x in candidate.recommendations if x) root_rows.append( f"{candidate.rank}{html_escape(candidate.category)}{html_escape(candidate.title)}" f"{html_escape(candidate.confidence)}{evidence}{recommendations}" ) step_rows = [] for step in result.steps: step_rows.append( "" f"{step.sequence}" f"{html_escape(step.step_name)}" f"{html_escape(step.program)}" f"{html_escape(step.return_code)}" f"{html_escape(step.abend_code or step.system_code or step.user_code)}" f"{html_escape(step.status)}" f"{html_escape(' | '.join(step.evidence[:3]))}" "" ) issue_rows = [] for issue in result.issues[:200]: issue_rows.append( "" f"{severity_badge(issue.severity)}" f"{html_escape(issue.code)}" f"{html_escape(issue.title)}" f"{html_escape(issue.step_name)}" f"{html_escape(issue.abend_code or issue.message_id)}" f"{html_escape(issue.dataset_name)}" f"{html_escape(issue.recommendation)}" f"{html_escape(issue.evidence)}" "" ) timeline_rows = [] for event in result.timeline[:300]: timeline_rows.append( "" f"{html_escape(event.get('line_number'))}" f"{html_escape(event.get('type'))}" f"{html_escape(event.get('step'))}" f"{html_escape(event.get('severity'))}" f"{html_escape(event.get('text'))}" "" ) message_family_counts: Dict[str, int] = {} for message in result.messages: message_family_counts[message.family] = message_family_counts.get(message.family, 0) + 1 family_rows = [] for family, count in sorted(message_family_counts.items(), key=lambda x: (-x[1], x[0])): family_rows.append(f"{html_escape(family)}{count}") return f""" MVS JobLog Doctor V2 Report

MVS JobLog Doctor V2 Report

Input: {html_escape(summary.input_name)} | Profile: {html_escape(summary.profile)} | Generated: {html_escape(summary.analyzed_at_utc)}

{html_escape(summary.status)}
Job status
{summary.risk_score}/100
Risk score
{html_escape(summary.max_return_code)}
Max return code
{', '.join(summary.abend_codes) or '-'}
Abend codes
{summary.step_count}
Steps
{summary.issue_count}
Issues
{summary.critical_count}
Critical
{summary.dataset_count}
Dataset mentions
Job: {html_escape(result.job.job_name)} | Job ID: {html_escape(result.job.job_id)} | Highest severity: {html_escape(summary.highest_severity)}

Root cause candidates

{''.join(root_rows)}
#CategoryCandidateConfidenceEvidenceRecommendations

Steps

{''.join(step_rows)}
#StepProgramRCAbendStatusEvidence

Issues

{''.join(issue_rows)}
SeverityCodeTitleStepMessage/AbendDatasetRecommendationEvidence

Message family distribution

{''.join(family_rows)}
FamilyCount

Timeline

{''.join(timeline_rows)}
LineTypeStepSeverityText
""" def write_html(result: AnalysisResult, path: str) -> None: with open(path, "w", encoding="utf-8") as fh: fh.write(render_html_report(result)) def print_console(result: AnalysisResult, evidence_limit: int = 5) -> None: s = result.summary print("MVS JobLog Doctor V2") print("=" * 72) print(f"Input : {s.input_name}") print(f"Job name : {result.job.job_name or '-'}") print(f"Job ID : {result.job.job_id or '-'}") print(f"Status : {s.status}") print(f"Highest severity : {s.highest_severity}") print(f"Risk score : {s.risk_score}/100") print(f"Max RC : {s.max_return_code if s.max_return_code is not None else '-'}") print(f"Abends : {', '.join(s.abend_codes) if s.abend_codes else '-'}") print(f"Steps : {s.step_count}") print(f"Issues : {s.issue_count} (critical={s.critical_count}, error={s.error_count}, warning={s.warning_count})") print(f"Datasets : {s.dataset_count}") print() print("ROOT CAUSE CANDIDATES") print("-" * 72) for candidate in result.root_causes: print(f"{candidate.rank}. [{candidate.confidence}] {candidate.title}") print(f" Area: {candidate.category}") if candidate.evidence: print(" Evidence:") for item in candidate.evidence[:evidence_limit]: if item: print(f" - {item}") if candidate.recommendations: print(" Recommendations:") for item in candidate.recommendations[:evidence_limit]: if item: print(f" - {item}") print() print("STEPS") print("-" * 72) for step in result.steps: abend = step.abend_code or step.system_code or step.user_code or "-" rc = f"{step.return_code:04d}" if step.return_code is not None else "-" program = step.program or "-" print(f"{step.sequence:02d} {step.step_name:8} PGM={program:8} RC={rc:>4} ABEND={abend:>5} STATUS={step.status}") print() print("TOP ISSUES") print("-" * 72) for issue in result.issues[:10]: line = f"L{issue.line_number}" if issue.line_number else "-" step = issue.step_name or "-" print(f"{issue.severity:8} {line:>6} STEP={step:8} {issue.title}") print(f" {issue.recommendation}") def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="MVS JobLog Doctor V2 - JES/JOBLOG analyzer") parser.add_argument("input_file", nargs="?", help="JES2/JESMSGLG/JESYSMSG/JESJCL/SYSOUT text file") parser.add_argument("--demo", action="store_true", help="Use the built-in S0C7 demo joblog") parser.add_argument("--demo-space", action="store_true", help="Use the built-in SB37 space demo joblog") parser.add_argument("--demo-security", action="store_true", help="Use the built-in S913 security demo joblog") parser.add_argument("--profile", default="production", choices=["production", "batch", "compiler", "security", "training"], help="Analysis profile label") parser.add_argument("--redact", action="store_true", help="Redact dataset names in parsed evidence") parser.add_argument("--json", dest="json_output", help="Write full analysis JSON") parser.add_argument("--csv-issues", help="Write issue CSV") parser.add_argument("--csv-steps", help="Write step CSV") parser.add_argument("--csv-messages", help="Write message CSV") parser.add_argument("--html", dest="html_output", help="Write HTML report") parser.add_argument("--fail-on-critical", action="store_true", help="Exit with code 2 if critical issues exist") parser.add_argument("--fail-on-error", action="store_true", help="Exit with code 1 if error or critical issues exist") parser.add_argument("--rc-warning", type=int, default=4, help="Warning return code threshold") parser.add_argument("--rc-error", type=int, default=8, help="Error return code threshold") parser.add_argument("--rc-critical", type=int, default=12, help="Critical return code threshold") parser.add_argument("--evidence-limit", type=int, default=5, help="Console evidence lines per root cause") return parser def load_input(args: argparse.Namespace) -> Tuple[str, str]: demos = [args.demo, args.demo_space, args.demo_security] if sum(1 for value in demos if value) > 1: raise SystemExit("Choose only one demo mode.") if args.demo: return DEMO_JOBLOG, "demo_s0c7_joblog.txt" if args.demo_space: return SPACE_JOBLOG, "demo_sb37_space_joblog.txt" if args.demo_security: return SECURITY_JOBLOG, "demo_s913_security_joblog.txt" if not args.input_file: raise SystemExit("Missing input_file. Use --demo, --demo-space, --demo-security, or provide a text file.") with open(args.input_file, "r", encoding="utf-8", errors="ignore") as fh: return fh.read(), args.input_file def main(argv: Optional[List[str]] = None) -> int: parser = build_arg_parser() args = parser.parse_args(argv) text, input_name = load_input(args) thresholds = Thresholds(rc_warning=args.rc_warning, rc_error=args.rc_error, rc_critical=args.rc_critical) doctor = MVSJobLogDoctor(thresholds=thresholds, profile=args.profile, redact=args.redact) result = doctor.analyze_text(text, input_name=input_name) print_console(result, evidence_limit=args.evidence_limit) if args.json_output: write_json(result, args.json_output) print(f"\nJSON exported: {args.json_output}") if args.csv_issues: write_csv_issues(result, args.csv_issues) print(f"CSV issues exported: {args.csv_issues}") if args.csv_steps: write_csv_steps(result, args.csv_steps) print(f"CSV steps exported: {args.csv_steps}") if args.csv_messages: write_csv_messages(result, args.csv_messages) print(f"CSV messages exported: {args.csv_messages}") if args.html_output: write_html(result, args.html_output) print(f"HTML exported: {args.html_output}") if args.fail_on_critical and result.summary.critical_count > 0: return 2 if args.fail_on_error and (result.summary.critical_count > 0 or result.summary.error_count > 0): return 1 return 0 if __name__ == "__main__": raise SystemExit(main())