#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SYSLOG / OPERLOG Incident Analyzer V2 Advanced incident timeline, correlation, and root-cause analyzer for z/OS SYSLOG, OPERLOG, JES, and subsystem logs. This tool intentionally works on text exports, not on binary SMF or proprietary spool formats. It is designed for offline diagnostics, training, incident support, and integration into audit pipelines. Code and comments are English-only by design. """ from __future__ import annotations import argparse import csv import datetime as dt import html import json import os import re import sys from collections import Counter, defaultdict from dataclasses import dataclass, field, asdict from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple SEVERITY_ORDER = {"INFO": 0, "WARNING": 1, "ERROR": 2, "CRITICAL": 3} SEVERITY_WEIGHT = {"INFO": 1, "WARNING": 8, "ERROR": 18, "CRITICAL": 32} MESSAGE_FAMILIES: Dict[str, Dict[str, str]] = { "$HASP": {"subsystem": "JES", "meaning": "JES2 message"}, "HASP": {"subsystem": "JES", "meaning": "JES message"}, "IEF": {"subsystem": "MVS", "meaning": "Job and step control"}, "IEC": {"subsystem": "DATASET", "meaning": "Data management / dataset"}, "IGD": {"subsystem": "SMS", "meaning": "SMS allocation"}, "ICH": {"subsystem": "SECURITY", "meaning": "RACF security"}, "IRR": {"subsystem": "SECURITY", "meaning": "RACF security"}, "CSV": {"subsystem": "LNKLST", "meaning": "Contents supervision / load module"}, "IOS": {"subsystem": "IO", "meaning": "I/O supervisor"}, "IOSH": {"subsystem": "IO", "meaning": "I/O supervisor"}, "IEA": {"subsystem": "SYSTEM", "meaning": "System error / recovery"}, "IEE": {"subsystem": "OPERATOR", "meaning": "Operator command / console"}, "IRA": {"subsystem": "STORAGE", "meaning": "Real storage manager"}, "IXC": {"subsystem": "SYSPLEX", "meaning": "XCF / sysplex"}, "IXG": {"subsystem": "LOGGER", "meaning": "System logger"}, "IXL": {"subsystem": "COUPLING", "meaning": "Coupling facility"}, "CEE": {"subsystem": "LE", "meaning": "Language Environment"}, "IDC": {"subsystem": "IDCAMS", "meaning": "Access method services"}, "ICE": {"subsystem": "SORT", "meaning": "DFSORT / sort"}, "WER": {"subsystem": "SORT", "meaning": "Syncsort / sort"}, "DFH": {"subsystem": "CICS", "meaning": "CICS"}, "DSN": {"subsystem": "DB2", "meaning": "Db2 for z/OS"}, "DFS": {"subsystem": "IMS", "meaning": "IMS"}, "BPX": {"subsystem": "USS", "meaning": "UNIX System Services"}, } ABEND_CATALOG: Dict[str, Dict[str, Any]] = { "S0C1": {"severity": "CRITICAL", "area": "Program", "title": "Operation exception", "recommendation": "Check load module, entry point, compiler/link-edit mismatch, and invalid instruction path."}, "S0C4": {"severity": "CRITICAL", "area": "Memory", "title": "Protection exception", "recommendation": "Check invalid address references, linkage section, table bounds, and dump registers."}, "S0C7": {"severity": "CRITICAL", "area": "Data quality", "title": "Data exception", "recommendation": "Validate packed/zoned decimal fields, input layout, copybook alignment, and upstream feed changes."}, "S0CB": {"severity": "CRITICAL", "area": "Numeric computation", "title": "Decimal divide exception", "recommendation": "Check divide-by-zero, decimal scaling, signs, and numeric field definitions."}, "S013": {"severity": "ERROR", "area": "Dataset open", "title": "Open failure", "recommendation": "Check DD statement, DCB attributes, RECFM, LRECL, BLKSIZE, dataset existence, and catalog entry."}, "S222": {"severity": "ERROR", "area": "Operations", "title": "Operator or system cancel", "recommendation": "Review cancellation source, job duration, console messages, and scheduler actions."}, "S322": {"severity": "ERROR", "area": "Runtime", "title": "CPU time limit exceeded", "recommendation": "Check TIME parameter, possible loop, data growth, sort/DB2 regression, and scheduler limits."}, "S522": {"severity": "ERROR", "area": "Runtime wait", "title": "Wait time exceeded", "recommendation": "Check enqueue contention, operator wait, DB2/CICS wait, and external dependencies."}, "S806": {"severity": "CRITICAL", "area": "Program library", "title": "Load module not found", "recommendation": "Check EXEC PGM, STEPLIB, JOBLIB, LNKLST, package deployment, and APF requirements."}, "S878": {"severity": "CRITICAL", "area": "Storage", "title": "Insufficient virtual storage", "recommendation": "Check REGION, LE options, memory growth, sort memory, and address space limits."}, "S80A": {"severity": "CRITICAL", "area": "Storage", "title": "Storage request could not be satisfied", "recommendation": "Check region size, subpool usage, memory pressure, and recent workload growth."}, "S913": {"severity": "CRITICAL", "area": "Security", "title": "Security violation", "recommendation": "Check RACF profile, user/group authority, dataset access list, and security class."}, "SB37": {"severity": "CRITICAL", "area": "Dataset space", "title": "Dataset ran out of space", "recommendation": "Check SPACE primary/secondary, volume free space, SMS class, extents, and expected data growth."}, "SD37": {"severity": "ERROR", "area": "Dataset space", "title": "No secondary allocation", "recommendation": "Add or increase secondary allocation and review dataset allocation standards."}, "SE37": {"severity": "ERROR", "area": "Dataset space", "title": "Extent or volume limit reached", "recommendation": "Review multi-volume allocation, SMS policy, extent limits, and dataset growth."}, "U4038": {"severity": "CRITICAL", "area": "Language Environment", "title": "Language Environment user abend", "recommendation": "Review CEE messages and the preceding application error; U4038 is often a wrapper around the real cause."}, "U0999": {"severity": "ERROR", "area": "Application", "title": "Application user abend", "recommendation": "Check application messages, return code policy, and site-specific abend catalog."}, } RULES: List[Dict[str, Any]] = [ { "id": "ABEND_DETECTED", "pattern": r"\b(?:ABEND=|SYSTEM=|ABEND CODE |COMPLETION CODE - SYSTEM=)?(S[0-9A-F]{3}|S[BE]37|U\d{4})\b", "severity": "CRITICAL", "category": "ABEND", "title": "ABEND detected", "recommendation": "Decode the ABEND and inspect messages immediately before the failure.", }, { "id": "SPACE_ABEND", "pattern": r"\b(SB37|SD37|SE37|B37|D37|E37)\b|DATA SET.*(?:SPACE|FULL|EXTENT)|NO SPACE|OUT OF SPACE|INSUFFICIENT SPACE", "severity": "CRITICAL", "category": "DATASET_SPACE", "title": "Dataset space problem", "recommendation": "Check SPACE allocation, volume free space, SMS storage class, secondary allocation, and dataset growth.", }, { "id": "SECURITY_DENIAL", "pattern": r"\b(S913|ICH\d+I|IRR\d+I|RACF|NOT AUTHORIZED|ACCESS DENIED|VIOLATION)\b", "severity": "CRITICAL", "category": "SECURITY", "title": "Security or RACF denial", "recommendation": "Check user authority, RACF dataset profile, access list, and security class.", }, { "id": "CATALOG_ERROR", "pattern": r"\b(CATALOG|NOT CATALOGED|CATLG|UNCATALOG|IDC\d+I).*\b(ERROR|FAILED|NOT FOUND|DUPLICATE)?", "severity": "ERROR", "category": "CATALOG", "title": "Catalog or IDCAMS issue", "recommendation": "Check catalog entry, dataset naming, IDCAMS output, aliases, and GDG base/generation status.", }, { "id": "ALLOCATION_ERROR", "pattern": r"\b(IGD\d+I|IEF\d+I).*\b(ALLOCATION|ALLOCATE|VOLUME|STORCLAS|DATACLAS|MGMTCLAS|UNIT|FAILED)\b|ALLOCATION FAILED", "severity": "ERROR", "category": "ALLOCATION", "title": "Allocation or SMS problem", "recommendation": "Review DD statement, SMS classes, storage group availability, unit, volume, and allocation policy.", }, { "id": "LOAD_MODULE_ERROR", "pattern": r"\b(S806|CSV\d+I|MODULE .*NOT FOUND|LOAD MODULE|STEPLIB|JOBLIB|LNKLST|PROGRAM .*NOT FOUND)\b", "severity": "CRITICAL", "category": "LOAD_MODULE", "title": "Load module or library problem", "recommendation": "Check EXEC PGM, STEPLIB/JOBLIB, LNKLST, deployment library, APF status, and module naming.", }, { "id": "STORAGE_SHORTAGE", "pattern": r"\b(S878|S80A|IRA\d+I|VIRTUAL STORAGE|REGION|GETMAIN|STORAGE SHORTAGE|INSUFFICIENT STORAGE)\b", "severity": "CRITICAL", "category": "STORAGE", "title": "Virtual storage shortage", "recommendation": "Review REGION, LE options, memory growth, sort memory, and recent workload changes.", }, { "id": "CPU_TIME_LIMIT", "pattern": r"\b(S322|TIME LIMIT|CPU TIME|EXCEEDED TIME|TIME=)\b", "severity": "ERROR", "category": "TIME_LIMIT", "title": "CPU time or execution time limit exceeded", "recommendation": "Check loops, input growth, DB2 access path regressions, sort size, and scheduler TIME limits.", }, { "id": "CICS_PROBLEM", "pattern": r"\b(DFH\d+|CICS|TRANSACTION|TASK ABEND|AICA|ASRA|AEI0|AEY9)\b", "severity": "ERROR", "category": "CICS", "title": "CICS-related incident signal", "recommendation": "Check transaction abend, CICS region health, program control, file control, DB2 waits, and terminal/user context.", }, { "id": "DB2_PROBLEM", "pattern": r"\b(DSN\d+|DB2|SQLCODE\s*[=-]\s*-?\d+|DEADLOCK|TIMEOUT|LOCK ESCALATION)\b", "severity": "ERROR", "category": "DB2", "title": "Db2-related incident signal", "recommendation": "Check SQLCODE, package, plan, locks, deadlocks, timeout, access path, and DB2 subsystem health.", }, { "id": "IMS_PROBLEM", "pattern": r"\b(DFS\d+|IMS|BMP|MPP|DL/I|DL1|DATABASE UNAVAILABLE)\b", "severity": "ERROR", "category": "IMS", "title": "IMS-related incident signal", "recommendation": "Check IMS region, PSB/PCB, database availability, transaction queue, and dependent region messages.", }, { "id": "SYSPLEX_PROBLEM", "pattern": r"\b(IXC\d+|IXL\d+|XCF|SYSPLEX|COUPLING FACILITY|STRUCTURE|CF\s+STRUCTURE|CONNECTIVITY)\b", "severity": "CRITICAL", "category": "SYSPLEX", "title": "Sysplex or coupling facility problem", "recommendation": "Check XCF, coupling facility structures, CFRM policy, connectivity, and impacted members.", }, { "id": "LOGGER_PROBLEM", "pattern": r"\b(IXG\d+|SYSTEM LOGGER|LOGSTREAM|LOGGER|OFFLOAD|STAGING DATA SET)\b", "severity": "ERROR", "category": "LOGGER", "title": "System logger issue", "recommendation": "Check log stream status, offload, staging datasets, coupling facility structures, and logger address space.", }, { "id": "IO_CHANNEL_PROBLEM", "pattern": r"\b(IOS\d+|I/O ERROR|CHANNEL|FICON|DEVICE END|UNIT CHECK|PATH|DASD|CUU|DEVICE\s+[0-9A-F]{3,5})\b", "severity": "ERROR", "category": "IO", "title": "I/O or channel path problem", "recommendation": "Check device pathing, channel status, storage subsystem alerts, volume health, and RMF/CMF DASD reports.", }, { "id": "JES_SPOOL_PROBLEM", "pattern": r"\b(\$HASP\d+|SPOOL|JES2|JES3|INITIATOR|OUTPUT QUEUE|HELD OUTPUT)\b.*\b(ERROR|FULL|SHORTAGE|FAILED|DELAY|HELD|DRAINED)?", "severity": "WARNING", "category": "JES", "title": "JES or spool signal", "recommendation": "Check JES initiators, spool utilization, output queues, held jobs, and job class constraints.", }, { "id": "OPERATOR_ACTION_REQUIRED", "pattern": r"\b(IEE\d+|WTOR|REPLY|RESPONSE REQUIRED|ACTION REQUIRED|WAITING FOR OPERATOR)\b", "severity": "WARNING", "category": "OPERATOR", "title": "Operator action required or console intervention", "recommendation": "Review outstanding replies, automation rules, console commands, and operational procedures.", }, { "id": "HIGH_RC", "pattern": r"\b(?:MAXCC|LASTCC|COND CODE|RC)\s*[=: ]\s*(0?0?0?[89]|0?0?1[0-9]|[2-9][0-9])\b", "severity": "ERROR", "category": "RETURN_CODE", "title": "High return code detected", "recommendation": "Review the failing step, application messages, utility output, and job condition handling.", }, ] PROFILE_RULE_BOOSTS: Dict[str, Dict[str, int]] = { "production": {}, "batch": {"DATASET_SPACE": 10, "ALLOCATION": 6, "RETURN_CODE": 4, "TIME_LIMIT": 4}, "security": {"SECURITY": 14}, "cics": {"CICS": 12, "DB2": 5}, "db2": {"DB2": 14, "CICS": 3}, "ims": {"IMS": 14}, "sysplex": {"SYSPLEX": 16, "LOGGER": 10, "IO": 4}, "storage": {"IO": 14, "DATASET_SPACE": 8, "STORAGE": 8, "ALLOCATION": 5}, "training": {"RETURN_CODE": -4, "JES": -3}, } DATASET_PATTERN = re.compile(r"\b(?:DSN=)?([A-Z0-9#$@][A-Z0-9#$@.-]{2,}(?:\.[A-Z0-9#$@][A-Z0-9#$@-]{0,}){1,})\b", re.IGNORECASE) ABEND_PATTERN = re.compile(r"\b(S[0-9A-F]{3}|S[BE]37|U(?!0000)\d{4})\b", re.IGNORECASE) RC_PATTERN = re.compile(r"\b(?:MAXCC|LASTCC|COND CODE|RC)\s*[=: ]\s*(\d{1,4})\b", re.IGNORECASE) JOB_NAME_PATTERN = re.compile(r"\b(?:JOB|JOBNAME=|\$HASP373\s+|\$HASP395\s+)([A-Z0-9#$@]{1,8})\b", re.IGNORECASE) JOB_STARTED_PATTERN = re.compile(r"\b\$HASP373\s+([A-Z0-9#$@]{1,8})\s+STARTED\b|\bIEF403I\s+([A-Z0-9#$@]{1,8})\s+-\s+STARTED\b", re.IGNORECASE) JOB_ENDED_PATTERN = re.compile(r"\b\$HASP395\s+([A-Z0-9#$@]{1,8})\s+ENDED\b|\bIEF404I\s+([A-Z0-9#$@]{1,8})\s+-\s+ENDED\b", re.IGNORECASE) STEP_PATTERN = re.compile(r"\b(IEF142I|IEF450I|IEF472I)\s+([A-Z0-9#$@]{1,8})\s+([A-Z0-9#$@]{1,8})\b", re.IGNORECASE) MESSAGE_ID_PATTERN = re.compile(r"\b(\$?HASP\d{3,4}|[A-Z]{3,4}\d{3,5}[A-Z]?)\b") def parse_dt_from_line(line: str, fallback_date: Optional[dt.date]) -> Tuple[Optional[dt.datetime], Optional[str]]: stripped = line.strip() patterns = [ (re.compile(r"^(\d{4}-\d{2}-\d{2})[ T](\d{2}:\d{2}:\d{2})(?:[.,]\d+)?"), "%Y-%m-%d %H:%M:%S"), (re.compile(r"^(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2}:\d{2})(?:[.,]\d+)?"), "%m/%d/%Y %H:%M:%S"), (re.compile(r"^(\d{4}\.\d{3})\s+(\d{2}:\d{2}:\d{2})(?:[.,]\d+)?"), "julian"), (re.compile(r"^(\d{2}:\d{2}:\d{2})(?:[.,]\d+)?"), "time_only"), ] for regex, fmt in patterns: match = regex.search(stripped) if not match: continue if fmt == "julian": year, day = match.group(1).split(".") base = dt.datetime(int(year), 1, 1) + dt.timedelta(days=int(day) - 1) h, m, s = map(int, match.group(2).split(":")) return base.replace(hour=h, minute=m, second=s), match.group(0) if fmt == "time_only": h, m, s = map(int, match.group(1).split(":")) date_part = fallback_date or dt.date(2000, 1, 1) return dt.datetime.combine(date_part, dt.time(h, m, s)), match.group(0) value = f"{match.group(1)} {match.group(2)}" try: return dt.datetime.strptime(value, fmt), match.group(0) except ValueError: continue return None, None def safe_float(value: Any, default: float = 0.0) -> float: try: if value is None or value == "": return default return float(value) except Exception: return default def normalize_code(code: str) -> str: code = code.strip().upper() if re.fullmatch(r"B37|D37|E37", code): return "S" + code return code def detect_family(line: str) -> Tuple[str, str, str]: match = MESSAGE_ID_PATTERN.search(line.upper()) if not match: return "UNKNOWN", "UNKNOWN", "Unclassified" msg_id = match.group(1).upper() if msg_id.startswith("$HASP"): prefix = "$HASP" else: prefix = msg_id[:3] for known in sorted(MESSAGE_FAMILIES.keys(), key=len, reverse=True): if known != "$HASP" and msg_id.startswith(known): prefix = known break meta = MESSAGE_FAMILIES.get(prefix, {"subsystem": "UNKNOWN", "meaning": "Unclassified"}) return prefix, meta["subsystem"], msg_id def redact_text(text: str) -> str: def repl(match: re.Match[str]) -> str: value = match.group(1) if "." not in value: return value return "DATASET.REDACTED" redacted = DATASET_PATTERN.sub(lambda m: repl(m), text) redacted = re.sub(r"\bUSER(?:ID)?[= ]+[A-Z0-9#$@]{2,8}\b", "USER=REDACTED", redacted, flags=re.IGNORECASE) return redacted @dataclass class LogEvent: line_no: int timestamp: Optional[str] raw_timestamp: Optional[str] family: str subsystem: str message_id: str severity_hint: str job_name: Optional[str] step_name: Optional[str] event_type: str message: str raw_line: str abend_codes: List[str] = field(default_factory=list) return_codes: List[int] = field(default_factory=list) datasets: List[str] = field(default_factory=list) @dataclass class Issue: issue_id: str severity: str category: str title: str line_no: int timestamp: Optional[str] family: str subsystem: str job_name: Optional[str] step_name: Optional[str] abend_codes: List[str] return_codes: List[int] datasets: List[str] recommendation: str evidence: str context_before: List[str] = field(default_factory=list) context_after: List[str] = field(default_factory=list) confidence: str = "MEDIUM" @dataclass class JobSummary: job_name: str started: bool = False ended: bool = False start_line: Optional[int] = None end_line: Optional[int] = None max_rc: int = 0 abends: List[str] = field(default_factory=list) steps: List[str] = field(default_factory=list) issue_count: int = 0 highest_severity: str = "INFO" @dataclass class IncidentCluster: cluster_id: str start_time: Optional[str] end_time: Optional[str] start_line: int end_line: int highest_severity: str risk_score: int categories: List[str] jobs: List[str] subsystems: List[str] abends: List[str] title: str recommendation: str evidence_lines: List[str] @dataclass class AnalysisResult: source_name: str profile: str status: str highest_severity: str risk_score: int line_count: int event_count: int issue_count: int cluster_count: int job_count: int dataset_count: int abends: List[str] max_rc: int started_at: Optional[str] ended_at: Optional[str] family_counts: Dict[str, int] subsystem_counts: Dict[str, int] category_counts: Dict[str, int] events: List[LogEvent] issues: List[Issue] jobs: List[JobSummary] clusters: List[IncidentCluster] datasets: List[str] root_causes: List[Dict[str, Any]] class SyslogOperlogAnalyzer: def __init__( self, profile: str = "production", window_minutes: int = 5, context_lines: int = 3, redact: bool = False, since: Optional[str] = None, until: Optional[str] = None, ) -> None: self.profile = profile self.window_minutes = window_minutes self.context_lines = context_lines self.redact = redact self.since = self._parse_cli_dt(since) if since else None self.until = self._parse_cli_dt(until) if until else None def _parse_cli_dt(self, value: str) -> Optional[dt.datetime]: for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%H:%M:%S"): try: parsed = dt.datetime.strptime(value, fmt) if fmt == "%H:%M:%S": return dt.datetime.combine(dt.date(2000, 1, 1), parsed.time()) return parsed except ValueError: continue raise ValueError(f"Invalid date/time: {value}") def analyze_text(self, text: str, source_name: str = "input") -> AnalysisResult: lines = text.splitlines() events: List[LogEvent] = [] issues: List[Issue] = [] jobs_by_name: Dict[str, JobSummary] = {} fallback_date: Optional[dt.date] = None parsed_datetimes: Dict[int, Optional[dt.datetime]] = {} for idx, original_line in enumerate(lines, start=1): working_line = redact_text(original_line) if self.redact else original_line timestamp, raw_ts = parse_dt_from_line(working_line, fallback_date) if timestamp and timestamp.year != 2000: fallback_date = timestamp.date() parsed_datetimes[idx] = timestamp if not self._within_time_filter(timestamp): continue event = self._parse_event(idx, working_line, timestamp, raw_ts) events.append(event) self._update_jobs(event, jobs_by_name) event_issues = self._match_rules(event, lines, parsed_datetimes) issues.extend(event_issues) for issue in issues: if issue.job_name: job = jobs_by_name.setdefault(issue.job_name, JobSummary(job_name=issue.job_name)) job.issue_count += 1 job.highest_severity = max([job.highest_severity, issue.severity], key=lambda s: SEVERITY_ORDER[s]) for code in issue.abend_codes: if code not in job.abends: job.abends.append(code) clusters = self._build_clusters(issues) root_causes = self._build_root_causes(issues, clusters) family_counts = Counter(e.family for e in events) subsystem_counts = Counter(e.subsystem for e in events) category_counts = Counter(i.category for i in issues) all_datasets = sorted({d for e in events for d in e.datasets} | {d for i in issues for d in i.datasets}) all_abends = sorted({c for e in events for c in e.abend_codes} | {c for i in issues for c in i.abend_codes}) all_rc = [rc for e in events for rc in e.return_codes] max_rc = max(all_rc or [0]) highest_severity = self._highest_severity([i.severity for i in issues]) risk_score = self._compute_risk_score(issues, clusters, max_rc) status = self._compute_status(highest_severity, all_abends, max_rc) timestamps = [e.timestamp for e in events if e.timestamp] return AnalysisResult( source_name=source_name, profile=self.profile, status=status, highest_severity=highest_severity, risk_score=risk_score, line_count=len(lines), event_count=len(events), issue_count=len(issues), cluster_count=len(clusters), job_count=len(jobs_by_name), dataset_count=len(all_datasets), abends=all_abends, max_rc=max_rc, started_at=min(timestamps) if timestamps else None, ended_at=max(timestamps) if timestamps else None, family_counts=dict(sorted(family_counts.items())), subsystem_counts=dict(sorted(subsystem_counts.items())), category_counts=dict(sorted(category_counts.items())), events=events, issues=issues, jobs=sorted(jobs_by_name.values(), key=lambda j: j.job_name), clusters=clusters, datasets=all_datasets, root_causes=root_causes, ) def _within_time_filter(self, timestamp: Optional[dt.datetime]) -> bool: if not timestamp: return True if self.since and timestamp < self.since: return False if self.until and timestamp > self.until: return False return True def _parse_event(self, line_no: int, line: str, timestamp: Optional[dt.datetime], raw_ts: Optional[str]) -> LogEvent: upper = line.upper() family, subsystem, message_id = detect_family(upper) abends = sorted({normalize_code(x) for x in ABEND_PATTERN.findall(upper)}) rcs = [int(x) for x in RC_PATTERN.findall(upper)] datasets = sorted({m.group(1).upper() for m in DATASET_PATTERN.finditer(upper) if self._looks_like_dataset(m.group(1))}) job_name = self._extract_job_name(upper) step_name = self._extract_step_name(upper) event_type = self._event_type(upper, abends, rcs) severity_hint = self._severity_hint(upper, abends, rcs) return LogEvent( line_no=line_no, timestamp=timestamp.isoformat(sep=" ") if timestamp else None, raw_timestamp=raw_ts, family=family, subsystem=subsystem, message_id=message_id, severity_hint=severity_hint, job_name=job_name, step_name=step_name, event_type=event_type, message=line.strip(), raw_line=line, abend_codes=abends, return_codes=rcs, datasets=datasets, ) def _looks_like_dataset(self, value: str) -> bool: if "." not in value: return False if re.search(r"\b(COMPLETION|REASON|SYSTEM|STARTED|ENDED)\b", value, re.IGNORECASE): return False return True def _extract_job_name(self, upper: str) -> Optional[str]: for regex in (JOB_STARTED_PATTERN, JOB_ENDED_PATTERN): m = regex.search(upper) if m: return next((g for g in m.groups() if g), None) m = STEP_PATTERN.search(upper) if m: return m.group(2) m = JOB_NAME_PATTERN.search(upper) if m: name = m.group(1) if name not in {"JOB", "STARTED", "ENDED"}: return name return None def _extract_step_name(self, upper: str) -> Optional[str]: m = STEP_PATTERN.search(upper) if m: return m.group(3) m = re.search(r"\bSTEP(?:NAME)?[= ]+([A-Z0-9#$@]{1,8})\b", upper) if m: return m.group(1) return None def _event_type(self, upper: str, abends: Sequence[str], rcs: Sequence[int]) -> str: if "STARTED" in upper and ("$HASP373" in upper or "IEF403I" in upper): return "JOB_START" if "ENDED" in upper and ("$HASP395" in upper or "IEF404I" in upper): return "JOB_END" if abends: return "ABEND" if rcs: return "RETURN_CODE" if "WTOR" in upper or "REPLY" in upper: return "OPERATOR_ACTION" if "RACF" in upper or "NOT AUTHORIZED" in upper: return "SECURITY" return "MESSAGE" def _severity_hint(self, upper: str, abends: Sequence[str], rcs: Sequence[int]) -> str: if abends: return "CRITICAL" if any(rc >= 12 for rc in rcs): return "CRITICAL" if any(rc >= 8 for rc in rcs): return "ERROR" if any(rc >= 4 for rc in rcs): return "WARNING" if re.search(r"\b(ERROR|FAILED|FAILURE|SEVERE|ABNORMAL)\b", upper): return "ERROR" if re.search(r"\b(WARNING|WARN|DELAY|HELD)\b", upper): return "WARNING" return "INFO" def _update_jobs(self, event: LogEvent, jobs_by_name: Dict[str, JobSummary]) -> None: if not event.job_name: return job = jobs_by_name.setdefault(event.job_name, JobSummary(job_name=event.job_name)) if event.event_type == "JOB_START": job.started = True job.start_line = job.start_line or event.line_no if event.event_type == "JOB_END": job.ended = True job.end_line = event.line_no if event.step_name and event.step_name not in job.steps: job.steps.append(event.step_name) if event.return_codes: job.max_rc = max(job.max_rc, max(event.return_codes)) for code in event.abend_codes: if code not in job.abends: job.abends.append(code) job.highest_severity = max([job.highest_severity, event.severity_hint], key=lambda s: SEVERITY_ORDER[s]) def _match_rules(self, event: LogEvent, original_lines: Sequence[str], parsed_datetimes: Dict[int, Optional[dt.datetime]]) -> List[Issue]: issues: List[Issue] = [] upper = event.raw_line.upper() for rule in RULES: if not re.search(rule["pattern"], upper, flags=re.IGNORECASE): continue severity = self._apply_profile(rule["severity"], rule["category"]) confidence = self._confidence(rule["id"], event) issue = Issue( issue_id=rule["id"], severity=severity, category=rule["category"], title=self._enhanced_title(rule, event), line_no=event.line_no, timestamp=event.timestamp, family=event.family, subsystem=event.subsystem, job_name=event.job_name, step_name=event.step_name, abend_codes=event.abend_codes, return_codes=event.return_codes, datasets=event.datasets, recommendation=self._enhanced_recommendation(rule, event), evidence=event.message, context_before=self._context(original_lines, event.line_no, before=True), context_after=self._context(original_lines, event.line_no, before=False), confidence=confidence, ) issues.append(issue) return self._dedupe_issues(issues) def _apply_profile(self, severity: str, category: str) -> str: boost = PROFILE_RULE_BOOSTS.get(self.profile, {}).get(category, 0) value = SEVERITY_ORDER[severity] if boost >= 10: value += 1 elif boost <= -4: value -= 1 value = max(0, min(3, value)) for name, order in SEVERITY_ORDER.items(): if order == value: return name return severity def _enhanced_title(self, rule: Dict[str, Any], event: LogEvent) -> str: if event.abend_codes: primary = event.abend_codes[0] meta = ABEND_CATALOG.get(primary) if meta: return f"{primary} - {meta['title']}" return rule["title"] def _enhanced_recommendation(self, rule: Dict[str, Any], event: LogEvent) -> str: if event.abend_codes: primary = event.abend_codes[0] meta = ABEND_CATALOG.get(primary) if meta: return str(meta["recommendation"]) return str(rule["recommendation"]) def _confidence(self, rule_id: str, event: LogEvent) -> str: if rule_id == "ABEND_DETECTED" and event.abend_codes: return "HIGH" if event.message_id != "Unclassified" and event.family != "UNKNOWN": return "HIGH" if event.datasets or event.return_codes: return "MEDIUM" return "LOW" def _context(self, original_lines: Sequence[str], line_no: int, before: bool) -> List[str]: if self.context_lines <= 0: return [] zero = line_no - 1 if before: start = max(0, zero - self.context_lines) end = zero else: start = zero + 1 end = min(len(original_lines), zero + 1 + self.context_lines) selected = [line.rstrip("\n") for line in original_lines[start:end]] if self.redact: selected = [redact_text(x) for x in selected] return selected def _dedupe_issues(self, issues: List[Issue]) -> List[Issue]: seen = set() out = [] for issue in issues: key = (issue.issue_id, issue.line_no, tuple(issue.abend_codes), tuple(issue.return_codes)) if key in seen: continue seen.add(key) out.append(issue) return out def _build_clusters(self, issues: List[Issue]) -> List[IncidentCluster]: critical = [i for i in issues if SEVERITY_ORDER[i.severity] >= SEVERITY_ORDER["ERROR"]] if not critical: return [] sorted_issues = sorted(critical, key=lambda i: (i.timestamp or "", i.line_no)) groups: List[List[Issue]] = [] current: List[Issue] = [] current_end: Optional[dt.datetime] = None for issue in sorted_issues: ts = self._parse_event_timestamp(issue.timestamp) if not current: current = [issue] current_end = ts continue if ts and current_end and (ts - current_end) <= dt.timedelta(minutes=self.window_minutes): current.append(issue) current_end = max(current_end, ts) elif ts is None or current_end is None: if issue.line_no - current[-1].line_no <= 5: current.append(issue) else: groups.append(current) current = [issue] current_end = ts else: groups.append(current) current = [issue] current_end = ts if current: groups.append(current) clusters: List[IncidentCluster] = [] for idx, group in enumerate(groups, start=1): highest = self._highest_severity([i.severity for i in group]) categories = sorted(set(i.category for i in group)) jobs = sorted({i.job_name for i in group if i.job_name}) subsystems = sorted(set(i.subsystem for i in group)) abends = sorted({a for i in group for a in i.abend_codes}) score = min(100, sum(SEVERITY_WEIGHT[i.severity] for i in group) + len(categories) * 3 + len(abends) * 8) evidence = [f"L{i.line_no}: {i.evidence}" for i in group[:8]] title, recommendation = self._cluster_summary(group) clusters.append( IncidentCluster( cluster_id=f"INC-{idx:03d}", start_time=min([i.timestamp for i in group if i.timestamp], default=None), end_time=max([i.timestamp for i in group if i.timestamp], default=None), start_line=min(i.line_no for i in group), end_line=max(i.line_no for i in group), highest_severity=highest, risk_score=score, categories=categories, jobs=jobs, subsystems=subsystems, abends=abends, title=title, recommendation=recommendation, evidence_lines=evidence, ) ) return clusters def _parse_event_timestamp(self, value: Optional[str]) -> Optional[dt.datetime]: if not value: return None try: return dt.datetime.fromisoformat(value) except ValueError: return None def _cluster_summary(self, group: List[Issue]) -> Tuple[str, str]: categories = Counter(i.category for i in group) if any(i.category == "SECURITY" for i in group): return "Security-driven incident", "Start with RACF/access authorization checks, then verify affected job and dataset context." if any(i.category == "DATASET_SPACE" for i in group): return "Dataset space or allocation incident", "Check SPACE parameters, SMS/storage group, volume free space, and dataset growth before rerun." if any(i.category == "SYSPLEX" for i in group): return "Sysplex or coupling facility incident", "Check XCF/CF messages, impacted members, structures, CFRM policy, and logger dependencies." if any(i.category == "DB2" for i in group): return "Db2-related incident", "Check SQLCODEs, lock/deadlock messages, plans/packages, and subsystem availability." if any(i.category == "CICS" for i in group): return "CICS-related incident", "Check transaction abends, region messages, file control, DB2 wait, and user/terminal context." if any(i.category == "IO" for i in group): return "I/O or storage path incident", "Check device/path messages and correlate with RMF/CMF DASD and storage subsystem alerts." if any(i.abend_codes for i in group): code = next(iter([a for i in group for a in i.abend_codes]), "ABEND") meta = ABEND_CATALOG.get(code, {}) return f"{code} incident", str(meta.get("recommendation", "Decode the ABEND and inspect preceding messages.")) most_common = categories.most_common(1)[0][0] if categories else "General" return f"{most_common} incident", "Inspect clustered evidence lines and correlate with job, subsystem, and dataset context." def _build_root_causes(self, issues: List[Issue], clusters: List[IncidentCluster]) -> List[Dict[str, Any]]: root_causes: List[Dict[str, Any]] = [] for cluster in clusters[:10]: root_causes.append( { "cluster_id": cluster.cluster_id, "title": cluster.title, "severity": cluster.highest_severity, "risk_score": cluster.risk_score, "jobs": cluster.jobs, "subsystems": cluster.subsystems, "categories": cluster.categories, "abends": cluster.abends, "recommendation": cluster.recommendation, "evidence": cluster.evidence_lines, } ) if not root_causes and issues: top = sorted(issues, key=lambda i: (SEVERITY_ORDER[i.severity], i.line_no), reverse=True)[0] root_causes.append( { "cluster_id": None, "title": top.title, "severity": top.severity, "risk_score": SEVERITY_WEIGHT[top.severity], "jobs": [top.job_name] if top.job_name else [], "subsystems": [top.subsystem], "categories": [top.category], "abends": top.abend_codes, "recommendation": top.recommendation, "evidence": [f"L{top.line_no}: {top.evidence}"], } ) return root_causes def _highest_severity(self, severities: Sequence[str]) -> str: if not severities: return "INFO" return max(severities, key=lambda s: SEVERITY_ORDER.get(s, 0)) def _compute_risk_score(self, issues: List[Issue], clusters: List[IncidentCluster], max_rc: int) -> int: score = 0 for issue in issues: score += SEVERITY_WEIGHT.get(issue.severity, 0) boost = PROFILE_RULE_BOOSTS.get(self.profile, {}).get(issue.category, 0) score += max(0, boost) score += len(clusters) * 8 if max_rc >= 16: score += 20 elif max_rc >= 12: score += 14 elif max_rc >= 8: score += 8 return max(0, min(100, score)) def _compute_status(self, highest: str, abends: List[str], max_rc: int) -> str: if abends or highest == "CRITICAL" or max_rc >= 12: return "FAILED" if highest == "ERROR" or max_rc >= 8: return "ERROR" if highest == "WARNING" or max_rc >= 4: return "WARNING" return "OK" def result_to_jsonable(result: AnalysisResult) -> Dict[str, Any]: return asdict(result) def write_json(result: AnalysisResult, path: str) -> None: with open(path, "w", encoding="utf-8") as fh: json.dump(result_to_jsonable(result), fh, indent=2, ensure_ascii=False) def write_csv(path: str, rows: Iterable[Dict[str, Any]], fields: Sequence[str]) -> None: with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=list(fields), extrasaction="ignore") writer.writeheader() for row in rows: clean = {} for key, value in row.items(): if isinstance(value, (list, dict)): clean[key] = json.dumps(value, ensure_ascii=False) else: clean[key] = value writer.writerow(clean) def write_events_csv(result: AnalysisResult, path: str) -> None: fields = ["line_no", "timestamp", "family", "subsystem", "message_id", "severity_hint", "job_name", "step_name", "event_type", "abend_codes", "return_codes", "datasets", "message"] write_csv(path, (asdict(e) for e in result.events), fields) def write_issues_csv(result: AnalysisResult, path: str) -> None: fields = ["issue_id", "severity", "category", "title", "line_no", "timestamp", "family", "subsystem", "job_name", "step_name", "abend_codes", "return_codes", "datasets", "confidence", "recommendation", "evidence"] write_csv(path, (asdict(i) for i in result.issues), fields) def write_jobs_csv(result: AnalysisResult, path: str) -> None: fields = ["job_name", "started", "ended", "start_line", "end_line", "max_rc", "abends", "steps", "issue_count", "highest_severity"] write_csv(path, (asdict(j) for j in result.jobs), fields) def write_clusters_csv(result: AnalysisResult, path: str) -> None: fields = ["cluster_id", "start_time", "end_time", "start_line", "end_line", "highest_severity", "risk_score", "categories", "jobs", "subsystems", "abends", "title", "recommendation", "evidence_lines"] write_csv(path, (asdict(c) for c in result.clusters), fields) def write_subsystems_csv(result: AnalysisResult, path: str) -> None: counts_by_subsystem = Counter(i.subsystem for i in result.issues) critical_by_subsystem = Counter(i.subsystem for i in result.issues if i.severity == "CRITICAL") rows = [] for subsystem, event_count in result.subsystem_counts.items(): rows.append( { "subsystem": subsystem, "event_count": event_count, "issue_count": counts_by_subsystem.get(subsystem, 0), "critical_count": critical_by_subsystem.get(subsystem, 0), } ) write_csv(path, rows, ["subsystem", "event_count", "issue_count", "critical_count"]) def severity_badge(severity: str) -> str: cls = { "INFO": "info", "WARNING": "warning", "ERROR": "error", "CRITICAL": "critical", }.get(severity, "info") return f'{html.escape(severity)}' def write_html(result: AnalysisResult, path: str) -> None: css = """ body { font-family: Segoe UI, Arial, sans-serif; margin: 0; background: #f4f7fb; color: #172033; } .page { max-width: 1220px; margin: 28px auto; background: #fff; border-radius: 14px; box-shadow: 0 10px 30px rgba(0,0,0,.08); overflow: hidden; } .hero { background: linear-gradient(135deg,#102d42,#0a6b83); color: #fff; padding: 28px 34px; } h1 { margin: 0 0 6px; font-size: 26px; } .sub { opacity: .88; } .content { padding: 26px 34px 40px; } .kpis { display: grid; grid-template-columns: repeat(5, minmax(0,1fr)); gap: 12px; margin: 20px 0 26px; } .kpi { border: 1px solid #dce3ea; border-radius: 12px; padding: 14px; background: #fbfdff; } .kpi .v { font-size: 22px; font-weight: 800; color: #0077b6; } .kpi .l { color: #596579; font-size: 13px; margin-top: 3px; } table { width: 100%; border-collapse: collapse; margin: 16px 0 28px; font-size: 13px; } th { background: #263849; color: #fff; text-align: left; padding: 9px; border: 1px solid #1e2b38; } td { border: 1px solid #dce3ea; padding: 9px; vertical-align: top; } tr:nth-child(even) td { background: #f8fafc; } .badge { display: inline-block; border-radius: 99px; padding: 3px 8px; font-size: 12px; font-weight: 700; } .critical { background:#ffe7e7; color:#8b0000; border:1px solid #ffb2b2; } .error { background:#fff0e1; color:#9a4a00; border:1px solid #ffd2a1; } .warning { background:#fff9db; color:#7a6000; border:1px solid #f2df85; } .info { background:#e9f7ff; color:#005a78; border:1px solid #bdeaff; } .section-title { border-left: 5px solid #00b4d8; padding-left: 12px; color: #2c3e50; margin-top: 30px; } .root { border:1px solid #dce3ea; border-left:5px solid #00b4d8; border-radius:10px; padding:14px; margin:12px 0; background:#fbfdff; } pre { background:#20242b; color:#d6deeb; border-radius:10px; padding:12px; overflow:auto; font-size:12px; } @media (max-width: 900px) { .kpis { grid-template-columns: repeat(2, minmax(0,1fr)); } .content { padding: 20px; } } """ def h(value: Any) -> str: return html.escape("" if value is None else str(value)) roots = [] for root in result.root_causes: evidence = "\n".join(root.get("evidence", [])) roots.append( f"""
{h(root.get('title'))} {severity_badge(root.get('severity', 'INFO'))}

Risk: {h(root.get('risk_score'))}/100   Jobs: {h(', '.join(root.get('jobs', [])) or '-')}   Subsystems: {h(', '.join(root.get('subsystems', [])) or '-')}

Recommendation: {h(root.get('recommendation'))}

{h(evidence)}
""" ) cluster_rows = [] for c in result.clusters[:20]: cluster_rows.append( f"{h(c.cluster_id)}{h(c.start_time)}{h(c.end_time)}{severity_badge(c.highest_severity)}{h(c.risk_score)}{h(c.title)}{h(', '.join(c.jobs))}{h(', '.join(c.categories))}{h(c.recommendation)}" ) issue_rows = [] for i in sorted(result.issues, key=lambda x: (SEVERITY_ORDER[x.severity], x.line_no), reverse=True)[:80]: issue_rows.append( f"{h(i.line_no)}{h(i.timestamp)}{severity_badge(i.severity)}{h(i.category)}{h(i.title)}{h(i.job_name)}{h(i.step_name)}{h(', '.join(i.abend_codes))}{h(i.recommendation)}
{h(i.evidence)}
" ) event_rows = [] for e in result.events[:120]: event_rows.append( f"{h(e.line_no)}{h(e.timestamp)}{h(e.family)}{h(e.subsystem)}{h(e.message_id)}{h(e.event_type)}{h(e.job_name)}{h(e.message)}" ) job_rows = [] for j in result.jobs: job_rows.append( f"{h(j.job_name)}{h(j.started)}{h(j.ended)}{h(j.max_rc)}{h(', '.join(j.abends))}{h(', '.join(j.steps))}{severity_badge(j.highest_severity)}{h(j.issue_count)}" ) subsystem_rows = [] for subsystem, count in sorted(result.subsystem_counts.items(), key=lambda x: x[1], reverse=True): subsystem_rows.append(f"{h(subsystem)}{h(count)}{h(result.category_counts)}") document = f""" SYSLOG / OPERLOG Incident Analyzer V2 Report

SYSLOG / OPERLOG Incident Analyzer V2

Source: {h(result.source_name)} | Profile: {h(result.profile)} | Status: {h(result.status)} | Window: {h(result.started_at)} → {h(result.ended_at)}
{h(result.status)}
Status
{h(result.risk_score)}
Risk score
{h(result.issue_count)}
Issues
{h(result.cluster_count)}
Incident clusters
{h(result.max_rc)}
Max RC

Root cause candidates

{''.join(roots) if roots else '

No root cause candidates detected.

'}

Incident clusters

{''.join(cluster_rows)}
IDStartEndSeverityRiskTitleJobsCategoriesRecommendation

Jobs

{''.join(job_rows)}
JobStartedEndedMax RCABENDsStepsSeverityIssues

Issues

{''.join(issue_rows)}
LineTimeSeverityCategoryTitleJobStepABENDRecommendationEvidence

Timeline events

{''.join(event_rows)}
LineTimeFamilySubsystemMessage IDTypeJobMessage
""" with open(path, "w", encoding="utf-8") as fh: fh.write(document) def console_summary(result: AnalysisResult, evidence_limit: int = 5) -> str: lines = [ "SYSLOG / OPERLOG Incident Analyzer V2", f"Source : {result.source_name}", f"Profile : {result.profile}", f"Status : {result.status}", f"Highest severity : {result.highest_severity}", f"Risk score : {result.risk_score}/100", f"Lines : {result.line_count}", f"Events : {result.event_count}", f"Issues : {result.issue_count}", f"Clusters : {result.cluster_count}", f"Jobs : {result.job_count}", f"Datasets : {result.dataset_count}", f"ABENDs : {', '.join(result.abends) if result.abends else '-'}", f"Max RC : {result.max_rc}", "", "Top root cause candidates:", ] if not result.root_causes: lines.append(" No root cause candidates detected.") for root in result.root_causes[:evidence_limit]: lines.append(f" - {root['title']} [{root['severity']}] risk={root['risk_score']} recommendation={root['recommendation']}") for ev in root.get("evidence", [])[:3]: lines.append(f" {ev}") lines.append("") lines.append("Subsystem counts:") for name, count in sorted(result.subsystem_counts.items(), key=lambda x: x[1], reverse=True)[:10]: lines.append(f" {name:12} {count}") return "\n".join(lines) def get_demo_text(kind: str = "mixed") -> str: demos = { "mixed": """2026-05-22 08:53:37.000 $HASP373 PAYROLL1 STARTED - INIT 1 - CLASS A - SYS ZOS1 2026-05-22 08:53:38.000 IEF403I PAYROLL1 - STARTED - TIME=08.53.38 2026-05-22 08:53:40.000 IEF236I ALLOC. FOR PAYROLL1 STEP010 2026-05-22 08:54:12.000 IGD101I SMS ALLOCATED TO DDNAME SYSUT2 DSN=PAYROLL.PROD.WORK.FILE 2026-05-22 08:54:16.000 IEC030I B37-04,IFG0554A,PAYROLL1,STEP030,SYSUT2,3390,VOL001,PAYROLL.PROD.WORK.FILE 2026-05-22 08:54:17.000 IEF450I PAYROLL1 STEP030 PAYPROG - ABEND=SB37 U0000 REASON=00000004 2026-05-22 08:54:18.000 $HASP395 PAYROLL1 ENDED - RC=0012 2026-05-22 08:55:02.000 $HASP373 ACCTSEC STARTED - INIT 2 - CLASS A - SYS ZOS1 2026-05-22 08:55:06.000 ICH408I USER(ACCT01) GROUP(FIN) NAME(ACCT BATCH) PAYROLL.SECRET.FILE CL(DATASET) INSUFFICIENT ACCESS AUTHORITY 2026-05-22 08:55:07.000 IEF450I ACCTSEC STEP010 READSEC - ABEND=S913 U0000 REASON=00000000 2026-05-22 08:55:08.000 $HASP395 ACCTSEC ENDED - RC=0012 2026-05-22 08:56:09.000 DSNT500I -DB2P RESOURCE UNAVAILABLE REASON 00C90088 TYPE 00000200 NAME PAYROLL.PAYTABLE 2026-05-22 08:56:10.000 DSNT376I -DB2P PLAN=PAYROLL LOCK TIMEOUT SQLCODE=-911 2026-05-22 08:56:11.000 DFHAC2206 05/22/2026 CICSP1 Transaction PAY1 failed with abend ASRA 2026-05-22 08:56:12.000 IXC431I SYSTEM ZOS2 LOST CONNECTIVITY TO COUPLING FACILITY CF01 2026-05-22 08:56:13.000 IXG251E SYSTEM LOGGER OFFLOAD DELAY DETECTED FOR LOGSTREAM PAYROLL.LOG 2026-05-22 08:56:14.000 IOS000I DEVICE 0A34 PATH 12 EXPERIENCED UNIT CHECK """, "sysplex": """2026-05-22 11:00:01 IXC402D REPLY I TO INITIALIZE SYSPLEX PARTITIONING, OR R TO RETRY 2026-05-22 11:00:06 IXC431I SYSTEM ZOS2 LOST CONNECTIVITY TO COUPLING FACILITY CF01 2026-05-22 11:00:08 IXL013I STRUCTURE PAYLOCK1 REBUILD REQUIRED 2026-05-22 11:00:10 IXG251E SYSTEM LOGGER OFFLOAD DELAY DETECTED FOR LOGSTREAM CICS.LOG1 2026-05-22 11:00:12 DFHAC2206 CICSP1 Transaction PAY1 failed with abend AICA 2026-05-22 11:00:15 DSNT500I -DB2P RESOURCE UNAVAILABLE REASON 00C90089 TYPE 00000200 NAME LOCKS """, "security": """2026-05-22 09:15:01 $HASP373 PAYSEC STARTED - INIT 3 - CLASS A - SYS ZOS1 2026-05-22 09:15:04 ICH408I USER(PAYU01) GROUP(PAY) NAME(PAY USER) PAYROLL.SECRET.FILE CL(DATASET) INSUFFICIENT ACCESS AUTHORITY 2026-05-22 09:15:05 IRR010I RACF CHECK FAILED FOR DATASET PAYROLL.SECRET.FILE 2026-05-22 09:15:06 IEF450I PAYSEC STEP010 READSEC - ABEND=S913 U0000 REASON=00000000 2026-05-22 09:15:07 $HASP395 PAYSEC ENDED - RC=0012 """, "storage": """2026-05-22 10:22:01 $HASP373 ARCHIVE1 STARTED - INIT 5 - CLASS B - SYS ZOS1 2026-05-22 10:22:12 IGD17272I VOLUME SELECTION HAS FAILED FOR DATA SET ARCHIVE.MONTHLY.G0001V00 2026-05-22 10:22:16 IEC030I E37-04,IFG0554A,ARCHIVE1,STEP020,OUTDD,3390,VOL009,ARCHIVE.MONTHLY.G0001V00 2026-05-22 10:22:17 IEF450I ARCHIVE1 STEP020 ARCHPGM - ABEND=SE37 U0000 REASON=00000004 2026-05-22 10:22:20 IOS000I DEVICE 0A34 PATH 10 EXPERIENCED UNIT CHECK 2026-05-22 10:22:21 $HASP395 ARCHIVE1 ENDED - RC=0012 """, } return demos.get(kind, demos["mixed"]) def load_text(path: str) -> str: with open(path, "r", encoding="utf-8", errors="ignore") as fh: return fh.read() def write_sample(path: str, kind: str) -> None: with open(path, "w", encoding="utf-8") as fh: fh.write(get_demo_text(kind)) def should_fail(result: AnalysisResult, fail_on: Optional[str]) -> bool: if not fail_on: return False threshold = SEVERITY_ORDER[fail_on.upper()] return SEVERITY_ORDER[result.highest_severity] >= threshold def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="SYSLOG / OPERLOG Incident Analyzer V2") parser.add_argument("input_file", nargs="?", help="SYSLOG/OPERLOG/JES text export to analyze") parser.add_argument("--demo", choices=["mixed", "sysplex", "security", "storage"], help="Run an embedded demo scenario") parser.add_argument("--write-sample", metavar="PATH", help="Write a sample SYSLOG/OPERLOG file") parser.add_argument("--sample-kind", default="mixed", choices=["mixed", "sysplex", "security", "storage"], help="Sample kind for --write-sample") parser.add_argument("--profile", default="production", choices=sorted(PROFILE_RULE_BOOSTS.keys()), help="Analysis profile") parser.add_argument("--window-minutes", type=int, default=5, help="Incident clustering time window") parser.add_argument("--context-lines", type=int, default=3, help="Evidence context lines before and after each issue") parser.add_argument("--since", help="Optional start filter: YYYY-MM-DD HH:MM:SS or HH:MM:SS") parser.add_argument("--until", help="Optional end filter: YYYY-MM-DD HH:MM:SS or HH:MM:SS") parser.add_argument("--redact", action="store_true", help="Redact dataset names and user IDs in outputs") parser.add_argument("--json", dest="json_path", help="Write JSON analysis") parser.add_argument("--html", dest="html_path", help="Write HTML report") parser.add_argument("--csv-events", help="Write timeline events CSV") parser.add_argument("--csv-issues", help="Write issues CSV") parser.add_argument("--csv-jobs", help="Write jobs CSV") parser.add_argument("--csv-clusters", help="Write incident clusters CSV") parser.add_argument("--csv-subsystems", help="Write subsystem summary CSV") parser.add_argument("--fail-on", choices=["WARNING", "ERROR", "CRITICAL"], help="Return non-zero if highest severity reaches this threshold") parser.add_argument("--evidence-limit", type=int, default=5, help="Console root-cause display limit") return parser def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_arg_parser() args = parser.parse_args(argv) if args.write_sample: write_sample(args.write_sample, args.sample_kind) print(f"Sample written: {args.write_sample}") return 0 if args.demo: text = get_demo_text(args.demo) source = f"demo:{args.demo}" elif args.input_file: text = load_text(args.input_file) source = args.input_file else: parser.error("Provide input_file, --demo, or --write-sample") analyzer = SyslogOperlogAnalyzer( profile=args.profile, window_minutes=args.window_minutes, context_lines=args.context_lines, redact=args.redact, since=args.since, until=args.until, ) result = analyzer.analyze_text(text, source_name=source) print(console_summary(result, evidence_limit=args.evidence_limit)) if args.json_path: write_json(result, args.json_path) print(f"JSON written: {args.json_path}") if args.html_path: write_html(result, args.html_path) print(f"HTML written: {args.html_path}") if args.csv_events: write_events_csv(result, args.csv_events) print(f"Events CSV written: {args.csv_events}") if args.csv_issues: write_issues_csv(result, args.csv_issues) print(f"Issues CSV written: {args.csv_issues}") if args.csv_jobs: write_jobs_csv(result, args.csv_jobs) print(f"Jobs CSV written: {args.csv_jobs}") if args.csv_clusters: write_clusters_csv(result, args.csv_clusters) print(f"Clusters CSV written: {args.csv_clusters}") if args.csv_subsystems: write_subsystems_csv(result, args.csv_subsystems) print(f"Subsystems CSV written: {args.csv_subsystems}") if should_fail(result, args.fail_on): return 2 return 0 if __name__ == "__main__": sys.exit(main())