mirror of
https://github.com/taylanbakircioglu/haproxy-openmanager.git
synced 2026-09-22 18:43:26 +00:00
taylanbakircioglu
648 lines
28 KiB
Python
648 lines
28 KiB
Python
"""
|
|
Dashboard Stats Service
|
|
Service layer for HAProxy dashboard statistics
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, List, Any, Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
from utils.haproxy_stats_cache import haproxy_stats_cache
|
|
from utils.haproxy_stats_parser import haproxy_stats_parser
|
|
from database.connection import get_database_connection, close_database_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DashboardStatsService:
|
|
"""Service for managing dashboard statistics"""
|
|
|
|
def __init__(self):
|
|
self.cache = haproxy_stats_cache
|
|
self.parser = haproxy_stats_parser
|
|
|
|
async def process_agent_stats(self, agent_name: str, cluster_id: int,
|
|
raw_stats_csv: Optional[str] = None,
|
|
server_statuses: Optional[Dict[str, Dict[str, str]]] = None):
|
|
"""
|
|
Process HAProxy stats from agent heartbeat
|
|
|
|
Args:
|
|
agent_name: Name of the agent
|
|
cluster_id: Cluster ID
|
|
raw_stats_csv: Raw CSV stats from HAProxy stats socket (optional)
|
|
server_statuses: Server statuses from agent {backend: {server: status}} (optional)
|
|
"""
|
|
try:
|
|
# Log what we received from agent
|
|
logger.info(f"📊 DASHBOARD STATS: Processing stats from agent '{agent_name}' (cluster {cluster_id})")
|
|
logger.info(f"📊 DASHBOARD STATS: Has raw_stats_csv: {raw_stats_csv is not None and len(raw_stats_csv) > 0}")
|
|
logger.info(f"📊 DASHBOARD STATS: Has server_statuses: {server_statuses is not None and len(server_statuses) > 0}")
|
|
|
|
# If we have raw CSV stats, parse and cache them
|
|
if raw_stats_csv:
|
|
parsed = self.parser.parse_csv_stats(raw_stats_csv)
|
|
logger.info(f"📊 DASHBOARD STATS: Parsed {len(parsed.get('frontends', {}))} frontends, "
|
|
f"{len(parsed.get('backends', {}))} backends, "
|
|
f"{len(parsed.get('servers', []))} servers")
|
|
|
|
if parsed.get('frontends'):
|
|
logger.info(f"📊 DASHBOARD STATS: Frontend names: {list(parsed.get('frontends', {}).keys())}")
|
|
if parsed.get('backends'):
|
|
logger.info(f"📊 DASHBOARD STATS: Backend names: {list(parsed.get('backends', {}).keys())}")
|
|
|
|
# Cache frontends
|
|
for frontend_name, frontend_stats in parsed.get('frontends', {}).items():
|
|
self.cache.cache_frontend_stats(cluster_id, frontend_name, frontend_stats)
|
|
|
|
# Add to timeseries - requests
|
|
requests_total = frontend_stats.get('requests_total', 0)
|
|
logger.info(f"📊 REDIS WRITE: frontend:{frontend_name}:requests = {requests_total}")
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:requests",
|
|
requests_total
|
|
)
|
|
|
|
# Add to timeseries - sessions
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:sessions",
|
|
frontend_stats.get('current_sessions', 0)
|
|
)
|
|
|
|
# Add to timeseries - errors
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:errors_4xx",
|
|
frontend_stats.get('hrsp_4xx', 0)
|
|
)
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:errors_5xx",
|
|
frontend_stats.get('hrsp_5xx', 0)
|
|
)
|
|
|
|
# Add to timeseries - bytes
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:bytes_in",
|
|
frontend_stats.get('bytes_in', 0)
|
|
)
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"frontend:{frontend_name}:bytes_out",
|
|
frontend_stats.get('bytes_out', 0)
|
|
)
|
|
|
|
# Cache backends
|
|
for backend_name, backend_stats in parsed.get('backends', {}).items():
|
|
self.cache.cache_backend_stats(cluster_id, backend_name, backend_stats)
|
|
|
|
# Add to timeseries - requests
|
|
backend_requests = backend_stats.get('requests_total', 0)
|
|
logger.info(f"📊 REDIS WRITE: backend:{backend_name}:requests = {backend_requests}")
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"backend:{backend_name}:requests",
|
|
backend_requests
|
|
)
|
|
|
|
# Add to timeseries - response time
|
|
backend_rtime = backend_stats.get('response_time_avg', 0)
|
|
logger.info(f"📊 REDIS WRITE: backend:{backend_name}:response_time = {backend_rtime}")
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"backend:{backend_name}:response_time",
|
|
backend_rtime
|
|
)
|
|
|
|
# Add to timeseries - sessions
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"backend:{backend_name}:sessions",
|
|
backend_stats.get('current_sessions', 0)
|
|
)
|
|
|
|
# Add to timeseries - queue
|
|
self.cache.add_timeseries_metric(
|
|
cluster_id,
|
|
f"backend:{backend_name}:queue",
|
|
backend_stats.get('queue_current', 0)
|
|
)
|
|
|
|
# Cache servers
|
|
for server in parsed.get('servers', []):
|
|
self.cache.cache_server_stats(
|
|
cluster_id,
|
|
server['backend'],
|
|
server['name'],
|
|
server
|
|
)
|
|
|
|
# Cache raw stats
|
|
self.cache.cache_raw_stats(cluster_id, agent_name, raw_stats_csv)
|
|
|
|
# Update summary stats
|
|
await self._update_summary_stats(cluster_id, parsed)
|
|
|
|
logger.debug(f"Processed stats from agent {agent_name} for cluster {cluster_id}")
|
|
|
|
# If we have server statuses from heartbeat, cache them
|
|
elif server_statuses:
|
|
servers = self.parser.parse_agent_heartbeat_stats(server_statuses)
|
|
for server in servers:
|
|
self.cache.cache_server_stats(
|
|
cluster_id,
|
|
server['backend'],
|
|
server['name'],
|
|
server
|
|
)
|
|
|
|
logger.debug(f"Processed server statuses from agent {agent_name} for cluster {cluster_id}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to process agent stats: {e}")
|
|
|
|
async def _update_summary_stats(self, cluster_id: int, parsed_data: Dict[str, Any]):
|
|
"""Update summary statistics for a cluster"""
|
|
try:
|
|
frontends = parsed_data.get('frontends', {})
|
|
backends = parsed_data.get('backends', {})
|
|
servers = parsed_data.get('servers', [])
|
|
|
|
# Aggregate stats
|
|
frontend_agg = self.parser.aggregate_frontend_stats(frontends)
|
|
backend_agg = self.parser.aggregate_backend_stats(backends)
|
|
|
|
# Server health
|
|
servers_total = len(servers)
|
|
servers_up = len([s for s in servers if s.get('status') == 'UP'])
|
|
|
|
summary = {
|
|
'cluster_id': cluster_id,
|
|
'timestamp': datetime.utcnow().isoformat(),
|
|
'frontends': {
|
|
'count': len(frontends),
|
|
'total_requests': frontend_agg.get('total_requests', 0),
|
|
'total_sessions': frontend_agg.get('total_sessions', 0),
|
|
'error_rate': frontend_agg.get('error_rate', 0)
|
|
},
|
|
'backends': {
|
|
'count': len(backends),
|
|
'total_requests': backend_agg.get('total_requests', 0),
|
|
'total_sessions': backend_agg.get('total_sessions', 0),
|
|
'avg_response_time': backend_agg.get('avg_response_time', 0)
|
|
},
|
|
'servers': {
|
|
'total': servers_total,
|
|
'up': servers_up,
|
|
'down': servers_total - servers_up,
|
|
'health_percentage': round((servers_up / servers_total * 100) if servers_total > 0 else 0, 1)
|
|
}
|
|
}
|
|
|
|
self.cache.cache_summary_stats(cluster_id, summary)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update summary stats: {e}")
|
|
|
|
async def get_cluster_stats(self, cluster_id: int,
|
|
frontend_filter: Optional[List[str]] = None,
|
|
backend_filter: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
"""
|
|
Get comprehensive stats for a cluster with optional filters
|
|
|
|
Args:
|
|
cluster_id: Cluster ID
|
|
frontend_filter: List of frontend names to filter (None = all)
|
|
backend_filter: List of backend names to filter (None = all)
|
|
|
|
Returns:
|
|
Comprehensive stats dictionary
|
|
"""
|
|
try:
|
|
# Get summary stats from cache
|
|
summary = self.cache.get_summary_stats(cluster_id)
|
|
|
|
# Get all frontends
|
|
all_frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
|
|
# Apply frontend filter
|
|
if frontend_filter:
|
|
frontends = {k: v for k, v in all_frontends.items() if k in frontend_filter}
|
|
else:
|
|
frontends = all_frontends
|
|
|
|
# Get all backends
|
|
all_backends = self.cache.get_all_backend_stats(cluster_id)
|
|
|
|
# Apply backend filter
|
|
if backend_filter:
|
|
backends = {k: v for k, v in all_backends.items() if k in backend_filter}
|
|
else:
|
|
backends = all_backends
|
|
|
|
# Get servers (filter by backend if needed)
|
|
all_servers = self.cache.get_all_server_stats(cluster_id)
|
|
if backend_filter:
|
|
servers = [s for s in all_servers if s.get('backend') in backend_filter]
|
|
else:
|
|
servers = all_servers
|
|
|
|
# Get additional counts from database
|
|
conn = await get_database_connection()
|
|
|
|
# Get SSL certificate count
|
|
ssl_count = await conn.fetchval(
|
|
"SELECT COUNT(*) FROM ssl_certificates WHERE cluster_id = $1",
|
|
cluster_id
|
|
) or 0
|
|
|
|
# Get WAF rules count
|
|
waf_count = await conn.fetchval(
|
|
"SELECT COUNT(*) FROM waf_rules WHERE cluster_id = $1",
|
|
cluster_id
|
|
) or 0
|
|
|
|
await close_database_connection(conn)
|
|
|
|
# Calculate aggregated metrics
|
|
frontend_agg = self.parser.aggregate_frontend_stats(frontends)
|
|
backend_agg = self.parser.aggregate_backend_stats(backends)
|
|
|
|
# Response time percentiles
|
|
response_time_percentiles = self.parser.calculate_response_time_percentiles(servers)
|
|
|
|
# HTTP response code distribution
|
|
total_2xx = sum(f.get('hrsp_2xx', 0) for f in frontends.values())
|
|
total_3xx = sum(f.get('hrsp_3xx', 0) for f in frontends.values())
|
|
total_4xx = sum(f.get('hrsp_4xx', 0) for f in frontends.values())
|
|
total_5xx = sum(f.get('hrsp_5xx', 0) for f in frontends.values())
|
|
total_responses = total_2xx + total_3xx + total_4xx + total_5xx
|
|
|
|
return {
|
|
'cluster_id': cluster_id,
|
|
'timestamp': datetime.utcnow().isoformat(),
|
|
'summary': summary or {},
|
|
'kpi': {
|
|
'frontends_count': len(all_frontends),
|
|
'backends_count': len(all_backends),
|
|
'servers_total': len(all_servers),
|
|
'servers_up': len([s for s in all_servers if s.get('status') == 'UP']),
|
|
'ssl_certificates': ssl_count,
|
|
'waf_rules': waf_count
|
|
},
|
|
'frontends': {
|
|
'count': len(frontends),
|
|
'data': list(frontends.values()),
|
|
'aggregated': frontend_agg
|
|
},
|
|
'backends': {
|
|
'count': len(backends),
|
|
'data': list(backends.values()),
|
|
'aggregated': backend_agg
|
|
},
|
|
'servers': {
|
|
'count': len(servers),
|
|
'data': servers
|
|
},
|
|
'metrics': {
|
|
'response_time': response_time_percentiles,
|
|
'http_responses': {
|
|
'2xx': total_2xx,
|
|
'3xx': total_3xx,
|
|
'4xx': total_4xx,
|
|
'5xx': total_5xx,
|
|
'total': total_responses,
|
|
'distribution': {
|
|
'2xx_pct': round((total_2xx / total_responses * 100) if total_responses > 0 else 0, 1),
|
|
'3xx_pct': round((total_3xx / total_responses * 100) if total_responses > 0 else 0, 1),
|
|
'4xx_pct': round((total_4xx / total_responses * 100) if total_responses > 0 else 0, 1),
|
|
'5xx_pct': round((total_5xx / total_responses * 100) if total_responses > 0 else 0, 1)
|
|
}
|
|
},
|
|
'error_rate': frontend_agg.get('error_rate', 0)
|
|
},
|
|
'filters_applied': {
|
|
'frontends': frontend_filter,
|
|
'backends': backend_filter
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get cluster stats: {e}")
|
|
return {
|
|
'cluster_id': cluster_id,
|
|
'timestamp': datetime.utcnow().isoformat(),
|
|
'error': str(e)
|
|
}
|
|
|
|
async def get_timeseries_data(self, cluster_id: int,
|
|
metric_name: str,
|
|
hours: int = 24,
|
|
limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get time series data for a specific metric
|
|
|
|
Args:
|
|
cluster_id: Cluster ID
|
|
metric_name: Metric name (e.g., "frontend:web:requests")
|
|
hours: Number of hours to look back
|
|
limit: Maximum number of data points to return (most recent)
|
|
|
|
Returns:
|
|
List of time series data points
|
|
"""
|
|
try:
|
|
end_time = datetime.utcnow()
|
|
start_time = end_time - timedelta(hours=hours)
|
|
|
|
data = self.cache.get_timeseries_metrics(
|
|
cluster_id,
|
|
metric_name,
|
|
start_time,
|
|
end_time,
|
|
limit
|
|
)
|
|
|
|
return data
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get timeseries data: {e}")
|
|
return []
|
|
|
|
async def get_frontend_list(self, cluster_id: int) -> List[str]:
|
|
"""Get list of available frontends for a cluster"""
|
|
try:
|
|
frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
return list(frontends.keys())
|
|
except Exception as e:
|
|
logger.error(f"Failed to get frontend list: {e}")
|
|
return []
|
|
|
|
async def get_backend_list(self, cluster_id: int) -> List[str]:
|
|
"""Get list of available backends for a cluster"""
|
|
try:
|
|
backends = self.cache.get_all_backend_stats(cluster_id)
|
|
return list(backends.keys())
|
|
except Exception as e:
|
|
logger.error(f"Failed to get backend list: {e}")
|
|
return []
|
|
|
|
async def get_slowest_backends(self, cluster_id: int, limit: int = 5) -> List[Dict[str, Any]]:
|
|
"""Get the slowest backends by response time"""
|
|
try:
|
|
backends = self.cache.get_all_backend_stats(cluster_id)
|
|
|
|
# Sort by response time
|
|
sorted_backends = sorted(
|
|
backends.items(),
|
|
key=lambda x: x[1].get('response_time_avg', 0),
|
|
reverse=True
|
|
)
|
|
|
|
return [
|
|
{
|
|
'name': name,
|
|
'response_time': stats.get('response_time_avg', 0),
|
|
'status': stats.get('status', 'UNKNOWN')
|
|
}
|
|
for name, stats in sorted_backends[:limit]
|
|
]
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get slowest backends: {e}")
|
|
return []
|
|
|
|
async def get_backend_health_matrix(self, cluster_id: int) -> List[Dict[str, Any]]:
|
|
"""Get backend health matrix for visualization"""
|
|
try:
|
|
backends = self.cache.get_all_backend_stats(cluster_id)
|
|
|
|
health_matrix = []
|
|
for name, stats in backends.items():
|
|
health_matrix.append({
|
|
'name': name,
|
|
'status': stats.get('status', 'UNKNOWN'),
|
|
'health_percentage': stats.get('health_percentage', 0),
|
|
'servers_active': stats.get('servers_active', 0),
|
|
'servers_total': stats.get('servers_total', 0),
|
|
'response_time': stats.get('response_time_avg', 0),
|
|
'queue_current': stats.get('queue_current', 0),
|
|
'queue_max': stats.get('queue_max', 0)
|
|
})
|
|
|
|
return health_matrix
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get backend health matrix: {e}")
|
|
return []
|
|
|
|
async def get_requests_timeseries(self, cluster_id: int, hours: int = 24,
|
|
frontend_filter: Optional[List[str]] = None,
|
|
limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""Get request rate time series data for frontends"""
|
|
try:
|
|
# Get all frontends if no filter specified
|
|
if not frontend_filter:
|
|
frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
frontend_filter = list(frontends.keys())
|
|
|
|
# Collect timeseries data for each frontend
|
|
all_data = []
|
|
for frontend_name in frontend_filter:
|
|
metric_name = f"frontend:{frontend_name}:requests"
|
|
data = await self.get_timeseries_data(cluster_id, metric_name, hours, limit)
|
|
|
|
# Add frontend name to each data point
|
|
for point in data:
|
|
point['frontend'] = frontend_name
|
|
all_data.append(point)
|
|
|
|
# Sort by timestamp
|
|
all_data.sort(key=lambda x: x.get('timestamp', ''))
|
|
|
|
return all_data
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get requests timeseries: {e}")
|
|
return []
|
|
|
|
async def get_response_time_timeseries(self, cluster_id: int, hours: int = 24,
|
|
backend_filter: Optional[List[str]] = None,
|
|
limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""Get response time trends (P50, P95, P99) for backends"""
|
|
try:
|
|
# Get all backends if no filter specified
|
|
if not backend_filter:
|
|
backends = self.cache.get_all_backend_stats(cluster_id)
|
|
backend_filter = list(backends.keys())
|
|
|
|
# Collect timeseries data for each backend
|
|
all_data = []
|
|
for backend_name in backend_filter:
|
|
metric_name = f"backend:{backend_name}:response_time"
|
|
data = await self.get_timeseries_data(cluster_id, metric_name, hours, limit)
|
|
|
|
# Add backend name to each data point
|
|
for point in data:
|
|
point['backend'] = backend_name
|
|
all_data.append(point)
|
|
|
|
# Sort by timestamp
|
|
all_data.sort(key=lambda x: x.get('timestamp', ''))
|
|
|
|
return all_data
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get response time timeseries: {e}")
|
|
return []
|
|
|
|
async def get_error_timeseries(self, cluster_id: int, hours: int = 24, limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""Get error rate trends (4xx and 5xx) over time"""
|
|
try:
|
|
end_time = datetime.utcnow()
|
|
start_time = end_time - timedelta(hours=hours)
|
|
|
|
# Get all frontends to aggregate error data
|
|
frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
|
|
# Collect error metrics from timeseries
|
|
error_4xx_data = []
|
|
error_5xx_data = []
|
|
|
|
for frontend_name in frontends.keys():
|
|
# Get 4xx errors
|
|
metric_4xx = f"frontend:{frontend_name}:errors_4xx"
|
|
data_4xx = self.cache.get_timeseries_metrics(cluster_id, metric_4xx, start_time, end_time, limit)
|
|
error_4xx_data.extend(data_4xx)
|
|
|
|
# Get 5xx errors
|
|
metric_5xx = f"frontend:{frontend_name}:errors_5xx"
|
|
data_5xx = self.cache.get_timeseries_metrics(cluster_id, metric_5xx, start_time, end_time, limit)
|
|
error_5xx_data.extend(data_5xx)
|
|
|
|
# Aggregate by timestamp
|
|
from collections import defaultdict
|
|
aggregated = defaultdict(lambda: {'errors_4xx': 0, 'errors_5xx': 0})
|
|
|
|
for point in error_4xx_data:
|
|
ts = point.get('timestamp')
|
|
aggregated[ts]['errors_4xx'] += point.get('value', 0)
|
|
aggregated[ts]['timestamp'] = ts
|
|
|
|
for point in error_5xx_data:
|
|
ts = point.get('timestamp')
|
|
aggregated[ts]['errors_5xx'] += point.get('value', 0)
|
|
aggregated[ts]['timestamp'] = ts
|
|
|
|
# Convert to list and sort
|
|
result = list(aggregated.values())
|
|
result.sort(key=lambda x: x.get('timestamp', ''))
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get error timeseries: {e}")
|
|
return []
|
|
|
|
async def get_realtime_throughput(self, cluster_id: int) -> Dict[str, Any]:
|
|
"""Get current bytes in/out rates and totals"""
|
|
try:
|
|
# Get all frontends for throughput calculation
|
|
frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
|
|
logger.info(f"THROUGHPUT: Cluster {cluster_id} - Found {len(frontends)} frontends in cache")
|
|
|
|
total_bytes_in = 0
|
|
total_bytes_out = 0
|
|
|
|
for frontend_name, frontend_stats in frontends.items():
|
|
bytes_in = frontend_stats.get('bytes_in', 0)
|
|
bytes_out = frontend_stats.get('bytes_out', 0)
|
|
total_bytes_in += bytes_in
|
|
total_bytes_out += bytes_out
|
|
logger.debug(f"THROUGHPUT: Frontend '{frontend_name}' - bytes_in={bytes_in}, bytes_out={bytes_out}")
|
|
|
|
# Convert to MB and GB
|
|
bytes_in_mb = round(total_bytes_in / (1024 * 1024), 2)
|
|
bytes_out_mb = round(total_bytes_out / (1024 * 1024), 2)
|
|
bytes_in_gb = round(total_bytes_in / (1024 * 1024 * 1024), 2)
|
|
bytes_out_gb = round(total_bytes_out / (1024 * 1024 * 1024), 2)
|
|
|
|
# Calculate rate (approximate based on last update)
|
|
# This is a simplified calculation - for production, you'd want to track deltas
|
|
rate_in_mbps = round(bytes_in_mb / 60, 2) # Rough estimate per minute
|
|
rate_out_mbps = round(bytes_out_mb / 60, 2)
|
|
|
|
result = {
|
|
'bytes_in': total_bytes_in,
|
|
'bytes_out': total_bytes_out,
|
|
'bytes_in_mb': bytes_in_mb,
|
|
'bytes_out_mb': bytes_out_mb,
|
|
'bytes_in_gb': bytes_in_gb,
|
|
'bytes_out_gb': bytes_out_gb,
|
|
'rate_in_mbps': rate_in_mbps,
|
|
'rate_out_mbps': rate_out_mbps,
|
|
'timestamp': datetime.utcnow().isoformat()
|
|
}
|
|
|
|
logger.info(f"THROUGHPUT: Cluster {cluster_id} - Result: {bytes_in_gb}GB in, {bytes_out_gb}GB out, {rate_in_mbps}MB/s in, {rate_out_mbps}MB/s out")
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get realtime throughput: {e}")
|
|
return {}
|
|
|
|
async def get_sessions_timeseries(self, cluster_id: int, hours: int = 24, limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""Get active sessions over time"""
|
|
try:
|
|
end_time = datetime.utcnow()
|
|
start_time = end_time - timedelta(hours=hours)
|
|
|
|
# Get all frontends
|
|
frontends = self.cache.get_all_frontend_stats(cluster_id)
|
|
|
|
# Collect session metrics from timeseries
|
|
all_session_data = []
|
|
|
|
for frontend_name in frontends.keys():
|
|
metric_name = f"frontend:{frontend_name}:sessions"
|
|
data = self.cache.get_timeseries_metrics(cluster_id, metric_name, start_time, end_time, limit)
|
|
all_session_data.extend(data)
|
|
|
|
# Aggregate by timestamp
|
|
from collections import defaultdict
|
|
aggregated = defaultdict(lambda: {'sessions': 0, 'count': 0})
|
|
|
|
for point in all_session_data:
|
|
ts = point.get('timestamp')
|
|
aggregated[ts]['sessions'] += point.get('value', 0)
|
|
aggregated[ts]['count'] += 1
|
|
aggregated[ts]['timestamp'] = ts
|
|
|
|
# Convert to list and calculate averages
|
|
result = []
|
|
for data in aggregated.values():
|
|
result.append({
|
|
'timestamp': data['timestamp'],
|
|
'sessions': data['sessions'],
|
|
'avg_sessions': round(data['sessions'] / data['count'], 2) if data['count'] > 0 else 0
|
|
})
|
|
|
|
# Sort by timestamp
|
|
result.sort(key=lambda x: x.get('timestamp', ''))
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get sessions timeseries: {e}")
|
|
return []
|
|
|
|
|
|
# Global service instance
|
|
dashboard_stats_service = DashboardStatsService()
|
|
|