-from .datamodel import KresConfig
+from .exceptions import KresBaseException
from .constants import VERSION
__version__ = VERSION
-__all__ = ["KresConfig"]
+__all__ = ["KresBaseException"]
--- /dev/null
+from typing import List
+
+from knot_resolver import KresBaseException
+
+
+class SubprocessControllerException(KresBaseException):
+ pass
+
+
+class SubprocessControllerExecException(Exception):
+ """
+ Exception that is used to deliberately terminate system startup
+ and make exec() of something else. This is used by the subprocess controller
+ as supervisord to run as the top-level process in a process tree hierarchy.
+ """
+
+ def __init__(self, exec_args: List[str], *args: object) -> None:
+ self.exec_args = exec_args
+ super().__init__(*args)
from knot_resolver.manager.constants import kresd_config_file, policy_loader_config_file
from knot_resolver.datamodel.config_schema import KresConfig
-from knot_resolver.manager.exceptions import SubprocessControllerException
+from knot_resolver.controller.exceptions import SubprocessControllerException
from knot_resolver.controller.registered_workers import register_worker, unregister_worker
from knot_resolver.utils.async_utils import writefile
import logging
from typing import TYPE_CHECKING, Dict, List, Tuple
-from knot_resolver.manager.exceptions import SubprocessControllerException
+from .exceptions import SubprocessControllerException
if TYPE_CHECKING:
from knot_resolver.controller.interface import KresID, Subprocess
from knot_resolver.compat.asyncio import async_in_a_thread
from knot_resolver.manager.constants import supervisord_config_file, supervisord_pid_file, supervisord_sock_file
from knot_resolver.datamodel.config_schema import KresConfig
-from knot_resolver.manager.exceptions import CancelStartupExecInsteadException, SubprocessControllerException
+from knot_resolver.controller.exceptions import SubprocessControllerExecException, SubprocessControllerException
from knot_resolver.controller.interface import (
KresID,
Subprocess,
logger.debug("Writing supervisord config")
await write_config_file(config)
logger.debug("Execing supervisord")
- raise CancelStartupExecInsteadException(
+ raise SubprocessControllerExecException(
[
str(which.which("supervisord")),
"supervisord",
--- /dev/null
+class KresBaseException(Exception):
+ """
+ Base class for all custom exceptions we use in Knot Resolver.
+ """
from typing import Any, Awaitable, Callable, List, Tuple
from knot_resolver.datamodel import KresConfig
-from knot_resolver.manager.exceptions import KresManagerException
+from knot_resolver import KresBaseException
from knot_resolver.utils.functional import Result
from knot_resolver.utils.modeling.exceptions import DataParsingError
from knot_resolver.utils.modeling.types import NoneType
err_res = filter(lambda r: r.is_err(), results)
errs = list(map(lambda r: r.unwrap_err(), err_res))
if len(errs) > 0:
- raise KresManagerException("Configuration validation failed. The reasons are:\n - " + "\n - ".join(errs))
+ raise KresBaseException("Configuration validation failed. The reasons are:\n - " + "\n - ".join(errs))
async with self._update_lock:
# update the stored config with the new version
+++ /dev/null
-from typing import List
-
-
-class CancelStartupExecInsteadException(Exception):
- """
- Exception used for terminating system startup and instead
- causing an exec of something else. Could be used by subprocess
- controllers such as supervisord to allow them to run as top-level
- process in a process tree.
- """
-
- def __init__(self, exec_args: List[str], *args: object) -> None:
- self.exec_args = exec_args
- super().__init__(*args)
-
-
-class KresManagerException(Exception):
- """
- Base class for all custom exceptions we use in our code
- """
-
-
-class SubprocessControllerException(KresManagerException):
- pass
-
-
-class SubprocessControllerTimeoutException(KresManagerException):
- pass
FIX_COUNTER_DECREASE_INTERVAL_SEC,
WATCHDOG_INTERVAL_SEC,
)
-from knot_resolver.manager.exceptions import SubprocessControllerException
+from knot_resolver.controller.exceptions import SubprocessControllerException
from knot_resolver.controller.interface import (
Subprocess,
SubprocessController,
from knot_resolver.utils.functional import Result
from knot_resolver.utils.modeling.types import NoneType
-from knot_resolver import KresConfig
+from knot_resolver.datamodel import KresConfig
logger = logging.getLogger(__name__)
from typing_extensions import Literal
import knot_resolver.utils.custom_atexit as atexit
+from knot_resolver import KresBaseException
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.datamodel.config_schema import KresConfig, get_rundir_without_validation
from knot_resolver.datamodel.globals import Context, set_global_validation_context
from knot_resolver.datamodel.management_schema import ManagementSchema
-from knot_resolver.manager.exceptions import CancelStartupExecInsteadException, KresManagerException
from knot_resolver.controller import get_best_controller_implementation
+from knot_resolver.controller.exceptions import SubprocessControllerExecException
from knot_resolver.controller.registered_workers import command_single_registered_worker
from knot_resolver.utils import ignore_exceptions_optional
from knot_resolver.utils.async_utils import readfile
return web.Response(text=f"validation of configuration failed:\n{e}", status=HTTPStatus.BAD_REQUEST)
except DataParsingError as e:
return web.Response(text=f"request processing error:\n{e}", status=HTTPStatus.BAD_REQUEST)
- except KresManagerException as e:
+ except KresBaseException as e:
return web.Response(text=f"request processing failed:\n{e}", status=HTTPStatus.INTERNAL_SERVER_ERROR)
except (DataParsingError, DataValidationError) as e:
logger.error(f"Failed to parse the updated configuration file: {e}")
logger.error("Configuration have NOT been changed.")
- except KresManagerException as e:
+ except KresBaseException as e:
logger.error(f"Reloading of the configuration file failed: {e}")
logger.error("Configuration have NOT been changed.")
nsite = web.TCPSite(self.runner, str(mgn.interface.addr), int(mgn.interface.port))
logger.info(f"Starting API HTTP server on http://{mgn.interface.addr}:{mgn.interface.port}")
else:
- raise KresManagerException("Requested API on unsupported configuration format.")
+ raise KresBaseException("Requested API on unsupported configuration format.")
await nsite.start()
# stop the old listen
# Initial configuration of the manager
if isinstance(config, Path):
if not config.exists():
- raise KresManagerException(
+ raise KresBaseException(
f"Manager is configured to load config file at {config} on startup, but the file does not exist."
)
else:
os.unlink(PID_FILE_NAME)
_lock_working_directory(attempt=attempt + 1)
return
- raise KresManagerException(
+ raise KresBaseException(
"Another manager is running in the same working directory."
f" PID file is located at {os.getcwd()}/{PID_FILE_NAME}"
) from e
else:
- raise KresManagerException(
+ raise KresBaseException(
"Another manager is running in the same working directory."
f" PID file is located at {os.getcwd()}/{PID_FILE_NAME}"
) from e
# After we have loaded the configuration, we can start worring about subprocess management.
manager = await _init_manager(config_store, server)
- except CancelStartupExecInsteadException as e:
+ except SubprocessControllerExecException as e:
# if we caught this exception, some component wants to perform a reexec during startup. Most likely, it would
# be a subprocess manager like supervisord, which wants to make sure the manager runs under supervisord in
# the process tree. So now we stop everything, and exec what we are told to. We are assuming, that the thing
# and finally exec what we were told to exec
os.execl(*e.exec_args)
- except KresManagerException as e:
+ except KresBaseException as e:
# We caught an error with a pretty error message. Just print it and exit.
logger.error(e)
return 1
from typing import Iterable, List
-from knot_resolver.manager.exceptions import KresManagerException
+from knot_resolver import KresBaseException
-class DataModelingBaseException(KresManagerException):
+class DataModelingBaseException(KresBaseException):
"""
Base class for all exceptions used in modelling.
"""
from pytest import raises
from knot_resolver.datamodel.types.base_types import IntRangeBase, StringLengthBase
-from knot_resolver.manager.exceptions import KresManagerException
+from knot_resolver import KresBaseException
@pytest.mark.parametrize("min,max", [(0, None), (None, 0), (1, 65535), (-65535, -1)])
invals.extend([random.randint(-sys.maxsize - 1, rmin - 1) for _ in range(n % 2)] if max else [])
for inval in invals:
- with raises(KresManagerException):
+ with raises(KresBaseException):
Test(inval)
invals.extend(["x" * random.randint(1, rmin - 1) for _ in range(n % 2)] if max else [])
for inval in invals:
- with raises(KresManagerException):
+ with raises(KresBaseException):
Test(inval)
projects / thirdparty / knot-resolver.git / commitdiff
