AIOpsLab/aiopslab/generators/fault/helpers.py at main · VitrixLab-Internship-Initiative-Program/AIOpsLab · 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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import subprocess

# SocialNetwork service process names
sn_svc_process_names = [
    "ComposePostServ",
    "HomeTimelineSer",
    "MediaService",
    "PostStorageServ",
    "SocialGraphServ",
    "TextService",
    "UserService",
    "UrlShortenServi",
    "UserMentionServ",
    "UserTimelineSer",
    "UniqueIdService",
]

# SocialNetwork MongoDB process names
sn_mongod_process_names = ["mongod"]

# SocialNetwork Redis process names
sn_redis_process_names = ["redis-server"]

# SocialNetwork Memcached process names
sn_memcached_process_names = ["memcached"]

# HotelResearvation service process names
hr_svc_process_names = [
    "geo",
    "frontend",
    "consul",
    "profile",
    "rate",
    "recommendation",
    "reservation",
    "search",
    "user",
]

# HotelResearvation MongoDB process names
hr_mongod_process_names = ["mongod"]

# HotelResearvation Memcached process names
hr_memcached_process_names = ["memcached"]


def get_pids_by_name_contain(search_term):
    """
    Get a list of PIDs for processes whose command contains the given search term.

    :param search_term: The term to search for in process names (case-sensitive).
    :return: A list of PIDs (integers) matching the search term.
    """
    try:
        result = subprocess.run(
            ["ps", "-e", "-o", "pid,comm"],
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise RuntimeError(f"Error running ps command: {result.stderr.strip()}")

        # Filter the output for lines containing the search term
        matching_pids = []
        for line in result.stdout.splitlines():
            if search_term in line:
                parts = line.split(maxsplit=1)
                if parts:  # Ensure we have at least one part
                    pid = parts[0]
                    if pid.isdigit():
                        matching_pids.append(int(pid))

        return matching_pids

    except Exception as e:
        print(f"Error: {e}")
        return []


def get_pids_by_name(search_term):
    """
    Get a list of PIDs for processes whose command exactly match the given search term.

    :param search_term: The term to search for in process names (case-sensitive).
    :return: A list of PIDs (integers) matching exactly the search term.
    """
    try:
        result = subprocess.run(
            ["ps", "-e", "-o", "pid,comm"],
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if result.returncode != 0:
            raise RuntimeError(f"Error running ps command: {result.stderr.strip()}")

        # Filter the output for lines containing the search term
        matching_pids = []
        for line in result.stdout.splitlines():
            parts = line.split(maxsplit=1)
            if len(parts) == 2:  # Ensure we have both PID and command
                pid, command = parts
                if command == search_term:  # Exact match check
                    if pid.isdigit():
                        matching_pids.append(int(pid))

        return matching_pids

    except Exception as e:
        print(f"Error: {e}")
        return []


if __name__ == "__main__":
    search_term = "HomeTimelineSer"
    pids = get_pids_by_name(search_term)
    print(f"Processes with '{search_term}' in their name: {pids}")