-rundir: etc/knot-resolver/runtime
+rundir: ./runtime
workers: 1
management:
interface: 127.0.0.1@5000
cache:
- storage: ../cache
+ storage: ./cache
logging:
level: notice
groups:
from knot_resolver_manager.datamodel.view_schema import ViewSchema
from knot_resolver_manager.datamodel.webmgmt_schema import WebmgmtSchema
from knot_resolver_manager.utils.modeling import ConfigSchema
+from knot_resolver_manager.utils.modeling.base_schema import lazy_default
logger = logging.getLogger(__name__)
rundir: UncheckedPath = UncheckedPath("/var/run/knot-resolver")
workers: Union[Literal["auto"], IntPositive] = IntPositive(1)
max_workers: IntPositive = IntPositive(_default_max_worker_count())
- management: ManagementSchema = ManagementSchema({"unix-socket": "./manager.sock"})
+ management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": "./manager.sock"})
webmgmt: Optional[WebmgmtSchema] = None
options: OptionsSchema = OptionsSchema()
network: NetworkSchema = NetworkSchema()
_ = KresConfig(data) # this should throw a descriptive error
assert False
- return UncheckedPath(rundir)
+ return UncheckedPath(rundir, object_path="/rundir")
--- /dev/null
+"""
+The parsing and validation of the datamodel is dependent on a global state:
+- a file system path used for resolving relative paths
+
+
+Commentary from @vsraier:
+=========================
+
+While this is not ideal, it is the best we can do at the moment. When I created this module,
+the datamodel was dependent on the global state implicitely. The validation procedures just read
+the current working directory. This module is the first step in removing the global dependency.
+
+At some point in the future, it might be interesting to add something like a "validation context"
+to the modelling tools. It is not technically complicated, but it requires
+massive model changes I am not willing to make at the moment. Ideally, when implementing this,
+the BaseSchema would turn into an empty class without any logic. Not even a constructor. All logic
+would be in the ObjectMapper class. Similar to how Gson works in Java or AutoMapper in C#.
+"""
+
+from pathlib import Path
+from typing import Optional
+
+
+class Context:
+ resolve_directory: Path
+
+ def __init__(self, resolve_directory: Path) -> None:
+ self.resolve_directory = resolve_directory
+
+
+_global_context: Optional[Context] = None
+
+
+def set_global_validation_context(context: Context) -> None:
+ global _global_context
+ _global_context = context
+
+
+def reset_global_validation_context() -> None:
+ global _global_context
+ _global_context = None
+
+
+def get_global_validation_context() -> Context:
+ if _global_context is None:
+ raise RuntimeError(
+ "Global validation context is not set! Before validation, you have to call `set_global_validation_context()` function!"
+ )
+
+ return _global_context
from pathlib import Path
from typing import Any, Dict, Tuple, Type, TypeVar
+from knot_resolver_manager.datamodel.globals import get_global_validation_context
from knot_resolver_manager.utils.modeling.base_value_type import BaseValueType
super().__init__(source_value, object_path=object_path)
self._object_path: str = object_path
self._parents: Tuple[UncheckedPath, ...] = parents
+
if isinstance(source_value, str):
+
+ # we do not load global validation context if the path is absolute
+ # this prevents errors when constructing defaults in the schema
+ if source_value.startswith("/"):
+ resolve_root = Path("/")
+ else:
+ resolve_root = get_global_validation_context().resolve_directory
+
self._raw_value: str = source_value
if self._parents:
pp = map(lambda p: p.to_path(), self._parents)
- self._value: Path = Path(*pp, source_value)
+ self._value: Path = Path(resolve_root, *pp, source_value)
else:
- self._value: Path = Path(source_value)
+ self._value: Path = Path(resolve_root, source_value)
else:
raise ValueError(f"expected file path in a string, got '{source_value}' with type '{type(source_value)}'.")
raise ValueError(f"path '{self._value}' does not point inside an existing directory")
if self._value.is_dir():
raise ValueError("path points to a directory when we expected a file")
-
-
-class CheckedPath(UncheckedPath):
- """
- Like UncheckedPath, but the file path is checked for being valid. So no non-existent directories in the middle,
- no symlink loops. This however means, that resolving of relative path happens while validating.
- """
-
- def __init__(
- self, source_value: Any, parents: Tuple["UncheckedPath", ...] = tuple(), object_path: str = "/"
- ) -> None:
- super().__init__(source_value, parents=parents, object_path=object_path)
- try:
- self._value = self._value.resolve(strict=False)
- except RuntimeError as e:
- raise ValueError("Failed to resolve given file path. Is there a symlink loop?") from e
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.datamodel.config_schema import KresConfig, get_rundir_without_validation
+from knot_resolver_manager.datamodel.globals import Context, set_global_validation_context
from knot_resolver_manager.datamodel.management_schema import ManagementSchema
from knot_resolver_manager.exceptions import CancelStartupExecInsteadException, KresManagerException
from knot_resolver_manager.kresd_controller import get_best_controller_implementation
sys.exit(128 + signal.SIGTERM)
-async def start_server(config: Union[Path, Dict[str, Any]] = DEFAULT_MANAGER_CONFIG_FILE) -> int:
+async def start_server(config: Path = DEFAULT_MANAGER_CONFIG_FILE) -> int:
# This function is quite long, but it describes how manager runs. So let's silence pylint
# pylint: disable=too-many-statements
# are fatal
try:
# Make sure that the config path does not change meaning when we change working directory
- if isinstance(config, Path):
- config = config.absolute()
+ config = config.absolute()
+
+ # before processing any configuration, set validation context
+ # - resolve_directory = root against which all relative paths will be resolved
+ set_global_validation_context(Context(resolve_directory=config.parent))
# Preprocess config - load from file or in general take it to the last step before validation.
config_raw = await _load_raw_config(config)
- # We want to change cwd as soon as possible. Especially before any real config validation, because cwd
- # is used for resolving relative paths. If we fail to read rundir from unparsed config, a full validation
- # error will come from here. If we are successfull, full validation will be done further on when initializing
- # the config store.
+ # We want to change cwd as soon as possible. Some parts of the codebase are using os.getcwd() to get the
+ # working directory.
+ #
+ # If we fail to read rundir from unparsed config, the first config validation error comes from here
_set_working_directory(config_raw)
# We don't want more than one manager in a single working directory. So we lock it with a PID file.
await statistics.init_monitoring(config_store)
# prepare instance of the server (no side effects)
- server = Server(config_store, config if isinstance(config, Path) else None)
+ server = Server(config_store, config)
# After we have loaded the configuration, we can start worring about subprocess management.
manager = await _init_manager(config_store, server)
import enum
import inspect
from abc import ABC, abstractmethod
-from typing import Any, Dict, List, Optional, Set, Tuple, Type, TypeVar, Union, cast
+from typing import Any, Callable, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union, cast
import yaml
return obj
+class _lazy_default(Generic[T], Serializable):
+ """
+ Wrapper for default values BaseSchema classes which deffers their instantiation until the schema
+ itself is being instantiated
+ """
+
+ def __init__(self, constructor: Callable[..., T], *args: Any, **kwargs: Any) -> None:
+ self._func = constructor
+ self._args = args
+ self._kwargs = kwargs
+
+ def instantiate(self) -> T:
+ return self._func(*self._args, **self._kwargs)
+
+ def to_dict(self) -> Dict[Any, Any]:
+ return Serializable.serialize(self.instantiate())
+
+
+def lazy_default(constructor: Callable[..., T], *args: Any, **kwargs: Any) -> T:
+ """We use a factory function because you can't lie about the return type in `__new__`"""
+ return _lazy_default(constructor, *args, **kwargs) # type: ignore
+
+
def _split_docstring(docstring: str) -> Tuple[str, Optional[str]]:
"""
Splits docstring into description of the class and description of attributes
msg = f"Failed to validate value against {tp} type"
raise DataValidationError(msg, object_path) from e
+ def _create_default(self, obj: Any) -> Any:
+ if isinstance(obj, _lazy_default):
+ return obj.instantiate() # type: ignore
+ else:
+ return obj
+
def map_object(
self,
tp: Type[Any],
# default values
if obj is None and use_default:
- return default
+ return self._create_default(default)
# NoneType
elif is_none_type(tp):
def _assign_default(self, obj: Any, name: str, python_type: Any, object_path: str) -> None:
cls = obj.__class__
- default = getattr(cls, name, None)
+ default = self._create_default(getattr(cls, name, None))
value = self.map_object(python_type, default, object_path=f"{object_path}/{name}")
setattr(obj, name, value)
+from pathlib import Path
+from knot_resolver_manager.datamodel.globals import Context, set_global_validation_context
+
+set_global_validation_context(Context(Path(".")))
projects / thirdparty / knot-resolver.git / commitdiff
