Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import Optional, Sequence
from .monitor_series_window import MetricSample ← from monitor.series-window ^1.0.0 · built alongside by fune
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)