from typing import List, Optional, Sequence from .math_round_div import round_div from .monitor_series_window import MetricSample, series_window from .monitor_rollup_types import Bucket _MAX_BUCKETS = 10000 def rollup(samples: Sequence[MetricSample], from_: int, to: int, bucket_seconds: int) -> List[Bucket]: """One row per bucket from `from` to `to`, empty buckets included so a chart shows gaps as gaps. Buckets are aligned to `from`; the last is clipped to `to`.""" if not isinstance(bucket_seconds, int) or isinstance(bucket_seconds, bool): raise ValueError("bucketSeconds must be a whole number of seconds") if bucket_seconds < 1: raise ValueError("bucketSeconds must be at least 1, received %d" % bucket_seconds) inside = series_window(samples, from_, to) n = -((from_ - to) // bucket_seconds) if n > _MAX_BUCKETS: raise ValueError("too many buckets: %d, at most %d" % (n, _MAX_BUCKETS)) out: List[Bucket] = [] j = 0 for i in range(n): start = from_ + i * bucket_seconds end = min(start + bucket_seconds, to) count = 0 total = 0 lo: Optional[int] = None hi: Optional[int] = None last: Optional[int] = None while j < len(inside) and inside[j].at < end: v = inside[j].value count += 1 total += v lo = v if lo is None or v < lo else lo hi = v if hi is None or v > hi else hi last = v j += 1 avg = None if count == 0 else round_div(total, count, "half-up") out.append(Bucket(start=start, end=end, count=count, min=lo, max=hi, avg=avg, sum=total, last=last)) return out