Store original NOTIFY_SOCKET value.
Remove obsolete systemd_notify.py module from utils.
from supervisor.states import SupervisorStates
from supervisor.supervisord import Supervisor
-from knot_resolver.utils.systemd_notify import systemd_notify
+from .notify_socket import send_notify_socket_message, NOTIFY_SOCKET
superd: Optional[Supervisor] = None
+SYSTEMD_NOTIFY_SOCKET: Optional[str] = os.environ.get(NOTIFY_SOCKET)
+
+
def check_for_fatal_manager(event: ProcessStateFatalEvent) -> None:
assert superd is not None
processname = as_string(proc.config.name)
if processname == "manager":
# manager has sucessfully started, report it upstream
- systemd_notify(STATUS="Starting services...")
+ send_notify_socket_message(SYSTEMD_NOTIFY_SOCKET, STATUS="Starting services...")
def check_for_runnning_manager(event: ProcessStateRunningEvent) -> None:
processname = as_string(proc.config.name)
if processname == "manager":
# manager has sucessfully started, report it upstream
- systemd_notify(READY="1", STATUS="Ready")
+ send_notify_socket_message(SYSTEMD_NOTIFY_SOCKET, READY="1", STATUS="Ready")
def get_server_options_signal(self):
# This status notification here unsets the env variable $NOTIFY_SOCKET provided by systemd
# and stores it locally. Therefore, it shouldn't clash with $NOTIFY_SOCKET we are providing
# downstream
- systemd_notify(STATUS="Initializing supervisord...")
+ send_notify_socket_message(SYSTEMD_NOTIFY_SOCKET, STATUS="Initializing supervisord...")
# register events
subscribe(ProcessStateFatalEvent, check_for_fatal_manager)
import signal
import time
from functools import partial
+ from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar
from supervisor.events import ProcessStateEvent, ProcessStateStartingEvent, subscribe
from supervisor.states import ProcessStates
from supervisor.supervisord import Supervisor
- from knot_resolver.controller.supervisord.plugin import notify
+ from knot_resolver.controller.supervisord.plugin.notify_socket import (
+ init_notify_socket,
+ read_notify_socket,
+ NOTIFY_SOCKET,
+ NOTIFY_SOCKET_NAME,
+ )
starting_processes: List[Subprocess] = []
def __init__(self, supervisor: Supervisor, fd: int):
self._supervisor = supervisor
self.fd = fd
- self.closed = False # True if close() has been called
+ self.closed: bool = False # True if close() has been called
def __repr__(self):
return f"<{self.__class__.__name__} with fd={self.fd}>"
def handle_read_event(self):
logger: Any = self._supervisor.options.logger
- res: Optional[Tuple[int, bytes]] = notify.read_message(self.fd)
- if res is None:
+ result: tuple[int, bytes, None] = read_notify_socket(self.fd)
+ if result is None:
return # there was some junk
- pid, data = res
+
+ pid, data = result
# pylint: disable=undefined-loop-variable
for proc in starting_processes:
def supervisord_get_process_map(supervisord: Any, mp: Dict[Any, Any]) -> Dict[Any, Any]:
global notify_dispatcher
if notify_dispatcher is None:
- notify_dispatcher = NotifySocketDispatcher(supervisord, notify.init_socket())
+ notify_dispatcher = NotifySocketDispatcher(supervisord, init_notify_socket())
supervisord.options.logger.info("notify: injected $NOTIFY_SOCKET into event loop")
# add our dispatcher to the result
def process_spawn_as_child_add_env(slf: Subprocess, *args: Any) -> Tuple[Any, ...]:
if is_type_notify(slf):
- slf.config.environment["NOTIFY_SOCKET"] = os.getcwd() + "/supervisor-notify-socket"
+ slf.config.environment[NOTIFY_SOCKET] = str(Path.cwd() / NOTIFY_SOCKET_NAME)
return (slf, *args)
T = TypeVar("T")
from knot_resolver.controller.exceptions import KresSubprocessControllerError, KresSubprocessControllerExec
from knot_resolver.controller.interface import SubprocessType
from knot_resolver.controller.registered_workers import command_single_registered_worker
+from knot_resolver.controller.supervisord.plugin.notify_socket import send_notify_socket_message
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.utils.modeling.parsing import DataFormat, data_combine, 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
async def sighup_handler(self) -> None:
logger.info("Received SIGHUP, reloading configuration file")
- systemd_notify(RELOADING="1")
+ send_notify_socket_message(RELOADING="1")
await self._reload_config()
- systemd_notify(READY="1")
+ send_notify_socket_message(READY="1")
@staticmethod
def all_handled_signals() -> Set[signal.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")
+ send_notify_socket_message(READY="1")
await server.wait_for_shutdown()
# notify systemd that we are shutting down
- systemd_notify(STOPPING="1")
+ send_notify_socket_message(STOPPING="1")
# Ok, now we are tearing everything down.
+++ /dev/null
-import logging
-import os
-import socket
-
-logger = logging.getLogger(__name__)
-
-
-def systemd_notify(**values: str) -> None:
- """
- Send systemd notify message to notify socket.
-
- Notify socket location (unix socket) should be saved in $NOTIFY_SOCKET environment variable.
- It is typically set by the processes supervisor (supervisord).
- If $NOTIFY_SOCKET is not configured, it is not possible to send a notification and the operation will fail.
- """
- socket_addr = os.getenv("NOTIFY_SOCKET")
- if socket_addr is None:
- logger.warning("Failed to get $NOTIFY_SOCKET environment variable")
- return
-
- if socket_addr.startswith("@"):
- socket_addr = socket_addr.replace("@", "\0", 1)
-
- try:
- notify_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
- notify_socket.connect(socket_addr)
- except OSError:
- logger.exception("Failed to connect to $NOTIFY_SOCKET at '%s'", socket_addr)
- return
-
- payload = "\n".join((f"{key}={value}" for key, value in values.items()))
- try:
- notify_socket.send(payload.encode("utf8"))
- except OSError:
- logger.exception("Failed to send systemd notification to $NOTIFY_SOCKET at '%s'", socket_addr)
-
- notify_socket.close()
projects / thirdparty / knot-resolver.git / commitdiff
