-from .datamodel.config_schema import KresConfig
+from .datamodel import KresConfig
__version__ = "0.1.0"
from typing import Dict, List, Optional, Tuple, Type, TypeVar
from urllib.parse import quote
-from knot_resolver.manager.constants import API_SOCK_ENV_VAR, CONFIG_FILE_ENV_VAR, DEFAULT_MANAGER_CONFIG_FILE
-from knot_resolver.datamodel.config_schema import DEFAULT_MANAGER_API_SOCK
+from knot_resolver.constants import (
+ CONFIG_FILE_PATH_DEFAULT,
+ CONFIG_FILE_PATH_ENV_VAR,
+ API_SOCK_PATH_ENV_VAR,
+ API_SOCK_PATH_DEFAULT,
+)
from knot_resolver.datamodel.types import IPAddressPort
from knot_resolver.utils.modeling import parsing
from knot_resolver.utils.modeling.exceptions import DataValidationError
if len(namespace.socket) > 0:
return SocketDesc(namespace.socket[0], "--socket argument")
- config_path = os.getenv(CONFIG_FILE_ENV_VAR)
- socket_env = os.getenv(API_SOCK_ENV_VAR)
+ config_path = os.getenv(CONFIG_FILE_PATH_ENV_VAR)
+ socket_env = os.getenv(API_SOCK_PATH_ENV_VAR)
socket: Optional[SocketDesc] = None
# 2) socket from config file ('--config' argument)
socket = get_socket_from_config(Path(config_path), False)
# 4) socket from environment variable
elif socket_env:
- socket = SocketDesc(socket_env, f'Environment variable "{API_SOCK_ENV_VAR}"')
+ socket = SocketDesc(socket_env, f'Environment variable "{API_SOCK_PATH_ENV_VAR}"')
# 5) socket from config file (default config file constant)
else:
- socket = get_socket_from_config(DEFAULT_MANAGER_CONFIG_FILE, True)
+ socket = get_socket_from_config(CONFIG_FILE_PATH_DEFAULT, True)
if socket:
return socket
# 6) socket default
- return SocketDesc(DEFAULT_MANAGER_API_SOCK, f'Default value "{DEFAULT_MANAGER_API_SOCK}"')
+ return SocketDesc(str(API_SOCK_PATH_DEFAULT), f'Default value "{API_SOCK_PATH_DEFAULT}"')
class CommandArgs:
--- /dev/null
+import logging
+
+from importlib.metadata import version
+from importlib.util import find_spec
+from pathlib import Path
+
+# Installed Knot Resolver build options from Meson is semi-optional.
+# They are needed to run the resolver, but not for its unit tests.
+if find_spec("knot_resolver_build_options"):
+ import knot_resolver_build_options as build_conf # type: ignore[import-not-found]
+else:
+ build_conf = None
+
+VERSION = version("knot_resolver") if find_spec("knot_resolver") else "6"
+WORKERS_MAX_DEFAULT = 256
+LOGGING_LEVEL_STARTUP = logging.DEBUG
+PID_FILE_NAME = "knot-resolver.pid"
+
+FIX_COUNTER_ATTEMPTS_MAX = 2
+FIX_COUNTER_DECREASE_INTERVAL_SEC = 30 * 60
+WATCHDOG_INTERVAL_SEC: float = 5
+
+USER_DEFAULT = build_conf.user if build_conf else "knot-resolver"
+GROUP_DEFAULT = build_conf.group if build_conf else "knot-resolver"
+
+RUN_DIR_DEFAULT: Path = build_conf.run_dir if build_conf else Path("/var/run/knot-resolver")
+ETC_DIR_DEFAULT: Path = build_conf.etc_dir if build_conf else Path("/etc/knot-resolver")
+CONFIG_FILE_PATH_DEFAULT = ETC_DIR_DEFAULT / "config.yaml"
+CONFIG_FILE_PATH_ENV_VAR = "KRES_MANAGER_CONFIG"
+API_SOCK_PATH_DEFAULT = RUN_DIR_DEFAULT / "kres-api.sock"
+API_SOCK_PATH_ENV_VAR = "KRES_MANAGER_API_SOCK"
+
+
+def kresd_executable() -> Path:
+ assert build_conf is not None
+ return build_conf.sbin_dir / "kresd"
+
+
+def kres_cache_gc_executable() -> Path:
+ assert build_conf is not None
+ return build_conf.sbin_dir / "kres-cache-gc"
from jinja2 import Template
from typing_extensions import Literal
+from knot_resolver.constants import (
+ kres_cache_gc_executable,
+ kresd_executable,
+)
from knot_resolver.manager.constants import (
- kres_gc_executable,
- kresd_cache_dir,
+ kres_cache_dir,
kresd_config_file_supervisord_pattern,
- kresd_executable,
policy_loader_config_file,
supervisord_config_file,
supervisord_config_file_tmp,
return ProcessTypeConfig( # type: ignore[call-arg]
logfile=supervisord_subprocess_log_dir(config) / "gc.log",
workdir=cwd,
- command=f"{kres_gc_executable()} -c {kresd_cache_dir(config)}{kres_cache_gc_args(config)}",
+ command=f"{kres_cache_gc_executable()} -c {kres_cache_dir(config)}{kres_cache_gc_args(config)}",
environment="",
)
from typing_extensions import Literal
-from knot_resolver.manager.constants import MAX_WORKERS
+from knot_resolver.constants import RUN_DIR_DEFAULT, API_SOCK_PATH_DEFAULT, WORKERS_MAX_DEFAULT
+
from knot_resolver.datamodel.cache_schema import CacheSchema
from knot_resolver.datamodel.dns64_schema import Dns64Schema
from knot_resolver.datamodel.dnssec_schema import DnssecSchema
from knot_resolver.utils.modeling.base_schema import lazy_default
from knot_resolver.utils.modeling.exceptions import AggregateDataValidationError, DataValidationError
-_DEFAULT_RUNDIR = "/var/run/knot-resolver"
-
-DEFAULT_MANAGER_API_SOCK = _DEFAULT_RUNDIR + "/manager.sock"
-
logger = logging.getLogger(__name__)
c = _cpu_count()
if c:
return c * 10
- return MAX_WORKERS
+ return WORKERS_MAX_DEFAULT
def _get_views_tags(views: List[ViewSchema]) -> List[str]:
version: int = 1
nsid: Optional[EscapedStr] = None
hostname: Optional[EscapedStr] = None
- rundir: WritableDir = lazy_default(WritableDir, _DEFAULT_RUNDIR)
+ rundir: WritableDir = lazy_default(WritableDir, str(RUN_DIR_DEFAULT))
workers: Union[Literal["auto"], IntPositive] = IntPositive(1)
max_workers: IntPositive = IntPositive(_default_max_worker_count())
- management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": DEFAULT_MANAGER_API_SOCK})
+ management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": str(API_SOCK_PATH_DEFAULT)})
webmgmt: Optional[WebmgmtSchema] = None
options: OptionsSchema = OptionsSchema()
network: NetworkSchema = NetworkSchema()
Used for initial manager startup.
"""
- return WritableDir(data["rundir"] if "rundir" in data else _DEFAULT_RUNDIR, object_path="/rundir")
+ return WritableDir(data["rundir"] if "rundir" in data else RUN_DIR_DEFAULT, object_path="/rundir")
from pwd import getpwnam
from typing import Any, Dict, Tuple, Type, TypeVar
-from knot_resolver.manager.constants import kresd_group, kresd_user
+from knot_resolver.constants import USER_DEFAULT, GROUP_DEFAULT
from knot_resolver.datamodel.globals import get_resolve_root, get_strict_validation
from knot_resolver.utils.modeling.base_value_type import BaseValueType
_PermissionMode.EXECUTE: [stat.S_IXUSR, stat.S_IXGRP, stat.S_IXOTH],
}
- username = kresd_user()
- groupname = kresd_group()
-
- if username is None or groupname is None:
- return True
-
- user_uid = getpwnam(username).pw_uid
- user_gid = getgrnam(groupname).gr_gid
+ user_uid = getpwnam(USER_DEFAULT).pw_uid
+ user_gid = getgrnam(GROUP_DEFAULT).gr_gid
dest_stat = os.stat(dest_path)
dest_uid = dest_stat.st_uid
super().__init__(source_value, parents=parents, object_path=object_path)
if self.strict_validation and not _kres_accessible(self._value, _PermissionMode.READ):
- raise ValueError(f"{kresd_user()}:{kresd_group()} has insufficient permissions to read '{self._value}'")
+ raise ValueError(f"{USER_DEFAULT}:{GROUP_DEFAULT} has insufficient permissions to read '{self._value}'")
class WritableDir(Dir):
self._value, _PermissionMode.WRITE | _PermissionMode.EXECUTE
):
raise ValueError(
- f"{kresd_user()}:{kresd_group()} has insufficient permissions to write/execute '{self._value}'"
+ f"{USER_DEFAULT}:{GROUP_DEFAULT} has insufficient permissions to write/execute '{self._value}'"
)
self._value.parent, _PermissionMode.WRITE | _PermissionMode.EXECUTE
):
raise ValueError(
- f"{kresd_user()}:{kresd_group()} has insufficient permissions to write/execute'{self._value.parent}'"
+ f"{USER_DEFAULT}:{GROUP_DEFAULT} has insufficient permissions to write/execute'{self._value.parent}'"
)
-import importlib.util
-import logging
from pathlib import Path
from typing import TYPE_CHECKING, Optional
-# Install config is semi-optional - only needed to actually run Manager, but not
-# for its unit tests.
-if importlib.util.find_spec("knot_resolver_build_options"):
- import knot_resolver_build_options # type: ignore[import-not-found]
-else:
- knot_resolver_build_options = None
-
if TYPE_CHECKING:
from knot_resolver.manager.config_store import ConfigStore
from knot_resolver.datamodel.config_schema import KresConfig
from knot_resolver.controller.interface import KresID
-STARTUP_LOG_LEVEL = logging.DEBUG
-DEFAULT_MANAGER_CONFIG_FILE = Path("/etc/knot-resolver/config.yaml")
-CONFIG_FILE_ENV_VAR = "KRES_MANAGER_CONFIG"
-API_SOCK_ENV_VAR = "KRES_MANAGER_API_SOCK"
-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:
- assert knot_resolver_build_options is not None
- return knot_resolver_build_options.sbin_dir / "kresd"
-
-
-def kres_gc_executable() -> Path:
- assert knot_resolver_build_options is not None
- return knot_resolver_build_options.sbin_dir / "kres-cache-gc"
-
-def kresd_user():
- return None if knot_resolver is None else knot_resolver.user
-
-
-def kresd_group():
- return None if knot_resolver is None else knot_resolver.group
-
-
-def kresd_cache_dir(config: "KresConfig") -> Path:
+def kres_cache_dir(config: "KresConfig") -> Path:
return config.cache.storage.to_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.
only_on_real_changes_update,
only_on_real_changes_verifier,
)
-from knot_resolver.manager.constants import (
+from knot_resolver.constants import (
+ FIX_COUNTER_ATTEMPTS_MAX,
FIX_COUNTER_DECREASE_INTERVAL_SEC,
- MANAGER_FIX_ATTEMPT_MAX_COUNTER,
- WATCHDOG_INTERVAL,
+ WATCHDOG_INTERVAL_SEC,
)
from knot_resolver.manager.exceptions import SubprocessControllerException
from knot_resolver.controller.interface import (
return str(self._counter)
def is_too_high(self) -> bool:
- return self._counter >= MANAGER_FIX_ATTEMPT_MAX_COUNTER
+ return self._counter >= FIX_COUNTER_ATTEMPTS_MAX
async def _deny_max_worker_changes(config_old: KresConfig, config_new: KresConfig) -> Result[None, str]:
async def _watchdog(self) -> None: # pylint: disable=too-many-branches
while True:
- await asyncio.sleep(WATCHDOG_INTERVAL)
+ await asyncio.sleep(WATCHDOG_INTERVAL_SEC)
self._fix_counter.try_decrease()
import sys
from typing import Optional
+from knot_resolver.constants import LOGGING_LEVEL_STARTUP
from knot_resolver.manager.config_store import ConfigStore, only_on_real_changes_update
-from knot_resolver.manager.constants import STARTUP_LOG_LEVEL
from knot_resolver.datamodel.config_schema import KresConfig
from knot_resolver.datamodel.logging_schema import LogTargetEnum
def logger_startup() -> None:
- logging.getLogger().setLevel(STARTUP_LOG_LEVEL)
+ logging.getLogger().setLevel(LOGGING_LEVEL_STARTUP)
err_handler = logging.StreamHandler(sys.stderr)
err_handler.setFormatter(logging.Formatter(logging.BASIC_FORMAT))
logging.getLogger().addHandler(logging.handlers.MemoryHandler(10_000, logging.ERROR, err_handler))
from typing import NoReturn
from knot_resolver import compat
-from knot_resolver.manager.constants import CONFIG_FILE_ENV_VAR, DEFAULT_MANAGER_CONFIG_FILE
+from knot_resolver.constants import CONFIG_FILE_PATH_DEFAULT, CONFIG_FILE_PATH_ENV_VAR
from knot_resolver.manager.log import logger_startup
from knot_resolver.manager.server import start_server
parser.add_argument(
"-c",
"--config",
- help="Config file to load. Overrides default config location at '" + str(DEFAULT_MANAGER_CONFIG_FILE) + "'",
+ help="Config file to load. Overrides default config location at '" + str(CONFIG_FILE_PATH_ENV_VAR) + "'",
type=str,
nargs=1,
required=False,
args = parse_args()
# where to look for config
- config_env = os.getenv(CONFIG_FILE_ENV_VAR)
+ config_env = os.getenv(CONFIG_FILE_PATH_ENV_VAR)
if args.config is not None:
config_path = Path(args.config[0])
elif config_env is not None:
config_path = Path(config_env)
else:
- config_path = DEFAULT_MANAGER_CONFIG_FILE
+ config_path = CONFIG_FILE_PATH_DEFAULT
exit_code = compat.asyncio.run(start_server(config=config_path))
sys.exit(exit_code)
from typing_extensions import Literal
import knot_resolver.utils.custom_atexit as atexit
+from knot_resolver.constants import CONFIG_FILE_PATH_DEFAULT, PID_FILE_NAME
from knot_resolver.manager import log, statistics
from knot_resolver.compat import asyncio as asyncio_compat
from knot_resolver.manager.config_store import ConfigStore
-from knot_resolver.manager.constants import DEFAULT_MANAGER_CONFIG_FILE, PID_FILE_NAME, init_user_constants
+from knot_resolver.manager.constants import init_user_constants
from knot_resolver.datamodel.cache_schema import CacheClearRPCSchema
from knot_resolver.datamodel.config_schema import KresConfig, get_rundir_without_validation
from knot_resolver.datamodel.globals import Context, set_global_validation_context
sys.exit(128 + signal.SIGTERM)
-async def start_server(config: Path = DEFAULT_MANAGER_CONFIG_FILE) -> int:
+async def start_server(config: Path = CONFIG_FILE_PATH_DEFAULT) -> int:
# This function is quite long, but it describes how manager runs. So let's silence pylint
# pylint: disable=too-many-statements
projects / thirdparty / knot-resolver.git / commitdiff
