Merge branch 'master' into rrl-wip

1 files changed, 639 insertions, 0 deletions

diff --git a/python/knot_resolver/manager/server.py b/python/knot_resolver/manager/server.py
new file mode 100644
index 00000000..972b167f
--- /dev/null
+++ b/python/knot_resolver/manager/server.py
@@ -0,0 +1,639 @@
+import asyncio
+import errno
+import json
+import logging
+import os
+import signal
+import sys
+from functools import partial
+from http import HTTPStatus
+from pathlib import Path
+from time import time
+from typing import Any, Dict, List, Literal, Optional, Set, Union, cast
+
+from aiohttp import web
+from aiohttp.web import middleware
+from aiohttp.web_app import Application
+from aiohttp.web_response import json_response
+from aiohttp.web_runner import AppRunner, TCPSite, UnixSite
+
+from knot_resolver.constants import CONFIG_FILE
+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.datamodel import kres_config_json_schema
+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
+from knot_resolver.datamodel.management_schema import ManagementSchema
+from knot_resolver.manager import metrics
+from knot_resolver.utils import custom_atexit as atexit
+from knot_resolver.utils import ignore_exceptions_optional
+from knot_resolver.utils.async_utils import readfile
+from knot_resolver.utils.compat import asyncio as asyncio_compat
+from knot_resolver.utils.etag import structural_etag
+from knot_resolver.utils.functional import Result
+from knot_resolver.utils.modeling.exceptions import AggregateDataValidationError, DataParsingError, DataValidationError
+from knot_resolver.utils.modeling.parsing import DataFormat, try_to_parse
+from knot_resolver.utils.modeling.query import query
+from knot_resolver.utils.modeling.types import NoneType
+from knot_resolver.utils.systemd_notify import systemd_notify
+
+from .config_store import ConfigStore
+from .constants import PID_FILE_NAME, init_user_constants
+from .exceptions import KresManagerException
+from .logging import logger_init
+from .manager import KresManager
+
+logger = logging.getLogger(__name__)
+
+
+@middleware
+async def error_handler(request: web.Request, handler: Any) -> web.Response:
+    """
+    Generic error handler for route handlers.
+
+    If an exception is thrown during request processing, this middleware catches it
+    and responds accordingly.
+    """
+
+    try:
+        return await handler(request)
+    except DataValidationError as e:
+        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:
+        return web.Response(text=f"request processing failed:\n{e}", status=HTTPStatus.INTERNAL_SERVER_ERROR)
+
+
+def from_mime_type(mime_type: str) -> DataFormat:
+    formats = {
+        "application/json": DataFormat.JSON,
+        "application/octet-stream": DataFormat.JSON,  # default in aiohttp
+    }
+    if mime_type not in formats:
+        raise DataParsingError(f"unsupported MIME type '{mime_type}', expected: {str(formats)[1:-1]}")
+    return formats[mime_type]
+
+
+def parse_from_mime_type(data: str, mime_type: str) -> Any:
+    return from_mime_type(mime_type).parse_to_dict(data)
+
+
+class Server:
+    # pylint: disable=too-many-instance-attributes
+    # This is top-level class containing pretty much everything. Instead of global
+    # variables, we use instance attributes. That's why there are so many and it's
+    # ok.
+    def __init__(self, store: ConfigStore, config_path: Optional[Path]):
+        # config store & server dynamic reconfiguration
+        self.config_store = store
+
+        # HTTP server
+        self.app = Application(middlewares=[error_handler])
+        self.runner = AppRunner(self.app)
+        self.listen: Optional[ManagementSchema] = None
+        self.site: Union[NoneType, TCPSite, UnixSite] = None
+        self.listen_lock = asyncio.Lock()
+        self._config_path: Optional[Path] = config_path
+        self._exit_code: int = 0
+        self._shutdown_event = asyncio.Event()
+
+    async def _reconfigure(self, config: KresConfig) -> None:
+        await self._reconfigure_listen_address(config)
+
+    async def _deny_management_changes(self, config_old: KresConfig, config_new: KresConfig) -> Result[None, str]:
+        if config_old.management != config_new.management:
+            return Result.err(
+                "/server/management: Changing management API address/uTruenix-socket dynamically is not allowed as it's really dangerous."
+                " If you really need this feature, please contact the developers and explain why. Technically,"
+                " there are no problems in supporting it. We are only blocking the dynamic changes because"
+                " we think the consequences of leaving this footgun unprotected are worse than its usefulness."
+            )
+        return Result.ok(None)
+
+    async def _reload_config(self) -> None:
+        if self._config_path is None:
+            logger.warning("The manager was started with inlined configuration - can't reload")
+        else:
+            try:
+                data = await readfile(self._config_path)
+                config = KresConfig(try_to_parse(data))
+                await self.config_store.update(config)
+                logger.info("Configuration file successfully reloaded")
+            except FileNotFoundError:
+                logger.error(
+                    f"Configuration file was not found at '{self._config_path}'."
+                    " Something must have happened to it while we were running."
+                )
+                logger.error("Configuration have NOT been changed.")
+            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:
+                logger.error(f"Reloading of the configuration file failed: {e}")
+                logger.error("Configuration have NOT been changed.")
+
+    async def sigint_handler(self) -> None:
+        logger.info("Received SIGINT, triggering graceful shutdown")
+        self.trigger_shutdown(0)
+
+    async def sigterm_handler(self) -> None:
+        logger.info("Received SIGTERM, triggering graceful shutdown")
+        self.trigger_shutdown(0)
+
+    async def sighup_handler(self) -> None:
+        logger.info("Received SIGHUP, reloading configuration file")
+        systemd_notify(RELOADING="1")
+        await self._reload_config()
+        systemd_notify(READY="1")
+
+    @staticmethod
+    def all_handled_signals() -> Set[signal.Signals]:
+        return {signal.SIGHUP, signal.SIGINT, signal.SIGTERM}
+
+    def bind_signal_handlers(self):
+        asyncio_compat.add_async_signal_handler(signal.SIGTERM, self.sigterm_handler)
+        asyncio_compat.add_async_signal_handler(signal.SIGINT, self.sigint_handler)
+        asyncio_compat.add_async_signal_handler(signal.SIGHUP, self.sighup_handler)
+
+    def unbind_signal_handlers(self):
+        asyncio_compat.remove_signal_handler(signal.SIGTERM)
+        asyncio_compat.remove_signal_handler(signal.SIGINT)
+        asyncio_compat.remove_signal_handler(signal.SIGHUP)
+
+    async def start(self) -> None:
+        self._setup_routes()
+        await self.runner.setup()
+        await self.config_store.register_verifier(self._deny_management_changes)
+        await self.config_store.register_on_change_callback(self._reconfigure)
+
+    async def wait_for_shutdown(self) -> None:
+        await self._shutdown_event.wait()
+
+    def trigger_shutdown(self, exit_code: int) -> None:
+        self._shutdown_event.set()
+        self._exit_code = exit_code
+
+    async def _handler_index(self, _request: web.Request) -> web.Response:
+        """
+        Dummy index handler to indicate that the server is indeed running...
+        """
+        return json_response(
+            {
+                "msg": "Knot Resolver Manager is running! The configuration endpoint is at /config",
+                "status": "RUNNING",
+            }
+        )
+
+    async def _handler_config_query(self, request: web.Request) -> web.Response:
+        """
+        Route handler for changing resolver configuration
+        """
+        # There are a lot of local variables in here, but they are usually immutable (almost SSA form :) )
+        # pylint: disable=too-many-locals
+
+        # parse the incoming data
+        if request.method == "GET":
+            update_with: Optional[Dict[str, Any]] = None
+        else:
+            update_with = parse_from_mime_type(await request.text(), request.content_type)
+        document_path = request.match_info["path"]
+        getheaders = ignore_exceptions_optional(List[str], None, KeyError)(request.headers.getall)
+        etags = getheaders("if-match")
+        not_etags = getheaders("if-none-match")
+        current_config: Dict[str, Any] = self.config_store.get().get_unparsed_data()
+
+        # stop processing if etags
+        def strip_quotes(s: str) -> str:
+            return s.strip('"')
+
+        # WARNING: this check is prone to race conditions. When changing, make sure that the current config
+        # is really the latest current config (i.e. no await in between obtaining the config and the checks)
+        status = HTTPStatus.NOT_MODIFIED if request.method in ("GET", "HEAD") else HTTPStatus.PRECONDITION_FAILED
+        if etags is not None and structural_etag(current_config) not in map(strip_quotes, etags):
+            return web.Response(status=status)
+        if not_etags is not None and structural_etag(current_config) in map(strip_quotes, not_etags):
+            return web.Response(status=status)
+
+        # run query
+        op = cast(Literal["get", "delete", "patch", "put"], request.method.lower())
+        new_config, to_return = query(current_config, op, document_path, update_with)
+
+        # update the config
+        if request.method != "GET":
+            # validate
+            config_validated = KresConfig(new_config)
+            # apply
+            await self.config_store.update(config_validated)
+
+        # serialize the response (the `to_return` object is a Dict/list/scalar, we want to return json)
+        resp_text: Optional[str] = json.dumps(to_return) if to_return is not None else None
+
+        # create the response and return it
+        res = web.Response(status=HTTPStatus.OK, text=resp_text, content_type="application/json")
+        res.headers.add("ETag", f'"{structural_etag(new_config)}"')
+        return res
+
+    async def _handler_metrics(self, request: web.Request) -> web.Response:
+        raise web.HTTPMovedPermanently("/metrics/json")
+
+    async def _handler_metrics_json(self, _request: web.Request) -> web.Response:
+
+        config = self.config_store.get()
+
+        return web.Response(
+            body=await metrics.report_json(config),
+            content_type="application/json",
+            charset="utf8",
+        )
+
+    async def _handler_metrics_prometheus(self, _request: web.Request) -> web.Response:
+
+        metrics_report = await metrics.report_prometheus()
+        if not metrics_report:
+            raise web.HTTPNotFound()
+
+        return web.Response(
+            body=metrics_report,
+            content_type="text/plain",
+            charset="utf8",
+        )
+
+    async def _handler_cache_clear(self, request: web.Request) -> web.Response:
+        data = parse_from_mime_type(await request.text(), request.content_type)
+
+        try:
+            config = CacheClearRPCSchema(data)
+        except (AggregateDataValidationError, DataValidationError) as e:
+            return web.Response(
+                body=e,
+                status=HTTPStatus.BAD_REQUEST,
+                content_type="text/plain",
+                charset="utf8",
+            )
+
+        _, result = await command_single_registered_worker(config.render_lua())
+        return web.Response(
+            body=json.dumps(result),
+            content_type="application/json",
+            charset="utf8",
+        )
+
+    async def _handler_schema(self, _request: web.Request) -> web.Response:
+        return web.json_response(
+            kres_config_json_schema(), headers={"Access-Control-Allow-Origin": "*"}, dumps=partial(json.dumps, indent=4)
+        )
+
+    async def _handle_view_schema(self, _request: web.Request) -> web.Response:
+        """
+        Provides a UI for visuallising and understanding JSON schema.
+
+        The feature in the Knot Resolver Manager to render schemas is unwanted, as it's completely
+        out of scope. However, it can be convinient. We therefore rely on a public web-based viewers
+        and provide just a redirect. If this feature ever breaks due to disapearance of the public
+        service, we can fix it. But we are not guaranteeing, that this will always work.
+        """
+
+        return web.Response(
+            text="""
+        <html>
+        <head><title>Redirect to schema viewer</title></head>
+        <body>
+        <script>
+          // we are using JS in order to use proper host
+          let protocol = window.location.protocol;
+          let host = window.location.host;
+          let url = encodeURIComponent(`${protocol}//${host}/schema`);
+          window.location.replace(`https://json-schema.app/view/%23?url=${url}`);
+        </script>
+        <h1>JavaScript required for a dynamic redirect...</h1>
+        </body>
+        </html>
+        """,
+            content_type="text/html",
+        )
+
+    async def _handler_stop(self, _request: web.Request) -> web.Response:
+        """
+        Route handler for shutting down the server (and whole manager)
+        """
+
+        self._shutdown_event.set()
+        logger.info("Shutdown event triggered...")
+        return web.Response(text="Shutting down...")
+
+    async def _handler_reload(self, _request: web.Request) -> web.Response:
+        """
+        Route handler for reloading the server
+        """
+
+        logger.info("Reloading event triggered...")
+        await self._reload_config()
+        return web.Response(text="Reloading...")
+
+    def _setup_routes(self) -> None:
+        self.app.add_routes(
+            [
+                web.get("/", self._handler_index),
+                web.get(r"/v1/config{path:.*}", self._handler_config_query),
+                web.put(r"/v1/config{path:.*}", self._handler_config_query),
+                web.delete(r"/v1/config{path:.*}", self._handler_config_query),
+                web.patch(r"/v1/config{path:.*}", self._handler_config_query),
+                web.post("/stop", self._handler_stop),
+                web.post("/reload", self._handler_reload),
+                web.get("/schema", self._handler_schema),
+                web.get("/schema/ui", self._handle_view_schema),
+                web.get("/metrics", self._handler_metrics),
+                web.get("/metrics/json", self._handler_metrics_json),
+                web.get("/metrics/prometheus", self._handler_metrics_prometheus),
+                web.post("/cache/clear", self._handler_cache_clear),
+            ]
+        )
+
+    async def _reconfigure_listen_address(self, config: KresConfig) -> None:
+        async with self.listen_lock:
+            mgn = config.management
+
+            # if the listen address did not change, do nothing
+            if self.listen == mgn:
+                return
+
+            # start the new listen address
+            nsite: Union[web.TCPSite, web.UnixSite]
+            if mgn.unix_socket:
+                nsite = web.UnixSite(self.runner, str(mgn.unix_socket))
+                logger.info(f"Starting API HTTP server on http+unix://{mgn.unix_socket}")
+            elif mgn.interface:
+                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.")
+            await nsite.start()
+
+            # stop the old listen
+            assert (self.listen is None) == (self.site is None)
+            if self.listen is not None and self.site is not None:
+                if self.listen.unix_socket:
+                    logger.info(f"Stopping API HTTP server on http+unix://{mgn.unix_socket}")
+                elif self.listen.interface:
+                    logger.info(
+                        f"Stopping API HTTP server on http://{self.listen.interface.addr}:{self.listen.interface.port}"
+                    )
+                await self.site.stop()
+
+            # save new state
+            self.listen = mgn
+            self.site = nsite
+
+    async def shutdown(self) -> None:
+        if self.site is not None:
+            await self.site.stop()
+        await self.runner.cleanup()
+
+    def get_exit_code(self) -> int:
+        return self._exit_code
+
+
+async def _load_raw_config(config: Union[Path, Dict[str, Any]]) -> Dict[str, Any]:
+    # Initial configuration of the manager
+    if isinstance(config, Path):
+        if not config.exists():
+            raise KresManagerException(
+                f"Manager is configured to load config file at {config} on startup, but the file does not exist."
+            )
+        else:
+            logger.info(f"Loading configuration from '{config}' file.")
+            config = try_to_parse(await readfile(config))
+
+    # validate the initial configuration
+    assert isinstance(config, dict)
+    return config
+
+
+async def _load_config(config: Dict[str, Any]) -> KresConfig:
+    config_validated = KresConfig(config)
+    return config_validated
+
+
+async def _init_config_store(config: Dict[str, Any]) -> ConfigStore:
+    config_validated = await _load_config(config)
+    config_store = ConfigStore(config_validated)
+    return config_store
+
+
+async def _init_manager(config_store: ConfigStore, server: Server) -> KresManager:
+    """
+    Called asynchronously when the application initializes.
+    """
+
+    # Instantiate subprocess controller (if we wanted to, we could switch it at this point)
+    controller = await get_best_controller_implementation(config_store.get())
+
+    # Create KresManager. This will perform autodetection of available service managers and
+    # select the most appropriate to use (or use the one configured directly)
+    manager = await KresManager.create(controller, config_store, server.trigger_shutdown)
+
+    logger.info("Initial configuration applied. Process manager initialized...")
+    return manager
+
+
+async def _deny_working_directory_changes(config_old: KresConfig, config_new: KresConfig) -> Result[None, str]:
+    if config_old.rundir != config_new.rundir:
+        return Result.err("Changing manager's `rundir` during runtime is not allowed.")
+
+    return Result.ok(None)
+
+
+def _set_working_directory(config_raw: Dict[str, Any]) -> None:
+    try:
+        rundir = get_rundir_without_validation(config_raw)
+    except ValueError as e:
+        raise DataValidationError(str(e), "/rundir") from e
+
+    logger.debug(f"Changing working directory to '{rundir.to_path().absolute()}'.")
+    os.chdir(rundir.to_path())
+
+
+def _lock_working_directory(attempt: int = 0) -> None:
+    # the following syscall is atomic, it's essentially the same as acquiring a lock
+    try:
+        pidfile_fd = os.open(PID_FILE_NAME, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+    except OSError as e:
+        if e.errno == errno.EEXIST and attempt == 0:
+            # the pid file exists, let's check PID
+            with open(PID_FILE_NAME, "r", encoding="utf-8") as f:
+                pid = int(f.read().strip())
+            try:
+                os.kill(pid, 0)
+            except OSError as e2:
+                if e2.errno == errno.ESRCH:
+                    os.unlink(PID_FILE_NAME)
+                    _lock_working_directory(attempt=attempt + 1)
+                    return
+            raise KresManagerException(
+                "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(
+                "Another manager is running in the same working directory."
+                f" PID file is located at {os.getcwd()}/{PID_FILE_NAME}"
+            ) from e
+
+    # now we know that we are the only manager running in this directory
+
+    # write PID to the pidfile and close it afterwards
+    pidfile = os.fdopen(pidfile_fd, "w")
+    pid = os.getpid()
+    pidfile.write(f"{pid}\n")
+    pidfile.close()
+
+    # make sure that the file is deleted on shutdown
+    atexit.register(lambda: os.unlink(PID_FILE_NAME))
+
+
+async def _sigint_while_shutting_down():
+    logger.warning(
+        "Received SIGINT while already shutting down. Ignoring."
+        " If you want to forcefully stop the manager right now, use SIGTERM."
+    )
+
+
+async def _sigterm_while_shutting_down():
+    logger.warning("Received SIGTERM. Invoking dirty shutdown!")
+    sys.exit(128 + signal.SIGTERM)
+
+
+async def start_server(config: Path = CONFIG_FILE) -> int:
+    # This function is quite long, but it describes how manager runs. So let's silence pylint
+    # pylint: disable=too-many-statements
+
+    start_time = time()
+    working_directory_on_startup = os.getcwd()
+    manager: Optional[KresManager] = None
+
+    # Block signals during initialization to force their processing once everything is ready
+    signal.pthread_sigmask(signal.SIG_BLOCK, Server.all_handled_signals())
+
+    # before starting server, initialize the subprocess controller, config store, etc. Any errors during inicialization
+    # are fatal
+    try:
+        # Make sure that the config path does not change meaning when we change working directory
+        config = config.absolute()
+
+        # Preprocess config - load from file or in general take it to the last step before validation.
+        config_raw = await _load_raw_config(config)
+
+        # before processing any configuration, set validation context
+        #  - resolve_root = root against which all relative paths will be resolved
+        set_global_validation_context(Context(config.parent, True))
+
+        # 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.
+        # Warning - this does not prevent multiple managers with the same naming of kresd service.
+        _lock_working_directory()
+
+        # set_global_validation_context(Context(config.parent))
+
+        # After the working directory is set, we can initialize proper config store with a newly parsed configuration.
+        config_store = await _init_config_store(config_raw)
+
+        # Some "constants" need to be loaded from the initial config, some need to be stored from the initial run conditions
+        await init_user_constants(config_store, working_directory_on_startup)
+
+        # This behaviour described above with paths means, that we MUST NOT allow `rundir` change after initialization.
+        # It would cause strange problems because every other path configuration depends on it. Therefore, we have to
+        # add a check to the config store, which disallows changes.
+        await config_store.register_verifier(_deny_working_directory_changes)
+
+        # Up to this point, we have been logging to memory buffer. But now, when we have the configuration loaded, we
+        # can flush the buffer into the proper place
+        await logger_init(config_store)
+
+        # With configuration on hand, we can initialize monitoring. We want to do this before any subprocesses are
+        # started, therefore before initializing manager
+        await metrics.init_prometheus(config_store)
+
+        # prepare instance of the server (no side effects)
+        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)
+
+    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
+        # we'll exec will invoke us again.
+        logger.info("Exec requested with arguments: %s", str(e.exec_args))
+
+        # unblock signals, this could actually terminate us straight away
+        signal.pthread_sigmask(signal.SIG_UNBLOCK, Server.all_handled_signals())
+
+        # run exit functions
+        atexit.run_callbacks()
+
+        # and finally exec what we were told to exec
+        os.execl(*e.exec_args)
+
+    except KresManagerException as e:
+        # We caught an error with a pretty error message. Just print it and exit.
+        logger.error(e)
+        return 1
+
+    except BaseException:
+        logger.error("Uncaught generic exception during manager inicialization...", exc_info=True)
+        return 1
+
+    # At this point, all backend functionality-providing components are initialized. It's therefore save to start
+    # the API server.
+    try:
+        await server.start()
+    except OSError as e:
+        if e.errno in (errno.EADDRINUSE, errno.EADDRNOTAVAIL):
+            # fancy error reporting of network binding errors
+            logger.error(str(e))
+            await manager.stop()
+            return 1
+        raise
+
+    # At this point, pretty much everything is ready to go. We should just make sure the user can shut
+    # the manager down with signals.
+    server.bind_signal_handlers()
+    signal.pthread_sigmask(signal.SIG_UNBLOCK, Server.all_handled_signals())
+
+    logger.info(f"Manager fully initialized and running in {round(time() - start_time, 3)} seconds")
+
+    # notify systemd/anything compatible that we are ready
+    systemd_notify(READY="1")
+
+    await server.wait_for_shutdown()
+
+    # notify systemd that we are shutting down
+    systemd_notify(STOPPING="1")
+
+    # Ok, now we are tearing everything down.
+
+    # First of all, let's block all unwanted interruptions. We don't want to be reconfiguring kresd's while
+    # shutting down.
+    signal.pthread_sigmask(signal.SIG_BLOCK, Server.all_handled_signals())
+    server.unbind_signal_handlers()
+    # on the other hand, we want to immediatelly stop when the user really wants us to stop
+    asyncio_compat.add_async_signal_handler(signal.SIGTERM, _sigterm_while_shutting_down)
+    asyncio_compat.add_async_signal_handler(signal.SIGINT, _sigint_while_shutting_down)
+    signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGTERM, signal.SIGINT})
+
+    # After triggering shutdown, we neet to clean everything up
+    logger.info("Stopping API service...")
+    await server.shutdown()
+    logger.info("Stopping kresd manager...")
+    await manager.stop()
+    logger.info(f"The manager run for {round(time() - start_time)} seconds...")
+    return server.get_exit_code()