84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
from threading import Lock
|
|
import typing
|
|
import time
|
|
import abc
|
|
|
|
from .path import Path
|
|
|
|
|
|
class Now(typing.NamedTuple):
|
|
def time(self) -> int:
|
|
return round(time.time() * 1000)
|
|
|
|
|
|
NOW = Now()
|
|
|
|
|
|
class MetricValue(typing.NamedTuple):
|
|
value: float
|
|
timestamp: typing.Optional[int]
|
|
|
|
|
|
_NAN = float('nan')
|
|
|
|
|
|
class MetricMutableValue:
|
|
# global lock for metrics by default
|
|
DEFAULT_LOCK: typing.ClassVar[Lock] = Lock()
|
|
|
|
def __init__(self, value: float = _NAN, timestamp: typing.Optional[int] = None, lock: Lock = DEFAULT_LOCK):
|
|
self._lock = lock
|
|
self._value = MetricValue(value, timestamp)
|
|
|
|
def set_no_lock(self, value: float, timestamp: typing.Optional[int]):
|
|
self._value = MetricValue(value, timestamp)
|
|
|
|
def set(self, value: float, timestamp: typing.Union[int, None, Now] = NOW):
|
|
if isinstance(timestamp, Now):
|
|
timestamp = timestamp.time()
|
|
with self._lock:
|
|
self.set_no_lock(value, timestamp)
|
|
|
|
def inc_no_lock(self, add: float, timestamp: typing.Optional[int]):
|
|
self._value = MetricValue(self._value.value + add, timestamp)
|
|
|
|
def inc(self, add: float = 1, timestamp: typing.Union[int, None, Now] = NOW):
|
|
if isinstance(timestamp, Now):
|
|
timestamp = timestamp.time()
|
|
with self._lock:
|
|
self.inc_no_lock(add, timestamp)
|
|
|
|
def get(self) -> MetricValue:
|
|
with self._lock:
|
|
return self._value
|
|
|
|
|
|
class MetricBase(MetricMutableValue):
|
|
metric_type: typing.ClassVar[str]
|
|
help_text: typing.Optional[str] = None
|
|
|
|
def __init__(self, value: float = _NAN, timestamp: typing.Optional[int] = None, help: typing.Optional[str] = None):
|
|
super().__init__(value, timestamp)
|
|
self.help_text = help
|
|
|
|
|
|
class MetricGroupDefinition(typing.NamedTuple):
|
|
reserved_suffixes: typing.FrozenSet[str]
|
|
reserved_labels: typing.FrozenSet[str]
|
|
|
|
|
|
EMPTY_GROUP_DEFINITION = MetricGroupDefinition(frozenset(), frozenset())
|
|
|
|
|
|
class MetricGroupBase(abc.ABC):
|
|
metric_type: typing.ClassVar[str]
|
|
group_definition: typing.ClassVar[MetricGroupDefinition]
|
|
help_text: typing.Optional[str] = None
|
|
|
|
def __init__(self, help: typing.Optional[str] = None):
|
|
self.help_text = help
|
|
|
|
@abc.abstractmethod
|
|
def metrics(self) -> typing.Iterator[typing.Tuple[(Path, MetricMutableValue)]]:
|
|
pass
|