One Hat Cyber Team

Your IP: 216.73.216.42
Server IP: 97.74.87.16
Server: Linux 16.87.74.97.host.secureserver.net 5.14.0-503.38.1.el9_5.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Apr 18 08:52:10 EDT 2025 x86_64
Server Software: Apache
PHP Version: 8.2.28
Buat File | Buat Folder

Dir : ~/var/opt/nydus/ops/primordial/service/

View File Name : base_daemon.py

# -*- coding: utf-8 -*-

import logging
from subprocess import PIPE, call, CalledProcessError
from typing import Any, Dict  # pylint: disable=W0611  # noqa: F401


LOGGER = logging.getLogger(__name__)


class BaseDaemon:
    """A base implementation of a daemon"""
    def __init__(self, name: str, command_fmt: str) -> None:
        self.name = name
        self.command_fmt = command_fmt

    def _run(self, action: str, check_return_code: bool = True, pipes: bool = True) -> Any:
        """Run an action for the daemon.

        :param action: The action to be run
        :param check_return_code: Whether to error on non-zero return codes
        :param pipes:
        :raises CalledProcessError: If `check_return_code` is True, and a non-zero return code occurs
        """
        LOGGER.info('%sing %s', action, self.name)
        kwargs = {'universal_newlines': True, 'timeout': 10}  # type: Dict[str, Any]
        if pipes:
            kwargs.update({'stdout': PIPE, 'stderr': PIPE})
        cmd = self.command_fmt.format(name=self.name, action=action)
        result = call(cmd.split(), **kwargs)
        LOGGER.info(result)
        if check_return_code:
            if result != 0:
                raise CalledProcessError(returncode=result, cmd=cmd)
        return result

    def status(self) -> Any:
        """Check the status of the daemon."""
        return self._run('status', check_return_code=False)

    def start(self) -> Any:
        """Start the daemon."""
        # Must not connect pipes to stdin/out, or subprocess will never
        # complete, hanging on communicate() call.
        return self._run('start', pipes=False)

    def stop(self) -> Any:
        """Stop the daemon."""
        return self._run('stop')

    def restart(self) -> Any:
        """Restart the daemon."""
        # Must not connect pipes to stdin/out, or subprocess will never
        # complete, hanging on communicate() call.
        return self._run('restart', pipes=False)