?HAMA

Edit File Size: 2.74 KB

/proc/thread-self/root/opt/cloudlinux/venv/lib64/python3.11/site-packages/lvestats/main.py

Text Base64

#!/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')

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
burstwatcher DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
core DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
eventloop DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
lib DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
orm DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
plugins DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
snapshots DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
utils DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
__pycache__ DIR	-	drwxr-xr-x	2026-03-05 22:36:53	Edit Download recursive recursive
main.py	2.74 KB	lrw-r--r--	2026-02-17 11:00:05	Edit Download recursive recursive
version.py	377 B	lrw-r--r--	2026-02-17 11:00:05	Edit Download recursive recursive
__init__.py	275 B	lrw-r--r--	2026-02-17 11:00:05	Edit Download recursive recursive

Directory Contents