5
5
import asyncio
6
6
import subprocess
7
7
import platform
8
+ import os
9
+ import shutil
8
10
9
11
from .logs import setup_logging
10
12
from .logs import logger
11
13
12
14
setup_logging ()
13
15
16
+ # dmesg process created at boot time
17
+ dmesg_proc = None
18
+
14
19
15
20
def get_kernel_messages ():
16
21
"""
@@ -25,12 +30,37 @@ def get_kernel_messages():
25
30
output , _ = process .communicate ()
26
31
return output .decode ("utf-8" )
27
32
elif current_platform == "Linux" :
28
- with open ("/var/log/dmesg" , "r" ) as file :
33
+ log_path = get_dmesg_log_path ()
34
+ with open (log_path , 'r' ) as file :
29
35
return file .read ()
30
36
else :
31
37
logger .info ("Unsupported platform." )
32
38
33
39
40
+ def get_dmesg_log_path ():
41
+ """
42
+ Check for the existence of a readable dmesg log file and return its path.
43
+ Create an accessible path if not found.
44
+ """
45
+ if os .access ('/var/log/dmesg' , os .F_OK | os .R_OK ):
46
+ return '/var/log/dmesg'
47
+
48
+ global dmesg_proc
49
+ dmesg_log_path = '/tmp/dmesg'
50
+ if dmesg_proc :
51
+ return dmesg_log_path
52
+
53
+ logger .info ("Created /tmp/dmesg." )
54
+ subprocess .run (['touch' , dmesg_log_path ])
55
+ dmesg_path = shutil .which ('dmesg' )
56
+ if dmesg_path :
57
+ logger .info (f"Writing to { dmesg_log_path } from dmesg." )
58
+ dmesg_proc = subprocess .Popen ([dmesg_path , '--follow' ], text = True , stdout = subprocess .PIPE )
59
+ subprocess .Popen (['tee' , dmesg_log_path ], text = True , stdin = dmesg_proc .stdout , stdout = subprocess .DEVNULL )
60
+
61
+ return dmesg_log_path
62
+
63
+
34
64
def custom_filter (message ):
35
65
# Check for {TO_INTERPRETER{ message here }TO_INTERPRETER} pattern
36
66
if "{TO_INTERPRETER{" in message and "}TO_INTERPRETER}" in message :
0 commit comments