SWAMPy

✎ CARVE FLESH

EDITING: main.py

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

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-11-09 22:53
📁 burstwatcher/	DIR	—	2026-11-09 22:53
📁 core/	DIR	—	2026-11-09 22:53
📁 eventloop/	DIR	—	2026-11-09 22:53
📁 lib/	DIR	—	2026-11-09 22:53
📁 orm/	DIR	—	2026-11-09 22:53
📁 plugins/	DIR	—	2026-11-09 22:53
📁 snapshots/	DIR	—	2026-11-09 22:53
📁 utils/	DIR	—	2026-11-09 22:53
📄 __init__.py	PY	275 B	2026-10-23 09:12	EDIT
📄 main.py	PY	2.7 KB	2026-10-23 09:12	EDIT
📄 version.py	PY	376 B	2026-10-23 09:12	EDIT