#!/usr/bin/env python3 """ RMF/CMF DASD report analyzer - V2. Purpose: - Parse IBM RMF DASD Activity and Shared DASD Activity text reports. - Parse BMC CMF Shared Device Activity text reports in a pragmatic way. - Normalize device metrics. - Detect likely I/O performance issues. - Export CSV, JSON, and a standalone HTML report. This script intentionally uses only the Python standard library. """ from __future__ import annotations import argparse import csv import dataclasses import datetime as dt import html import json import math import re import statistics import sys from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple # --------------------------------------------------------------------------- # Data model # --------------------------------------------------------------------------- class Severity(str, Enum): INFO = "INFO" WARN = "WARN" CRITICAL = "CRITICAL" @dataclass class ReportMeta: source_file: str parser_mode: str detected_report_type: str = "unknown" system_id: Optional[str] = None sysplex: Optional[str] = None report_date: Optional[str] = None report_time: Optional[str] = None interval: Optional[str] = None rmf_version: Optional[str] = None raw_headers: List[str] = field(default_factory=list) @dataclass class DeviceMetricRecord: record_type: str source: str line_number: int raw_line: str device_number: Optional[str] = None device_type: Optional[str] = None volume_serial: Optional[str] = None storage_group: Optional[str] = None pav: Optional[str] = None smf_system_id: Optional[str] = None iodf_suffix: Optional[str] = None lcu: Optional[str] = None activity_rate: Optional[float] = None response_time_ms: Optional[float] = None iosq_time_ms: Optional[float] = None cmr_delay_ms: Optional[float] = None db_delay_ms: Optional[float] = None interrupt_delay_ms: Optional[float] = None pending_time_ms: Optional[float] = None disconnect_time_ms: Optional[float] = None connect_time_ms: Optional[float] = None device_connect_pct: Optional[float] = None device_util_pct: Optional[float] = None device_reserve_pct: Optional[float] = None avg_allocated: Optional[float] = None total_service_ms: Optional[float] = None bmc_init_command_response_ms: Optional[float] = None bmc_device_busy_delay_ms: Optional[float] = None parent_device_number: Optional[str] = None parent_volume_serial: Optional[str] = None score: float = 0.0 @dataclass class AnalysisIssue: severity: Severity code: str title: str message: str device_number: Optional[str] volume_serial: Optional[str] lcu: Optional[str] value: Optional[float] = None threshold: Optional[float] = None line_number: Optional[int] = None @dataclass class AnalysisResult: meta: ReportMeta records: List[DeviceMetricRecord] issues: List[AnalysisIssue] summary: Dict[str, Any] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _NUMBER_RE = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)$") _HEX_DEVICE_RE = re.compile(r"^[0-9A-Fa-f]{3,5}$") _VOL_RE = re.compile(r"^[A-Z0-9@$#]{1,8}$", re.IGNORECASE) def clean_text_line(line: str) -> str: return line.rstrip("\n\r") def normalize_spaces(line: str) -> str: return re.sub(r"\s+", " ", line.strip()) def is_number_token(token: str) -> bool: token = token.strip().replace(",", "") return bool(_NUMBER_RE.match(token)) def to_float(token: Optional[str]) -> Optional[float]: if token is None: return None text = token.strip().replace(",", "") if text in {"", "-", "--", "*", "N/A", "NA"}: return None if not is_number_token(text): return None try: return float(text) except ValueError: return None def round_or_none(value: Optional[float], digits: int = 6) -> Optional[float]: if value is None: return None if math.isnan(value) or math.isinf(value): return None return round(value, digits) def safe_ratio(part: Optional[float], total: Optional[float]) -> Optional[float]: if part is None or total is None or total == 0: return None return part / total def as_plain_dict(obj: Any) -> Any: if dataclasses.is_dataclass(obj): result = {} for f in dataclasses.fields(obj): value = getattr(obj, f.name) if isinstance(value, Enum): result[f.name] = value.value else: result[f.name] = as_plain_dict(value) return result if isinstance(obj, list): return [as_plain_dict(x) for x in obj] if isinstance(obj, dict): return {k: as_plain_dict(v) for k, v in obj.items()} return obj def percentile(values: Sequence[float], pct: float) -> Optional[float]: cleaned = sorted(v for v in values if v is not None) if not cleaned: return None if len(cleaned) == 1: return cleaned[0] rank = (len(cleaned) - 1) * pct low = math.floor(rank) high = math.ceil(rank) if low == high: return cleaned[int(rank)] return cleaned[low] + (cleaned[high] - cleaned[low]) * (rank - low) # --------------------------------------------------------------------------- # Header and report detection # --------------------------------------------------------------------------- def extract_meta(lines: Sequence[str], source_file: str, parser_mode: str) -> ReportMeta: meta = ReportMeta(source_file=source_file, parser_mode=parser_mode) upper_joined = "\n".join(lines[:80]).upper() if ("S H A R E D" in upper_joined and "D I R E C T" in upper_joined and "A C C E S S" in upper_joined) or "SHARED DIRECT ACCESS DEVICE ACTIVITY" in upper_joined: meta.detected_report_type = "ibm_shared_dasd_activity" elif "D I R E C T A C C E S S D E V I C E A C T I V I T Y" in upper_joined or "DIRECT ACCESS DEVICE ACTIVITY" in upper_joined: meta.detected_report_type = "ibm_dasd_activity" elif "SHARED DEVICE ACTIVITY REPORT" in upper_joined and "CMF" in upper_joined: meta.detected_report_type = "bmc_cmf_shared_device_activity" elif "DEVICE ACTIVITY REPORT" in upper_joined and "BMC" in upper_joined: meta.detected_report_type = "bmc_cmf_device_activity" for raw in lines[:120]: line = clean_text_line(raw) if not line.strip(): continue if len(meta.raw_headers) < 25: meta.raw_headers.append(line) m = re.search(r"\bSYSTEM ID\s+([A-Z0-9@$#_-]+)", line, flags=re.IGNORECASE) if m: meta.system_id = m.group(1) m = re.search(r"\bSYSPLEX\s+([A-Z0-9@$#_-]+)", line, flags=re.IGNORECASE) if m: meta.sysplex = m.group(1) m = re.search(r"\bDATE\s+([0-9]{1,2}/[0-9]{1,2}/[0-9]{2,4}|[0-9]{4}-[0-9]{2}-[0-9]{2})", line, flags=re.IGNORECASE) if m: meta.report_date = m.group(1) m = re.search(r"\bTIME\s+([0-9]{1,2}[.:][0-9]{2}(?:[.:][0-9]{2})?)", line, flags=re.IGNORECASE) if m: meta.report_time = m.group(1) m = re.search(r"\bINTERVAL\s+([0-9.:\s]+)", line, flags=re.IGNORECASE) if m: meta.interval = normalize_spaces(m.group(1)) m = re.search(r"\bRPT VERSION\s+([A-Z0-9.]+)\s+RMF", line, flags=re.IGNORECASE) if m: meta.rmf_version = m.group(1) return meta # --------------------------------------------------------------------------- # Parsers # --------------------------------------------------------------------------- def parse_ibm_shared_dasd_line( line: str, line_number: int, last_parent: Optional[DeviceMetricRecord], ) -> Tuple[Optional[DeviceMetricRecord], Optional[DeviceMetricRecord]]: """ Parses IBM Shared Direct Access Device Activity report lines. Parent line example: 02180 33909 SYST10 *ALL 0.036 0.248 0.000 0.000 0.000 0.005 0.120 0.000 0.128 0.00 0.00 0.0 0.0 Child line example: 1 SYSD 00 0062 0.000 0.000 ... """ raw = clean_text_line(line) stripped = raw.strip() if not stripped: return None, last_parent tokens = stripped.split() if len(tokens) < 8: return None, last_parent first = tokens[0] if _HEX_DEVICE_RE.match(first): numbers = [t for t in tokens if is_number_token(t)] if len(numbers) < 8: return None, last_parent rec = DeviceMetricRecord( record_type="ibm_shared_dasd_parent", source="IBM_RMF", line_number=line_number, raw_line=raw, device_number=tokens[0], device_type=tokens[1] if len(tokens) > 1 else None, volume_serial=tokens[2] if len(tokens) > 2 else None, ) # In the standard layout, the last 13 numeric values are: # activity, resp, iosq, cmr, db, int, pend, disc, conn, dev_conn, dev_util, dev_resv, avg_alloc. nums = [to_float(t) for t in numbers] nums = [n for n in nums if n is not None] tail = nums[-13:] if len(nums) >= 13 else nums if len(tail) >= 9: rec.activity_rate = tail[0] rec.response_time_ms = tail[1] rec.iosq_time_ms = tail[2] rec.cmr_delay_ms = tail[3] rec.db_delay_ms = tail[4] rec.interrupt_delay_ms = tail[5] rec.pending_time_ms = tail[6] rec.disconnect_time_ms = tail[7] rec.connect_time_ms = tail[8] if len(tail) >= 13: rec.device_connect_pct = tail[9] rec.device_util_pct = tail[10] rec.device_reserve_pct = tail[11] rec.avg_allocated = tail[12] return rec, rec # IBM shared child lines often start with PAV, SYSID, IODF, LCU, then metrics. if last_parent is not None and len(tokens) >= 12: if tokens[0].replace(".", "", 1).isdigit() and re.match(r"^[A-Z0-9@$#_-]{2,8}$", tokens[1], re.IGNORECASE): numbers = [to_float(t) for t in tokens if to_float(t) is not None] if len(numbers) >= 9: rec = DeviceMetricRecord( record_type="ibm_shared_dasd_system", source="IBM_RMF", line_number=line_number, raw_line=raw, device_number=last_parent.device_number, device_type=last_parent.device_type, volume_serial=last_parent.volume_serial, pav=tokens[0], smf_system_id=tokens[1], iodf_suffix=tokens[2] if len(tokens) > 2 else None, lcu=tokens[3] if len(tokens) > 3 else None, parent_device_number=last_parent.device_number, parent_volume_serial=last_parent.volume_serial, ) tail = numbers[-13:] if len(numbers) >= 13 else numbers if len(tail) >= 9: rec.activity_rate = tail[0] rec.response_time_ms = tail[1] rec.iosq_time_ms = tail[2] rec.cmr_delay_ms = tail[3] rec.db_delay_ms = tail[4] rec.interrupt_delay_ms = tail[5] rec.pending_time_ms = tail[6] rec.disconnect_time_ms = tail[7] rec.connect_time_ms = tail[8] if len(tail) >= 13: rec.device_connect_pct = tail[9] rec.device_util_pct = tail[10] rec.device_reserve_pct = tail[11] rec.avg_allocated = tail[12] return rec, last_parent return None, last_parent def parse_ibm_dasd_activity_line(line: str, line_number: int) -> Optional[DeviceMetricRecord]: """ Parses IBM Direct Access Device Activity report lines. The full report can include storage group before DEV NUM. This parser is pragmatic: it finds the first plausible device number and then maps the numeric tail. """ raw = clean_text_line(line) stripped = raw.strip() if not stripped: return None upper = stripped.upper() header_words = ["DIRECT ACCESS", "DEVICE", "VOLUME", "STORAGE", "AVG", "RESP", "IOSQ", "PEND", "CONN"] if sum(1 for word in header_words if word in upper) >= 3: return None tokens = stripped.split() if len(tokens) < 10: return None # Avoid parsing IBM shared child rows as standalone DASD rows. if ( len(tokens) >= 4 and to_float(tokens[0]) is not None and re.match(r"^[A-Z0-9@$#_-]{2,8}$", tokens[1], re.IGNORECASE) and re.match(r"^[0-9A-Fa-f]{1,4}$", tokens[2]) and re.match(r"^[0-9A-Fa-f]{3,5}$", tokens[3]) ): return None device_idx = None for idx, token in enumerate(tokens[:8]): if _HEX_DEVICE_RE.match(token): # A valid DASD line normally has a device type and volser after the device number. if idx + 2 < len(tokens): device_idx = idx break if device_idx is None: return None numbers = [to_float(t) for t in tokens if to_float(t) is not None] if len(numbers) < 9: return None rec = DeviceMetricRecord( record_type="ibm_dasd_activity", source="IBM_RMF", line_number=line_number, raw_line=raw, storage_group=" ".join(tokens[:device_idx]) if device_idx > 0 else None, device_number=tokens[device_idx], device_type=tokens[device_idx + 1] if device_idx + 1 < len(tokens) else None, volume_serial=tokens[device_idx + 2] if device_idx + 2 < len(tokens) else None, ) tail = numbers[-13:] if len(numbers) >= 13 else numbers if len(tail) >= 9: rec.activity_rate = tail[0] rec.response_time_ms = tail[1] rec.iosq_time_ms = tail[2] rec.cmr_delay_ms = tail[3] rec.db_delay_ms = tail[4] rec.interrupt_delay_ms = tail[5] rec.pending_time_ms = tail[6] rec.disconnect_time_ms = tail[7] rec.connect_time_ms = tail[8] if len(tail) >= 13: rec.device_connect_pct = tail[9] rec.device_util_pct = tail[10] rec.device_reserve_pct = tail[11] rec.avg_allocated = tail[12] return rec def parse_bmc_cmf_shared_device_line(line: str, line_number: int) -> Optional[DeviceMetricRecord]: """ Parses BMC CMF Shared Device Activity report lines. Example: 0A87B 87 00B7 SJSC 1.0 HPV 0 0.146 0 0 0 0 0.055 0 0.091 0.0 0.0 0.0 86.1 0.0 0.0 """ raw = clean_text_line(line) stripped = raw.strip() if not stripped: return None upper = stripped.upper() if "BMC" in upper or "PRODUCED BY" in upper or "SHARED DEVICE" in upper: return None if "VOLSER" in upper or "DEV NUM" in upper or "AVERAGE TIME" in upper: return None tokens = stripped.split() if len(tokens) < 14: return None if not _HEX_DEVICE_RE.match(tokens[0]): return None # BMC reports can contain a device type/group line just before device rows. numeric_count = sum(1 for t in tokens if to_float(t) is not None) if numeric_count < 8: return None rec = DeviceMetricRecord( record_type="bmc_cmf_shared_device", source="BMC_CMF", line_number=line_number, raw_line=raw, device_number=tokens[0], iodf_suffix=tokens[1] if len(tokens) > 1 else None, lcu=tokens[2] if len(tokens) > 2 else None, smf_system_id=tokens[3] if len(tokens) > 3 else None, pav=tokens[4] if len(tokens) > 4 else None, ) nums = [to_float(t) for t in tokens if to_float(t) is not None] nums = [n for n in nums if n is not None] # Expected relevant BMC metric sequence after PAV/type fields: # ssch/sec, total_service, iosq, init_cmd_resp, dev_busy_delay, # interrupt_delay, pend, disc, conn, dev_conn, in_use, dev_resv, # dev_alloc, mount_pend, avg_dsets_alloc. tail = nums[-15:] if len(nums) >= 15 else nums if len(tail) >= 9: rec.activity_rate = tail[0] rec.total_service_ms = tail[1] rec.response_time_ms = tail[1] rec.iosq_time_ms = tail[2] rec.bmc_init_command_response_ms = tail[3] rec.bmc_device_busy_delay_ms = tail[4] rec.interrupt_delay_ms = tail[5] rec.pending_time_ms = tail[6] rec.disconnect_time_ms = tail[7] rec.connect_time_ms = tail[8] if len(tail) >= 15: rec.device_connect_pct = tail[9] rec.device_util_pct = tail[10] rec.device_reserve_pct = tail[11] rec.avg_allocated = tail[14] return rec def parse_records(lines: Sequence[str], meta: ReportMeta, mode: str) -> List[DeviceMetricRecord]: records: List[DeviceMetricRecord] = [] last_shared_parent: Optional[DeviceMetricRecord] = None effective_mode = mode if mode == "auto": if meta.detected_report_type == "ibm_shared_dasd_activity": effective_mode = "ibm-shared-dasd" elif meta.detected_report_type == "ibm_dasd_activity": effective_mode = "ibm-dasd" elif meta.detected_report_type in {"bmc_cmf_shared_device_activity", "bmc_cmf_device_activity"}: effective_mode = "bmc-cmf-shared-device" else: effective_mode = "mixed" for line_number, line in enumerate(lines, start=1): record: Optional[DeviceMetricRecord] = None if effective_mode in {"ibm-shared-dasd", "mixed"}: record, last_shared_parent = parse_ibm_shared_dasd_line(line, line_number, last_shared_parent) if record is not None: records.append(record) continue if effective_mode in {"ibm-dasd", "mixed"}: record = parse_ibm_dasd_activity_line(line, line_number) if record is not None: records.append(record) continue if effective_mode in {"bmc-cmf-shared-device", "mixed"}: record = parse_bmc_cmf_shared_device_line(line, line_number) if record is not None: records.append(record) continue return records # --------------------------------------------------------------------------- # Analysis engine # --------------------------------------------------------------------------- @dataclass class ThresholdProfile: warn_response_ms: float = 10.0 critical_response_ms: float = 25.0 warn_iosq_ms: float = 1.0 critical_iosq_ms: float = 5.0 warn_pending_ms: float = 1.0 critical_pending_ms: float = 5.0 warn_device_util_pct: float = 50.0 critical_device_util_pct: float = 75.0 warn_connect_ms: float = 3.0 critical_connect_ms: float = 10.0 warn_component_ratio: float = 0.55 critical_component_ratio: float = 0.75 def classify_threshold(value: Optional[float], warn: float, critical: float) -> Optional[Severity]: if value is None: return None if value >= critical: return Severity.CRITICAL if value >= warn: return Severity.WARN return None def add_issue( issues: List[AnalysisIssue], severity: Severity, code: str, title: str, message: str, record: DeviceMetricRecord, value: Optional[float], threshold: Optional[float], ) -> None: issues.append( AnalysisIssue( severity=severity, code=code, title=title, message=message, device_number=record.device_number, volume_serial=record.volume_serial, lcu=record.lcu, value=round_or_none(value), threshold=threshold, line_number=record.line_number, ) ) def score_device(record: DeviceMetricRecord, profile: ThresholdProfile) -> float: score = 0.0 if record.response_time_ms is not None: score += min(record.response_time_ms / max(profile.warn_response_ms, 0.001), 10.0) * 18.0 if record.iosq_time_ms is not None: score += min(record.iosq_time_ms / max(profile.warn_iosq_ms, 0.001), 10.0) * 24.0 if record.pending_time_ms is not None: score += min(record.pending_time_ms / max(profile.warn_pending_ms, 0.001), 10.0) * 22.0 if record.device_util_pct is not None: score += min(record.device_util_pct / max(profile.warn_device_util_pct, 0.001), 4.0) * 14.0 if record.connect_time_ms is not None: score += min(record.connect_time_ms / max(profile.warn_connect_ms, 0.001), 5.0) * 8.0 activity = record.activity_rate or 0.0 if activity > 100: score += 8.0 elif activity > 20: score += 4.0 return round(score, 3) def analyze_record(record: DeviceMetricRecord, profile: ThresholdProfile) -> List[AnalysisIssue]: issues: List[AnalysisIssue] = [] sev = classify_threshold(record.response_time_ms, profile.warn_response_ms, profile.critical_response_ms) if sev: add_issue( issues, sev, "HIGH_RESPONSE_TIME", "High response time", "The average DASD response time is above the configured threshold.", record, record.response_time_ms, profile.critical_response_ms if sev == Severity.CRITICAL else profile.warn_response_ms, ) sev = classify_threshold(record.iosq_time_ms, profile.warn_iosq_ms, profile.critical_iosq_ms) if sev: add_issue( issues, sev, "HIGH_IOSQ_TIME", "High IOSQ time", "The device shows queueing delay while busy from the local system. This often points to device contention, insufficient parallelism, or poor volume placement.", record, record.iosq_time_ms, profile.critical_iosq_ms if sev == Severity.CRITICAL else profile.warn_iosq_ms, ) sev = classify_threshold(record.pending_time_ms, profile.warn_pending_ms, profile.critical_pending_ms) if sev: add_issue( issues, sev, "HIGH_PENDING_TIME", "High pending time", "The device shows pending delay before I/O acceptance. This can indicate control-unit, path, backend, or shared-device contention.", record, record.pending_time_ms, profile.critical_pending_ms if sev == Severity.CRITICAL else profile.warn_pending_ms, ) sev = classify_threshold(record.device_util_pct, profile.warn_device_util_pct, profile.critical_device_util_pct) if sev: add_issue( issues, sev, "HIGH_DEVICE_UTILIZATION", "High device utilization", "The device utilization percentage is high. Interpret together with IOSQ and response time to avoid false positives.", record, record.device_util_pct, profile.critical_device_util_pct if sev == Severity.CRITICAL else profile.warn_device_util_pct, ) sev = classify_threshold(record.connect_time_ms, profile.warn_connect_ms, profile.critical_connect_ms) if sev: add_issue( issues, sev, "HIGH_CONNECT_TIME", "High connect time", "The connect time is high. This can indicate large transfers, channel pressure, or backend service-time issues.", record, record.connect_time_ms, profile.critical_connect_ms if sev == Severity.CRITICAL else profile.warn_connect_ms, ) response = record.response_time_ms if response and response > 0: iosq_ratio = safe_ratio(record.iosq_time_ms, response) pending_ratio = safe_ratio(record.pending_time_ms, response) connect_ratio = safe_ratio(record.connect_time_ms, response) disconnect_ratio = safe_ratio(record.disconnect_time_ms, response) if iosq_ratio is not None and iosq_ratio >= profile.warn_component_ratio: sev = Severity.CRITICAL if iosq_ratio >= profile.critical_component_ratio else Severity.WARN add_issue( issues, sev, "IOSQ_DOMINATES_RESPONSE", "IOSQ dominates response time", "Most of the response time is queueing delay. This is a strong sign of contention rather than raw device service time.", record, iosq_ratio, profile.critical_component_ratio if sev == Severity.CRITICAL else profile.warn_component_ratio, ) if pending_ratio is not None and pending_ratio >= profile.warn_component_ratio: sev = Severity.CRITICAL if pending_ratio >= profile.critical_component_ratio else Severity.WARN add_issue( issues, sev, "PENDING_DOMINATES_RESPONSE", "Pending time dominates response time", "Most of the response time is pending delay. Investigate control unit, paths, remote sharing, backend pressure, or device acceptance delays.", record, pending_ratio, profile.critical_component_ratio if sev == Severity.CRITICAL else profile.warn_component_ratio, ) if connect_ratio is not None and connect_ratio >= profile.warn_component_ratio: sev = Severity.CRITICAL if connect_ratio >= profile.critical_component_ratio else Severity.WARN add_issue( issues, sev, "CONNECT_DOMINATES_RESPONSE", "Connect time dominates response time", "Most of the response time is connect time. Check transfer size, channel usage, and I/O pattern characteristics.", record, connect_ratio, profile.critical_component_ratio if sev == Severity.CRITICAL else profile.warn_component_ratio, ) if disconnect_ratio is not None and disconnect_ratio >= profile.warn_component_ratio: sev = Severity.CRITICAL if disconnect_ratio >= profile.critical_component_ratio else Severity.WARN add_issue( issues, sev, "DISCONNECT_DOMINATES_RESPONSE", "Disconnect time dominates response time", "Most of the response time is disconnect time. This may be normal for some workloads, but can also reveal backend wait or cache miss behavior.", record, disconnect_ratio, profile.critical_component_ratio if sev == Severity.CRITICAL else profile.warn_component_ratio, ) components = [ record.iosq_time_ms, record.cmr_delay_ms, record.db_delay_ms, record.interrupt_delay_ms, record.pending_time_ms, record.disconnect_time_ms, record.connect_time_ms, ] known_sum = sum(v for v in components if v is not None) if known_sum > 0 and abs(known_sum - response) > max(1.0, response * 0.35): add_issue( issues, Severity.INFO, "COMPONENT_SUM_MISMATCH", "Timing component sum differs from response time", "The known timing components do not closely add up to response time. This can be caused by report format differences, missing columns, rounding, or parser layout mismatch.", record, known_sum, response, ) if record.activity_rate is not None and record.activity_rate < 0.1 and response is not None and response >= profile.warn_response_ms: add_issue( issues, Severity.WARN, "LOW_ACTIVITY_HIGH_RESPONSE", "Low activity but high response time", "The device has low activity rate but high response time. This can indicate sporadic severe latency, path issues, or an outlier interval.", record, response, profile.warn_response_ms, ) return issues def build_summary(records: List[DeviceMetricRecord], issues: List[AnalysisIssue], top_n: int) -> Dict[str, Any]: def values(attr: str) -> List[float]: out = [] for r in records: value = getattr(r, attr) if value is not None: out.append(value) return out def stats(attr: str) -> Dict[str, Optional[float]]: vals = values(attr) if not vals: return {"count": 0, "avg": None, "p50": None, "p95": None, "max": None} return { "count": len(vals), "avg": round_or_none(statistics.mean(vals)), "p50": round_or_none(percentile(vals, 0.50)), "p95": round_or_none(percentile(vals, 0.95)), "max": round_or_none(max(vals)), } issue_counts: Dict[str, int] = {} severity_counts: Dict[str, int] = {s.value: 0 for s in Severity} for issue in issues: issue_counts[issue.code] = issue_counts.get(issue.code, 0) + 1 severity_counts[issue.severity.value] = severity_counts.get(issue.severity.value, 0) + 1 def top_by(attr: str) -> List[Dict[str, Any]]: return [ compact_record(r) for r in sorted(records, key=lambda x: getattr(x, attr) or -1.0, reverse=True)[:top_n] ] return { "record_count": len(records), "device_count": len({r.device_number for r in records if r.device_number}), "volume_count": len({r.volume_serial for r in records if r.volume_serial}), "issue_count": len(issues), "severity_counts": severity_counts, "issue_counts": dict(sorted(issue_counts.items(), key=lambda kv: kv[1], reverse=True)), "metrics": { "response_time_ms": stats("response_time_ms"), "iosq_time_ms": stats("iosq_time_ms"), "pending_time_ms": stats("pending_time_ms"), "disconnect_time_ms": stats("disconnect_time_ms"), "connect_time_ms": stats("connect_time_ms"), "device_util_pct": stats("device_util_pct"), "activity_rate": stats("activity_rate"), }, "top_by_score": [compact_record(r) for r in sorted(records, key=lambda x: x.score, reverse=True)[:top_n]], "top_by_response_time": top_by("response_time_ms"), "top_by_iosq_time": top_by("iosq_time_ms"), "top_by_pending_time": top_by("pending_time_ms"), "top_by_device_util": top_by("device_util_pct"), } def compact_record(record: DeviceMetricRecord) -> Dict[str, Any]: return { "record_type": record.record_type, "line": record.line_number, "device": record.device_number, "volume": record.volume_serial, "lcu": record.lcu, "system": record.smf_system_id, "activity": round_or_none(record.activity_rate), "resp_ms": round_or_none(record.response_time_ms), "iosq_ms": round_or_none(record.iosq_time_ms), "pend_ms": round_or_none(record.pending_time_ms), "disc_ms": round_or_none(record.disconnect_time_ms), "conn_ms": round_or_none(record.connect_time_ms), "util_pct": round_or_none(record.device_util_pct), "score": round_or_none(record.score), } def analyze(records: List[DeviceMetricRecord], meta: ReportMeta, profile: ThresholdProfile, top_n: int) -> AnalysisResult: issues: List[AnalysisIssue] = [] for record in records: record.score = score_device(record, profile) issues.extend(analyze_record(record, profile)) severity_order = {Severity.CRITICAL: 0, Severity.WARN: 1, Severity.INFO: 2} issues.sort(key=lambda i: (severity_order[i.severity], i.code, -(i.value or 0))) summary = build_summary(records, issues, top_n=top_n) return AnalysisResult(meta=meta, records=records, issues=issues, summary=summary) # --------------------------------------------------------------------------- # Rendering # --------------------------------------------------------------------------- def format_num(value: Optional[float], digits: int = 3) -> str: if value is None: return "-" return f"{value:.{digits}f}" def print_console_report(result: AnalysisResult, top_n: int) -> None: meta = result.meta summary = result.summary print() print("=" * 96) print("RMF/CMF DASD ANALYSIS REPORT") print("=" * 96) print(f"Source file : {meta.source_file}") print(f"Parser mode : {meta.parser_mode}") print(f"Detected type : {meta.detected_report_type}") if meta.system_id: print(f"System ID : {meta.system_id}") if meta.sysplex: print(f"Sysplex : {meta.sysplex}") if meta.report_date or meta.report_time: print(f"Report timestamp : {(meta.report_date or '-') + ' ' + (meta.report_time or '-')}") if meta.interval: print(f"Interval : {meta.interval}") print("-" * 96) print(f"Records : {summary['record_count']}") print(f"Devices : {summary['device_count']}") print(f"Volumes : {summary['volume_count']}") print(f"Issues : {summary['issue_count']}") print(f"Severity : CRITICAL={summary['severity_counts'].get('CRITICAL', 0)} " f"WARN={summary['severity_counts'].get('WARN', 0)} " f"INFO={summary['severity_counts'].get('INFO', 0)}") print() print("METRIC SNAPSHOT") print("-" * 96) print(f"{'Metric':26} {'Count':>8} {'Avg':>12} {'P50':>12} {'P95':>12} {'Max':>12}") for metric, stat in summary["metrics"].items(): print( f"{metric:26} {stat['count']:>8} " f"{format_num(stat['avg']):>12} {format_num(stat['p50']):>12} " f"{format_num(stat['p95']):>12} {format_num(stat['max']):>12}" ) print_top_table("TOP BY SCORE", summary["top_by_score"], top_n) print_top_table("TOP BY RESPONSE TIME", summary["top_by_response_time"], top_n) print_top_table("TOP BY IOSQ TIME", summary["top_by_iosq_time"], top_n) print_top_table("TOP BY PENDING TIME", summary["top_by_pending_time"], top_n) print() print("TOP ISSUES") print("-" * 96) if not result.issues: print("No issue detected with the current threshold profile.") else: for issue in result.issues[:top_n]: target = issue.volume_serial or issue.device_number or "-" print( f"{issue.severity.value:8} {issue.code:28} " f"DEV={issue.device_number or '-':>6} VOL={target:>8} " f"LCU={issue.lcu or '-':>5} VALUE={format_num(issue.value)} " f"THRESHOLD={format_num(issue.threshold)} LINE={issue.line_number or '-'}" ) print(f" {issue.message}") def print_top_table(title: str, rows: List[Dict[str, Any]], top_n: int) -> None: print() print(title) print("-" * 96) if not rows: print("No data.") return print( f"{'#':>3} {'Device':>7} {'Volume':>8} {'LCU':>5} {'Sys':>6} " f"{'Act':>10} {'Resp':>8} {'IOSQ':>8} {'Pend':>8} {'Disc':>8} {'Conn':>8} {'Util':>8} {'Score':>8}" ) for idx, row in enumerate(rows[:top_n], start=1): print( f"{idx:>3} {str(row.get('device') or '-'):>7} {str(row.get('volume') or '-'):>8} " f"{str(row.get('lcu') or '-'):>5} {str(row.get('system') or '-'):>6} " f"{format_num(row.get('activity')):>10} {format_num(row.get('resp_ms')):>8} " f"{format_num(row.get('iosq_ms')):>8} {format_num(row.get('pend_ms')):>8} " f"{format_num(row.get('disc_ms')):>8} {format_num(row.get('conn_ms')):>8} " f"{format_num(row.get('util_pct')):>8} {format_num(row.get('score')):>8}" ) def export_json(result: AnalysisResult, output_path: Path) -> None: output_path.write_text(json.dumps(as_plain_dict(result), indent=2), encoding="utf-8") def export_csv_records(records: List[DeviceMetricRecord], output_path: Path) -> None: fields = [f.name for f in dataclasses.fields(DeviceMetricRecord)] with output_path.open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for record in records: writer.writerow(as_plain_dict(record)) def export_csv_issues(issues: List[AnalysisIssue], output_path: Path) -> None: fields = [f.name for f in dataclasses.fields(AnalysisIssue)] with output_path.open("w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() for issue in issues: writer.writerow(as_plain_dict(issue)) def render_html_report(result: AnalysisResult) -> str: meta = result.meta summary = result.summary generated_at = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") css = """ body{font-family:Arial,Helvetica,sans-serif;margin:0;background:#0f172a;color:#e5e7eb} .wrap{max-width:1240px;margin:0 auto;padding:28px} h1{font-size:28px;margin:0 0 8px} h2{font-size:20px;margin-top:28px;border-bottom:1px solid #334155;padding-bottom:8px} .muted{color:#94a3b8} .grid{display:grid;grid-template-columns:repeat(4,1fr);gap:14px;margin:18px 0} .card{background:#111827;border:1px solid #334155;border-radius:14px;padding:16px;box-shadow:0 10px 25px rgba(0,0,0,.25)} .big{font-size:28px;font-weight:700} table{width:100%;border-collapse:collapse;background:#111827;border-radius:14px;overflow:hidden} th,td{padding:8px 10px;border-bottom:1px solid #334155;text-align:right;font-size:13px} th{background:#1e293b;color:#cbd5e1} td:first-child,th:first-child{text-align:left} .sev-CRITICAL{color:#fecaca;font-weight:700} .sev-WARN{color:#fde68a;font-weight:700} .sev-INFO{color:#bfdbfe;font-weight:700} .pill{display:inline-block;border-radius:999px;padding:3px 8px;background:#334155;color:#e5e7eb;font-size:12px} .footer{margin-top:28px;color:#94a3b8;font-size:12px} """ def metric_card(title: str, value: Any) -> str: return f"
{html.escape(title)}
{html.escape(str(value))}
" def records_table(title: str, rows: List[Dict[str, Any]]) -> str: out = [f"

{html.escape(title)}

"] out.append("") headers = ["Device", "Volume", "LCU", "System", "Activity", "Resp", "IOSQ", "Pend", "Disc", "Conn", "Util", "Score"] for h in headers: out.append(f"") out.append("") for r in rows: out.append("") values = [ r.get("device"), r.get("volume"), r.get("lcu"), r.get("system"), format_num(r.get("activity")), format_num(r.get("resp_ms")), format_num(r.get("iosq_ms")), format_num(r.get("pend_ms")), format_num(r.get("disc_ms")), format_num(r.get("conn_ms")), format_num(r.get("util_pct")), format_num(r.get("score")), ] for v in values: out.append(f"") out.append("") out.append("
{html.escape(h)}
{html.escape(str(v if v is not None else '-'))}
") return "\n".join(out) def issues_table() -> str: out = ["

Top issues

", ""] headers = ["Severity", "Code", "Device", "Volume", "LCU", "Value", "Threshold", "Line", "Message"] for h in headers: out.append(f"") out.append("") for issue in result.issues[:100]: sev = issue.severity.value out.append("") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append(f"") out.append("") out.append("
{html.escape(h)}
{html.escape(sev)}{html.escape(issue.code)}{html.escape(issue.device_number or '-')}{html.escape(issue.volume_serial or '-')}{html.escape(issue.lcu or '-')}{html.escape(format_num(issue.value))}{html.escape(format_num(issue.threshold))}{html.escape(str(issue.line_number or '-'))}{html.escape(issue.message)}
") return "\n".join(out) parts = [ "", "RMF/CMF DASD Analysis Report", f"", "
", "

RMF/CMF DASD Analysis Report

", f"
Generated at {html.escape(generated_at)}
", "
", f"
Source {html.escape(meta.source_file)}
", f"
Detected {html.escape(meta.detected_report_type)}
", f"
Parser {html.escape(meta.parser_mode)}
", f"
System {html.escape(meta.system_id or '-')}
", f"
Sysplex {html.escape(meta.sysplex or '-')}
", "
", "
", metric_card("Records", summary["record_count"]), metric_card("Devices", summary["device_count"]), metric_card("Volumes", summary["volume_count"]), metric_card("Issues", summary["issue_count"]), metric_card("Critical", summary["severity_counts"].get("CRITICAL", 0)), metric_card("Warnings", summary["severity_counts"].get("WARN", 0)), metric_card("Avg response ms", format_num(summary["metrics"]["response_time_ms"]["avg"])), metric_card("P95 response ms", format_num(summary["metrics"]["response_time_ms"]["p95"])), "
", records_table("Top by score", summary["top_by_score"]), records_table("Top by response time", summary["top_by_response_time"]), records_table("Top by IOSQ time", summary["top_by_iosq_time"]), records_table("Top by pending time", summary["top_by_pending_time"]), issues_table(), "
This report is heuristic. Confirm findings with native RMF/CMF tooling and site-specific thresholds.
", "
", ] return "\n".join(parts) def export_html(result: AnalysisResult, output_path: Path) -> None: output_path.write_text(render_html_report(result), encoding="utf-8") # --------------------------------------------------------------------------- # Demo sample # --------------------------------------------------------------------------- DEMO_REPORT = """ S H A R E D D I R E C T A C C E S S D E V I C E A C T I V I T Y PAGE 1 z/OS 3.1 SYSPLEX SYSDPLEX DATE 09/30/2021 INTERVAL 15.00.000 RPT VERSION 3.1 RMF TIME 12.30.00 CYCLE 1.000 SECONDS SMF DEVICE AVG AVG AVG AVG AVG AVG AVG AVG % % % AVG DEV DEVICE VOLUME PAV SYS IODF LCU ACTIVITY RESP IOSQ CMR DB INT PEND DISC CONN DEV DEV DEV NUMBER NUM TYPE SERIAL ID SUFF RATE TIME TIME DLY DLY DLY TIME TIME TIME CONN UTIL RESV ALLOC 02180 33909 SYST10 *ALL 0.036 0.248 0.000 0.000 0.000 0.005 0.120 0.000 0.128 0.00 0.00 0.0 0.0 1 SYSD 00 0062 0.000 0.000 0.000 0.000 0.000 0.000 0.000 0.000 0.000 0.00 0.00 0.0 0.0 1 SYSE 00 0062 0.036 0.248 0.000 0.000 0.000 0.005 0.120 0.000 0.128 0.00 0.00 0.0 0.0 02181 33909 SYST11 *ALL 88.071 33.264 8.500 0.006 0.000 0.000 19.130 0.000 5.134 0.00 82.00 0.0 0.0 1 SYSD 00 0062 44.036 12.280 6.012 0.012 0.000 0.000 1.136 0.000 4.144 0.00 74.00 0.0 0.0 1 SYSE 00 0062 44.036 25.248 2.488 0.000 0.000 0.000 18.124 0.000 4.124 0.00 90.00 0.0 0.0 """ def write_demo_report(path: Path) -> None: path.write_text(DEMO_REPORT, encoding="utf-8") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Analyze IBM RMF and BMC CMF DASD activity reports.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("input_file", nargs="?", help="Input RMF/CMF text report.") parser.add_argument( "--mode", choices=["auto", "mixed", "ibm-dasd", "ibm-shared-dasd", "bmc-cmf-shared-device"], default="auto", help="Parser mode.", ) parser.add_argument("--json", dest="json_output", help="Write full JSON analysis.") parser.add_argument("--csv-records", dest="csv_records_output", help="Write normalized records CSV.") parser.add_argument("--csv-issues", dest="csv_issues_output", help="Write detected issues CSV.") parser.add_argument("--html", dest="html_output", help="Write standalone HTML report.") parser.add_argument("--top", type=int, default=20, help="Number of top rows to display/export in summaries.") parser.add_argument("--warn-response-ms", type=float, default=10.0) parser.add_argument("--critical-response-ms", type=float, default=25.0) parser.add_argument("--warn-iosq-ms", type=float, default=1.0) parser.add_argument("--critical-iosq-ms", type=float, default=5.0) parser.add_argument("--warn-pending-ms", type=float, default=1.0) parser.add_argument("--critical-pending-ms", type=float, default=5.0) parser.add_argument("--warn-util-pct", type=float, default=50.0) parser.add_argument("--critical-util-pct", type=float, default=75.0) parser.add_argument("--warn-connect-ms", type=float, default=3.0) parser.add_argument("--critical-connect-ms", type=float, default=10.0) parser.add_argument("--demo", action="store_true", help="Create and analyze a built-in demo report.") parser.add_argument("--demo-output", default="rmf_demo_report.txt", help="Path used with --demo.") parser.add_argument("--quiet", action="store_true", help="Do not print the console report.") parser.add_argument("--fail-on-critical", action="store_true", help="Return exit code 2 when critical issues are found.") return parser def load_input(args: argparse.Namespace) -> Tuple[Path, List[str]]: if args.demo: demo_path = Path(args.demo_output) write_demo_report(demo_path) return demo_path, demo_path.read_text(encoding="utf-8", errors="ignore").splitlines() if not args.input_file: raise SystemExit("Missing input file. Use --demo to generate a demo report.") path = Path(args.input_file) if not path.exists(): raise SystemExit(f"Input file not found: {path}") return path, path.read_text(encoding="utf-8", errors="ignore").splitlines() def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_arg_parser() args = parser.parse_args(argv) input_path, lines = load_input(args) profile = ThresholdProfile( warn_response_ms=args.warn_response_ms, critical_response_ms=args.critical_response_ms, warn_iosq_ms=args.warn_iosq_ms, critical_iosq_ms=args.critical_iosq_ms, warn_pending_ms=args.warn_pending_ms, critical_pending_ms=args.critical_pending_ms, warn_device_util_pct=args.warn_util_pct, critical_device_util_pct=args.critical_util_pct, warn_connect_ms=args.warn_connect_ms, critical_connect_ms=args.critical_connect_ms, ) meta = extract_meta(lines, source_file=str(input_path), parser_mode=args.mode) records = parse_records(lines, meta=meta, mode=args.mode) result = analyze(records, meta=meta, profile=profile, top_n=args.top) if not args.quiet: print_console_report(result, top_n=args.top) if args.json_output: export_json(result, Path(args.json_output)) print(f"JSON written: {args.json_output}") if args.csv_records_output: export_csv_records(result.records, Path(args.csv_records_output)) print(f"Records CSV written: {args.csv_records_output}") if args.csv_issues_output: export_csv_issues(result.issues, Path(args.csv_issues_output)) print(f"Issues CSV written: {args.csv_issues_output}") if args.html_output: export_html(result, Path(args.html_output)) print(f"HTML written: {args.html_output}") if args.fail_on_critical and result.summary["severity_counts"].get("CRITICAL", 0) > 0: return 2 return 0 if __name__ == "__main__": raise SystemExit(main())