Viewing file:  utils.py (1.7 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# 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)
|