blob: 3595c0cf2e24a3e9fc99c5d5eb095ce8d934054b (
plain)
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
|
import time
import psutil
from datetime import datetime
from ATRI.service import Service
from ATRI.log import logger as log
from ATRI.rule import is_in_service
from ATRI.exceptions import GetStatusError
_status_msg = """
> Status Overview
[CPU: {cpu}%]
[Memory: {mem}%]
[Disk usage: {disk}%]
[Net sent: {inteSENT}MB]
[Net recv: {inteRECV}MB]
[Runtime: {up_time}]
{msg}
""".strip()
class IsSurvive(Service):
def __init__(self):
Service.__init__(self, "状态", "检查自身状态", rule=is_in_service("状态"))
@staticmethod
def ping() -> str:
return "I'm fine."
@staticmethod
def get_status():
log.info("开始检查资源消耗...")
try:
cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory().percent
disk = psutil.disk_usage("/").percent
inteSENT = psutil.net_io_counters().bytes_sent / 1000000 # type: ignore
inteRECV = psutil.net_io_counters().bytes_recv / 1000000 # type: ignore
now = time.time()
boot = psutil.boot_time()
up_time = str(
datetime.utcfromtimestamp(now).replace(microsecond=0)
- datetime.utcfromtimestamp(boot).replace(microsecond=0)
)
except GetStatusError:
raise GetStatusError("Failed to get status.")
msg = "アトリは、高性能ですから！"
if cpu > 90: # type: ignore
msg = "咱感觉有些头晕..."
is_ok = False
if mem > 90:
msg = "咱感觉有点头晕并且有点累..."
is_ok = False
elif mem > 90:
msg = "咱感觉有点累..."
is_ok = False
elif disk > 90:
msg = "咱感觉身体要被塞满了..."
is_ok = False
else:
log.info("资源占用正常")
is_ok = True
msg0 = _status_msg.format(
cpu=cpu,
mem=mem,
disk=disk,
inteSENT=inteSENT,
inteRECV=inteRECV,
up_time=up_time,
msg=msg,
)
return msg0, is_ok
|
