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
|
import os
from typing import Any, List, Literal, Optional, Set, Type, Union, cast
from knot_resolver.datamodel.types import WritableFilePath
from knot_resolver.utils.modeling import ConfigSchema
from knot_resolver.utils.modeling.base_schema import is_obj_type_valid
LogLevelEnum = Literal["crit", "err", "warning", "notice", "info", "debug"]
LogTargetEnum = Literal["syslog", "stderr", "stdout"]
LogGroupsProcessesEnum = Literal[
"manager",
"supervisord",
"policy-loader",
"kresd",
"cache-gc",
]
LogGroupsManagerEnum = Literal[
"files",
"metrics",
"server",
]
LogGroupsKresdEnum = Literal[
## Now the LOG_GRP_*_TAG defines, exactly from ../../../lib/log.h
"system",
"cache",
"io",
"net",
"ta",
"tasent",
"tasign",
"taupd",
"tls",
"gnutls",
"tls_cl",
"xdp",
"doh",
"dnssec",
"hint",
"plan",
"iterat",
"valdtr",
"resolv",
"select",
"zoncut",
"cookie",
"statis",
"rebind",
"worker",
"policy",
"daf",
"timejm",
"timesk",
"graphi",
"prefil",
"primin",
"srvstl",
"wtchdg",
"nsid",
"dnstap",
"tests",
"dotaut",
"http",
"contrl",
"module",
"devel",
"renum",
"exterr",
"rules",
"prlayr",
"defer",
# "reqdbg",... (non-displayed section of the enum)
]
LogGroupsEnum = Literal[LogGroupsProcessesEnum, LogGroupsManagerEnum, LogGroupsKresdEnum]
class DnstapSchema(ConfigSchema):
"""
Logging DNS queries and responses to a unix socket.
---
enable: Enable/disable DNS queries logging.
unix_socket: Path to unix domain socket where dnstap messages will be sent.
log_queries: Log queries from downstream in wire format.
log_responses: Log responses to downstream in wire format.
log_tcp_rtt: Log TCP RTT (Round-trip time).
"""
enable: bool = False
unix_socket: Optional[WritableFilePath] = None
log_queries: bool = False
log_responses: bool = False
log_tcp_rtt: bool = False
def _validate(self) -> None:
if self.enable and self.unix_socket is None:
raise ValueError("DNS queries logging enabled, but 'unix-socket' not specified")
class LoggingSchema(ConfigSchema):
class Raw(ConfigSchema):
"""
Logging and debugging configuration.
---
level: Global logging level.
target: Global logging stream target. "from-env" uses $KRES_LOGGING_TARGET and defaults to "stdout".
groups: List of groups for which 'debug' logging level is set.
dnstap: Logging DNS requests and responses to a unix socket.
"""
level: LogLevelEnum = "notice"
target: Union[LogTargetEnum, Literal["from-env"]] = "from-env"
groups: Optional[List[LogGroupsEnum]] = None
dnstap: DnstapSchema = DnstapSchema()
_LAYER = Raw
level: LogLevelEnum
target: LogTargetEnum
groups: Optional[List[LogGroupsEnum]]
dnstap: DnstapSchema
def _target(self, raw: Raw) -> LogTargetEnum:
if raw.target == "from-env":
target = os.environ.get("KRES_LOGGING_TARGET") or "stdout"
if not is_obj_type_valid(target, cast(Type[Any], LogTargetEnum)):
raise ValueError(f"logging target '{target}' read from $KRES_LOGGING_TARGET is invalid")
return cast(LogTargetEnum, target)
return raw.target
def _validate(self):
if self.groups is None:
return
checked: Set[str] = set()
for i, g in enumerate(self.groups):
if g in checked:
raise ValueError(f"duplicate logging group '{g}' on index {i}")
checked.add(g)
|
