1#@!#!123s

: / opt / cloudlinux / venv / lib / python3.11 / site-packages / lvestats / plugins / generic / burster /
Filename : lve_sm.py
back
# 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

# pylint: disable=not-an-iterable

import inspect
import contextlib
from dataclasses import dataclass
from datetime import timedelta
from typing import TypeAlias, NamedTuple, AbstractSet, Self

from lvestats.orm import BurstingEventType

from .utils import bootstrap_gen
from ._logs import logger
from .common import (
    BurstingMultipliers,
    LveId,
    Timestamp,
    LveLimits,
    ApplyLveSettings,
    LveStats,
    InvalidStateError,
    LveUsage,
    LveState,
    calc_bursted_limits,
    infer_lve_state,
)
from .history import LveHistory
from .notify import Emitter, Signal


class LveStateManager:
    @dataclass(frozen=True, slots=True)
    class Disappered(RuntimeError):
        now: Timestamp

    @dataclass(frozen=True, slots=True)
    class Burst:
        now: Timestamp

    @dataclass(frozen=True, slots=True)
    class Unburst:
        now: Timestamp

    # NOTE(vlebedev): Seems that namedtuples are ~20% faster to create compared to
    #                 classes - dataclasses or usual ones.
    class UpdateReadings(NamedTuple):
        now: Timestamp
        normal_limits: LveLimits | None
        stats: LveStats | None
        usage: LveUsage | None

    Request: TypeAlias = UpdateReadings | Disappered | Burst | Unburst

    @dataclass(frozen=True, slots=True)
    class _StateSwitched(RuntimeError):
        pass

    def __init__(
        self,
        now: Timestamp,
        lve_id: LveId,
        bursting_multipliers: BurstingMultipliers,
        initial_normal_limits: LveLimits,
        initial_stats: LveStats,
        initial_usage: LveUsage,
        quota: timedelta,
        quota_window: timedelta,
        apply_lve_settings: ApplyLveSettings,
        initial_history: LveHistory = LveHistory(),
        fail_fast: bool = True,
    ):
        if quota > quota_window:
            raise ValueError('Bursting quota must not exceed bursting quota window!')

        # NOTE(vlebedev): Constants start here.
        self._lve_id = lve_id
        self._bursting_multipliers = bursting_multipliers
        self._quota_sec = quota.total_seconds()
        self._quota_window_sec = quota_window.total_seconds()
        self._fail_fast = fail_fast
        # NOTE(vlebedev): Constants end here.

        # NOTE(vlebedev): "Input" variables start here.
        #                 "Input" variables are those that are read from external source (Request) and must not
        #                  be altered by internal logic.
        self._now = now
        self._normal_limits = initial_normal_limits
        self._stats = initial_stats
        self._usage = initial_usage
        self._state = infer_lve_state(
            initial_stats,
            initial_normal_limits,
            initial_usage,
        )
        # NOTE(vlebedev): "Input" variables end here.

        # NOTE(vlebedev): "Output" variables start here.
        #                 "Output" variables are a channel that communicates internal state to external world.
        #                 Internal logic is responsible for mutating these vairables.
        self._history = initial_history
        # NOTE(vlebedev): "Output" variables end here.

        # NOTE(vlebedev): Utility stuff starts here.
        #                 These variables are not supposed to rewritten by internal logic.
        def apply_lve_settings_wrapper(cmd) -> None:
            if isinstance(cmd, self.Burst):
                limits = calc_bursted_limits(self._normal_limits, bursting_multipliers)
            elif isinstance(cmd, self.Unburst):
                limits = self._normal_limits
            else:
                raise ValueError(f'Unexpected command: {cmd}')

            apply_lve_settings(self._lve_id, limits)
        self._apply_lve_settings = apply_lve_settings_wrapper

        root_handler_gen = self._create_root_handler_gen()

        def step(request):
            response = root_handler_gen.send(request)
            if isinstance(response, Exception):
                raise response
            return response
        self.step = step

        self._on_state_changed = Emitter()

    @property
    def lve_id(self) -> LveId:
        return self._lve_id

    @property
    def now(self) -> Timestamp:
        return self._now

    @property
    def state(self) -> LveState:
        return self._state

    @property
    def is_bursted(self) -> bool:
        return self._state in {LveState.BURSTED, LveState.OVERUSING}

    @property
    def is_unbursted(self) -> bool:
        return self._state == LveState.UNBURSTED

    @property
    def is_overusing(self) -> bool:
        return self._state == LveState.OVERUSING

    @property
    def history_contains_overusing(self) -> bool:
        return self._history.contains_overusing

    @property
    def history(self) -> LveHistory:
        return self._history

    def check_quota_exceeded(self, now: float) -> bool:
        return self._history.get_overusing_duration(now) >= self._quota_sec

    @property
    def on_state_changed(self) -> Signal:
        return self._on_state_changed

    def trim_history(self, now: Timestamp) -> None:
        cutoff = Timestamp(int(now - self._quota_window_sec))
        self._history = self._history.trim(cutoff)

    @bootstrap_gen
    def _create_root_handler_gen(self):
        # NOTE(vlebedev): This coroutine is solely responsible for modifying "input" variables.
        #                 No mutations of "input" variables must happen outside of it.

        state_to_handler_factory = {
            LveState.EXISTED: self._create_existed_state_handler_gen,
            LveState.UNBURSTED: self._create_unbursted_state_handler_gen,
            LveState.BURSTED: self._create_bursted_state_handler_gen,
            LveState.OVERUSING: self._create_overusing_state_handler_gen,
        }

        prev_state = next_state = self._state
        subhandler = state_to_handler_factory[next_state](next_state)
        request, response = None, None
        while True:
            try:
                request, response = (yield response), None

                # NOTE(vlebedev): "Input" variables mutations start here.
                #                 It's important that it happens before invoking state handler
                #                 so that state handler sees updated state.
                assert request.now >= self._now, "Time is not expected to go backward!"
                self._now = request.now
                if isinstance(request, self.UpdateReadings):
                    if request.normal_limits is not None:
                        self._normal_limits = request.normal_limits
                    if request.stats is not None:
                        self._stats = request.stats
                    if request.usage is not None:
                        self._usage = request.usage
                    next_state = infer_lve_state(
                        self._stats,
                        self._normal_limits,
                        self._usage,
                    )
                elif isinstance(request, self.Disappered):
                    next_state = LveState.EXISTED
                self._state = next_state
                # NOTE(vlebedev): "Input" variables mutations end here.

                # TODO(vlebedev): Refactor subhandler management - it turned out to be quite messy.
                if inspect.getgeneratorstate(subhandler) != inspect.GEN_CLOSED:
                    subhandler_exc = None
                    if isinstance(request, self.Disappered):
                        subhandler_exc = request
                    elif next_state != prev_state:
                        subhandler_exc = self._StateSwitched()

                    if subhandler_exc is not None:
                        try:
                            with contextlib.suppress(StopIteration, type(subhandler_exc)):
                                subhandler.throw(subhandler_exc)
                        except (StopIteration, type(subhandler_exc)):
                            logger.debug(
                                'LVE "%s": state subhandler for "%s" state terminated',
                                self._lve_id, next_state
                            )
                        except Exception:
                            if self._fail_fast:
                                raise
                            logger.exception(
                                'LVE "%s": state subhandler for "%s" state unexpectedly failed during termination!',
                                self._lve_id, next_state
                            )
                        else:
                            try:
                                subhandler.close()
                            except Exception:
                                if self._fail_fast:
                                    raise
                                logger.exception(
                                    'LVE "%s": state subhandler for "%s" state did not terminate properly '
                                    'and failed during force-close!',
                                    self._lve_id, next_state
                                )

                if inspect.getgeneratorstate(subhandler) == inspect.GEN_CLOSED and next_state != prev_state:
                    subhandler = state_to_handler_factory[next_state](next_state)
                    logger.debug('LVE "%s": state subhandler for "%s" state created', self._lve_id, next_state)

                if inspect.getgeneratorstate(subhandler) != inspect.GEN_CLOSED:
                    try:
                        response = subhandler.send(request)
                    except StopIteration:
                        logger.debug('LVE "%s": state subhandler for "%s" state finished', self._lve_id, next_state)
                    except Exception:
                        if self._fail_fast:
                            raise
                        logger.exception(
                            'LVE "%s": state subhandler for "%s" state failed while handling "%s" request!',
                            self._lve_id, next_state, request,
                        )

                # NOTE(vlebedev): Trigger signal handlers only after all internal state has been updated.
                if next_state != prev_state:
                    logger.debug(
                        'LVE "%s": transitioned from "%s" to "%s" state',
                        self._lve_id, prev_state, next_state,
                    )
                    try:
                        self._on_state_changed(prev_state, next_state)
                    except Exception:
                        if self._fail_fast:
                            raise
                        logger.exception('LVE "%s": some "on_state_changed" listeners failed!', self._lve_id)

                prev_state = next_state
            except Exception as e:
                if self._fail_fast:
                    raise e
                logger.critical('LVE "%s": top handler unexpectadly failed!', self._lve_id, exc_info=e)

    @bootstrap_gen
    def _create_unbursted_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, self.Unburst):
                    response = InvalidStateError('Already unbursted!')
                elif isinstance(request, self.Burst):
                    try:
                        self._apply_lve_settings(request)
                    except Exception as e:
                        response = e
                    else:
                        break
        except self._StateSwitched:
            if self._state == LveState.BURSTED:
                logger.warning(
                    'LVE "%s": was set to bursted externally, without internal request from adjuster',
                    self._lve_id,
                )
            raise

        logger.debug('LVE "%s": waiting for bursting enabled confirmation', self._lve_id)

        try:
            yield from self._wait_for_confirmation_from_readings({LveState.BURSTED, LveState.OVERUSING})
        except self.Disappered:
            logger.debug(
                'LVE "%s": disappeared while waiting for bursting enabled or overusing started confirmation',
                self._lve_id,
            )
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting enabled or overusing started confirmation',
                self._lve_id, self._state,
            )
            raise

        assert self._state in {LveState.BURSTED, LveState.OVERUSING}

        logger.debug(
            'LVE "%s": confirmed from readings that bursting switched to enabled%s',
            self._lve_id,
            " and overusing has started" if self._state == LveState.OVERUSING else "",
        )

    @bootstrap_gen
    def _create_bursted_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, self.Unburst):
                    try:
                        self._apply_lve_settings(request)
                    except Exception as e:
                        response = e
                    else:
                        break
                elif isinstance(request, self.Burst):
                    response = InvalidStateError('Already bursted!')
        except self._StateSwitched:
            if self._state == LveState.UNBURSTED:
                logger.warning(
                    'LVE "%s": was set to unbursted externally, without internal request from adjuster',
                    self._lve_id,
                )
            raise

        logger.debug('LVE "%s": waiting for bursting disabled confirmation', self._lve_id)

        try:
            yield from self._wait_for_confirmation_from_readings({LveState.UNBURSTED})
        except self.Disappered:
            logger.debug('LVE "%s": disappeared while waiting for bursting disabled confirmation', self._lve_id)
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting disabled confirmation',
                self._lve_id, self._state,
            )
            raise

        logger.debug('LVE "%s": confirmed from readings that bursting switched to disabled', self._lve_id)

    @bootstrap_gen
    def _create_overusing_state_handler_gen(self, prev_state: LveState):
        # TODO(vlebedev): Do not use absolute time but relative to the latest event with `now`.
        self._history = self._history.append(self._now, BurstingEventType.STARTED)

        request, response = None, None
        try:
            try:
                while True:
                    request, response = (yield response), None

                    if isinstance(request, self.Unburst):
                        try:
                            self._apply_lve_settings(request)
                        except Exception as e:
                            response = e
                        else:
                            break
                    elif isinstance(request, self.Burst):
                        response = InvalidStateError('LVE is overusing i.e. is also already bursted!')
            except self._StateSwitched:
                if self._state == LveState.UNBURSTED:
                    logger.warning(
                        'LVE "%s": was set to unbursted externally, without internal request from adjuster',
                        self._lve_id,
                    )
                raise

            yield from self._wait_for_unbursted_confirmation_from_readings(prev_state)
        finally:
            self._history = self._history.append(self._now, BurstingEventType.STOPPED)

    @bootstrap_gen
    def _create_existed_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                try:
                    request, response = (yield response), None
                except self.Disappered:
                    logger.warning(
                        'LVE "%s": disappeared again while waiting for LVE to re-appear',
                        self._lve_id,
                    )

                if isinstance(request, (self.Burst, self.Unburst)):
                    response = InvalidStateError('LVE does not exist!')
        except self._StateSwitched:
            logger.debug('LVE "%s": got back from the dead', self._lve_id)
            return

    def _wait_for_unbursted_confirmation_from_readings(self, prev_state: LveState):
        try:
            yield from self._wait_for_confirmation_from_readings({LveState.UNBURSTED})
        except self.Disappered:
            logger.debug('LVE "%s": disappeared while waiting for bursting disabled confirmation', self._lve_id)
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting disabled confirmation',
                self._lve_id, self._state,
            )
            raise

        logger.debug('LVE "%s": confirmed from readings that bursting switched to disabled', self._lve_id)

    def _wait_for_confirmation_from_readings(self, expected_states: AbstractSet[LveState]):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, (self.Burst, self.Unburst)):
                    response = InvalidStateError('Waiting for confirmation from readings that limits have switched!')
        except self._StateSwitched:
            if self._state not in expected_states:
                raise


_state_to_code = {
    LveState.EXISTED: "X",
    LveState.UNBURSTED: "U",
    LveState.BURSTED: "B",
    LveState.OVERUSING: "O",
}

assert _state_to_code.keys() == set(LveState)


class LveStateSummary(NamedTuple):
    @classmethod
    def for_lve(cls, manager: LveStateManager) -> Self:
        return cls(manager.state, manager.history, manager.now)

    state: LveState
    history: LveHistory
    now: Timestamp

    def __str__(self) -> str:
        intervals = self.history.get_overusing_intervals(self.now)
        duration = self.history.get_overusing_duration(self.now)
        state_code = _state_to_code[self.state]
        return f"{state_code};[{intervals}~={round(duration)}s]"