k.eaven/pterodactyl-discord-bot
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Update: Code formatting
All checks were successful
CI/CD Pipeline / Unit Tests (Python 3.10) (push) Successful in 9m23s
Details
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 9m19s
Details
CI/CD Pipeline / Unit Tests (Python 3.9) (push) Successful in 9m21s
Details
CI/CD Pipeline / Security Scanning (push) Successful in 16s
Details
CI/CD Pipeline / Code Quality & Linting (push) Successful in 44s
Details
CI/CD Pipeline / Integration Tests (push) Successful in 9m14s
Details
CI/CD Pipeline / Build Docker Image (push) Successful in 35s
Details
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
Details
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
Details

Browse Source
This commit is contained in:
Eaven Kimura
2025-10-26 13:26:49 +00:00
parent 11ee1447de
commit 5e61c838d5
4 changed files with 1103 additions and 704 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.flake8 Normal file
View File

@@ -0,0 +1,2 @@
[flake8]
max-line-length = 140

8
.gitea/workflows/ci-cd.yml
View File

@@ -113,7 +113,7 @@ jobs:
- name: Run flake8 - name: Run flake8
run: | run: |
flake8 pterodisbot.py server_metrics_graphs.py \ flake8 pterodisbot.py server_metrics_graphs.py \
--max-line-length=120 \ --max-line-length=140 \
--ignore=E501,W503,E203 \ --ignore=E501,W503,E203 \
--exclude=venv,__pycache__,build,dist \ --exclude=venv,__pycache__,build,dist \
--statistics \ --statistics \
@@ -124,19 +124,19 @@ jobs:
run: | run: |
pylint pterodisbot.py server_metrics_graphs.py \ pylint pterodisbot.py server_metrics_graphs.py \
--disable=C0111,C0103,R0913,R0914,R0915,W0718 \ --disable=C0111,C0103,R0913,R0914,R0915,W0718 \
--max-line-length=120 \ --max-line-length=140 \
--output-format=text \ --output-format=text \
--reports=y > pylint-report.txt || true --reports=y > pylint-report.txt || true
continue-on-error: true continue-on-error: true
- name: Check code formatting with black - name: Check code formatting with black
run: | run: |
black --check --line-length=120 --diff pterodisbot.py server_metrics_graphs.py | tee black-report.txt black --check --line-length=140 --diff pterodisbot.py server_metrics_graphs.py | tee black-report.txt
continue-on-error: true continue-on-error: true
- name: Check import ordering - name: Check import ordering
run: | run: |
isort --check-only --profile black --line-length=120 pterodisbot.py server_metrics_graphs.py isort --check-only --profile black --line-length=140 pterodisbot.py server_metrics_graphs.py
continue-on-error: true continue-on-error: true
- name: Type checking with mypy - name: Type checking with mypy

1318
pterodisbot.py
View File

File diff suppressed because it is too large Load Diff

479
server_metrics_graphs.py
View File

@@ -6,23 +6,25 @@ Generates line graphs as PNG images for embedding in Discord messages.
""" """
import matplotlib import matplotlib
matplotlib.use('Agg') # Use non-interactive backend for server environments
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
import matplotlib.dates as mdates import matplotlib.dates as mdates
from collections import deque from collections import deque
from datetime import datetime, timedelta from datetime import datetime
from typing import Dict, Tuple, Optional from typing import Dict, Optional
import io import io
import logging import logging
import math import math
matplotlib.use("Agg") # Use non-interactive backend for server environments
# Get the logger from the main bot module # Get the logger from the main bot module
logger = logging.getLogger('pterodisbot') logger = logging.getLogger("pterodisbot")
class ServerMetricsGraphs: class ServerMetricsGraphs:
""" """
Manages CPU and memory usage graphs for individual servers. Manages CPU and memory usage graphs for individual servers.
Features: Features:
- Stores last 6 data points (1 minute of history at 10-second intervals) - Stores last 6 data points (1 minute of history at 10-second intervals)
- Generates PNG images of line graphs for Discord embedding - Generates PNG images of line graphs for Discord embedding
@@ -31,31 +33,35 @@ class ServerMetricsGraphs:
- Dynamic CPU scaling in 100% increments for multi-vCPU servers - Dynamic CPU scaling in 100% increments for multi-vCPU servers
- Clean graph styling optimized for Discord dark theme - Clean graph styling optimized for Discord dark theme
""" """
def __init__(self, server_id: str, server_name: str): def __init__(self, server_id: str, server_name: str):
""" """
Initialize metrics tracking for a server. Initialize metrics tracking for a server.
Args: Args:
server_id: Pterodactyl server identifier server_id: Pterodactyl server identifier
server_name: Human-readable server name server_name: Human-readable server name
""" """
self.server_id = server_id self.server_id = server_id
self.server_name = server_name self.server_name = server_name
# Use deque with maxlen=6 for automatic FIFO rotation # Use deque with maxlen=6 for automatic FIFO rotation
# Each entry is a tuple: (timestamp, cpu_percent, memory_mb) # Each entry is a tuple: (timestamp, cpu_percent, memory_mb)
self.data_points = deque(maxlen=6) self.data_points = deque(maxlen=6)
# Track if we have enough data for meaningful graphs (at least 2 points) # Track if we have enough data for meaningful graphs (at least 2 points)
self.has_sufficient_data = False self.has_sufficient_data = False
logger.debug(f"Initialized metrics tracking for server {server_name} ({server_id})") logger.debug(
f"Initialized metrics tracking for server {server_name} ({server_id})"
def add_data_point(self, cpu_percent: float, memory_mb: float, timestamp: Optional[datetime] = None): )
def add_data_point(
self, cpu_percent: float, memory_mb: float, timestamp: Optional[datetime] = None
):
""" """
Add a new data point to the metrics history. Add a new data point to the metrics history.
Args: Args:
cpu_percent: Current CPU usage percentage cpu_percent: Current CPU usage percentage
memory_mb: Current memory usage in megabytes memory_mb: Current memory usage in megabytes
@@ -63,351 +69,430 @@ class ServerMetricsGraphs:
""" """
if timestamp is None: if timestamp is None:
timestamp = datetime.now() timestamp = datetime.now()
# Add new data point (automatically rotates old data due to maxlen=6) # Add new data point (automatically rotates old data due to maxlen=6)
self.data_points.append((timestamp, cpu_percent, memory_mb)) self.data_points.append((timestamp, cpu_percent, memory_mb))
# Update sufficient data flag # Update sufficient data flag
self.has_sufficient_data = len(self.data_points) >= 2 self.has_sufficient_data = len(self.data_points) >= 2
logger.debug(f"Added metrics data point for {self.server_name}: CPU={cpu_percent}%, Memory={memory_mb}MB") logger.debug(
f"Added metrics data point for {self.server_name}: CPU={cpu_percent}%, Memory={memory_mb}MB"
)
def _calculate_cpu_scale_limit(self, max_cpu_value: float) -> int: def _calculate_cpu_scale_limit(self, max_cpu_value: float) -> int:
""" """
Calculate appropriate CPU scale limit in 100% increments. Calculate appropriate CPU scale limit in 100% increments.
Args: Args:
max_cpu_value: Maximum CPU value in the dataset max_cpu_value: Maximum CPU value in the dataset
Returns: Returns:
Scale limit rounded up to nearest 100% increment Scale limit rounded up to nearest 100% increment
""" """
if max_cpu_value <= 100: if max_cpu_value <= 100:
return 100 return 100
# Round up to nearest 100% increment # Round up to nearest 100% increment
# e.g., 150% -> 200%, 250% -> 300%, 350% -> 400% # e.g., 150% -> 200%, 250% -> 300%, 350% -> 400%
return math.ceil(max_cpu_value / 100) * 100 return math.ceil(max_cpu_value / 100) * 100
def generate_cpu_graph(self) -> Optional[io.BytesIO]: def generate_cpu_graph(self) -> Optional[io.BytesIO]:
""" """
Generate a CPU usage line graph as a PNG image. Generate a CPU usage line graph as a PNG image.
Returns: Returns:
BytesIO object containing PNG image data, or None if insufficient data BytesIO object containing PNG image data, or None if insufficient data
""" """
if not self.has_sufficient_data: if not self.has_sufficient_data:
logger.debug(f"Insufficient data for CPU graph generation: {self.server_name}") logger.debug(
f"Insufficient data for CPU graph generation: {self.server_name}"
)
return None return None
try: try:
# Extract timestamps and CPU data # Extract timestamps and CPU data
timestamps = [point[0] for point in self.data_points] timestamps = [point[0] for point in self.data_points]
cpu_values = [point[1] for point in self.data_points] cpu_values = [point[1] for point in self.data_points]
# Calculate dynamic CPU scale limit # Calculate dynamic CPU scale limit
max_cpu = max(cpu_values) max_cpu = max(cpu_values)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu) cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
# Create figure with dark theme styling # Create figure with dark theme styling
plt.style.use('dark_background') plt.style.use("dark_background")
fig, ax = plt.subplots(figsize=(8, 4), dpi=100) fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
fig.patch.set_facecolor('#2f3136') # Discord dark theme background fig.patch.set_facecolor("#2f3136") # Discord dark theme background
ax.set_facecolor('#36393f') # Slightly lighter for graph area ax.set_facecolor("#36393f") # Slightly lighter for graph area
# Plot CPU line with gradient fill # Plot CPU line with gradient fill
line = ax.plot(timestamps, cpu_values, color='#7289da', linewidth=2.5, marker='o', markersize=4) ax.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
ax.fill_between(timestamps, cpu_values, alpha=0.3, color='#7289da')
# Customize axes with dynamic scaling # Customize axes with dynamic scaling
ax.set_ylabel('CPU Usage (%)', color='#ffffff', fontsize=10) ax.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
ax.set_ylim(0, cpu_scale_limit) ax.set_ylim(0, cpu_scale_limit)
# Add horizontal grid lines at 100% increments for better readability # Add horizontal grid lines at 100% increments for better readability
for i in range(100, cpu_scale_limit + 1, 100): for i in range(100, cpu_scale_limit + 1, 100):
ax.axhline(y=i, color='#ffffff', alpha=0.2, linestyle='--', linewidth=0.8) ax.axhline(
y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8
)
# Format time axis # Format time axis
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20)) ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45, ha='right', color='#ffffff', fontsize=8) plt.setp(
ax.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style the graph # Style the graph
ax.tick_params(colors='#ffffff', labelsize=8) ax.tick_params(colors="#ffffff", labelsize=8)
ax.grid(True, alpha=0.3, color='#ffffff') ax.grid(True, alpha=0.3, color="#ffffff")
ax.spines['bottom'].set_color('#ffffff') ax.spines["bottom"].set_color("#ffffff")
ax.spines['left'].set_color('#ffffff') ax.spines["left"].set_color("#ffffff")
ax.spines['top'].set_visible(False) ax.spines["top"].set_visible(False)
ax.spines['right'].set_visible(False) ax.spines["right"].set_visible(False)
# Add title with scale info for multi-vCPU servers # Add title with scale info for multi-vCPU servers
title = f'{self.server_name} - CPU Usage' title = f"{self.server_name} - CPU Usage"
if cpu_scale_limit > 100: if cpu_scale_limit > 100:
estimated_vcpus = cpu_scale_limit // 100 estimated_vcpus = cpu_scale_limit // 100
title += f' (~{estimated_vcpus} vCPU cores)' title += f" (~{estimated_vcpus} vCPU cores)"
ax.set_title(title, color='#ffffff', fontsize=12, pad=20) ax.set_title(title, color="#ffffff", fontsize=12, pad=20)
# Tight layout to prevent label cutoff # Tight layout to prevent label cutoff
plt.tight_layout() plt.tight_layout()
# Save to BytesIO # Save to BytesIO
img_buffer = io.BytesIO() img_buffer = io.BytesIO()
plt.savefig(img_buffer, format='png', facecolor='#2f3136', edgecolor='none', plt.savefig(
bbox_inches='tight', dpi=100) img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0) img_buffer.seek(0)
# Clean up matplotlib resources # Clean up matplotlib resources
plt.close(fig) plt.close(fig)
logger.debug(f"Generated CPU graph for {self.server_name} (scale: 0-{cpu_scale_limit}%)") logger.debug(
f"Generated CPU graph for {self.server_name} (scale: 0-{cpu_scale_limit}%)"
)
return img_buffer return img_buffer
except Exception as e: except Exception as e:
logger.error(f"Failed to generate CPU graph for {self.server_name}: {str(e)}") logger.error(
plt.close('all') # Clean up any remaining figures f"Failed to generate CPU graph for {self.server_name}: {str(e)}"
)
plt.close("all") # Clean up any remaining figures
return None return None
def generate_memory_graph(self) -> Optional[io.BytesIO]: def generate_memory_graph(self) -> Optional[io.BytesIO]:
""" """
Generate a memory usage line graph as a PNG image. Generate a memory usage line graph as a PNG image.
Returns: Returns:
BytesIO object containing PNG image data, or None if insufficient data BytesIO object containing PNG image data, or None if insufficient data
""" """
if not self.has_sufficient_data: if not self.has_sufficient_data:
logger.debug(f"Insufficient data for memory graph generation: {self.server_name}") logger.debug(
f"Insufficient data for memory graph generation: {self.server_name}"
)
return None return None
try: try:
# Extract timestamps and memory data # Extract timestamps and memory data
timestamps = [point[0] for point in self.data_points] timestamps = [point[0] for point in self.data_points]
memory_values = [point[2] for point in self.data_points] memory_values = [point[2] for point in self.data_points]
# Create figure with dark theme styling # Create figure with dark theme styling
plt.style.use('dark_background') plt.style.use("dark_background")
fig, ax = plt.subplots(figsize=(8, 4), dpi=100) fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
fig.patch.set_facecolor('#2f3136') # Discord dark theme background fig.patch.set_facecolor("#2f3136") # Discord dark theme background
ax.set_facecolor('#36393f') # Slightly lighter for graph area ax.set_facecolor("#36393f") # Slightly lighter for graph area
# Plot memory line with gradient fill # Plot memory line with gradient fill
line = ax.plot(timestamps, memory_values, color='#43b581', linewidth=2.5, marker='o', markersize=4) ax.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
ax.fill_between(timestamps, memory_values, alpha=0.3, color='#43b581')
# Customize axes # Customize axes
ax.set_ylabel('Memory Usage (MB)', color='#ffffff', fontsize=10) ax.set_ylabel("Memory Usage (MB)", color="#ffffff", fontsize=10)
ax.set_ylim(0, max(memory_values) * 1.1) # Dynamic scaling with 10% padding ax.set_ylim(0, max(memory_values) * 1.1) # Dynamic scaling with 10% padding
# Format time axis # Format time axis
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20)) ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45, ha='right', color='#ffffff', fontsize=8) plt.setp(
ax.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style the graph # Style the graph
ax.tick_params(colors='#ffffff', labelsize=8) ax.tick_params(colors="#ffffff", labelsize=8)
ax.grid(True, alpha=0.3, color='#ffffff') ax.grid(True, alpha=0.3, color="#ffffff")
ax.spines['bottom'].set_color('#ffffff') ax.spines["bottom"].set_color("#ffffff")
ax.spines['left'].set_color('#ffffff') ax.spines["left"].set_color("#ffffff")
ax.spines['top'].set_visible(False) ax.spines["top"].set_visible(False)
ax.spines['right'].set_visible(False) ax.spines["right"].set_visible(False)
# Add title # Add title
ax.set_title(f'{self.server_name} - Memory Usage', color='#ffffff', fontsize=12, pad=20) ax.set_title(
f"{self.server_name} - Memory Usage",
color="#ffffff",
fontsize=12,
pad=20,
)
# Tight layout to prevent label cutoff # Tight layout to prevent label cutoff
plt.tight_layout() plt.tight_layout()
# Save to BytesIO # Save to BytesIO
img_buffer = io.BytesIO() img_buffer = io.BytesIO()
plt.savefig(img_buffer, format='png', facecolor='#2f3136', edgecolor='none', plt.savefig(
bbox_inches='tight', dpi=100) img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0) img_buffer.seek(0)
# Clean up matplotlib resources # Clean up matplotlib resources
plt.close(fig) plt.close(fig)
logger.debug(f"Generated memory graph for {self.server_name}") logger.debug(f"Generated memory graph for {self.server_name}")
return img_buffer return img_buffer
except Exception as e: except Exception as e:
logger.error(f"Failed to generate memory graph for {self.server_name}: {str(e)}") logger.error(
plt.close('all') # Clean up any remaining figures f"Failed to generate memory graph for {self.server_name}: {str(e)}"
)
plt.close("all") # Clean up any remaining figures
return None return None
def generate_combined_graph(self) -> Optional[io.BytesIO]: def generate_combined_graph(self) -> Optional[io.BytesIO]:
""" """
Generate a combined CPU and memory usage graph as a PNG image. Generate a combined CPU and memory usage graph as a PNG image.
Returns: Returns:
BytesIO object containing PNG image data, or None if insufficient data BytesIO object containing PNG image data, or None if insufficient data
""" """
if not self.has_sufficient_data: if not self.has_sufficient_data:
logger.debug(f"Insufficient data for combined graph generation: {self.server_name}") logger.debug(
f"Insufficient data for combined graph generation: {self.server_name}"
)
return None return None
try: try:
# Extract data # Extract data
timestamps = [point[0] for point in self.data_points] timestamps = [point[0] for point in self.data_points]
cpu_values = [point[1] for point in self.data_points] cpu_values = [point[1] for point in self.data_points]
memory_values = [point[2] for point in self.data_points] memory_values = [point[2] for point in self.data_points]
# Calculate dynamic CPU scale limit # Calculate dynamic CPU scale limit
max_cpu = max(cpu_values) max_cpu = max(cpu_values)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu) cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
# Create figure with two subplots # Create figure with two subplots
plt.style.use('dark_background') plt.style.use("dark_background")
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 6), dpi=100, sharex=True) fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 6), dpi=100, sharex=True)
fig.patch.set_facecolor('#2f3136') fig.patch.set_facecolor("#2f3136")
# CPU subplot # CPU subplot
ax1.set_facecolor('#36393f') ax1.set_facecolor("#36393f")
ax1.plot(timestamps, cpu_values, color='#7289da', linewidth=2.5, marker='o', markersize=4) ax1.plot(
ax1.fill_between(timestamps, cpu_values, alpha=0.3, color='#7289da') timestamps,
ax1.set_ylabel('CPU Usage (%)', color='#ffffff', fontsize=10) cpu_values,
color="#7289da",
linewidth=2.5,
marker="o",
markersize=4,
)
ax1.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
ax1.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
ax1.set_ylim(0, cpu_scale_limit) ax1.set_ylim(0, cpu_scale_limit)
ax1.tick_params(colors='#ffffff', labelsize=8) ax1.tick_params(colors="#ffffff", labelsize=8)
ax1.grid(True, alpha=0.3, color='#ffffff') ax1.grid(True, alpha=0.3, color="#ffffff")
# Add horizontal grid lines at 100% increments for CPU subplot # Add horizontal grid lines at 100% increments for CPU subplot
for i in range(100, cpu_scale_limit + 1, 100): for i in range(100, cpu_scale_limit + 1, 100):
ax1.axhline(y=i, color='#ffffff', alpha=0.2, linestyle='--', linewidth=0.8) ax1.axhline(
y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8
)
# Title with vCPU info if applicable # Title with vCPU info if applicable
title = f'{self.server_name} - Resource Usage' title = f"{self.server_name} - Resource Usage"
if cpu_scale_limit > 100: if cpu_scale_limit > 100:
estimated_vcpus = cpu_scale_limit // 100 estimated_vcpus = cpu_scale_limit // 100
title += f' (~{estimated_vcpus} vCPU cores)' title += f" (~{estimated_vcpus} vCPU cores)"
ax1.set_title(title, color='#ffffff', fontsize=12) ax1.set_title(title, color="#ffffff", fontsize=12)
# Memory subplot # Memory subplot
ax2.set_facecolor('#36393f') ax2.set_facecolor("#36393f")
ax2.plot(timestamps, memory_values, color='#43b581', linewidth=2.5, marker='o', markersize=4) ax2.plot(
ax2.fill_between(timestamps, memory_values, alpha=0.3, color='#43b581') timestamps,
ax2.set_ylabel('Memory (MB)', color='#ffffff', fontsize=10) memory_values,
color="#43b581",
linewidth=2.5,
marker="o",
markersize=4,
)
ax2.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
ax2.set_ylabel("Memory (MB)", color="#ffffff", fontsize=10)
ax2.set_ylim(0, max(memory_values) * 1.1) ax2.set_ylim(0, max(memory_values) * 1.1)
ax2.tick_params(colors='#ffffff', labelsize=8) ax2.tick_params(colors="#ffffff", labelsize=8)
ax2.grid(True, alpha=0.3, color='#ffffff') ax2.grid(True, alpha=0.3, color="#ffffff")
# Format time axis (only on bottom subplot) # Format time axis (only on bottom subplot)
ax2.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) ax2.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax2.xaxis.set_major_locator(mdates.SecondLocator(interval=20)) ax2.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(ax2.xaxis.get_majorticklabels(), rotation=45, ha='right', color='#ffffff', fontsize=8) plt.setp(
ax2.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style both subplots # Style both subplots
for ax in [ax1, ax2]: for ax in [ax1, ax2]:
ax.spines['bottom'].set_color('#ffffff') ax.spines["bottom"].set_color("#ffffff")
ax.spines['left'].set_color('#ffffff') ax.spines["left"].set_color("#ffffff")
ax.spines['top'].set_visible(False) ax.spines["top"].set_visible(False)
ax.spines['right'].set_visible(False) ax.spines["right"].set_visible(False)
plt.tight_layout() plt.tight_layout()
# Save to BytesIO # Save to BytesIO
img_buffer = io.BytesIO() img_buffer = io.BytesIO()
plt.savefig(img_buffer, format='png', facecolor='#2f3136', edgecolor='none', plt.savefig(
bbox_inches='tight', dpi=100) img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0) img_buffer.seek(0)
plt.close(fig) plt.close(fig)
logger.debug(f"Generated combined graph for {self.server_name} (CPU scale: 0-{cpu_scale_limit}%)") logger.debug(
f"Generated combined graph for {self.server_name} (CPU scale: 0-{cpu_scale_limit}%)"
)
return img_buffer return img_buffer
except Exception as e: except Exception as e:
logger.error(f"Failed to generate combined graph for {self.server_name}: {str(e)}") logger.error(
plt.close('all') f"Failed to generate combined graph for {self.server_name}: {str(e)}"
)
plt.close("all")
return None return None
def get_data_summary(self) -> Dict[str, any]: def get_data_summary(self) -> Dict[str, any]:
""" """
Get summary statistics for the current data points. Get summary statistics for the current data points.
Returns: Returns:
Dictionary containing data point count, latest values, and trends Dictionary containing data point count, latest values, and trends
""" """
if not self.data_points: if not self.data_points:
return { return {
'point_count': 0, "point_count": 0,
'has_data': False, "has_data": False,
'latest_cpu': 0, "latest_cpu": 0,
'latest_memory': 0 "latest_memory": 0,
} }
# Get latest values # Get latest values
latest_point = self.data_points[-1] latest_point = self.data_points[-1]
latest_cpu = latest_point[1] latest_cpu = latest_point[1]
latest_memory = latest_point[2] latest_memory = latest_point[2]
# Calculate CPU scale info # Calculate CPU scale info
max_cpu = max(point[1] for point in self.data_points) max_cpu = max(point[1] for point in self.data_points)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu) cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
estimated_vcpus = cpu_scale_limit // 100 estimated_vcpus = cpu_scale_limit // 100
# Calculate trends if we have multiple points # Calculate trends if we have multiple points
cpu_trend = 'stable' cpu_trend = "stable"
memory_trend = 'stable' memory_trend = "stable"
if len(self.data_points) >= 2: if len(self.data_points) >= 2:
first_point = self.data_points[0] first_point = self.data_points[0]
cpu_change = latest_cpu - first_point[1] cpu_change = latest_cpu - first_point[1]
memory_change = latest_memory - first_point[2] memory_change = latest_memory - first_point[2]
# Determine trends (>5% change considered significant) # Determine trends (>5% change considered significant)
if abs(cpu_change) > 5: if abs(cpu_change) > 5:
cpu_trend = 'increasing' if cpu_change > 0 else 'decreasing' cpu_trend = "increasing" if cpu_change > 0 else "decreasing"
if abs(memory_change) > 50: # 50MB change threshold if abs(memory_change) > 50: # 50MB change threshold
memory_trend = 'increasing' if memory_change > 0 else 'decreasing' memory_trend = "increasing" if memory_change > 0 else "decreasing"
return { return {
'point_count': len(self.data_points), "point_count": len(self.data_points),
'has_data': self.has_sufficient_data, "has_data": self.has_sufficient_data,
'latest_cpu': latest_cpu, "latest_cpu": latest_cpu,
'latest_memory': latest_memory, "latest_memory": latest_memory,
'cpu_trend': cpu_trend, "cpu_trend": cpu_trend,
'memory_trend': memory_trend, "memory_trend": memory_trend,
'cpu_scale_limit': cpu_scale_limit, "cpu_scale_limit": cpu_scale_limit,
'estimated_vcpus': estimated_vcpus, "estimated_vcpus": estimated_vcpus,
'time_span_minutes': len(self.data_points) * 10 / 60 # Convert to minutes "time_span_minutes": len(self.data_points) * 10 / 60, # Convert to minutes
} }
class ServerMetricsManager: class ServerMetricsManager:
""" """
Global manager for all server metrics graphs. Global manager for all server metrics graphs.
Handles: Handles:
- Creation and cleanup of ServerMetricsGraphs instances - Creation and cleanup of ServerMetricsGraphs instances
- Bulk operations across all tracked servers - Bulk operations across all tracked servers
- Memory management for graph storage - Memory management for graph storage
""" """
def __init__(self): def __init__(self):
"""Initialize the metrics manager.""" """Initialize the metrics manager."""
self.server_graphs: Dict[str, ServerMetricsGraphs] = {} self.server_graphs: Dict[str, ServerMetricsGraphs] = {}
logger.info("Initialized ServerMetricsManager") logger.info("Initialized ServerMetricsManager")
def get_or_create_server_graphs(self, server_id: str, server_name: str) -> ServerMetricsGraphs: def get_or_create_server_graphs(
self, server_id: str, server_name: str
) -> ServerMetricsGraphs:
""" """
Get existing ServerMetricsGraphs instance or create a new one. Get existing ServerMetricsGraphs instance or create a new one.
Args: Args:
server_id: Pterodactyl server identifier server_id: Pterodactyl server identifier
server_name: Human-readable server name server_name: Human-readable server name
Returns: Returns:
ServerMetricsGraphs instance for the specified server ServerMetricsGraphs instance for the specified server
""" """
if server_id not in self.server_graphs: if server_id not in self.server_graphs:
self.server_graphs[server_id] = ServerMetricsGraphs(server_id, server_name) self.server_graphs[server_id] = ServerMetricsGraphs(server_id, server_name)
logger.debug(f"Created new metrics graphs for server {server_name}") logger.debug(f"Created new metrics graphs for server {server_name}")
return self.server_graphs[server_id] return self.server_graphs[server_id]
def add_server_data(self, server_id: str, server_name: str, cpu_percent: float, memory_mb: float): def add_server_data(
self, server_id: str, server_name: str, cpu_percent: float, memory_mb: float
):
""" """
Add data point to a server's metrics tracking. Add data point to a server's metrics tracking.
Args: Args:
server_id: Pterodactyl server identifier server_id: Pterodactyl server identifier
server_name: Human-readable server name server_name: Human-readable server name
@@ -416,34 +501,34 @@ class ServerMetricsManager:
""" """
graphs = self.get_or_create_server_graphs(server_id, server_name) graphs = self.get_or_create_server_graphs(server_id, server_name)
graphs.add_data_point(cpu_percent, memory_mb) graphs.add_data_point(cpu_percent, memory_mb)
def remove_server(self, server_id: str): def remove_server(self, server_id: str):
""" """
Remove a server from metrics tracking. Remove a server from metrics tracking.
Args: Args:
server_id: Pterodactyl server identifier to remove server_id: Pterodactyl server identifier to remove
""" """
if server_id in self.server_graphs: if server_id in self.server_graphs:
del self.server_graphs[server_id] del self.server_graphs[server_id]
logger.debug(f"Removed metrics tracking for server {server_id}") logger.debug(f"Removed metrics tracking for server {server_id}")
def get_server_graphs(self, server_id: str) -> Optional[ServerMetricsGraphs]: def get_server_graphs(self, server_id: str) -> Optional[ServerMetricsGraphs]:
""" """
Get ServerMetricsGraphs instance for a specific server. Get ServerMetricsGraphs instance for a specific server.
Args: Args:
server_id: Pterodactyl server identifier server_id: Pterodactyl server identifier
Returns: Returns:
ServerMetricsGraphs instance or None if not found ServerMetricsGraphs instance or None if not found
""" """
return self.server_graphs.get(server_id) return self.server_graphs.get(server_id)
def cleanup_old_servers(self, active_server_ids: list): def cleanup_old_servers(self, active_server_ids: list):
""" """
Remove tracking for servers that no longer exist. Remove tracking for servers that no longer exist.
Args: Args:
active_server_ids: List of currently active server IDs active_server_ids: List of currently active server IDs
""" """
@@ -451,22 +536,30 @@ class ServerMetricsManager:
for server_id in self.server_graphs: for server_id in self.server_graphs:
if server_id not in active_server_ids: if server_id not in active_server_ids:
servers_to_remove.append(server_id) servers_to_remove.append(server_id)
for server_id in servers_to_remove: for server_id in servers_to_remove:
self.remove_server(server_id) self.remove_server(server_id)
if servers_to_remove: if servers_to_remove:
logger.info(f"Cleaned up metrics for {len(servers_to_remove)} inactive servers") logger.info(
f"Cleaned up metrics for {len(servers_to_remove)} inactive servers"
)
def get_summary(self) -> Dict[str, any]: def get_summary(self) -> Dict[str, any]:
""" """
Get summary of all tracked servers. Get summary of all tracked servers.
Returns: Returns:
Dictionary with tracking statistics Dictionary with tracking statistics
""" """
return { return {
'total_servers': len(self.server_graphs), "total_servers": len(self.server_graphs),
'servers_with_data': sum(1 for graphs in self.server_graphs.values() if graphs.has_sufficient_data), "servers_with_data": sum(
'total_data_points': sum(len(graphs.data_points) for graphs in self.server_graphs.values()) 1
} for graphs in self.server_graphs.values()
if graphs.has_sufficient_data
),
"total_data_points": sum(
len(graphs.data_points) for graphs in self.server_graphs.values()
),
}