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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
import argparse
import json
import os
import sys
from pathlib import Path
from typing import List, Optional, Tuple, Type
from knot_resolver.client.command import Command, CommandArgs, CompWords, comp_get_words, register_command
from knot_resolver.utils import which
from knot_resolver.utils.requests import request
PROCS_TYPE = List
@register_command
class DebugCommand(Command):
def __init__(self, namespace: argparse.Namespace) -> None:
self.proc_type: Optional[str] = namespace.proc_type
self.sudo: bool = namespace.sudo
self.gdb: str = namespace.gdb
self.print_only: bool = namespace.print_only
self.gdb_args: List[str] = namespace.extra if namespace.extra is not None else []
super().__init__(namespace)
@staticmethod
def register_args_subparser(
subparser: "argparse._SubParsersAction[argparse.ArgumentParser]",
) -> Tuple[argparse.ArgumentParser, "Type[Command]"]:
debug = subparser.add_parser(
"debug",
help="Run GDB on the manager's subprocesses",
)
debug.add_argument(
"--sudo",
dest="sudo",
help="Run GDB with sudo",
action="store_true",
default=False,
)
debug.add_argument(
"--gdb",
help="Custom GDB executable (may be a command on PATH, or an absolute path)",
type=str,
default=None,
)
debug.add_argument(
"--print-only",
help="Prints the GDB command line into stderr as a Python array, does not execute GDB",
action="store_true",
default=False,
)
debug.add_argument(
"proc_type",
help="Optional, the type of process to debug. May be 'kresd', 'gc', or 'all'.",
choices=["kresd", "gc", "all"],
type=str,
nargs="?",
default="kresd",
)
return debug, DebugCommand
@staticmethod
def completion(args: List[str], parser: argparse.ArgumentParser) -> CompWords:
return comp_get_words(args, parser)
def run(self, args: CommandArgs) -> None: # noqa: PLR0912, PLR0915
if self.gdb is None:
try:
gdb_cmd = str(which.which("gdb"))
except RuntimeError:
print("Could not find 'gdb' in $PATH. Is GDB installed?", file=sys.stderr)
sys.exit(1)
elif "/" not in self.gdb:
try:
gdb_cmd = str(which.which(self.gdb))
except RuntimeError:
print(f"Could not find '{self.gdb}' in $PATH.", file=sys.stderr)
sys.exit(1)
else:
gdb_cmd_path = Path(self.gdb).absolute()
if not gdb_cmd_path.exists():
print(f"Could not find '{self.gdb}'.", file=sys.stderr)
sys.exit(1)
gdb_cmd = str(gdb_cmd_path)
response = request(args.socket, "GET", f"processes/{self.proc_type}")
if response.status != 200:
print(response, file=sys.stderr)
sys.exit(1)
procs = json.loads(response.body)
if not isinstance(procs, PROCS_TYPE):
print(
f"Unexpected response type '{type(procs).__name__}' from manager. Expected '{PROCS_TYPE.__name__}'",
file=sys.stderr,
)
sys.exit(1)
if len(procs) == 0:
print(
f"There are no processes of type '{self.proc_type}' available to debug",
file=sys.stderr,
)
exec_args = []
# Put `sudo --` at the beginning of the command.
if self.sudo:
try:
sudo_cmd = str(which.which("sudo"))
except RuntimeError:
print("Could not find 'sudo' in $PATH. Is sudo installed?", file=sys.stderr)
sys.exit(1)
exec_args.extend([sudo_cmd, "--"])
# Attach GDB to processes - the processes are attached using the `add-inferior` and `attach` GDB
# commands. This way, we can debug multiple processes.
exec_args.extend([gdb_cmd, "--"])
exec_args.extend(["-init-eval-command", "set detach-on-fork off"])
exec_args.extend(["-init-eval-command", "set schedule-multiple on"])
exec_args.extend(["-init-eval-command", f'attach {procs[0]["pid"]}'])
inferior = 2
for proc in procs[1:]:
exec_args.extend(["-init-eval-command", "add-inferior"])
exec_args.extend(["-init-eval-command", f"inferior {inferior}"])
exec_args.extend(["-init-eval-command", f'attach {proc["pid"]}'])
inferior += 1
num_inferiors = inferior - 1
if num_inferiors > 1:
# Now we switch back to the first process and add additional provided GDB arguments.
exec_args.extend(["-init-eval-command", "inferior 1"])
exec_args.extend(
[
"-init-eval-command",
"echo \\n\\nYou are now debugging multiple Knot Resolver processes. To switch between "
"them, use the 'inferior <n>' command, where <n> is an integer from 1 to "
f"{num_inferiors}.\\n\\n",
]
)
exec_args.extend(self.gdb_args)
if self.print_only:
print(f"{exec_args}")
else:
os.execl(*exec_args)
|
