C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # 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 import math import enum import json import typing from enum import StrEnum from types import SimpleNamespace from collections.abc import Sequence, Mapping from typing import NewType, TypedDict, Protocol, NamedTuple, TYPE_CHECKING if TYPE_CHECKING: from lveapi import PyLve SerializedLveId = NewType('SerializedLveId', int) LveId = NewType('LveId', int) Timestamp = NewType('Timestamp', int) class BurstingMultipliers(NamedTuple): cpu: float io: float @enum.unique class LveState(StrEnum): EXISTED = enum.auto() UNBURSTED = enum.auto() BURSTED = enum.auto() OVERUSING = enum.auto() def __str__(self) -> str: return self.name class LveStats(Protocol): cpu: int io: int # in KB reseller_id: int empty_stats = typing.cast(LveStats, SimpleNamespace( cpu=0, cpu_usage=0, io=0, io_usage=0, reseller_id=0, )) class LveUsage(Protocol): cpu_usage: int io_usage: int # in bytes empty_usage = typing.cast(LveUsage, SimpleNamespace( cpu_usage=0, io_usage=0, )) class AdjustStepData(NamedTuple): now: Timestamp lve_active_ids: Sequence[SerializedLveId] stats: Mapping[SerializedLveId, LveStats] lve_usages_by_id: Mapping[SerializedLveId, LveUsage] class LveLimits(TypedDict): cpu: int io: int class InvalidStateError(RuntimeError): pass class ApplyLveSettings(Protocol): def __call__(self, lve_id: LveId, lve_limits: LveLimits) -> None: ... class GetNormalLimits(Protocol): def __call__(self) -> Mapping[LveId, LveLimits]: ... def read_normal_limits_from_proc() -> Mapping[LveId, LveLimits]: with open('/var/run/cloudlinux/effective-normal-limits', 'r', encoding='utf-8') as f: result = json.load(f) result = {LveId(int(k)): v for k, v in result.items()} return result class PyLveSettingsApplier: def __init__(self, pylve: 'PyLve') -> None: self._pylve = pylve def __call__(self, lve_id: LveId, lve_limits: LveLimits) -> None: if lve_id == 0: raise RuntimeError('Cannot alter LVE with id 0') lve_settings = self._pylve.liblve_settings() lve_settings.ls_io = int(lve_limits['io']) lve_settings.ls_cpu = int(lve_limits['cpu']) lve_settings.ls_cpus = int(lve_limits['ncpu']) lve_settings.ls_memory = int(lve_limits['mem']) lve_settings.ls_enters = int(lve_limits['ep']) lve_settings.ls_memory_phy = int(lve_limits['pmem']) lve_settings.ls_nproc = int(lve_limits['nproc']) lve_settings.ls_iops = int(lve_limits['iops']) self._pylve.lve_setup( lve_id, lve_settings, err_msg=f