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
|
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"]
LogGroupsEnum = Literal[
"manager",
"supervisord",
"cache-gc",
## 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)
]
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)
|
