from typing import Optional, Sequence from .monitor_series_window import MetricSample from .monitor_counter_increase_types import CounterIncrease def counter_increase(samples: Sequence[MetricSample]) -> CounterIncrease: """How much a counter grew over the samples, Prometheus-style: a drop means the counter restarted from zero, so the value after it is all increase. No extrapolation to window edges, so the answer is a whole number.""" increase = 0 resets = 0 for i, s in enumerate(samples): if i > 0 and s.at <= samples[i - 1].at: raise ValueError("samples must be in strictly ascending time order: %d follows %d" % (s.at, samples[i - 1].at)) if s.value < 0: raise ValueError("counter value must not be negative: %d at %d" % (s.value, s.at)) if i == 0: continue prev = samples[i - 1].value if s.value >= prev: increase += s.value - prev else: resets += 1 increase += s.value if not samples: return CounterIncrease(increase=0, resets=0, rate_per_second_milli=None, first_at=None, last_at=None) first_at = samples[0].at last_at = samples[-1].at rate: Optional[int] = None if len(samples) >= 2: n = increase * 1000 d = last_at - first_at rate = (2 * n + d) // (2 * d) return CounterIncrease(increase=increase, resets=resets, rate_per_second_milli=rate, first_at=first_at, last_at=last_at)