admin_tool/resource_monitor.py at main · OpenChamp/admin_tool · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
"""
Resource Monitor for OpenChamp Admin Tool
Monitors container and process statistics
"""

from datetime import datetime
from typing import Dict, List, Optional

from service_manager import ServiceManager


class ResourceMonitor:
    """Monitors system and container resources"""

    def __init__(self, service_manager: ServiceManager):
        self.service_manager = service_manager

    def get_container_info(self) -> List[Dict]:
        """Get comprehensive info about containers"""
        containers = self.service_manager.get_podman_containers()
        info_list = []

        for container in containers:
            try:
                container_id = container.get('Id')
                if not container_id:
                    continue

                # Get container name
                names = container.get('Names', [])
                if isinstance(names, list):
                    name = names[0] if names else 'unknown'
                else:
                    name = names if names else 'unknown'

                if name.startswith('/'):
                    name = name[1:]

                # Get state
                state = container.get('State', 'unknown').lower()
                status = container.get('Status', state)

                # Calculate uptime
                up_time = 'N/A'
                started_at = container.get('StartedAt')
                if started_at and state == 'running':
                    try:
                        start_dt = datetime.fromisoformat(started_at.replace('Z', '+00:00'))
                        elapsed = datetime.now(start_dt.tzinfo) - start_dt
                        hours = int(elapsed.total_seconds() // 3600)
                        minutes = int((elapsed.total_seconds() % 3600) // 60)
                        up_time = f"{hours}h {minutes}m"
                    except Exception:
                        up_time = 'N/A'

                # Get stats
                stats = self.service_manager.get_container_stats(container_id[:12])
                cpu = '0%'
                memory = '0 MB'

                if stats:
                    cpu_str = stats.get('CPUPerc', '0%').replace('%', '')
                    mem_str = stats.get('MemUsage', '0B').split('/')[0]
                    cpu = f"{cpu_str}%"
                    memory = mem_str

                info = {
                    'name': name,
                    'id': container_id[:12],
                    'state': state,
                    'status': status,
                    'cpu': cpu,
                    'memory': memory,
                    'up_time': up_time
                }

                info_list.append(info)
            except Exception as e:
                continue

        return info_list

    def get_matchmaking_info(self) -> Optional[Dict]:
        """Get info about matchmaking server"""
        return self.service_manager.get_matchmaking_process_info()