github/krkn
1
0
Fork 0
mirror of https://github.com/krkn-chaos/krkn.git synced 2026-08-25 09:27:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

changed pdf library to reportlab (#1532)

Browse Source 
Signed-off-by: Teju Gangisetty <tgangise@redhat.com>
This commit is contained in:
Teju Gangisetty
2026-07-31 12:30:55 -04:00
committed by GitHub
parent 09fa54164f
commit 84b360c83b
7 changed files with 623 additions and 792 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CI/run_test.sh
+5
View File
@@ -38,3 +38,8 @@ else
# return value for run.sh
echo 1
fi
# Preserve per-test PDF report
if [ -f kraken.report.pdf ]; then
cp kraken.report.pdf "CI/out/${ci_test}.report.pdf"
fi
krkn/summarized_reports/templates/report.html
-700
View File
@@ -1,700 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>KRKN Run Summary</title>
<style>
@page {
size: A4;
margin: 1.5cm;
}
body {
font-family: "Helvetica Neue", Helvetica, Arial, sans-serif;
font-size: 10pt;
color: #1a1a1a;
line-height: 1.4;
margin: 0;
padding: 0;
}
h1 {
text-align: center;
font-size: 18pt;
margin: 0 0 4px 0;
color: #111;
}
.subtitle {
text-align: center;
font-size: 9pt;
color: #666;
margin-bottom: 16px;
}
h2 {
font-size: 12pt;
color: #222;
border-bottom: 2px solid #cc0000;
padding-bottom: 3px;
margin: 18px 0 8px 0;
}
h3 {
font-size: 10pt;
color: #333;
margin: 12px 0 4px 0;
}
table {
width: 100%;
border-collapse: collapse;
margin-bottom: 12px;
page-break-inside: auto;
}
tr {
page-break-inside: avoid;
page-break-after: auto;
}
th, td {
text-align: left;
padding: 5px 8px;
border: 1px solid #ccc;
font-size: 9pt;
}
th {
background-color: #e8e8e8;
font-weight: bold;
}
h2, h3 {
page-break-after: avoid;
}
.pass { color: #1a7f37; font-weight: bold; }
.fail { color: #cf222e; font-weight: bold; }
.score-green { color: #1a7f37; font-weight: bold; }
.score-yellow { color: #9a6700; font-weight: bold; }
.score-red { color: #cf222e; font-weight: bold; }
.badge {
display: inline-block;
padding: 2px 8px;
border-radius: 3px;
font-size: 8pt;
font-weight: bold;
color: #fff;
}
.badge-pass { background-color: #1a7f37; }
.badge-fail { background-color: #cf222e; }
.badge-warn { background-color: #9a6700; }
.meta-table td:first-child {
width: 160px;
font-weight: bold;
background-color: #f5f5f5;
}
.recovery-sub {
color: #555;
font-size: 8.5pt;
}
.overall-score {
text-align: center;
font-size: 16pt;
font-weight: bold;
padding: 10px;
margin-top: 8px;
border: 2px solid #ccc;
border-radius: 6px;
page-break-inside: avoid;
}
.meta-table {
page-break-inside: avoid;
}
.pod-list {
margin: 0;
padding-left: 16px;
}
.pod-list li {
font-size: 8.5pt;
margin-bottom: 1px;
}
.error-note {
color: #cf222e;
font-size: 8.5pt;
font-style: italic;
}
.event-meta {
color: #666;
font-size: 8pt;
}
</style>
</head>
<body>
<h1>KRKN Run Summary</h1>
<p class="subtitle">Generated {{ generated_at }}</p>
<!-- Run Metadata -->
<h2>Run Metadata</h2>
<table class="meta-table">
<tr><td>Run UUID</td><td>{{ run_uuid }}</td></tr>
<tr><td>Cluster Version</td><td>{{ cluster_version }}</td></tr>
<tr><td>Infrastructure</td><td>{{ cloud_infrastructure }}</td></tr>
<tr><td>Cloud Type</td><td>{{ cloud_type }}</td></tr>
<tr><td>Time Window</td><td>{{ time_window }}</td></tr>
<tr><td>Total Nodes</td><td>{{ total_node_count }}</td></tr>
{% if network_plugins %}
<tr><td>Network Plugins</td><td>{{ network_plugins | join(', ') }}</td></tr>
{% endif %}
{% if security_flags %}
<tr><td>Security</td><td>{{ security_flags | join(', ') }}</td></tr>
{% endif %}
</table>
<!-- Cluster Overview -->
{% if node_summary_infos %}
<h2>Cluster Overview</h2>
<table>
<tr>
<th>Type</th>
<th>Count</th>
<th>Instance</th>
<th>Architecture</th>
<th>Kubelet</th>
<th>OS</th>
</tr>
{% for ni in node_summary_infos %}
<tr>
<td>{{ ni.nodes_type or 'N/A' }}</td>
<td>{{ ni.count or 'N/A' }}</td>
<td>{{ ni.instance_type or 'N/A' }}</td>
<td>{{ ni.architecture or 'N/A' }}</td>
<td>{{ ni.kubelet_version or 'N/A' }}</td>
<td>{{ ni.os_version or 'N/A' }}</td>
</tr>
{% endfor %}
</table>
{% endif %}
<!-- Targets -->
<h2>Targets</h2>
{% for s in scenarios %}
<table>
<tr>
{% set doc_url = scenario_type_docs.get(s.scenario_type, '') %}
<th>{{ s.scenario }} ({% if doc_url %}<a href="{{ doc_url }}">{{ s.scenario_type }}</a>{% else %}{{ s.scenario_type }}{% endif %})</th>
<th style="width: 80px; text-align: right;">
{% if s.exit_status == "0" %}
<span class="badge badge-pass">PASS</span>
{% else %}
<span class="badge badge-fail">FAIL</span>
{% endif %}
</th>
</tr>
{% if s.selectors %}
<tr>
<td style="width:160px; font-weight:bold;">Label Selector</td>
<td>{{ s.selectors | join(', ') }}</td>
</tr>
{% endif %}
{% if s.namespaces %}
<tr>
<td style="font-weight:bold;">Namespace</td>
<td>{{ s.namespaces | join(', ') }}</td>
</tr>
{% endif %}
{% if s.exclude_labels %}
<tr>
<td style="font-weight:bold;">Exclude Label</td>
<td>{{ s.exclude_labels | join(', ') }}</td>
</tr>
{% endif %}
{% if s.cloud_types %}
<tr>
<td style="font-weight:bold;">Cloud Type</td>
<td>{{ s.cloud_types | join(', ') }}</td>
</tr>
{% endif %}
{% if s.all_pods %}
<tr>
<td style="font-weight:bold;">Disrupted Pods</td>
<td>
<ul class="pod-list">
{% for pod in s.all_pods %}
<li>{{ pod }}</li>
{% endfor %}
</ul>
{% if s.pods_error %}
<p class="error-note">Monitoring error: {{ s.pods_error }}</p>
{% endif %}
</td>
</tr>
{% elif s.pods_error %}
<tr>
<td style="font-weight:bold;">Pod Monitoring</td>
<td><span class="error-note">Error: {{ s.pods_error }}</span></td>
</tr>
{% endif %}
{% if s.all_vmis %}
<tr>
<td style="font-weight:bold;">Disrupted VMIs</td>
<td>
<ul class="pod-list">
{% for vmi in s.all_vmis %}
<li>{{ vmi }}</li>
{% endfor %}
</ul>
{% if s.vmis_error %}
<p class="error-note">Monitoring error: {{ s.vmis_error }}</p>
{% endif %}
</td>
</tr>
{% elif s.vmis_error %}
<tr>
<td style="font-weight:bold;">VMI Monitoring</td>
<td><span class="error-note">Error: {{ s.vmis_error }}</span></td>
</tr>
{% endif %}
{% if s.affected_nodes %}
<tr>
<td style="font-weight:bold;">Affected Nodes</td>
<td>
<ul class="pod-list">
{% for node in s.affected_nodes %}
<li>{{ node.node_name }}{% if node.node_id %} ({{ node.node_id }}){% endif %}</li>
{% endfor %}
</ul>
</td>
</tr>
{% endif %}
</table>
{% endfor %}
<!-- Key Metrics -->
{% set has_pods = scenarios | selectattr('all_pods') | list | length > 0 %}
{% if has_pods %}
<h2>Key Metrics</h2>
<table>
<tr>
<th>Scenario</th>
<th>Pods Recovered</th>
<th>Pods Unrecovered</th>
<th>Total Recovery Time</th>
</tr>
{% for s in scenarios %}
{% if s.recovered_count or s.unrecovered_count %}
<tr>
<td>{{ s.scenario }}</td>
<td>{{ s.recovered_count }}</td>
<td>{{ s.unrecovered_count }}</td>
<td>
{% if s.total_recovery_time is not none %}
{{ "%.2f"|format(s.total_recovery_time) }}s
<br><span class="recovery-sub">Rescheduling: {{ "%.2f"|format(s.rescheduling_time) }}s | Readiness: {{ "%.2f"|format(s.readiness_time) }}s</span>
{% endif %}
</td>
</tr>
{% endif %}
{% endfor %}
</table>
{% endif %}
{% set has_vmis = scenarios | selectattr('all_vmis') | list | length > 0 %}
{% if has_vmis %}
<h3>VMI Recovery</h3>
<table>
<tr>
<th>Scenario</th>
<th>VMIs Recovered</th>
<th>VMIs Unrecovered</th>
<th>Total Recovery Time</th>
</tr>
{% for s in scenarios %}
{% if s.all_vmis %}
<tr>
<td>{{ s.scenario }}</td>
<td>{{ s.vmi_recovered_count }}</td>
<td>{{ s.vmi_unrecovered_count }}</td>
<td>
{% if s.vmi_total_recovery_time is not none %}
{{ "%.2f"|format(s.vmi_total_recovery_time) }}s
<br><span class="recovery-sub">Rescheduling: {{ "%.2f"|format(s.vmi_rescheduling_time) }}s | Readiness: {{ "%.2f"|format(s.vmi_readiness_time) }}s</span>
{% endif %}
</td>
</tr>
{% endif %}
{% endfor %}
</table>
{% endif %}
{% set has_nodes = scenarios | selectattr('affected_nodes') | list | length > 0 %}
{% if has_nodes %}
<h3>Node Recovery</h3>
{% for s in scenarios %}
{% if s.affected_nodes %}
<table>
<tr>
<th colspan="{% if s.affected_nodes | selectattr('node_id') | list | length > 0 %}7{% else %}6{% endif %}">{{ s.scenario }}</th>
</tr>
<tr>
<th>Node</th>
{% if s.affected_nodes | selectattr('node_id') | list | length > 0 %}
<th>Instance ID</th>
{% endif %}
<th>Stopped</th>
<th>Running</th>
<th>Terminated</th>
<th>Not Ready</th>
<th>Ready</th>
</tr>
{% for node in s.affected_nodes %}
<tr>
<td>{{ node.node_name }}</td>
{% if s.affected_nodes | selectattr('node_id') | list | length > 0 %}
<td>{{ node.node_id }}</td>
{% endif %}
<td>{% if node.stopped_time %}{{ "%.2f"|format(node.stopped_time) }}s{% endif %}</td>
<td>{% if node.running_time %}{{ "%.2f"|format(node.running_time) }}s{% endif %}</td>
<td>{% if node.terminating_time %}{{ "%.2f"|format(node.terminating_time) }}s{% endif %}</td>
<td>{% if node.not_ready_time %}{{ "%.2f"|format(node.not_ready_time) }}s{% endif %}</td>
<td>{% if node.ready_time %}{{ "%.2f"|format(node.ready_time) }}s{% endif %}</td>
</tr>
{% endfor %}
</table>
{% endif %}
{% endfor %}
{% endif %}
<!-- Additional Telemetry (HTTP Load Test) -->
{% set has_additional = scenarios | selectattr('additional_telemetry') | list | length > 0 %}
{% if has_additional %}
<h3>Load Test Metrics</h3>
{% for s in scenarios %}
{% if s.additional_telemetry %}
<table>
<tr><th colspan="2">{{ s.scenario }}</th></tr>
{% for key, val in s.additional_telemetry.items() %}
<tr>
<td style="width: 200px; font-weight: bold;">{{ key }}</td>
<td>{{ val }}</td>
</tr>
{% endfor %}
</table>
{% endif %}
{% endfor %}
{% endif %}
<!-- Cluster Events (per scenario) -->
{% set has_events = scenarios | selectattr('cluster_events') | list | length > 0 %}
{% if has_events %}
<h3>Cluster Events</h3>
{% for s in scenarios %}
{% if s.cluster_events %}
<table>
<tr>
<th colspan="5">{{ s.scenario }} ({{ s.cluster_events | length }} events)</th>
</tr>
<tr>
<th>Type</th>
<th>Reason</th>
<th>Object</th>
<th>Message</th>
<th>Namespace</th>
</tr>
{% for event in s.cluster_events[:10] %}
<tr>
{% if event is mapping %}
<td>
{% if event.type == 'Warning' %}
<span class="badge badge-warn">{{ event.type }}</span>
{% elif event.type %}
{{ event.type }}
{% endif %}
</td>
<td>{{ event.reason or '' }}</td>
<td>{% if event.involved_object_kind %}{{ event.involved_object_kind }}/{{ event.involved_object_name }}{% endif %}</td>
<td>{{ event.message or '' }}</td>
<td>{{ event.namespace or '' }}</td>
{% else %}
<td colspan="5">{{ event }}</td>
{% endif %}
</tr>
{% endfor %}
{% if s.cluster_events | length > 10 %}
<tr><td colspan="5" style="color: #888; font-size: 8pt;">... and {{ s.cluster_events | length - 10 }} more</td></tr>
{% endif %}
</table>
{% endif %}
{% endfor %}
{% endif %}
<!-- Health Checks -->
{% if health_checks %}
<h2>Health Checks</h2>
<table>
<tr>
<th>URL / Endpoint</th>
<th>Status Code</th>
<th>Duration</th>
<th>Result</th>
</tr>
{% for check in health_checks %}
<tr>
{% if check is mapping %}
<td>{{ check.url or check.name or check.check_name or '' }}</td>
<td>{{ check.status_code or '' }}</td>
<td>{% if check.duration is not none and check.duration != '' %}{{ "%.2f"|format(check.duration|float) }}s{% endif %}</td>
<td>
{% if check.status or check.passed %}
<span class="badge badge-pass">PASS</span>
{% else %}
<span class="badge badge-fail">FAIL</span>
{% endif %}
</td>
{% else %}
<td colspan="4">{{ check }}</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% endif %}
<!-- KubeVirt Health Checks (pre-chaos) -->
{% if virt_checks %}
<h2>KubeVirt Health Checks (Pre-Chaos)</h2>
<table>
<tr>
<th>VM Name</th>
<th>Namespace</th>
<th>Node</th>
<th>IP Address</th>
<th>Duration</th>
<th>Result</th>
</tr>
{% for check in virt_checks %}
<tr>
{% if check is mapping %}
<td>{{ check.vm_name or check.vmi_name or check.name or '' }}</td>
<td>{{ check.namespace or '' }}</td>
<td>{{ check.node_name or '' }}</td>
<td>{{ check.ip_address or '' }}</td>
<td>{% if check.duration is not none and check.duration != '' %}{{ "%.2f"|format(check.duration|float) }}s{% endif %}</td>
<td>
{% if check.status is defined and not check.status %}
<span class="badge badge-fail">FAIL</span>
{% else %}
<span class="badge badge-pass">PASS</span>
{% endif %}
</td>
{% else %}
<td colspan="6">{{ check }}</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% endif %}
<!-- KubeVirt Health Checks (post-chaos) -->
{% if post_virt_checks %}
<h2>KubeVirt Health Checks (Post-Chaos)</h2>
<table>
<tr>
<th>VM Name</th>
<th>Namespace</th>
<th>Node</th>
<th>IP Address</th>
<th>New IP</th>
<th>Duration</th>
<th>Result</th>
</tr>
{% for check in post_virt_checks %}
<tr>
{% if check is mapping %}
<td>{{ check.vm_name or check.vmi_name or check.name or '' }}</td>
<td>{{ check.namespace or '' }}</td>
<td>{{ check.node_name or '' }}</td>
<td>{{ check.ip_address or '' }}</td>
<td>{% if check.new_ip_address and check.new_ip_address != check.ip_address %}{{ check.new_ip_address }}{% endif %}</td>
<td>{% if check.duration is not none and check.duration != '' %}{{ "%.2f"|format(check.duration|float) }}s{% endif %}</td>
<td>
{% if check.status is defined and not check.status %}
<span class="badge badge-fail">FAIL</span>
{% else %}
<span class="badge badge-pass">PASS</span>
{% endif %}
</td>
{% else %}
<td colspan="7">{{ check }}</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% endif %}
<!-- Alerts & SLOs -->
<h2>Alerts &amp; SLOs</h2>
<table class="meta-table">
<tr><td>SLOs Evaluated</td><td>{{ total_slos }}</td></tr>
<tr>
<td>SLOs Passed</td>
<td>{{ passed_slos }} / {{ total_slos }}</td>
</tr>
<tr>
<td>SLOs Failed</td>
<td>{% if failed_slos > 0 %}<span class="fail">{{ failed_slos }}</span>{% else %}{{ failed_slos }}{% endif %}</td>
</tr>
<tr>
<td>Critical Alerts</td>
<td>{% if critical_alert_count > 0 %}<span class="fail">{{ critical_alert_count }}</span>{% else %}None{% endif %}</td>
</tr>
</table>
<!-- Critical Alert Details -->
{% if chaos_alerts %}
<h3>Critical Alerts (During Chaos)</h3>
<table>
<tr>
<th>Alert Name</th>
<th>Severity</th>
<th>Namespace</th>
<th>State</th>
</tr>
{% for alert in chaos_alerts %}
<tr>
{% if alert is mapping %}
<td>{{ alert.alertname or 'N/A' }}</td>
<td>{{ alert.severity or 'N/A' }}</td>
<td>{{ alert.namespace or 'N/A' }}</td>
<td>{{ alert.alertstate or 'N/A' }}</td>
{% else %}
<td colspan="4">{{ alert }}</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% endif %}
{% if post_chaos_alerts %}
<h3>Critical Alerts (Post Chaos)</h3>
<table>
<tr>
<th>Alert Name</th>
<th>Severity</th>
<th>Namespace</th>
<th>State</th>
</tr>
{% for alert in post_chaos_alerts %}
<tr>
{% if alert is mapping %}
<td>{{ alert.alertname or 'N/A' }}</td>
<td>{{ alert.severity or 'N/A' }}</td>
<td>{{ alert.namespace or 'N/A' }}</td>
<td>{{ alert.alertstate or 'N/A' }}</td>
{% else %}
<td colspan="4">{{ alert }}</td>
{% endif %}
</tr>
{% endfor %}
</table>
{% endif %}
<!-- Error Logs -->
{% if error_logs %}
<h3>Error Logs ({{ error_logs | length }})</h3>
<table>
<tr>
<th style="width: 160px;">Timestamp</th>
<th>Message</th>
</tr>
{% for log in error_logs[:20] %}
<tr>
{% if log is mapping %}
<td>{{ log.timestamp or '-' }}</td>
<td>{{ log.message or log }}</td>
{% else %}
<td>-</td>
<td>{{ log }}</td>
{% endif %}
</tr>
{% endfor %}
{% if error_logs | length > 20 %}
<tr><td colspan="2" style="color: #888; font-size: 8pt;">... and {{ error_logs | length - 20 }} more</td></tr>
{% endif %}
</table>
{% endif %}
<!-- Failed SLOs -->
{% set failed_entries = [] %}
{% for entry in scenario_slo_details %}
{% set failed_slos_list = entry.slo_details | selectattr('passed', 'false') | list %}
{% if failed_slos_list %}
{% set _ = failed_entries.append({"scenario": entry.scenario, "slo_details": failed_slos_list}) %}
{% endif %}
{% endfor %}
{% if failed_entries %}
<h2>Failed SLOs</h2>
{% for entry in failed_entries %}
<h3>{{ entry.scenario }}</h3>
<table>
<tr>
<th>SLO</th>
<th>Severity</th>
<th>Status</th>
</tr>
{% for slo in entry.slo_details %}
<tr>
<td>{{ slo.name }}</td>
<td>{{ slo.severity }}</td>
<td><span class="badge badge-fail">FAIL</span></td>
</tr>
{% endfor %}
</table>
{% endfor %}
{% endif %}
<!-- Resiliency Score -->
<h2>Resiliency Score</h2>
{% if per_scenario_scores %}
<table>
<tr>
<th>Scenario</th>
<th>Score</th>
</tr>
{% for name, score in per_scenario_scores.items() %}
<tr>
<td>{{ name }}</td>
<td>
{% if score >= 90 %}
<span class="score-green">{{ score }} / 100</span>
{% elif score >= 70 %}
<span class="score-yellow">{{ score }} / 100</span>
{% else %}
<span class="score-red">{{ score }} / 100</span>
{% endif %}
</td>
</tr>
{% endfor %}
</table>
{% endif %}
<div class="overall-score
{%- if overall_score is number and overall_score >= 90 %} score-green
{%- elif overall_score is number and overall_score >= 70 %} score-yellow
{%- elif overall_score is number %} score-red
{%- endif %}">
Overall: {{ overall_score }} / 100
</div>
</body>
</html>
krkn/summarized_reports/transform.py
+516 -56
View File
@@ -2,8 +2,15 @@ import logging
import os
from datetime import datetime
from pathlib import Path
from xml.sax.saxutils import escape as _xml_escape
from jinja2 import Environment, FileSystemLoader
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import cm
from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
from reportlab.platypus.flowables import HRFlowable
SCENARIO_TYPE_DOCS = {
"pod_disruption_scenarios": "https://krkn-chaos.dev/docs/scenarios/pod-disruption/",
@@ -27,6 +34,142 @@ SCENARIO_TYPE_DOCS = {
"storage_throttle_scenarios": "https://krkn-chaos.dev/docs/scenarios/storage-throttle/",
}
# --- ReportLab PDF constants ---
_RED_HEADER = colors.HexColor("#cc0000")
_PASS_GREEN = colors.HexColor("#1a7f37")
_FAIL_RED = colors.HexColor("#cf222e")
_WARN_YELLOW = colors.HexColor("#9a6700")
_HEADER_BG = colors.HexColor("#e8e8e8")
_BORDER_COLOR = colors.HexColor("#cccccc")
_TEXT_COLOR = colors.HexColor("#1a1a1a")
_STYLES = getSampleStyleSheet()
_STYLE_TITLE = ParagraphStyle(
"ReportTitle", parent=_STYLES["Title"],
fontSize=18, alignment=TA_CENTER, textColor=_TEXT_COLOR, spaceAfter=4,
)
_STYLE_SUBTITLE = ParagraphStyle(
"ReportSubtitle", parent=_STYLES["Normal"],
fontSize=9, alignment=TA_CENTER, textColor=colors.HexColor("#666666"),
spaceAfter=16,
)
_STYLE_H2 = ParagraphStyle(
"SectionHeader", parent=_STYLES["Heading2"],
fontSize=12, textColor=colors.HexColor("#222222"),
spaceBefore=18, spaceAfter=4,
)
_STYLE_H3 = ParagraphStyle(
"SubSectionHeader", parent=_STYLES["Heading3"],
fontSize=10, textColor=colors.HexColor("#333333"),
spaceBefore=12, spaceAfter=4,
)
_STYLE_CELL = ParagraphStyle(
"CellText", parent=_STYLES["Normal"],
fontSize=9, leading=11, textColor=_TEXT_COLOR,
)
_STYLE_CELL_BOLD = ParagraphStyle(
"CellTextBold", parent=_STYLE_CELL, fontName="Helvetica-Bold",
)
_STYLE_CELL_SMALL = ParagraphStyle(
"CellTextSmall", parent=_STYLES["Normal"],
fontSize=7, leading=9, textColor=_TEXT_COLOR,
)
_STYLE_OVERALL = ParagraphStyle(
"OverallScore", parent=_STYLES["Normal"],
fontSize=16, alignment=TA_CENTER, fontName="Helvetica-Bold",
borderWidth=2, borderColor=_BORDER_COLOR, borderPadding=10,
spaceBefore=8,
)
def _p(text, style=None):
return Paragraph(_xml_escape(str(text)), style or _STYLE_CELL)
def _section_header(title):
return [
Spacer(1, 10),
Paragraph(_xml_escape(title), _STYLE_H2),
HRFlowable(width="100%", thickness=2, color=_RED_HEADER, spaceAfter=8),
]
def _subsection_header(title):
return [Paragraph(_xml_escape(title), _STYLE_H3)]
def _badge(text, passed):
c = _PASS_GREEN if passed else _FAIL_RED
return Paragraph(f'<font color="{c}"><b>{_xml_escape(text)}</b></font>', _STYLE_CELL)
def _score_color(score):
if isinstance(score, (int, float)):
if score >= 90:
return _PASS_GREEN
if score >= 70:
return _WARN_YELLOW
return _FAIL_RED
return _TEXT_COLOR
def _make_kv_table(rows):
if not rows:
return []
avail = A4[0] - 3 * cm
data = []
for k, v in rows:
k_cell = _p(k, _STYLE_CELL_BOLD) if not isinstance(k, Paragraph) else k
v_cell = _p(v) if not isinstance(v, Paragraph) else v
data.append([k_cell, v_cell])
t = Table(data, colWidths=[avail * 0.35, avail * 0.65])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (0, -1), _HEADER_BG),
("GRID", (0, 0), (-1, -1), 0.5, _BORDER_COLOR),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
("LEFTPADDING", (0, 0), (-1, -1), 8),
("RIGHTPADDING", (0, 0), (-1, -1), 8),
]))
return [t, Spacer(1, 12)]
def _make_data_table(headers, rows, col_widths=None, small=False, span_header=None):
cell_style = _STYLE_CELL_SMALL if small else _STYLE_CELL
def _cell(val):
if isinstance(val, Paragraph):
return val
return _p(str(val), cell_style)
data = []
if span_header:
data.append([_p(span_header, _STYLE_CELL_BOLD)])
data.append([_p(h, _STYLE_CELL_BOLD if not small else cell_style) for h in headers])
for row in rows:
data.append([_cell(v) for v in row])
repeat = 2 if span_header else 1
t = Table(data, colWidths=col_widths, repeatRows=repeat)
cmds = [
("GRID", (0, 0), (-1, -1), 0.5, _BORDER_COLOR),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 5 if not small else 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 5 if not small else 3),
("LEFTPADDING", (0, 0), (-1, -1), 8 if not small else 4),
("RIGHTPADDING", (0, 0), (-1, -1), 8 if not small else 4),
]
header_row = 0
if span_header:
cmds.append(("SPAN", (0, 0), (-1, 0)))
cmds.append(("BACKGROUND", (0, 0), (-1, 0), _HEADER_BG))
header_row = 1
cmds.append(("BACKGROUND", (0, header_row), (-1, header_row), _HEADER_BG))
t.setStyle(TableStyle(cmds))
return [t, Spacer(1, 12)]
def format_ts(unix_ts):
return datetime.fromtimestamp(unix_ts).strftime("%Y-%m-%d %H:%M:%S")
@@ -493,9 +636,6 @@ def build_chaos_report(chaos_output: dict) -> str:
def build_chaos_report_pdf(chaos_output: dict, output_path: str) -> str:
from weasyprint import HTML
logging.getLogger("weasyprint").setLevel(logging.WARNING)
telemetry = chaos_output.get("telemetry", {})
scenarios_raw = telemetry.get("scenarios", [])
@@ -617,10 +757,6 @@ def build_chaos_report_pdf(chaos_output: dict, output_path: str) -> str:
error_logs = telemetry.get("error_logs") or []
template_dir = Path(__file__).parent / "templates"
env = Environment(loader=FileSystemLoader(str(template_dir)), autoescape=True)
template = env.get_template("report.html")
scenario_slo_details = chaos_output.get("scenario_slo_details", [])
security_flags = []
@@ -635,55 +771,379 @@ def build_chaos_report_pdf(chaos_output: dict, output_path: str) -> str:
health_checks = telemetry.get("health_checks")
virt_checks = telemetry.get("virt_checks")
post_virt_checks = telemetry.get("post_virt_checks")
failed_slos = total_slos - passed_slos
per_scenario_scores = resiliency.get("scenarios", {})
overall_score = resiliency.get("resiliency_score", "N/A")
SCENARIO_TYPE_DOCS = {
"hog_scenarios": "https://krkn-chaos.dev/docs/scenarios/hog-scenarios/",
"application_outages_scenarios": "https://krkn-chaos.dev/docs/scenarios/application-outages/",
"container_scenarios": "https://krkn-chaos.dev/docs/scenarios/container-scenarios/",
"pod_network_scenarios": "https://krkn-chaos.dev/docs/scenarios/pod-network-scenario/",
"pod_disruption_scenarios": "https://krkn-chaos.dev/docs/scenarios/service-disruption-scenarios/",
"node_scenarios": "https://krkn-chaos.dev/docs/scenarios/node-scenarios/",
"time_scenarios": "https://krkn-chaos.dev/docs/scenarios/time-scenarios/",
"cluster_shut_down_scenarios": "https://krkn-chaos.dev/docs/scenarios/power-outage-scenarios/",
"service_disruption_scenarios": "https://krkn-chaos.dev/docs/scenarios/service-disruption-scenarios/",
"zone_outages_scenarios": "https://krkn-chaos.dev/docs/scenarios/zone-outage-scenarios/",
"pvc_scenarios": "https://krkn-chaos.dev/docs/scenarios/pvc-scenario/",
"storage_throttle_scenarios": "https://krkn-chaos.dev/docs/scenarios/storage-throttle-scenario/",
"network_chaos_scenarios": "https://krkn-chaos.dev/docs/scenarios/network-chaos-scenario/",
"service_hijacking_scenarios": "https://krkn-chaos.dev/docs/scenarios/service-hijacking-scenario/",
"syn_flood_scenarios": "https://krkn-chaos.dev/docs/scenarios/syn-flood-scenario/",
"network_chaos_ng_scenarios": "https://krkn-chaos.dev/docs/scenarios/network-chaos-ng-scenarios/",
"kubevirt_vm_outage": "https://krkn-chaos.dev/docs/scenarios/kubevirt-vm-outage-scenario/",
"http_load_scenarios": "https://krkn-chaos.dev/docs/scenarios/http-load-scenario/"
}
html_content = template.render(
generated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
run_uuid=telemetry.get("run_uuid", "N/A"),
cluster_version=telemetry.get("cluster_version", "N/A"),
cloud_infrastructure=telemetry.get("cloud_infrastructure", "N/A"),
cloud_type=telemetry.get("cloud_type", "N/A"),
time_window=time_window,
total_node_count=telemetry.get("total_node_count", "N/A"),
network_plugins=telemetry.get("network_plugins") or [],
security_flags=security_flags,
node_summary_infos=node_infos,
health_checks=health_checks,
virt_checks=virt_checks,
post_virt_checks=post_virt_checks,
scenarios=scenarios,
total_slos=total_slos,
passed_slos=passed_slos,
failed_slos=total_slos - passed_slos,
scenario_slo_details=scenario_slo_details,
critical_alert_count=total_alert_count,
chaos_alerts=chaos_alerts,
post_chaos_alerts=post_chaos_alerts,
error_logs=error_logs,
per_scenario_scores=resiliency.get("scenarios", {}),
overall_score=resiliency.get("resiliency_score", "N/A"),
scenario_type_docs=SCENARIO_TYPE_DOCS
avail = A4[0] - 3 * cm
doc = SimpleDocTemplate(
output_path, pagesize=A4,
leftMargin=1.5 * cm, rightMargin=1.5 * cm,
topMargin=1.5 * cm, bottomMargin=1.5 * cm,
)
f = []
HTML(string=html_content).write_pdf(output_path)
# 1. Title
f.append(Paragraph("KRKN Run Summary", _STYLE_TITLE))
f.append(Paragraph(
f"Generated {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
_STYLE_SUBTITLE,
))
# 2. Run Metadata
f.extend(_section_header("Run Metadata"))
meta_rows = [
("Run UUID", telemetry.get("run_uuid", "N/A")),
("Cluster Version", telemetry.get("cluster_version", "N/A")),
("Infrastructure", telemetry.get("cloud_infrastructure", "N/A")),
("Cloud Type", telemetry.get("cloud_type", "N/A")),
("Time Window", time_window),
("Total Nodes", str(telemetry.get("total_node_count", "N/A"))),
]
network_plugins = telemetry.get("network_plugins") or []
if network_plugins:
meta_rows.append(("Network Plugins", ", ".join(network_plugins)))
if security_flags:
meta_rows.append(("Security", ", ".join(security_flags)))
f.extend(_make_kv_table(meta_rows))
# 3. Cluster Overview
if node_infos:
f.extend(_section_header("Cluster Overview"))
rows = []
for ni in node_infos:
rows.append([
ni.get("nodes_type", "N/A"),
str(ni.get("count", "N/A")),
ni.get("instance_type", "N/A"),
ni.get("architecture", "N/A"),
ni.get("kubelet_version", "N/A"),
ni.get("os_version", "N/A"),
])
f.extend(_make_data_table(
["Type", "Count", "Instance", "Architecture", "Kubelet", "OS"], rows,
))
# 4. Targets
f.extend(_section_header("Targets"))
for s in scenarios:
passed = s["exit_status"] == "0"
scenario_label = _xml_escape(s["scenario"])
type_label = _xml_escape(s["scenario_type"])
doc_url = SCENARIO_TYPE_DOCS.get(s["scenario_type"], "")
if doc_url:
type_link = f'<a href="{_xml_escape(doc_url)}" color="blue"><u>{type_label}</u></a>'
else:
type_link = type_label
target_rows = [
(Paragraph(f"{scenario_label} ({type_link})", _STYLE_CELL),
_badge("PASS" if passed else "FAIL", passed)),
]
if s["selectors"]:
target_rows.append(("Label Selector", ", ".join(s["selectors"])))
if s["namespaces"]:
target_rows.append(("Namespace", ", ".join(s["namespaces"])))
if s["exclude_labels"]:
target_rows.append(("Exclude Label", ", ".join(s["exclude_labels"])))
if s["cloud_types"]:
target_rows.append(("Cloud Type", ", ".join(s["cloud_types"])))
if s["all_pods"]:
pod_text = "<br/>".join(_xml_escape(p) for p in s["all_pods"])
if s["pods_error"]:
pod_text += f'<br/><i><font color="{_FAIL_RED}">Monitoring error: {_xml_escape(s["pods_error"])}</font></i>'
target_rows.append(("Disrupted Pods", Paragraph(pod_text, _STYLE_CELL)))
elif s.get("pods_error"):
target_rows.append(("Pod Monitoring",
Paragraph(f'<font color="{_FAIL_RED}">Error: {_xml_escape(s["pods_error"])}</font>', _STYLE_CELL)))
if s["all_vmis"]:
vmi_text = "<br/>".join(_xml_escape(v) for v in s["all_vmis"])
if s.get("vmis_error"):
vmi_text += f'<br/><i><font color="{_FAIL_RED}">Monitoring error: {_xml_escape(s["vmis_error"])}</font></i>'
target_rows.append(("Disrupted VMIs", Paragraph(vmi_text, _STYLE_CELL)))
elif s.get("vmis_error"):
target_rows.append(("VMI Monitoring",
Paragraph(f'<font color="{_FAIL_RED}">Error: {_xml_escape(s["vmis_error"])}</font>', _STYLE_CELL)))
if s["affected_nodes"]:
node_lines = []
for n in s["affected_nodes"]:
label = n["node_name"]
if n.get("node_id"):
label += f" ({n['node_id']})"
node_lines.append(_xml_escape(label))
target_rows.append(("Affected Nodes", Paragraph("<br/>".join(node_lines), _STYLE_CELL)))
f.extend(_make_kv_table(target_rows))
# 5. Key Metrics (pod recovery)
has_pods = any(s["recovered_count"] or s["unrecovered_count"] for s in scenarios)
if has_pods:
f.extend(_section_header("Key Metrics"))
rows = []
for s in scenarios:
if not (s["recovered_count"] or s["unrecovered_count"]):
continue
recovery_cell = ""
if s["total_recovery_time"] is not None:
rt = f'{s["total_recovery_time"]:.2f}s'
rt += f'<br/><font size="7" color="#555555">Rescheduling: {(s["rescheduling_time"] or 0):.2f}s<br/>Readiness: {(s["readiness_time"] or 0):.2f}s</font>'
recovery_cell = Paragraph(rt, _STYLE_CELL)
rows.append([
s["scenario"],
str(s["recovered_count"]),
str(s["unrecovered_count"]),
recovery_cell or "",
])
f.extend(_make_data_table(
["Scenario", "Pods Recovered", "Pods Unrecovered", "Total Recovery Time"],
rows,
))
# 6. VMI Recovery
has_vmis = any(s["all_vmis"] for s in scenarios)
if has_vmis:
f.extend(_subsection_header("VMI Recovery"))
rows = []
for s in scenarios:
if not s["all_vmis"]:
continue
recovery_cell = ""
if s["vmi_total_recovery_time"] is not None:
rt = f'{s["vmi_total_recovery_time"]:.2f}s'
rt += f'<br/><font size="7" color="#555555">Rescheduling: {(s["vmi_rescheduling_time"] or 0):.2f}s<br/>Readiness: {(s["vmi_readiness_time"] or 0):.2f}s</font>'
recovery_cell = Paragraph(rt, _STYLE_CELL)
rows.append([
s["scenario"],
str(s["vmi_recovered_count"]),
str(s["vmi_unrecovered_count"]),
recovery_cell or "",
])
f.extend(_make_data_table(
["Scenario", "VMIs Recovered", "VMIs Unrecovered", "Total Recovery Time"],
rows,
))
# 7. Node Recovery
has_nodes = any(s["affected_nodes"] for s in scenarios)
if has_nodes:
f.extend(_subsection_header("Node Recovery"))
for s in scenarios:
if not s["affected_nodes"]:
continue
has_id = any(n.get("node_id") for n in s["affected_nodes"])
if has_id:
headers = ["Node", "Instance", "Stopped", "Running", "Terminated", "Not Ready", "Ready"]
cw = [avail * 0.26, avail * 0.15, avail * 0.10, avail * 0.10, avail * 0.13, avail * 0.13, avail * 0.13]
else:
headers = ["Node", "Stopped", "Running", "Terminated", "Not Ready", "Ready"]
cw = [avail * 0.40, avail * 0.12, avail * 0.12, avail * 0.12, avail * 0.12, avail * 0.12]
rows = []
for n in s["affected_nodes"]:
row = [n["node_name"]]
if has_id:
row.append(n.get("node_id", ""))
for key in ["stopped_time", "running_time", "terminating_time", "not_ready_time", "ready_time"]:
val = n.get(key)
row.append(f"{val:.2f}s" if val else "")
rows.append(row)
f.extend(_make_data_table(headers, rows, col_widths=cw, small=True, span_header=s["scenario"]))
# 8. Load Test Metrics
has_additional = any(s.get("additional_telemetry") for s in scenarios)
if has_additional:
f.extend(_subsection_header("Load Test Metrics"))
for s in scenarios:
if not s.get("additional_telemetry"):
continue
rows = [(k, str(v)) for k, v in s["additional_telemetry"].items()]
f.extend(_subsection_header(s["scenario"]))
f.extend(_make_kv_table(rows))
# 9. Cluster Events
has_events = any(s["cluster_events"] for s in scenarios)
if has_events:
f.extend(_subsection_header("Cluster Events"))
for s in scenarios:
if not s["cluster_events"]:
continue
events = s["cluster_events"][:10]
rows = []
for e in events:
obj_ref = ""
if e.get("involved_object_kind"):
obj_ref = f'{e["involved_object_kind"]}/{e.get("involved_object_name", "")}'
type_cell = e.get("type", "")
if type_cell == "Warning":
type_cell = Paragraph(f'<font color="{_WARN_YELLOW}"><b>Warning</b></font>', _STYLE_CELL)
rows.append([
type_cell,
e.get("reason", ""),
obj_ref,
e.get("message", ""),
e.get("namespace", ""),
])
f.extend(_make_data_table(
["Type", "Reason", "Object", "Message", "Namespace"],
rows,
span_header=f'{s["scenario"]} ({len(s["cluster_events"])} events)',
))
if len(s["cluster_events"]) > 10:
f.append(_p(f"... and {len(s['cluster_events']) - 10} more"))
# 10. Health Checks
if health_checks:
f.extend(_section_header("Health Checks"))
rows = []
for check in health_checks:
if isinstance(check, dict):
url = check.get("url") or check.get("name") or check.get("check_name", "")
status_code = str(check.get("status_code", ""))
duration = ""
if check.get("duration") is not None and check.get("duration") != "":
duration = f"{float(check['duration']):.2f}s"
passed = check.get("status") or check.get("passed")
rows.append([url, status_code, duration, _badge("PASS" if passed else "FAIL", bool(passed))])
else:
rows.append([str(check), "", "", ""])
f.extend(_make_data_table(
["URL / Endpoint", "Status Code", "Duration", "Result"], rows,
))
# 11. KubeVirt Health Checks (Pre-Chaos)
if virt_checks:
f.extend(_section_header("KubeVirt Health Checks (Pre-Chaos)"))
rows = []
for check in virt_checks:
if isinstance(check, dict):
passed = not (check.get("status") is not None and not check.get("status"))
rows.append([
check.get("vm_name") or check.get("vmi_name") or check.get("name", ""),
check.get("namespace", ""),
check.get("node_name", ""),
check.get("ip_address", ""),
f"{float(check['duration']):.2f}s" if check.get("duration") not in (None, "") else "",
_badge("PASS" if passed else "FAIL", passed),
])
else:
rows.append([str(check), "", "", "", "", ""])
f.extend(_make_data_table(
["VM Name", "Namespace", "Node", "IP Address", "Duration", "Result"], rows,
))
# 12. KubeVirt Health Checks (Post-Chaos)
if post_virt_checks:
f.extend(_section_header("KubeVirt Health Checks (Post-Chaos)"))
rows = []
for check in post_virt_checks:
if isinstance(check, dict):
passed = not (check.get("status") is not None and not check.get("status"))
new_ip = ""
if check.get("new_ip_address") and check.get("new_ip_address") != check.get("ip_address"):
new_ip = check["new_ip_address"]
rows.append([
check.get("vm_name") or check.get("vmi_name") or check.get("name", ""),
check.get("namespace", ""),
check.get("node_name", ""),
check.get("ip_address", ""),
new_ip,
f"{float(check['duration']):.2f}s" if check.get("duration") not in (None, "") else "",
_badge("PASS" if passed else "FAIL", passed),
])
else:
rows.append([str(check), "", "", "", "", "", ""])
f.extend(_make_data_table(
["VM Name", "Namespace", "Node", "IP Address", "New IP", "Duration", "Result"], rows,
))
# 13. Alerts & SLOs
f.extend(_section_header("Alerts & SLOs"))
failed_slo_val = _p(str(failed_slos)) if failed_slos == 0 else Paragraph(
f'<font color="{_FAIL_RED}"><b>{failed_slos}</b></font>', _STYLE_CELL)
alert_val = _p("None") if total_alert_count == 0 else Paragraph(
f'<font color="{_FAIL_RED}"><b>{total_alert_count}</b></font>', _STYLE_CELL)
f.extend(_make_kv_table([
("SLOs Evaluated", str(total_slos)),
("SLOs Passed", f"{passed_slos} / {total_slos}"),
("SLOs Failed", failed_slo_val),
("Critical Alerts", alert_val),
]))
def _build_alert_table(title, alerts):
if not alerts:
return
f.extend(_subsection_header(title))
rows = []
for alert in alerts:
if isinstance(alert, dict):
rows.append([
alert.get("alertname", "N/A"),
alert.get("severity", "N/A"),
alert.get("namespace", "N/A"),
alert.get("alertstate", "N/A"),
])
else:
rows.append([str(alert), "", "", ""])
f.extend(_make_data_table(["Alert Name", "Severity", "Namespace", "State"], rows))
_build_alert_table("Critical Alerts (During Chaos)", chaos_alerts)
_build_alert_table("Critical Alerts (Post Chaos)", post_chaos_alerts)
# 14. Error Logs
if error_logs:
f.extend(_subsection_header(f"Error Logs ({len(error_logs)})"))
rows = []
for log_entry in error_logs[:20]:
if isinstance(log_entry, dict):
rows.append([
log_entry.get("timestamp", "-"),
(log_entry.get("message") or str(log_entry))[:300],
])
else:
rows.append(["-", str(log_entry)[:300]])
f.extend(_make_data_table(
["Timestamp", "Message"], rows,
col_widths=[avail * 0.25, avail * 0.75],
))
if len(error_logs) > 20:
f.append(_p(f"... and {len(error_logs) - 20} more"))
# 15. Failed SLOs
if scenario_slo_details:
failed_entries = []
for entry in scenario_slo_details:
failed = [s for s in entry.get("slo_details", []) if not s["passed"]]
if failed:
failed_entries.append({"scenario": entry["scenario"], "slo_details": failed})
if failed_entries:
f.extend(_section_header("Failed SLOs"))
for entry in failed_entries:
f.extend(_subsection_header(entry["scenario"]))
rows = []
for slo in entry["slo_details"]:
rows.append([slo["name"], slo.get("severity", "unknown"), _badge("FAIL", False)])
f.extend(_make_data_table(["SLO", "Severity", "Status"], rows))
# 16. Resiliency Score
f.extend(_section_header("Resiliency Score"))
if per_scenario_scores:
rows = []
for name, score in per_scenario_scores.items():
c = _score_color(score)
rows.append([
name,
Paragraph(f'<font color="{c}"><b>{score} / 100</b></font>', _STYLE_CELL),
])
f.extend(_make_data_table(["Scenario", "Score"], rows))
c = _score_color(overall_score)
overall_style = ParagraphStyle(
"OverallScoreBox", parent=_STYLE_OVERALL,
textColor=c,
)
f.append(Paragraph(f"Overall: {_xml_escape(str(overall_score))} / 100", overall_style))
doc.build(f)
return output_path
requirements.txt
+3 -3
View File
@@ -1,3 +1,5 @@
setuptools<82.0.0 # Has pkg_resources (required by VMware SDK) + newer vendored jaraco-context
wheel>=0.46.2 # Fixes GHSA-8rrh-rw8j-w5fx
aliyun-python-sdk-core==2.13.36
aliyun-python-sdk-ecs==4.24.25
arcaflow-plugin-sdk==0.14.3
@@ -16,7 +18,7 @@ ibm-cloud-sdk-core>=3.24.4 # Requires requests>=2.32.4
ibm_vpc==0.26.3 # Requires ibm_cloud_sdk_core
jinja2==3.1.6
jaraco-context>=6.1.0 # Fixes GHSA-58pv-8j8x-9vj2
weasyprint>=63.0
reportlab>=4.0
cbor2<5.7.0 # Pinned by arcaflow-plugin-sdk
lxml==6.1.0
kubernetes>=35.0.0,<36.0.0
@@ -35,8 +37,6 @@ requests>=2.32.4 # Fixes GHSA-9hjg-9r4m-mvj7, GHSA-9wx4-h78v-vm56, GHSA-gc5v-m9
urllib3>=2.7.0 # Fixes GHSA-qccp-gfcp-xxvc, GHSA-38jv-5279-wg99, GHSA-gm62-xv2j-4w53, GHSA-2xpw-w6gg-jr37
service_identity==24.1.0
PyYAML==6.0.1
setuptools<82.0.0 # Has pkg_resources (required by VMware SDK) + newer vendored jaraco-context
wheel>=0.46.2 # Fixes GHSA-8rrh-rw8j-w5fx
colorlog==6.10.1
git+https://github.com/vmware/vsphere-automation-sdk-python.git@v8.0.0.0
run_kraken.py
+2 -2
View File
@@ -682,7 +682,7 @@ def main(options, command: Optional[str], out: Optional[dict] = None) -> int:
if out is not None:
out["text_summary"] = text_summary
except Exception as e:
logging.error(f"Failed to build text summary: {e}")
logging.exception("Failed to build text summary: %s", e)
if generate_pdf_report:
pdf_path = report_file + ".pdf"
@@ -692,7 +692,7 @@ def main(options, command: Optional[str], out: Optional[dict] = None) -> int:
logging.info("PDF report generated: %s", abs_pdf_path)
print(f"\nfile://{abs_pdf_path}\n")
except Exception as e:
logging.error(f"Failed to generate PDF report: {e}")
logging.exception("Failed to generate PDF report: %s", e)
if enable_elastic:
result = elastic_search.push_telemetry(
tests/test_resiliency.py
+76
View File
@@ -407,6 +407,82 @@ class TestResiliencyScenarioReports(unittest.TestCase):
self.res.get_detailed_report()
class TestGetScenarioSloDetails(unittest.TestCase):
"""Test cases for get_scenario_slo_details method."""
def setUp(self):
"""Set up test fixtures with two SLOs."""
alerts_data = [
{"expr": "up == 0", "severity": "critical", "description": "slo1"},
{"expr": "cpu > 80", "severity": "warning", "description": "slo2"},
]
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
import yaml
yaml.dump(alerts_data, f)
self.temp_file = f.name
self.res = Resiliency(alerts_yaml_path=self.temp_file)
def tearDown(self):
"""Clean up temp files."""
if os.path.exists(self.temp_file):
os.unlink(self.temp_file)
def test_severity_lookup_from_slos(self):
"""Test that severity is correctly looked up from the SLO definitions."""
self.res.scenario_reports = [
{
"name": "pod_scenario",
"slo_results": {"slo1": True, "slo2": False},
}
]
result = self.res.get_scenario_slo_details()
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["scenario"], "pod_scenario")
details = {d["name"]: d for d in result[0]["slo_details"]}
self.assertEqual(details["slo1"]["severity"], "critical")
self.assertTrue(details["slo1"]["passed"])
self.assertEqual(details["slo2"]["severity"], "warning")
self.assertFalse(details["slo2"]["passed"])
def test_unknown_severity_for_missing_slo(self):
"""Test that SLOs not in the alerts file get severity 'unknown'."""
self.res.scenario_reports = [
{
"name": "scenario_x",
"slo_results": {"nonexistent_slo": True},
}
]
result = self.res.get_scenario_slo_details()
self.assertEqual(result[0]["slo_details"][0]["severity"], "unknown")
def test_multiple_scenarios(self):
"""Test that details are returned for each scenario in order."""
self.res.scenario_reports = [
{"name": "first", "slo_results": {"slo1": True}},
{"name": "second", "slo_results": {"slo2": False}},
]
result = self.res.get_scenario_slo_details()
self.assertEqual(len(result), 2)
self.assertEqual(result[0]["scenario"], "first")
self.assertEqual(result[1]["scenario"], "second")
def test_empty_scenario_reports(self):
"""Test that empty scenario_reports returns an empty list."""
self.res.scenario_reports = []
result = self.res.get_scenario_slo_details()
self.assertEqual(result, [])
class TestResiliencyCompactBreakdown(unittest.TestCase):
"""Test cases for compact_breakdown static method."""
tests/test_summarized_reports.py
+21 -31
View File
@@ -807,11 +807,8 @@ class TestBuildChaosReportEdgeCases(unittest.TestCase):
class TestBuildChaosReportPdf(unittest.TestCase):
@patch("weasyprint.HTML")
def test_pdf_generated(self, mock_html_cls):
mock_html_instance = MagicMock()
mock_html_cls.return_value = mock_html_instance
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_generated(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
parameters=[{"id": "kill", "config": {"label_selector": "app=etcd",
@@ -835,15 +832,13 @@ class TestBuildChaosReportPdf(unittest.TestCase):
try:
result = build_chaos_report_pdf(output, pdf_path)
self.assertEqual(result, pdf_path)
mock_html_cls.assert_called_once()
mock_html_instance.write_pdf.assert_called_once_with(pdf_path)
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_with_node_recovery(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_with_node_recovery(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
affected_nodes=[{
@@ -855,14 +850,13 @@ class TestBuildChaosReportPdf(unittest.TestCase):
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_with_vmi_recovery(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_with_vmi_recovery(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
affected_vmis={
@@ -877,14 +871,13 @@ class TestBuildChaosReportPdf(unittest.TestCase):
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_with_none_recovery_times(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_with_none_recovery_times(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
affected_pods={
@@ -904,14 +897,13 @@ class TestBuildChaosReportPdf(unittest.TestCase):
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_with_cluster_events(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_with_cluster_events(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
cluster_events=[
@@ -923,14 +915,13 @@ class TestBuildChaosReportPdf(unittest.TestCase):
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_with_additional_telemetry(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_with_additional_telemetry(self, mock_build):
output = _minimal_chaos_output()
output["telemetry"]["scenarios"] = [_make_scenario(
additional_telemetry={"rps": 100},
@@ -939,20 +930,19 @@ class TestBuildChaosReportPdf(unittest.TestCase):
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)
@patch("weasyprint.HTML")
def test_pdf_empty_scenarios(self, mock_html_cls):
mock_html_cls.return_value = MagicMock()
@patch("reportlab.platypus.SimpleDocTemplate.build")
def test_pdf_empty_scenarios(self, mock_build):
output = _minimal_chaos_output()
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
pdf_path = f.name
try:
build_chaos_report_pdf(output, pdf_path)
mock_html_cls.assert_called_once()
mock_build.assert_called_once()
finally:
if os.path.exists(pdf_path):
os.unlink(pdf_path)