XMAN Replicator

Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/ssa/internal/

Name	Size	Rights	Date	Actions
[.]	<DIR>	drwxr-xr-x	2026-06-03 05:00	🔄 🗑️ 🗑️📁 ⬇️ 🔒
[..]	<DIR>	drwxr-xr-x	2026-06-03 05:00	🔄 🗑️ 🗑️📁 ⬇️ 🔒
__init__.py	171 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒
[__pycache__]	<DIR>	drwxr-xr-x	2026-06-03 05:01	🔄 🗑️ 🗑️📁 ⬇️ 🔒
constants.py	915 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒
exceptions.py	1,288 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒
mailer.py	5,589 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒
tools.py	2,317 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒
utils.py	16,051 B	-rw-r--r--	2026-05-13 14:10	✏️ 🔄 🗑️ ⬇️ 🔒

Edit: tools.py

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

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import logging
import signal
import sys
from typing import Union, Iterable, Callable

from .utils import switch_schedstats, no_xray_active_tasks
from ..modules.processor import RequestProcessor

logger = logging.getLogger('tools')

def log_on_receive(signum: int, frame) -> None:
    """
    Signal handler to log received signal
    """
    logger.info('Received sig: %i (%s)', signum, signum)

def sigterm_received(signum: int, frame) -> None:
    """
    Signal handler to dump collected statistics and switch off schedstats
    """
    logger.info('Received sig: %i', signum)
    RequestProcessor().flush_buffer()
    logger.info('Dumped successfully after %i signal, exiting...', signum)
    # disable throttling detection kernel mechanism
    # if there is no X-Ray tracing tasks active
    if no_xray_active_tasks():
        logger.info('No X-Ray tasks running, switching schedstats off')
        switch_schedstats(enabled=False)
    sys.exit(0)

def register_a_signal(signum: int, sighandler: Callable) -> None:
    """
    Register given sighandler for given signum
    """
    try:
        signal.signal(signum, sighandler)
        logger.info('Registered %s', signum)
    except (OSError, ValueError) as e:
        logger.info('Skipped %s because of %s', signum, e)

def register_sigterm():
    """
    Register a dump handler for SIGTERM signal
    """
    register_a_signal(signal.SIGTERM, sigterm_received)

def register_signals(signals: Union[int, Iterable] = None,
                     handler: Callable = log_on_receive) -> None:
    """
    Register given handler for given signal or range of signals
    If no signals given, add given handler for all available signals
    """
    if signals is None:
        # register all available signals
        for sig in signal.Signals:
            register_a_signal(sig, handler)
    else:
        if isinstance(signals, int):
            # register single signal
            register_a_signal(signals, handler)
        else:
            # register range of signals
            for sig in signals:
                register_a_signal(sig, handler)

🛸 XMAN Replicator

Edit: tools.py

PHP Console

SQL Console