| Server IP : 185.124.137.75 / Your IP : 216.73.217.69 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/cloudlinux/venv/lib64/python3.11/site-packages/xray/continuous/ |
Upload File : |
# -*- 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
"""
This module contains basic part of continuous tracingg implementation
"""
import logging
import shelve
from typing import List
from xray import gettext as _
from ..internal.constants import continuous_storage
from ..internal.exceptions import XRayError
from ..internal.types import ContinuousTask
class ContinuousCommon:
"""
Base class for continuous tracing
"""
def __init__(self):
self.storage = continuous_storage
self.logger = logging.getLogger('continuous')
self.tracing_conf = self.load_tracing_configuration()
def load_tracing_configuration(self) -> dict:
"""
Load full continuous tracing configuration from the local storage
"""
self.logger.info('Loading continuous tracing configuration')
try:
with shelve.open(self.storage) as db:
return {item: db[item] for item in db.keys()}
except OSError as e:
self.logger.error(
'Failed to load continuous tracing configuration',
extra={'err': str(e)})
return dict()
def get_continuous_tasks(self) -> List[ContinuousTask]:
"""
Return existing configuration in the form of list
"""
return sorted(
{ContinuousTask(**item) for item in self.tracing_conf.values()})
def dump_tracing_configuration(self) -> List[dict]:
"""
Dump continuous tracing configuration into local storage
"""
try:
with shelve.open(self.storage) as db:
for item in self.tracing_conf:
db[item] = self.tracing_conf[item]
return [db[item] for item in db.keys()]
except OSError as e:
raise XRayError(
_('Failed to dump continuous configuration: %s') % str(e))
def remove_tracing_configuration(self, domain: str) -> None:
"""
Remove continuous tracing configuration for domain
"""
try:
del self.tracing_conf[domain]
with shelve.open(self.storage) as db:
del db[domain]
except KeyError:
self.logger.error('Continuous monitoring is not enabled',
extra={'domain': domain,
'tracing_entries': self.tracing_conf})
raise XRayError(
_('Continuous monitoring for %s is not enabled') % domain)
except OSError as e:
raise XRayError(
_('Failed to remove continuous configuration for {}: {}'.format(domain, str(e))))