Module glean.metrics.memory_distribution
Expand source code
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from typing import List, Optional
from .. import _ffi
from .._dispatcher import Dispatcher
from ..testing import ErrorType
from .distribution_data import DistributionData
from .lifetime import Lifetime
from .memoryunit import MemoryUnit
class MemoryDistributionMetricType:
"""
This implements the developer facing API for recording memory distribution
metrics.
Instances of this class type are automatically generated by
`glean.load_metrics`, allowing developers to record values that were
previously registered in the metrics.yaml file.
"""
def __init__(
self,
disabled: bool,
category: str,
lifetime: Lifetime,
name: str,
send_in_pings: List[str],
memory_unit: MemoryUnit,
):
self._disabled = disabled
self._send_in_pings = send_in_pings
self._handle = _ffi.lib.glean_new_memory_distribution_metric(
_ffi.ffi_encode_string(category),
_ffi.ffi_encode_string(name),
_ffi.ffi_encode_vec_string(send_in_pings),
len(send_in_pings),
lifetime.value,
disabled,
memory_unit.value,
)
def __del__(self):
if getattr(self, "_handle", 0) != 0:
_ffi.lib.glean_destroy_memory_distribution_metric(self._handle)
def accumulate(self, sample: int) -> None:
"""
Record a single value, in the unit specified by `memory_unit`, to the distribution.
Args:
sample: The value to record.
"""
if self._disabled:
return None
@Dispatcher.launch
def accumulate():
return _ffi.lib.glean_memory_distribution_accumulate(self._handle, sample)
def test_has_value(self, ping_name: Optional[str] = None) -> bool:
"""
Tests whether a value is stored for the metric for testing purposes
only.
Args:
ping_name (str): (default: first value in send_in_pings) The name
of the ping to retrieve the metric for.
Returns:
has_value (bool): True if the metric value exists.
"""
if ping_name is None:
ping_name = self._send_in_pings[0]
return bool(
_ffi.lib.glean_memory_distribution_test_has_value(
self._handle, _ffi.ffi_encode_string(ping_name)
)
)
def test_get_value(self, ping_name: Optional[str] = None) -> DistributionData:
"""
Returns the stored value for testing purposes only.
Args:
ping_name (str): (default: first value in send_in_pings) The name
of the ping to retrieve the metric for.
Returns:
value (DistriubutionData): value of the stored metric.
"""
if ping_name is None:
ping_name = self._send_in_pings[0]
if not self.test_has_value(ping_name):
raise ValueError("metric has no value")
return DistributionData.from_json_string(
_ffi.ffi_decode_string(
_ffi.lib.glean_memory_distribution_test_get_value_as_json_string(
self._handle, _ffi.ffi_encode_string(ping_name)
)
)
)
def test_get_num_recorded_errors(
self, error_type: ErrorType, ping_name: Optional[str] = None
) -> int:
"""
Returns the number of errors recorded for the given metric.
Args:
error_type (ErrorType): The type of error recorded.
ping_name (str): (default: first value in send_in_pings) The name
of the ping to retrieve the metric for.
Returns:
num_errors (int): The number of errors recorded for the metric for
the given error type.
"""
if ping_name is None:
ping_name = self._send_in_pings[0]
return _ffi.lib.glean_memory_distribution_test_get_num_recorded_errors(
self._handle,
error_type.value,
_ffi.ffi_encode_string(ping_name),
)
__all__ = ["MemoryDistributionMetricType"]
Classes
class MemoryDistributionMetricType (disabled: bool, category: str, lifetime: Lifetime, name: str, send_in_pings: List[str], memory_unit: MemoryUnit)
-
This implements the developer facing API for recording memory distribution metrics.
Instances of this class type are automatically generated by
load_metrics()
, allowing developers to record values that were previously registered in the metrics.yaml file.Expand source code
class MemoryDistributionMetricType: """ This implements the developer facing API for recording memory distribution metrics. Instances of this class type are automatically generated by `glean.load_metrics`, allowing developers to record values that were previously registered in the metrics.yaml file. """ def __init__( self, disabled: bool, category: str, lifetime: Lifetime, name: str, send_in_pings: List[str], memory_unit: MemoryUnit, ): self._disabled = disabled self._send_in_pings = send_in_pings self._handle = _ffi.lib.glean_new_memory_distribution_metric( _ffi.ffi_encode_string(category), _ffi.ffi_encode_string(name), _ffi.ffi_encode_vec_string(send_in_pings), len(send_in_pings), lifetime.value, disabled, memory_unit.value, ) def __del__(self): if getattr(self, "_handle", 0) != 0: _ffi.lib.glean_destroy_memory_distribution_metric(self._handle) def accumulate(self, sample: int) -> None: """ Record a single value, in the unit specified by `memory_unit`, to the distribution. Args: sample: The value to record. """ if self._disabled: return None @Dispatcher.launch def accumulate(): return _ffi.lib.glean_memory_distribution_accumulate(self._handle, sample) def test_has_value(self, ping_name: Optional[str] = None) -> bool: """ Tests whether a value is stored for the metric for testing purposes only. Args: ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: has_value (bool): True if the metric value exists. """ if ping_name is None: ping_name = self._send_in_pings[0] return bool( _ffi.lib.glean_memory_distribution_test_has_value( self._handle, _ffi.ffi_encode_string(ping_name) ) ) def test_get_value(self, ping_name: Optional[str] = None) -> DistributionData: """ Returns the stored value for testing purposes only. Args: ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: value (DistriubutionData): value of the stored metric. """ if ping_name is None: ping_name = self._send_in_pings[0] if not self.test_has_value(ping_name): raise ValueError("metric has no value") return DistributionData.from_json_string( _ffi.ffi_decode_string( _ffi.lib.glean_memory_distribution_test_get_value_as_json_string( self._handle, _ffi.ffi_encode_string(ping_name) ) ) ) def test_get_num_recorded_errors( self, error_type: ErrorType, ping_name: Optional[str] = None ) -> int: """ Returns the number of errors recorded for the given metric. Args: error_type (ErrorType): The type of error recorded. ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: num_errors (int): The number of errors recorded for the metric for the given error type. """ if ping_name is None: ping_name = self._send_in_pings[0] return _ffi.lib.glean_memory_distribution_test_get_num_recorded_errors( self._handle, error_type.value, _ffi.ffi_encode_string(ping_name), )
Methods
def accumulate(self, sample: int) ‑> NoneType
-
Record a single value, in the unit specified by
memory_unit
, to the distribution.Args
sample
- The value to record.
Expand source code
def accumulate(self, sample: int) -> None: """ Record a single value, in the unit specified by `memory_unit`, to the distribution. Args: sample: The value to record. """ if self._disabled: return None @Dispatcher.launch def accumulate(): return _ffi.lib.glean_memory_distribution_accumulate(self._handle, sample)
def test_get_num_recorded_errors(self, error_type: ErrorType, ping_name: Union[str, NoneType] = None) ‑> int
-
Returns the number of errors recorded for the given metric.
Args
error_type
:ErrorType
- The type of error recorded.
ping_name
:str
- (default: first value in send_in_pings) The name of the ping to retrieve the metric for.
Returns
num_errors (int): The number of errors recorded for the metric for the given error type.
Expand source code
def test_get_num_recorded_errors( self, error_type: ErrorType, ping_name: Optional[str] = None ) -> int: """ Returns the number of errors recorded for the given metric. Args: error_type (ErrorType): The type of error recorded. ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: num_errors (int): The number of errors recorded for the metric for the given error type. """ if ping_name is None: ping_name = self._send_in_pings[0] return _ffi.lib.glean_memory_distribution_test_get_num_recorded_errors( self._handle, error_type.value, _ffi.ffi_encode_string(ping_name), )
def test_get_value(self, ping_name: Union[str, NoneType] = None) ‑> DistributionData
-
Returns the stored value for testing purposes only.
Args
ping_name
:str
- (default: first value in send_in_pings) The name of the ping to retrieve the metric for.
Returns
value (DistriubutionData): value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> DistributionData: """ Returns the stored value for testing purposes only. Args: ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: value (DistriubutionData): value of the stored metric. """ if ping_name is None: ping_name = self._send_in_pings[0] if not self.test_has_value(ping_name): raise ValueError("metric has no value") return DistributionData.from_json_string( _ffi.ffi_decode_string( _ffi.lib.glean_memory_distribution_test_get_value_as_json_string( self._handle, _ffi.ffi_encode_string(ping_name) ) ) )
def test_has_value(self, ping_name: Union[str, NoneType] = None) ‑> bool
-
Tests whether a value is stored for the metric for testing purposes only.
Args
ping_name
:str
- (default: first value in send_in_pings) The name of the ping to retrieve the metric for.
Returns
has_value (bool): True if the metric value exists.
Expand source code
def test_has_value(self, ping_name: Optional[str] = None) -> bool: """ Tests whether a value is stored for the metric for testing purposes only. Args: ping_name (str): (default: first value in send_in_pings) The name of the ping to retrieve the metric for. Returns: has_value (bool): True if the metric value exists. """ if ping_name is None: ping_name = self._send_in_pings[0] return bool( _ffi.lib.glean_memory_distribution_test_has_value( self._handle, _ffi.ffi_encode_string(ping_name) ) )