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
|
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from knot_resolver.controller.interface import KresID
from knot_resolver.datamodel.config_schema import KresConfig
from knot_resolver.manager.config_store import ConfigStore
LOGGING_LEVEL_STARTUP = logging.DEBUG
PID_FILE_NAME = "knot-resolver.pid"
FIX_COUNTER_ATTEMPTS_MAX = 2
FIX_COUNTER_DECREASE_INTERVAL_SEC = 30 * 60
PROCESSES_WATCHDOG_INTERVAL_SEC: float = 5
def kres_cache_dir(config: "KresConfig") -> Path:
return config.cache.storage.to_path()
def policy_loader_config_file(_config: "KresConfig") -> Path:
return Path("policy-loader.conf")
def kresd_config_file(_config: "KresConfig", kres_id: "KresID") -> Path:
return Path(f"kresd{int(kres_id)}.conf")
def kresd_config_file_supervisord_pattern(_config: "KresConfig") -> Path:
return Path("kresd%(process_num)d.conf")
def supervisord_config_file(_config: "KresConfig") -> Path:
return Path("supervisord.conf")
def supervisord_config_file_tmp(_config: "KresConfig") -> Path:
return Path("supervisord.conf.tmp")
def supervisord_pid_file(_config: "KresConfig") -> Path:
return Path("supervisord.pid")
def supervisord_sock_file(_config: "KresConfig") -> Path:
return Path("supervisord.sock")
def supervisord_subprocess_log_dir(_config: "KresConfig") -> Path:
return Path("logs")
class _UserConstants:
"""
Class for accessing constants, which are technically not constants as they are user configurable.
"""
def __init__(self, config_store: "ConfigStore", working_directory_on_startup: str) -> None:
self._config_store = config_store
self.working_directory_on_startup = working_directory_on_startup
_user_constants: Optional[_UserConstants] = None
async def init_user_constants(config_store: "ConfigStore", working_directory_on_startup: str) -> None:
global _user_constants
_user_constants = _UserConstants(config_store, working_directory_on_startup)
def user_constants() -> _UserConstants:
assert _user_constants is not None
return _user_constants
|
