File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/lvestats/plugins/generic/burster

opt / cloudlinux / venv / lib / python3.11 / site-packages / lvestats / plugins / generic / burster /

📁 ..
📄 __init__.py(8.53 KB)
📁 __pycache__
📄 _logs.py(597 B)
📄 adjust.py(7.36 KB)
📄 common.py(4.56 KB)
📄 config.py(7.96 KB)
📄 history.py(9.79 KB)
📄 lve_sm.py(18.4 KB)
📄 lves_tracker.py(8.92 KB)
📄 notify.py(1.64 KB)
📄 overload.py(3.23 KB)
📄 profiler.py(10.3 KB)
📁 storage
📄 utils.py(1.63 KB)

Editing: overload.py

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from collections import deque
from typing import NamedTuple, Self, Protocol

from ._logs import logger

class StatsTimes(NamedTuple):
    @classmethod
    def from_proc_stat(cls, line: str) -> Self:
        # NOTE(vlebedev): First element of the line is a name of a CPU core, not time of any kind.
        return cls(*(int(v) for v in line.strip().split()[1:]))

user: int
    nice: int
    system: int
    idle: int
    iowait: int
    irq: int
    softirq: int
    steal: int
    guest: int
    guest_nice: int

class GetStats(Protocol):
    def __call__(self) -> StatsTimes:
        ...

class OverloadCheckResult(NamedTuple):
    is_overloaded: bool
    server_load: float

def __bool__(self) -> bool:
        return self.is_overloaded

class OverloadChecker:
    def __init__(
        self,
        idle_time_threshold: float,
        get_stats: GetStats,
        max_samples_number: int,
    ) -> None:
        self._prev_stats = None
        self._idle_time_threshold = idle_time_threshold
        self._weights = _get_weights(num=max_samples_number)
        self._idle_times_history = deque(maxlen=max_samples_number)
        self._get_stats = get_stats
        self._max_samples_number = max_samples_number

def __call__(self) -> OverloadCheckResult:
        stats = self._get_stats()

# NOTE(vlebedev): 'delta' represents total time spent by CPU
        #                 on all kinds of activities since the previous measurement.
        delta = stats if self._prev_stats is None else StatsTimes(*(
            new - old
            for (new, old)
            in zip(stats, self._prev_stats)
        ))
        self._prev_stats = stats

# NOTE(vlebedev): Fraction of total delta time spent by CPU on being 'idle'.
        try:
            idle_time = delta.idle / sum(delta)
        except ZeroDivisionError:
            idle_time = 0.0

self._idle_times_history.append(idle_time)

samples_num = len(self._idle_times_history)
        relevant_weights = self._weights[:samples_num]
        weighted_idle_times = [
            weight * idle_time
            for weight, idle_time
            in zip(relevant_weights, self._idle_times_history)
        ]
        weighted_idle_time = sum(weighted_idle_times)
        if samples_num < self._max_samples_number:
            weighted_idle_time /= sum(relevant_weights)

logger.debug('Current idle time: %02.2f, threshold %s', weighted_idle_time, self._idle_time_threshold)

return OverloadCheckResult(
            is_overloaded=weighted_idle_time < self._idle_time_threshold,
            server_load=1 - weighted_idle_time,
        )

def _get_weights(num: int = 10, alpha: float = 0.9):
    # NOTE(vlebedev): Generate an exponentially decaying weight list
    weights = [alpha ** i for i in range(num)]
    # NOTE(vlebedev): Normalize the weights so they sum to 1
    total = sum(weights)
    return [weight / total for weight in weights]

def read_times_from_proc() -> StatsTimes:
    with open('/proc/stat', encoding='ascii') as fd:
        return StatsTimes.from_proc_stat(fd.readline())

X7ROOT File Manager

Editing: overload.py

Upload File

Create Folder