No data.
"
head = "".join(f"
{escape(summary.get('record_count'))}
Records
{escape(summary.get('estimated_cost'))}
Estimated cost
{escape(summary.get('cpu_seconds'))}
CPU seconds
{escape(summary.get('issue_count'))}
Issues
{escape(summary.get('applications'))}
Applications
{escape(summary.get('departments'))}
Departments
{escape(summary.get('ziip_offload_pct'))}%
zIIP offload
{escape(summary.get('abend_count'))}
ABENDs
Findings
{table(findings, finding_cols, 50)}
Top applications by estimated cost
{table(analysis.get('top_applications', []), agg_cols, 20)}
Top jobs by estimated cost
{table(analysis.get('top_jobs', []), agg_cols, 20)}
Departments / chargeback view
{table(analysis.get('top_departments', []), agg_cols, 20)}
Service classes
{table(analysis.get('top_service_classes', []), agg_cols, 20)}
Hourly activity
{table(analysis.get('hourly', []), agg_cols, 30)}
Baseline comparison
{table(compare, compare_cols, 30)}
"""
with open(path, "w", encoding="utf-8") as fh:
fh.write(html_doc)
def print_console_summary(analysis: Dict[str, Any]) -> None:
summary = analysis["summary"]
print("SMF Cost & Activity Analyzer V2")
print("-" * 72)
print(f"Profile : {analysis.get('profile')}")
print(f"Records : {summary.get('record_count')}")
print(f"Record types : {summary.get('record_types')}")
print(f"Jobs : {summary.get('jobs')}")
print(f"Users : {summary.get('users')}")
print(f"Applications : {summary.get('applications')}")
print(f"Departments : {summary.get('departments')}")
print(f"CPU seconds : {summary.get('cpu_seconds')}")
print(f"zIIP seconds : {summary.get('ziip_seconds')}")
print(f"zIIP offload % : {summary.get('ziip_offload_pct')}")
print(f"Elapsed seconds : {summary.get('elapsed_seconds')}")
print(f"EXCP count : {summary.get('excp_count')}")
print(f"DB2 getpages : {summary.get('db2_getpages')}")
print(f"API requests : {summary.get('api_requests')}")
print(f"Estimated cost : {summary.get('estimated_cost')}")
print(f"Max RC : {summary.get('max_rc')}")
print(f"ABENDs : {summary.get('abend_count')}")
print(f"Issues : {summary.get('issue_count')} (critical={summary.get('critical_count')}, error={summary.get('error_count')}, warning={summary.get('warning_count')})")
print()
print("TOP APPLICATIONS BY COST")
print("-" * 72)
for row in analysis.get("top_applications", [])[:10]:
print(f"{row['key']:<18} cost={row['estimated_cost']:<10} cpu={row['cpu_seconds']:<10} elapsed={row['elapsed_seconds']:<10} issues_rc={row['max_rc']}")
print()
print("TOP FINDINGS")
print("-" * 72)
for finding in analysis.get("findings", [])[:12]:
print(f"[{finding['severity']:<8}] {finding['code']:<28} {finding['entity_type']}={finding['entity_id']} value={finding['value']}")
def build_cost_model(args: argparse.Namespace) -> Dict[str, float]:
model = dict(DEFAULT_COST_MODEL)
for key in list(model.keys()):
value = getattr(args, key, None)
if value is not None:
model[key] = value
return model
def build_thresholds(args: argparse.Namespace) -> Dict[str, float]:
thresholds = dict(DEFAULT_THRESHOLDS)
for key in list(thresholds.keys()):
value = getattr(args, key, None)
if value is not None:
thresholds[key] = value
return thresholds
def add_cost_model_arguments(parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group("cost model")
for key, value in DEFAULT_COST_MODEL.items():
group.add_argument(f"--{key.replace('_', '-')}", dest=key, type=float, default=None, help=f"Default: {value}")
def add_threshold_arguments(parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group("thresholds")
for key, value in DEFAULT_THRESHOLDS.items():
group.add_argument(f"--{key.replace('_', '-')}", dest=key, type=float, default=None, help=f"Default: {value}")
def create_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="SMF Cost & Activity Analyzer V2 - normalized SMF CSV/JSONL analyzer")
parser.add_argument("input_file", nargs="?", help="Normalized SMF export file, CSV/JSON/JSONL")
parser.add_argument("--demo", action="store_true", help="Run with built-in demo records")
parser.add_argument("--write-sample-csv", help="Write the current demo CSV and exit")
parser.add_argument("--write-baseline-csv", help="Write the baseline demo CSV and exit")
parser.add_argument("--compare-base", help="Baseline normalized SMF export for regression comparison")
parser.add_argument("--profile", default="production", choices=["production", "batch", "api", "db2", "capacity", "chargeback", "training"], help="Analysis profile")
parser.add_argument("--redact", action="store_true", help="Redact identifying values in output")
parser.add_argument("--json", dest="json_output", help="Export full analysis JSON")
parser.add_argument("--html", dest="html_output", help="Export HTML report")
parser.add_argument("--csv-jobs", help="Export job aggregate CSV")
parser.add_argument("--csv-users", help="Export user aggregate CSV")
parser.add_argument("--csv-apps", help="Export application aggregate CSV")
parser.add_argument("--csv-departments", help="Export department aggregate CSV")
parser.add_argument("--csv-service-classes", help="Export service class aggregate CSV")
parser.add_argument("--csv-hourly", help="Export hourly aggregate CSV")
parser.add_argument("--csv-findings", help="Export findings CSV")
parser.add_argument("--csv-compare", help="Export baseline comparison CSV")
parser.add_argument("--fail-on-critical", action="store_true", help="Return non-zero exit code if critical findings exist")
parser.add_argument("--fail-on-error", action="store_true", help="Return non-zero exit code if error or critical findings exist")
add_cost_model_arguments(parser)
add_threshold_arguments(parser)
return parser
def analyze_records(records: List[SMFRecord], args: argparse.Namespace, cost_model: Dict[str, float], thresholds: Dict[str, float]) -> Dict[str, Any]:
redactor = Redactor(args.redact)
redacted_records = [redactor.record(rec) for rec in records]
analyzer = SMFCostActivityAnalyzer(redacted_records, cost_model, thresholds, args.profile)
analysis = analyzer.analyze()
if args.compare_base:
baseline_records = read_records(args.compare_base)
baseline_records = [redactor.record(rec) for rec in baseline_records]
baseline_analyzer = SMFCostActivityAnalyzer(baseline_records, cost_model, thresholds, args.profile)
baseline_analysis = baseline_analyzer.analyze()
compare_rows, compare_findings = compare_analyses(analysis, baseline_analysis, thresholds)
analysis["baseline_compare"] = compare_rows
analysis["findings"].extend([asdict(f) for f in compare_findings])
analysis["summary"]["issue_count"] = len(analysis["findings"])
analysis["summary"]["critical_count"] = sum(1 for f in analysis["findings"] if f.get("severity") == "CRITICAL")
analysis["summary"]["error_count"] = sum(1 for f in analysis["findings"] if f.get("severity") == "ERROR")
analysis["summary"]["warning_count"] = sum(1 for f in analysis["findings"] if f.get("severity") == "WARNING")
else:
analysis["baseline_compare"] = []
return analysis
def main(argv: Optional[List[str]] = None) -> int:
parser = create_arg_parser()
args = parser.parse_args(argv)
if args.write_sample_csv:
write_sample_csv(args.write_sample_csv, SAMPLE_ROWS)
print(f"Sample CSV written: {args.write_sample_csv}")
return 0
if args.write_baseline_csv:
write_sample_csv(args.write_baseline_csv, BASELINE_ROWS)
print(f"Baseline CSV written: {args.write_baseline_csv}")
return 0
if args.demo:
records = [normalize_row(row) for row in SAMPLE_ROWS]
elif args.input_file:
records = read_records(args.input_file)
else:
parser.error("Provide input_file or use --demo")
return 2
cost_model = build_cost_model(args)
thresholds = build_thresholds(args)
analysis = analyze_records(records, args, cost_model, thresholds)
print_console_summary(analysis)
if args.json_output:
write_json(args.json_output, analysis)
print(f"JSON exported: {args.json_output}")
if args.html_output:
write_html_report(args.html_output, analysis)
print(f"HTML exported: {args.html_output}")
if args.csv_jobs:
write_csv(args.csv_jobs, analysis.get("top_jobs", []))
print(f"Jobs CSV exported: {args.csv_jobs}")
if args.csv_users:
write_csv(args.csv_users, analysis.get("top_users", []))
print(f"Users CSV exported: {args.csv_users}")
if args.csv_apps:
write_csv(args.csv_apps, analysis.get("top_applications", []))
print(f"Applications CSV exported: {args.csv_apps}")
if args.csv_departments:
write_csv(args.csv_departments, analysis.get("top_departments", []))
print(f"Departments CSV exported: {args.csv_departments}")
if args.csv_service_classes:
write_csv(args.csv_service_classes, analysis.get("top_service_classes", []))
print(f"Service classes CSV exported: {args.csv_service_classes}")
if args.csv_hourly:
write_csv(args.csv_hourly, analysis.get("hourly", []))
print(f"Hourly CSV exported: {args.csv_hourly}")
if args.csv_findings:
write_csv(args.csv_findings, analysis.get("findings", []))
print(f"Findings CSV exported: {args.csv_findings}")
if args.csv_compare:
write_csv(args.csv_compare, analysis.get("baseline_compare", []))
print(f"Comparison CSV exported: {args.csv_compare}")
critical = analysis["summary"].get("critical_count", 0)
errors = analysis["summary"].get("error_count", 0)
if args.fail_on_critical and critical:
return 10
if args.fail_on_error and (critical or errors):
return 11
return 0
if __name__ == "__main__":
sys.exit(main())