blob: f9f73449bfff3c59f72feafeb41cd61f38d221b7 (
plain)
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
|
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Optional
from knot_resolver_manager.utils import which
if TYPE_CHECKING:
from knot_resolver_manager.config_store import ConfigStore
from knot_resolver_manager.datamodel.config_schema import KresConfig
from knot_resolver_manager.kresd_controller.interface import KresID
STARTUP_LOG_LEVEL = logging.DEBUG
DEFAULT_MANAGER_CONFIG_FILE = Path("/etc/knot-resolver/config.yml")
MANAGER_FIX_ATTEMPT_MAX_COUNTER = 2
FIX_COUNTER_DECREASE_INTERVAL_SEC = 30 * 60
PID_FILE_NAME = "manager.pid"
MAX_WORKERS = 256
def kresd_executable() -> Path:
return which.which("kresd")
def kres_gc_executable() -> Path:
return which.which("kres-cache-gc")
def kresd_cache_dir(config: "KresConfig") -> Path:
return config.cache.storage.to_path()
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")
WATCHDOG_INTERVAL: float = 5
"""
Used in KresdManager. It's a number of seconds in between system health checks.
"""
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
|
