Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.last_update_timestamp = time_ns()
self.data += value
class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
def set(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.GaugeHandle.set`."""
if self._validate_update(value):
if self.monotonic and value < self.data:
logger.warning("Monotonic gauge cannot descend.")
return
self.last_update_timestamp = time_ns()
self.data = value
class MeasureHandle(metrics_api.MeasureHandle, BaseHandle):
def record(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.MeasureHandle.record`."""
if self._validate_update(value):
if self.monotonic and value < 0:
logger.warning("Monotonic measure cannot accept negatives.")
return
self.last_update_timestamp = time_ns()
# TODO: record
class Metric(metrics_api.Metric):
"""See `opentelemetry.metrics.Metric`."""
HANDLE_TYPE = BaseHandle
def __init__(
type(self).__name__, self.data, self.last_update_timestamp
)
class CounterHandle(metrics_api.CounterHandle, BaseHandle):
def add(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.CounterHandle.add`."""
if self._validate_update(value):
if self.monotonic and value < 0:
logger.warning("Monotonic counter cannot descend.")
return
self.last_update_timestamp = time_ns()
self.data += value
class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
def set(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.GaugeHandle.set`."""
if self._validate_update(value):
if self.monotonic and value < self.data:
logger.warning("Monotonic gauge cannot descend.")
return
self.last_update_timestamp = time_ns()
self.data = value
class MeasureHandle(metrics_api.MeasureHandle, BaseHandle):
def record(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.MeasureHandle.record`."""
if self._validate_update(value):
if self.monotonic and value < 0:
logger.warning("Monotonic measure cannot accept negatives.")
if not self.enabled:
return False
if not isinstance(value, self.value_type):
logger.warning(
"Invalid value passed for %s.", self.value_type.__name__
)
return False
return True
def __repr__(self):
return '{}(data="{}", last_update_timestamp={})'.format(
type(self).__name__, self.data, self.last_update_timestamp
)
class CounterHandle(metrics_api.CounterHandle, BaseHandle):
def add(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.CounterHandle.add`."""
if self._validate_update(value):
if self.monotonic and value < 0:
logger.warning("Monotonic counter cannot descend.")
return
self.last_update_timestamp = time_ns()
self.data += value
class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
def set(self, value: metrics_api.ValueT) -> None:
"""See `opentelemetry.metrics.GaugeHandle.set`."""
if self._validate_update(value):
if self.monotonic and value < self.data:
logger.warning("Monotonic gauge cannot descend.")