| Server IP : 88.223.91.55 / Your IP : 216.73.216.91 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/proc/self/root/opt/cloudlinux/venv/lib64/python3.11/site-packages/lvestats/ |
Upload File : |
#!/opt/cloudlinux/venv/bin/python3 -bb
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import sys
from typing import List, Optional # NOQA
from lvestats.core.plugin import LveStatsPlugin # NOQA
from lvestats.core.plugin_loader import PluginLoader
from lvestats.eventloop.main_loop import MainLoop
from lvestats.plugins import default_plugins
DEFAULT_INTERVAL = '5' # seconds
DEFAULT_TIMEOUT = '5' # seconds
LOW_PRIORITY = sys.maxsize
class LveStatsServer(object):
def __init__(self, plugins, config, singleprocess, profiling_log, times):
self.singleprocess = singleprocess
self.plugins = plugins
self.config = config
self.profiling_log = profiling_log
self.times = times
def run(self):
loop = MainLoop(self.config,
interval=int(self.config.get('interval', DEFAULT_INTERVAL)),
plugins=self.plugins,
plugin_timeout=int(self.config.get('timeout', DEFAULT_TIMEOUT)),
multiproc=not self.singleprocess,
profiling_log=self.profiling_log)
loop.run(times=self.times)
def init_plugins(config, user_specified_list_of_plugins=None):
# type: (dict, Optional[List[str]]) -> List[LveStatsPlugin]
if user_specified_list_of_plugins is None:
user_specified_list_of_plugins = []
def_plugins_corteges = default_plugins()
user_plugins = []
plugins_folder = config.get('plugins', None)
if plugins_folder:
plugin_loader = PluginLoader(plugins_folder)
user_plugins.extend(plugin_loader.plugins)
for cls in user_plugins:
setattr(cls, '__is_user_plugin__', True)
plugins = []
plugins.extend(user_plugins)
for def_plugin_cortege in def_plugins_corteges:
cls, plugin_order = def_plugin_cortege
setattr(cls, '__is_user_plugin__', False)
setattr(cls, 'order', plugin_order)
plugins.append(cls)
if not user_specified_list_of_plugins:
plugins.sort(key=lambda pligin: pligin.order if hasattr(pligin, 'order') else LOW_PRIORITY)
else:
plugins = [p for p in plugins if p.__name__ in user_specified_list_of_plugins]
plugins.sort(key=lambda x: user_specified_list_of_plugins.index(x.__name__))
return plugins
def main(config, singleprocess, user_specified_list_of_plugins, profiling_log, times):
plugins = init_plugins(config, user_specified_list_of_plugins)
server = LveStatsServer(plugins, config, singleprocess, profiling_log, times)
try:
server.run()
except KeyboardInterrupt:
print('\nKeyboardInterrupt')