관리-도구

편집 파일: utils.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
import typing
import itertools
from typing import Collection, Iterator

from lvestats.orm import BurstingEventType
from .base import InBurstingEventRow, Timestamp
from ..common import LveId

def lves_historical_events_iter(
    lves_ids: Collection[LveId],
    events_ts_and_type_iter: Iterator[tuple[Timestamp, BurstingEventType]],
):
    for ts, event_type in events_ts_and_type_iter:
        for lve_id in lves_ids:
            yield InBurstingEventRow(
                lve_id=lve_id,
                timestamp=ts,
                event_type=event_type,
            )

def lve_historical_events_ts_and_type_iter(
    start_stop_pairs_iter: Iterator[tuple[Timestamp, Timestamp]],
) -> Iterator[tuple[Timestamp, BurstingEventType]]:
    timestemps_iter = itertools.chain.from_iterable(start_stop_pairs_iter)
    event_types_iter = itertools.cycle([BurstingEventType.STARTED, BurstingEventType.STOPPED])
    for timestamp, event_type in zip(timestemps_iter, event_types_iter):
        yield timestamp, event_type

def lve_historical_intervals_iter(
    overusing_intervals: int,
    interval_duration: int,
    cooldown: int = 1,
) -> Iterator[tuple[Timestamp, Timestamp]]:
    assert overusing_intervals >= 0
    assert interval_duration >= 1
    assert cooldown >= 1
    now = Timestamp(0)
    for _ in range(overusing_intervals):
        started_at, stopped_at = now, now + interval_duration
        yield typing.cast(tuple[Timestamp, Timestamp], (started_at, stopped_at))
        now = typing.cast(Timestamp, stopped_at + cooldown)