Module glean.metrics.counter
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 .lifetime import Lifetime
class CounterMetricType:
"""
This implements the developer facing API for recording counter 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.
The counter API only exposes the `CounterMetricType.add` method, which
takes care of validating the input data and making sure that limits are
enforced.
"""
def __init__(
self,
disabled: bool,
category: str,
lifetime: Lifetime,
name: str,
send_in_pings: List[str],
):
self._disabled = disabled
self._send_in_pings = send_in_pings
self._handle = _ffi.lib.glean_new_counter_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,
)
def __del__(self):
if getattr(self, "_handle", 0) != 0:
_ffi.lib.glean_destroy_counter_metric(self._handle)
def add(self, amount: int = 1) -> None:
"""
Add to counter value.
Args:
amount (int): (default: 1) This is the amount to increment the
counter by.
"""
if self._disabled:
return
@Dispatcher.launch
def add():
_ffi.lib.glean_counter_add(self._handle, amount)
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_counter_test_has_value(
self._handle, _ffi.ffi_encode_string(ping_name)
)
)
def test_get_value(self, ping_name: Optional[str] = None) -> int:
"""
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 (int): 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 _ffi.lib.glean_counter_test_get_value(
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_counter_test_get_num_recorded_errors(
self._handle,
error_type.value,
_ffi.ffi_encode_string(ping_name),
)
__all__ = ["CounterMetricType"]
Classes
class CounterMetricType (disabled: bool, category: str, lifetime: Lifetime, name: str, send_in_pings: List[str])
-
This implements the developer facing API for recording counter 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.The counter API only exposes the
CounterMetricType.add()
method, which takes care of validating the input data and making sure that limits are enforced.Expand source code
class CounterMetricType: """ This implements the developer facing API for recording counter 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. The counter API only exposes the `CounterMetricType.add` method, which takes care of validating the input data and making sure that limits are enforced. """ def __init__( self, disabled: bool, category: str, lifetime: Lifetime, name: str, send_in_pings: List[str], ): self._disabled = disabled self._send_in_pings = send_in_pings self._handle = _ffi.lib.glean_new_counter_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, ) def __del__(self): if getattr(self, "_handle", 0) != 0: _ffi.lib.glean_destroy_counter_metric(self._handle) def add(self, amount: int = 1) -> None: """ Add to counter value. Args: amount (int): (default: 1) This is the amount to increment the counter by. """ if self._disabled: return @Dispatcher.launch def add(): _ffi.lib.glean_counter_add(self._handle, amount) 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_counter_test_has_value( self._handle, _ffi.ffi_encode_string(ping_name) ) ) def test_get_value(self, ping_name: Optional[str] = None) -> int: """ 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 (int): 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 _ffi.lib.glean_counter_test_get_value( 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_counter_test_get_num_recorded_errors( self._handle, error_type.value, _ffi.ffi_encode_string(ping_name), )
Methods
def add(self, amount: int = 1) ‑> NoneType
-
Add to counter value.
Args
amount
:int
- (default: 1) This is the amount to increment the counter by.
Expand source code
def add(self, amount: int = 1) -> None: """ Add to counter value. Args: amount (int): (default: 1) This is the amount to increment the counter by. """ if self._disabled: return @Dispatcher.launch def add(): _ffi.lib.glean_counter_add(self._handle, amount)
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_counter_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) ‑> int
-
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 (int): value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> int: """ 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 (int): 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 _ffi.lib.glean_counter_test_get_value( 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_counter_test_has_value( self._handle, _ffi.ffi_encode_string(ping_name) ) )