smueller/Ceph-Calculator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Erstelle die Grundstruktur für die Ceph Max Storage Rechner-Anwendung mit Flask und HTMX. Füge Konfigurationsdateien, Routen, Modelle, Templates und statische Dateien hinzu. Implementiere grundlegende Funktionen zur Berechnung der Speichernutzung und zur PDF-Generierung. Integriere CSRF-Schutz und Logging. Stelle sicher, dass die Anwendung modular und wartbar ist.

Browse Source
This commit is contained in:
Samuel Müller
2025-03-26 08:40:32 +01:00
commit f00c1f48e9
26 changed files with 1744 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
app/utils/pdf_generator.py Normal file
View File

@ -0,0 +1,217 @@
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
def validate_data(data):
"""Überprüft, ob alle erforderlichen Daten vorhanden sind."""
required_fields = ['replication_type', 'nodes', 'result']
for field in required_fields:
if field not in data:
raise ValueError(f"Fehlendes Feld: {field}")
if not data['nodes']:
raise ValueError("Keine Nodes definiert")
if not data['result'].get('raw_total'):
raise ValueError("Keine Berechnungsergebnisse vorhanden")
def safe_int(value, default=0):
"""Konvertiert einen Wert sicher in einen Integer."""
if value is None:
return default
try:
return int(str(value).strip())
except (ValueError, TypeError):
logger.warning(f"Konnte Wert '{value}' nicht in Integer konvertieren. Verwende Standardwert {default}")
return default
def safe_float(value, default=0.0):
"""Konvertiert einen Wert sicher in einen Float."""
if value is None:
return default
try:
return float(str(value).strip())
except (ValueError, TypeError):
logger.warning(f"Konnte Wert '{value}' nicht in Float konvertieren. Verwende Standardwert {default}")
return default
def generate_pdf_report(data):
"""
Generiert einen PDF-Report mit den Ceph-Konfigurationsdaten und Ergebnissen.
Args:
data (dict): Dictionary mit Konfigurations- und Ergebnisdaten
Returns:
bytes: PDF-Datei als Bytes
Raises:
ValueError: Wenn erforderliche Daten fehlen
"""
try:
# Validiere Eingabedaten
validate_data(data)
# Erstelle einen temporären Buffer für die PDF
from io import BytesIO
buffer = BytesIO()
# Erstelle das PDF-Dokument
doc = SimpleDocTemplate(
buffer,
pagesize=A4,
rightMargin=72,
leftMargin=72,
topMargin=72,
bottomMargin=72,
title="Ceph Storage Kapazitätsreport"
)
# Styles definieren
styles = getSampleStyleSheet()
title_style = styles['Heading1']
heading_style = styles['Heading2']
normal_style = styles['Normal']
# Custom Style für Warnungen
warning_style = ParagraphStyle(
'Warning',
parent=styles['Normal'],
textColor=colors.red,
fontSize=10
)
# Story (Inhalt) erstellen
story = []
# Titel
story.append(Paragraph("Ceph Storage Kapazitätsreport", title_style))
story.append(Spacer(1, 12))
story.append(Paragraph(f"Generiert am: {datetime.now().strftime('%d.%m.%Y %H:%M')}", normal_style))
story.append(Spacer(1, 20))
# Konfigurationsübersicht
story.append(Paragraph("Konfigurationsübersicht", heading_style))
story.append(Spacer(1, 12))
# Replikationstyp
config_data = [
["Replikationstyp", data['replication_type']],
["Speichereinheit", data.get('storage_unit', 'GB')]
]
if data['replication_type'] == 'replication':
config_data.extend([
["Anzahl Replikate", str(safe_int(data.get('replicas', 3)))],
["Min. Replikate (min_size)", str(safe_int(data.get('min_size', 2)))]
])
else:
config_data.extend([
["Datenchunks (k)", str(safe_int(data.get('k', 4)))],
["Codierungschunks (m)", str(safe_int(data.get('m', 2)))]
])
# Nodes und OSDs
story.append(Paragraph("Nodes und OSDs", heading_style))
story.append(Spacer(1, 12))
for i, node in enumerate(data['nodes'], 1):
story.append(Paragraph(f"Node {i}", heading_style))
story.append(Spacer(1, 6))
osd_count = safe_int(node.get('osd_count', 1))
osd_size = safe_float(node.get('osd_size_gb', 0))
if osd_count <= 0:
logger.warning(f"Ungültige OSD-Anzahl für Node {i}: {osd_count}. Verwende Standardwert 1.")
osd_count = 1
osd_data = []
for j in range(osd_count):
osd_data.append([f"OSD {j+1}", f"{osd_size} {data.get('storage_unit', 'GB')}"])
t = Table(osd_data, colWidths=[2*inch, 2*inch])
t.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, -1), colors.white),
('TEXTCOLOR', (0, 0), (-1, -1), colors.black),
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
('FONTNAME', (0, 0), (-1, -1), 'Helvetica'),
('FONTSIZE', (0, 0), (-1, -1), 10),
('BOTTOMPADDING', (0, 0), (-1, -1), 12),
('GRID', (0, 0), (-1, -1), 1, colors.black)
]))
story.append(t)
story.append(Spacer(1, 12))
# Ergebnisse
story.append(Paragraph("Ergebnisse", heading_style))
story.append(Spacer(1, 12))
result_data = [
["Gesamt-Rohspeicher", f"{safe_float(data['result']['raw_total'])} GB ({(safe_float(data['result']['raw_total']) / 1024):.2f} TB)"],
["Max. empfohlene Nutzung", f"{safe_float(data['result'].get('max_usage_percent', 0))}%"],
["Max. nutzbarer Speicher", f"{safe_float(data['result'].get('max_usage_gb', 0))} GB ({safe_float(data['result'].get('max_usage_tb', 0))} TB)"]
]
t = Table(result_data, colWidths=[3*inch, 3*inch])
t.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, -1), colors.white),
('TEXTCOLOR', (0, 0), (-1, -1), colors.black),
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
('FONTNAME', (0, 0), (-1, -1), 'Helvetica'),
('FONTSIZE', (0, 0), (-1, -1), 10),
('BOTTOMPADDING', (0, 0), (-1, -1), 12),
('GRID', (0, 0), (-1, -1), 1, colors.black)
]))
story.append(t)
# Ausfalltoleranz
story.append(Spacer(1, 20))
story.append(Paragraph("Ausfalltoleranz-Analyse", heading_style))
story.append(Spacer(1, 12))
# Node-Ausfalltoleranz
node_tolerance = "Vorhanden" if data['result'].get('node_failure_tolerance', False) else "Nicht vorhanden"
story.append(Paragraph(f"Node-Ausfalltoleranz: {node_tolerance}", normal_style))
story.append(Paragraph(data['result'].get('node_failure_info', 'Keine Informationen verfügbar'), normal_style))
if data['result'].get('multi_failure_tolerance', False):
story.append(Paragraph(f"Multi-Node Ausfalltoleranz: {safe_int(data['result'].get('max_failure_nodes', 0))} Nodes", normal_style))
# OSD-Ausfalltoleranz
story.append(Spacer(1, 12))
osd_tolerance = "Vorhanden" if data['result'].get('osd_failure_tolerance', False) else "Eingeschränkt"
story.append(Paragraph(f"OSD-Ausfalltoleranz: {osd_tolerance}", normal_style))
story.append(Paragraph(data['result'].get('osd_failure_info', 'Keine Informationen verfügbar'), normal_style))
# Empfehlungen
story.append(Spacer(1, 20))
story.append(Paragraph("Empfehlungen", heading_style))
story.append(Spacer(1, 12))
recommendations = [
"Stelle sicher, dass nach Nodeausfällen genügend Kapazität vorhanden ist, um die 'full ratio' nicht zu erreichen.",
"Bei Replikation sollte die Anzahl der Nodes größer sein als der Replikationsfaktor.",
"Bei Erasure Coding sollte die Anzahl der Nodes größer sein als k+m (Datenchunks + Codierungschunks).",
"Die Performance bei der Wiederherstellung hängt von der Netzwerk- und Speichergeschwindigkeit ab.",
"Beachte, dass nach einem Nodeausfall das Wiederausbalancieren des Clusters länger dauert, je größer der ausgefallene Node ist."
]
for rec in recommendations:
story.append(Paragraph(f"• {rec}", normal_style))
# PDF generieren
doc.build(story)
# PDF-Bytes zurückgeben
return buffer.getvalue()
except Exception as e:
logger.error(f"Fehler bei der PDF-Generierung: {str(e)}")
raise