Module glean.metrics

This module contains all of the metric types.

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/.


"""
This module contains all of the metric types.
"""


from .boolean import BooleanMetricType
from .counter import CounterMetricType
from .datetime import DatetimeMetricType
from .event import EventMetricType, RecordedEventData
from .experiment import RecordedExperimentData
from .jwe import JweMetricType
from .labeled import (
    LabeledBooleanMetricType,
    LabeledCounterMetricType,
    LabeledStringMetricType,
)
from .lifetime import Lifetime
from .memory_distribution import MemoryDistributionMetricType
from .memoryunit import MemoryUnit
from .ping import PingType
from .string import StringMetricType
from .string_list import StringListMetricType
from .timespan import TimespanMetricType
from .timeunit import TimeUnit
from .timing_distribution import TimingDistributionMetricType
from .uuid import UuidMetricType


__all__ = [
    "BooleanMetricType",
    "CounterMetricType",
    "DatetimeMetricType",
    "EventMetricType",
    "JweMetricType",
    "LabeledBooleanMetricType",
    "LabeledCounterMetricType",
    "LabeledStringMetricType",
    "Lifetime",
    "MemoryDistributionMetricType",
    "MemoryUnit",
    "PingType",
    "RecordedEventData",
    "RecordedExperimentData",
    "StringMetricType",
    "StringListMetricType",
    "TimespanMetricType",
    "TimeUnit",
    "TimingDistributionMetricType",
    "UuidMetricType",
]

Sub-modules

glean.metrics.boolean
glean.metrics.counter
glean.metrics.datetime
glean.metrics.distribution_data
glean.metrics.event
glean.metrics.experiment
glean.metrics.jwe
glean.metrics.labeled
glean.metrics.lifetime
glean.metrics.memory_distribution
glean.metrics.memoryunit
glean.metrics.ping
glean.metrics.string
glean.metrics.string_list
glean.metrics.timespan
glean.metrics.timeunit
glean.metrics.timing_distribution
glean.metrics.uuid

Classes

class BooleanMetricType (disabled, category, lifetime, name, send_in_pings)

This implements the developer facing API for recording boolean 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 boolean API only exposes the BooleanMetricType.set() method

Expand source code
class BooleanMetricType:
    """
    This implements the developer facing API for recording boolean 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 boolean API only exposes the `BooleanMetricType.set` method
    """

    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_boolean_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_boolean_metric(self._handle)

    def set(self, value: bool) -> None:
        """
        Set a boolean value.

        Args:
            value (bool): This is a user-defined boolean value.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_boolean_set(self._handle, value)

    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_boolean_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> bool:
        """
        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 (bool): 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 bool(
            _ffi.lib.glean_boolean_test_get_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

Methods

def set(self, value)

Set a boolean value.

Args

value : bool
This is a user-defined boolean value.
Expand source code
def set(self, value: bool) -> None:
    """
    Set a boolean value.

    Args:
        value (bool): This is a user-defined boolean value.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_boolean_set(self._handle, value)
def test_get_value(self, ping_name=None)

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 : bool
value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> bool:
    """
    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 (bool): 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 bool(
        _ffi.lib.glean_boolean_test_get_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
def test_has_value(self, ping_name=None)

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_boolean_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class CounterMetricType (disabled, category, lifetime, name, send_in_pings)

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=1)

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, ping_name=None)

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=None)

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=None)

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)
        )
    )
class DatetimeMetricType (disabled, category, lifetime, name, send_in_pings, time_unit)

This implements the developer facing API for recording datetime 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 datetime API only exposes the DatetimeMetricType.set() method.

Expand source code
class DatetimeMetricType:
    """
    This implements the developer facing API for recording datetime 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 datetime API only exposes the `DatetimeMetricType.set` method.
    """

    def __init__(
        self,
        disabled: bool,
        category: str,
        lifetime: Lifetime,
        name: str,
        send_in_pings: List[str],
        time_unit: TimeUnit,
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_datetime_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,
            time_unit.value,
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_datetime_metric(self._handle)

    def set(self, value: Optional[datetime.datetime] = None) -> None:
        """
        Set a datetime value, truncating it to the metric's resolution.

        Args:
            value (datetime.datetime): (default: now) The `datetime.datetime`
                value to set. If not provided, will record the current time.
        """
        if self._disabled:
            return

        if value is None:
            value = datetime.datetime.now()

        @Dispatcher.launch
        def set():
            tzinfo = value.tzinfo
            if tzinfo is not None:
                offset = tzinfo.utcoffset(value).seconds
            else:
                offset = 0
            _ffi.lib.glean_datetime_set(
                self._handle,
                value.year,
                value.month,
                value.day,
                value.hour,
                value.minute,
                value.second,
                value.microsecond * 1000,
                offset,
            )

    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_datetime_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value_as_str(self, ping_name: Optional[str] = None) -> str:
        """
        Returns the stored value for testing purposes only, as an ISO8601 string.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (str): 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.ffi_decode_string(
            _ffi.lib.glean_datetime_test_get_value_as_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> datetime.datetime:
        """
        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 (datetime.datetime): value of the stored metric.
        """
        if sys.version_info < (3, 7):
            return iso8601.parse_date(self.test_get_value_as_str(ping_name))
        else:
            return datetime.datetime.fromisoformat(
                self.test_get_value_as_str(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_datetime_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def set(self, value=None)

Set a datetime value, truncating it to the metric's resolution.

Args

value : datetime.datetime
(default: now) The datetime.datetime value to set. If not provided, will record the current time.
Expand source code
def set(self, value: Optional[datetime.datetime] = None) -> None:
    """
    Set a datetime value, truncating it to the metric's resolution.

    Args:
        value (datetime.datetime): (default: now) The `datetime.datetime`
            value to set. If not provided, will record the current time.
    """
    if self._disabled:
        return

    if value is None:
        value = datetime.datetime.now()

    @Dispatcher.launch
    def set():
        tzinfo = value.tzinfo
        if tzinfo is not None:
            offset = tzinfo.utcoffset(value).seconds
        else:
            offset = 0
        _ffi.lib.glean_datetime_set(
            self._handle,
            value.year,
            value.month,
            value.day,
            value.hour,
            value.minute,
            value.second,
            value.microsecond * 1000,
            offset,
        )
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_datetime_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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 : datetime.datetime
value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> datetime.datetime:
    """
    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 (datetime.datetime): value of the stored metric.
    """
    if sys.version_info < (3, 7):
        return iso8601.parse_date(self.test_get_value_as_str(ping_name))
    else:
        return datetime.datetime.fromisoformat(
            self.test_get_value_as_str(ping_name)
        )
def test_get_value_as_str(self, ping_name=None)

Returns the stored value for testing purposes only, as an ISO8601 string.

Args

ping_name : str
(default: first value in send_in_pings) The name of the ping to retrieve the metric for.

Returns

value : str
value of the stored metric.
Expand source code
def test_get_value_as_str(self, ping_name: Optional[str] = None) -> str:
    """
    Returns the stored value for testing purposes only, as an ISO8601 string.

    Args:
        ping_name (str): (default: first value in send_in_pings) The name
            of the ping to retrieve the metric for.

    Returns:
        value (str): 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.ffi_decode_string(
        _ffi.lib.glean_datetime_test_get_value_as_string(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
def test_has_value(self, ping_name=None)

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_datetime_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class EventMetricType (disabled, category, lifetime, name, send_in_pings, allowed_extra_keys)

This implements the developer facing API for recording events.

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 event API only exposes the EventMetricType.record() method, which takes care of validating the input data and making sure that limits are enforced.

Expand source code
class EventMetricType:
    """
    This implements the developer facing API for recording events.

    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 event API only exposes the `EventMetricType.record` 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],
        allowed_extra_keys: List[str],
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_event_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,
            _ffi.ffi_encode_vec_string(allowed_extra_keys),
            len(allowed_extra_keys),
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_event_metric(self._handle)

    def record(self, extra: Optional[Dict[int, str]] = None) -> None:
        """
        Record an event by using the information provided by the instance of
        this class.

        Args:
            extra (dict of (int, str)): optional. This is a map from keys
                (which are enumerations) to values. This is used for events
                where additional richer context is needed. The maximum length
                for values is 100.
        """
        if self._disabled:
            return

        timestamp = _util.time_ms()

        @Dispatcher.launch
        def record():
            if extra is None:
                keys = []
                values = []
                nextra = 0
            else:
                keys, values = zip(*list(extra.items()))
                keys = [x.value for x in keys]
                nextra = len(extra)

            _ffi.lib.glean_event_record(
                self._handle,
                timestamp,
                _ffi.ffi_encode_vec_int32(keys),
                _ffi.ffi_encode_vec_string(values),
                nextra,
            )

    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_event_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(
        self, ping_name: Optional[str] = None
    ) -> List[RecordedEventData]:
        """
        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 (list of RecordedEventData): value of the stored events.
        """
        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")

        json_string = _ffi.ffi_decode_string(
            _ffi.lib.glean_event_test_get_value_as_json_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

        json_content = json.loads(json_string)

        return [RecordedEventData(**x) for x in json_content]

    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_event_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def record(self, extra=None)

Record an event by using the information provided by the instance of this class.

Args

extra (dict of (int, str)): optional. This is a map from keys (which are enumerations) to values. This is used for events where additional richer context is needed. The maximum length for values is 100.

Expand source code
def record(self, extra: Optional[Dict[int, str]] = None) -> None:
    """
    Record an event by using the information provided by the instance of
    this class.

    Args:
        extra (dict of (int, str)): optional. This is a map from keys
            (which are enumerations) to values. This is used for events
            where additional richer context is needed. The maximum length
            for values is 100.
    """
    if self._disabled:
        return

    timestamp = _util.time_ms()

    @Dispatcher.launch
    def record():
        if extra is None:
            keys = []
            values = []
            nextra = 0
        else:
            keys, values = zip(*list(extra.items()))
            keys = [x.value for x in keys]
            nextra = len(extra)

        _ffi.lib.glean_event_record(
            self._handle,
            timestamp,
            _ffi.ffi_encode_vec_int32(keys),
            _ffi.ffi_encode_vec_string(values),
            nextra,
        )
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_event_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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 : list of RecordedEventData
value of the stored events.
Expand source code
def test_get_value(
    self, ping_name: Optional[str] = None
) -> List[RecordedEventData]:
    """
    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 (list of RecordedEventData): value of the stored events.
    """
    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")

    json_string = _ffi.ffi_decode_string(
        _ffi.lib.glean_event_test_get_value_as_json_string(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )

    json_content = json.loads(json_string)

    return [RecordedEventData(**x) for x in json_content]
def test_has_value(self, ping_name=None)

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_event_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class JweMetricType (disabled, category, lifetime, name, send_in_pings)

This implements the developer facing API for recording JWE 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 string API exposes the JweMetricType.set() and JweMetricType.setWithCompactRepresentation methods, which take care of validating the input data and making sure that limits are enforced.

Expand source code
class JweMetricType:
    """
    This implements the developer facing API for recording JWE 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 string API exposes the `JweMetricType.set` and
    `JweMetricType.setWithCompactRepresentation` methods,
    which take 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_jwe_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_jwe_metric(self._handle)

    def set_with_compact_representation(self, value: str):
        """
        Set to the specified JWE value.

        Args:
            value (str): the compact representation of a JWE value.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set_with_compact_representation(
                self._handle, _ffi.ffi_encode_string(value)
            )

    def set(
        self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
    ) -> None:
        """
        Build a JWE value from its elements and set to it.

        Args:
            header (str): A variable-size JWE protected header.
            key (str): A variable-size encrypted key.
                This can be an empty octet sequence.
            init_vector (str): A fixed-size, 96-bit, base64 encoded Jwe initialization vector.
                If not required by the encryption algorithm, can be an empty octet sequence.
            cipher_text (str): The variable-size base64 encoded cipher text.
            auth_tag (str): A fixed-size, 132-bit, base64 encoded authentication tag.
                Can be an empty octet sequence.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_jwe_set(
                self._handle,
                _ffi.ffi_encode_string(header),
                _ffi.ffi_encode_string(key),
                _ffi.ffi_encode_string(init_vector),
                _ffi.ffi_encode_string(cipher_text),
                _ffi.ffi_encode_string(auth_tag),
            )

    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_jwe_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> JweData:
        """
        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 (JweData): 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")

        json_payload = json.loads(
            _ffi.ffi_decode_string(
                _ffi.lib.glean_jwe_test_get_value_as_json_string(
                    self._handle, _ffi.ffi_encode_string(ping_name)
                )
            )
        )

        return JweData(
            json_payload["header"],
            json_payload["key"],
            json_payload["init_vector"],
            json_payload["cipher_text"],
            json_payload["auth_tag"],
        )

    def test_get_compact_representation(self, ping_name: Optional[str] = None) -> str:
        """
        Returns the stored value as the compact representation
        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 (JweData): 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.ffi_decode_string(
            _ffi.lib.glean_jwe_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_jwe_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def set(self, header, key, init_vector, cipher_text, auth_tag)

Build a JWE value from its elements and set to it.

Args

header : str
A variable-size JWE protected header.
key : str
A variable-size encrypted key. This can be an empty octet sequence.
init_vector : str
A fixed-size, 96-bit, base64 encoded Jwe initialization vector. If not required by the encryption algorithm, can be an empty octet sequence.
cipher_text : str
The variable-size base64 encoded cipher text.
auth_tag : str
A fixed-size, 132-bit, base64 encoded authentication tag. Can be an empty octet sequence.
Expand source code
def set(
    self, header: str, key: str, init_vector: str, cipher_text: str, auth_tag: str
) -> None:
    """
    Build a JWE value from its elements and set to it.

    Args:
        header (str): A variable-size JWE protected header.
        key (str): A variable-size encrypted key.
            This can be an empty octet sequence.
        init_vector (str): A fixed-size, 96-bit, base64 encoded Jwe initialization vector.
            If not required by the encryption algorithm, can be an empty octet sequence.
        cipher_text (str): The variable-size base64 encoded cipher text.
        auth_tag (str): A fixed-size, 132-bit, base64 encoded authentication tag.
            Can be an empty octet sequence.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_jwe_set(
            self._handle,
            _ffi.ffi_encode_string(header),
            _ffi.ffi_encode_string(key),
            _ffi.ffi_encode_string(init_vector),
            _ffi.ffi_encode_string(cipher_text),
            _ffi.ffi_encode_string(auth_tag),
        )
def set_with_compact_representation(self, value)

Set to the specified JWE value.

Args

value : str
the compact representation of a JWE value.
Expand source code
def set_with_compact_representation(self, value: str):
    """
    Set to the specified JWE value.

    Args:
        value (str): the compact representation of a JWE value.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_jwe_set_with_compact_representation(
            self._handle, _ffi.ffi_encode_string(value)
        )
def test_get_compact_representation(self, ping_name=None)

Returns the stored value as the compact representation 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 : JweData
value of the stored metric.
Expand source code
def test_get_compact_representation(self, ping_name: Optional[str] = None) -> str:
    """
    Returns the stored value as the compact representation
    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 (JweData): 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.ffi_decode_string(
        _ffi.lib.glean_jwe_test_get_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_jwe_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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 : JweData
value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> JweData:
    """
    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 (JweData): 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")

    json_payload = json.loads(
        _ffi.ffi_decode_string(
            _ffi.lib.glean_jwe_test_get_value_as_json_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )
    )

    return JweData(
        json_payload["header"],
        json_payload["key"],
        json_payload["init_vector"],
        json_payload["cipher_text"],
        json_payload["auth_tag"],
    )
def test_has_value(self, ping_name=None)

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_jwe_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class LabeledBooleanMetricType (disabled, category, lifetime, name, send_in_pings, labels=None)

This implements the developer-facing API for labeled 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.

Unlike most metric types, LabeledMetricType does not have its own corresponding storage, but records metrics for the underlying metric type T in the storage for that type. The only difference is that labeled metrics are stored with the special key $category.$name/$label. The collect method knows how to pull these special values back out of the individual metric storage and rearrange them correctly in the ping.

Expand source code
class LabeledBooleanMetricType(LabeledMetricBase):
    _submetric_type = BooleanMetricType
    _metric_type_instantiator = _ffi.lib.glean_new_labeled_boolean_metric
    _submetric_type_instantiator = _ffi.lib.glean_labeled_boolean_metric_get
    _test_get_num_recorded_errors_ffi = (
        _ffi.lib.glean_labeled_boolean_test_get_num_recorded_errors
    )

Ancestors

  • glean.metrics.labeled.LabeledMetricBase
class LabeledCounterMetricType (disabled, category, lifetime, name, send_in_pings, labels=None)

This implements the developer-facing API for labeled 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.

Unlike most metric types, LabeledMetricType does not have its own corresponding storage, but records metrics for the underlying metric type T in the storage for that type. The only difference is that labeled metrics are stored with the special key $category.$name/$label. The collect method knows how to pull these special values back out of the individual metric storage and rearrange them correctly in the ping.

Expand source code
class LabeledCounterMetricType(LabeledMetricBase):
    _submetric_type = CounterMetricType
    _metric_type_instantiator = _ffi.lib.glean_new_labeled_counter_metric
    _submetric_type_instantiator = _ffi.lib.glean_labeled_counter_metric_get
    _test_get_num_recorded_errors_ffi = (
        _ffi.lib.glean_labeled_counter_test_get_num_recorded_errors
    )

Ancestors

  • glean.metrics.labeled.LabeledMetricBase
class LabeledStringMetricType (disabled, category, lifetime, name, send_in_pings, labels=None)

This implements the developer-facing API for labeled 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.

Unlike most metric types, LabeledMetricType does not have its own corresponding storage, but records metrics for the underlying metric type T in the storage for that type. The only difference is that labeled metrics are stored with the special key $category.$name/$label. The collect method knows how to pull these special values back out of the individual metric storage and rearrange them correctly in the ping.

Expand source code
class LabeledStringMetricType(LabeledMetricBase):
    _submetric_type = StringMetricType
    _metric_type_instantiator = _ffi.lib.glean_new_labeled_string_metric
    _submetric_type_instantiator = _ffi.lib.glean_labeled_string_metric_get
    _test_get_num_recorded_errors_ffi = (
        _ffi.lib.glean_labeled_string_test_get_num_recorded_errors
    )

Ancestors

  • glean.metrics.labeled.LabeledMetricBase
class Lifetime (*args, **kwargs)

An enumeration for the different metric lifetimes that Glean supports.

Metric lifetimes define when a metric is reset.

Expand source code
class Lifetime(Enum):
    """
    An enumeration for the different metric lifetimes that Glean supports.

    Metric lifetimes define when a metric is reset.
    """

    PING = _ffi.lib.Lifetime_Ping
    """
    The metric is reset with each sent ping
    """

    APPLICATION = _ffi.lib.Lifetime_Application
    """
    The metric is reset on application restart
    """

    USER = _ffi.lib.Lifetime_User
    """
    The metric is reset with each user profile
    """

Ancestors

  • enum.Enum

Class variables

var APPLICATION

The metric is reset on application restart

var PING

The metric is reset with each sent ping

var USER

The metric is reset with each user profile

class MemoryDistributionMetricType (disabled, category, lifetime, name, send_in_pings, memory_unit)

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)

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, ping_name=None)

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=None)

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=None)

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)
        )
    )
class MemoryUnit (*args, **kwargs)

An enumeration of different resolutions supported by the glean.metrics.MemoryDistribution metric type.

These use the power-of-2 values of these units, that is, Kilobyte is pedantically a Kibibyte.

Expand source code
class MemoryUnit(IntEnum):
    """
    An enumeration of different resolutions supported by the
    `glean.metrics.MemoryDistribution` metric type.

    These use the power-of-2 values of these units, that is, Kilobyte is
    pedantically a Kibibyte.
    """

    BYTE = _ffi.lib.MemoryUnit_Byte
    """
    Byte: 1 byte.
    """

    KILOBYTE = _ffi.lib.MemoryUnit_Kilobyte
    """
    Kilobyte: 2^10 bytes
    """

    MEGABYTE = _ffi.lib.MemoryUnit_Megabyte
    """
    Megabyte: 2^20 bytes
    """

    GIGABYTE = _ffi.lib.MemoryUnit_Gigabyte
    """
    Gigabyte: 2^30 bytes
    """

Ancestors

  • enum.IntEnum
  • builtins.int
  • enum.Enum

Class variables

var BYTE

Byte: 1 byte.

var GIGABYTE

Gigabyte: 2^30 bytes

var KILOBYTE

Kilobyte: 2^10 bytes

var MEGABYTE

Megabyte: 2^20 bytes

class PingType (name, include_client_id, send_if_empty, reason_codes)

This implements the developer facing API for custom pings.

The Ping API only exposes the PingType.submit() method, which schedules a ping for eventual uploading.

Expand source code
class PingType:
    def __init__(
        self,
        name: str,
        include_client_id: bool,
        send_if_empty: bool,
        reason_codes: List[str],
    ):
        """
        This implements the developer facing API for custom pings.

        The Ping API only exposes the `PingType.submit` method, which schedules a
        ping for eventual uploading.
        """
        self._name = name
        self._reason_codes = reason_codes
        self._handle = _ffi.lib.glean_new_ping_type(
            _ffi.ffi_encode_string(name),
            include_client_id,
            send_if_empty,
            _ffi.ffi_encode_vec_string(reason_codes),
            len(reason_codes),
        )
        Glean.register_ping_type(self)

    def __del__(self):
        if self._handle != 0:
            _ffi.lib.glean_destroy_ping_type(self._handle)

    @property
    def name(self) -> str:
        """
        Get the name of the ping.
        """
        return self._name

    def submit(self, reason: Optional[int] = None) -> None:
        """
        Collect and submit the ping for eventual uploading.

        If the ping currently contains no content, it will not be sent.

        Args:
            reason (enum, optional): The reason the ping was submitted.
        """
        reason_string: Optional[str] = None
        if reason is not None:
            reason_string = self._reason_codes[reason]
        else:
            reason_string = None
        Glean._submit_ping(self, reason_string)

Instance variables

var name

Get the name of the ping.

Expand source code
@property
def name(self) -> str:
    """
    Get the name of the ping.
    """
    return self._name

Methods

def submit(self, reason=None)

Collect and submit the ping for eventual uploading.

If the ping currently contains no content, it will not be sent.

Args

reason : enum, optional
The reason the ping was submitted.
Expand source code
def submit(self, reason: Optional[int] = None) -> None:
    """
    Collect and submit the ping for eventual uploading.

    If the ping currently contains no content, it will not be sent.

    Args:
        reason (enum, optional): The reason the ping was submitted.
    """
    reason_string: Optional[str] = None
    if reason is not None:
        reason_string = self._reason_codes[reason]
    else:
        reason_string = None
    Glean._submit_ping(self, reason_string)
class RecordedEventData (category, name, timestamp, extra=None)

Deserialized event data.

Args

category : str
The event's category, part of the full identifier.
name : str
The event's name, part of the full identifier.
timestamp : int
The event's timestamp, in milliseconds.

extra (dict of str->str): Optional. Any extra data recorded for the event.

Expand source code
class RecordedEventData:
    """
    Deserialized event data.
    """

    def __init__(
        self,
        category: str,
        name: str,
        timestamp: int,
        extra: Optional[Dict[str, str]] = None,
    ):
        """
        Args:
            category (str): The event's category, part of the full identifier.
            name (str): The event's name, part of the full identifier.
            timestamp (int): The event's timestamp, in milliseconds.
            extra (dict of str->str): Optional. Any extra data recorded for
                the event.
        """
        self._category = category
        self._name = name
        self._timestamp = timestamp
        if extra is None:
            extra = {}
        self._extra = extra

    @property
    def category(self) -> str:
        """The event's category, part of the full identifier."""
        return self._category

    @property
    def name(self) -> str:
        """The event's name, part of the full identifier."""
        return self._name

    @property
    def timestamp(self) -> int:
        """The event's timestamp."""
        return self._timestamp

    @property
    def extra(self) -> Optional[Dict[str, str]]:
        """Any extra data recorded for the event."""
        return self._extra

    @property
    def identifier(self) -> str:
        if self.category == "":
            return self.name
        else:
            return ".".join([self.category, self.name])

Instance variables

var category

The event's category, part of the full identifier.

Expand source code
@property
def category(self) -> str:
    """The event's category, part of the full identifier."""
    return self._category
var extra

Any extra data recorded for the event.

Expand source code
@property
def extra(self) -> Optional[Dict[str, str]]:
    """Any extra data recorded for the event."""
    return self._extra
var identifier
Expand source code
@property
def identifier(self) -> str:
    if self.category == "":
        return self.name
    else:
        return ".".join([self.category, self.name])
var name

The event's name, part of the full identifier.

Expand source code
@property
def name(self) -> str:
    """The event's name, part of the full identifier."""
    return self._name
var timestamp

The event's timestamp.

Expand source code
@property
def timestamp(self) -> int:
    """The event's timestamp."""
    return self._timestamp
class RecordedExperimentData (branch, extra=None)

Deserialized experiment data.

Args

branch : str
The experiment's branch.

extra (dict of str->str): Any extra data associated with this experiment.

Expand source code
class RecordedExperimentData:
    """
    Deserialized experiment data.
    """

    def __init__(self, branch: str, extra: Dict[str, str] = None):
        """
        Args:
            branch (str): The experiment's branch.
            extra (dict of str->str): Any extra data associated with this
                experiment.
        """
        self._branch = branch
        if extra is None:
            extra = {}
        self._extra = extra

    @property
    def branch(self) -> str:
        """The experiment's branch."""
        return self._branch

    @property
    def extra(self) -> Dict[str, str]:
        """Any extra data associated with this experiment."""
        return self._extra

Instance variables

var branch

The experiment's branch.

Expand source code
@property
def branch(self) -> str:
    """The experiment's branch."""
    return self._branch
var extra

Any extra data associated with this experiment.

Expand source code
@property
def extra(self) -> Dict[str, str]:
    """Any extra data associated with this experiment."""
    return self._extra
class StringListMetricType (disabled, category, lifetime, name, send_in_pings)

This implements the developer facing API for recording string list 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 stringlist API only exposes the StringListMetricType.set() and StringListMetricType.add() methods, which takes case of validating the input data and making sure that limits are enforced.

Expand source code
class StringListMetricType:
    """
    This implements the developer facing API for recording string list 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 stringlist API only exposes the `StringListMetricType.set` and
    `StringListMetricType.add` methods, which takes case 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_string_list_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_string_list_metric(self._handle)

    def add(self, value: str) -> None:
        """
        Appends a string value to one or more string list metric stores. If the
        string exceeds the maximum string length or if the list exceeds the
        maximum length it will be truncated.

        Args:
            value (str): The is the user defined string value.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_string_list_add(self._handle, _ffi.ffi_encode_string(value))

    def set(self, value: Iterable[str]) -> None:
        """
        Sets a string list to one or more metric stores. If any string exceeds
        the maximum string length or if the list exceeds the maximum length it
        will be truncated.

        Args:
            value (iterator over str): This is a user-defined string list.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            values = list(value)
            _ffi.lib.glean_string_list_set(
                self._handle, _ffi.ffi_encode_vec_string(values), len(values)
            )

    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_string_list_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> List[str]:
        """
        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 (bool): 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 json.loads(
            _ffi.ffi_decode_string(
                _ffi.lib.glean_string_list_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_string_list_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def add(self, value)

Appends a string value to one or more string list metric stores. If the string exceeds the maximum string length or if the list exceeds the maximum length it will be truncated.

Args

value : str
The is the user defined string value.
Expand source code
def add(self, value: str) -> None:
    """
    Appends a string value to one or more string list metric stores. If the
    string exceeds the maximum string length or if the list exceeds the
    maximum length it will be truncated.

    Args:
        value (str): The is the user defined string value.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_string_list_add(self._handle, _ffi.ffi_encode_string(value))
def set(self, value)

Sets a string list to one or more metric stores. If any string exceeds the maximum string length or if the list exceeds the maximum length it will be truncated.

Args

value : iterator over str
This is a user-defined string list.
Expand source code
def set(self, value: Iterable[str]) -> None:
    """
    Sets a string list to one or more metric stores. If any string exceeds
    the maximum string length or if the list exceeds the maximum length it
    will be truncated.

    Args:
        value (iterator over str): This is a user-defined string list.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        values = list(value)
        _ffi.lib.glean_string_list_set(
            self._handle, _ffi.ffi_encode_vec_string(values), len(values)
        )
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_string_list_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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 : bool
value of the stored metric.
Expand source code
def test_get_value(self, ping_name: Optional[str] = None) -> List[str]:
    """
    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 (bool): 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 json.loads(
        _ffi.ffi_decode_string(
            _ffi.lib.glean_string_list_test_get_value_as_json_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )
    )
def test_has_value(self, ping_name=None)

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_string_list_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class StringMetricType (disabled, category, lifetime, name, send_in_pings)

This implements the developer facing API for recording string 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 string API only exposes the StringMetricType.set() method, which takes care of validating the input data and making sure that limits are enforced.

Expand source code
class StringMetricType:
    """
    This implements the developer facing API for recording string 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 string API only exposes the `StringMetricType.set` 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_string_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_string_metric(self._handle)

    def set(self, value: str) -> None:
        """
        Set a string value.

        Args:
            value (str): This is a user-defined string value. If the length of
                the string exceeds the maximum length, it will be truncated.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_string_set(self._handle, _ffi.ffi_encode_string(value))

    def _set_sync(self, value: str) -> None:
        """
        Set a string value, synchronously.

        Args:
            value (str): This is a user-defined string value. If the length of
                the string exceeds the maximum length, it will be truncated.
        """
        if self._disabled:
            return

        _ffi.lib.glean_string_set(self._handle, _ffi.ffi_encode_string(value))

    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_string_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> str:
        """
        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.ffi_decode_string(
            _ffi.lib.glean_string_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_string_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def set(self, value)

Set a string value.

Args

value : str
This is a user-defined string value. If the length of the string exceeds the maximum length, it will be truncated.
Expand source code
def set(self, value: str) -> None:
    """
    Set a string value.

    Args:
        value (str): This is a user-defined string value. If the length of
            the string exceeds the maximum length, it will be truncated.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_string_set(self._handle, _ffi.ffi_encode_string(value))
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_string_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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) -> str:
    """
    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.ffi_decode_string(
        _ffi.lib.glean_string_test_get_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
def test_has_value(self, ping_name=None)

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_string_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class TimeUnit (*args, **kwargs)

An enumeration of different resolutions supported by the time-related metric types.

Expand source code
class TimeUnit(IntEnum):
    """
    An enumeration of different resolutions supported by the time-related
    metric types.
    """

    NANOSECOND = _ffi.lib.TimeUnit_Nanosecond
    """
    Represents nanosecond precision.
    """

    MICROSECOND = _ffi.lib.TimeUnit_Microsecond
    """
    Represents microsecond precision.
    """

    MILLISECOND = _ffi.lib.TimeUnit_Millisecond
    """
    Represents millisecond precision.
    """

    SECOND = _ffi.lib.TimeUnit_Second
    """
    Represents second precision.
    """

    MINUTE = _ffi.lib.TimeUnit_Minute
    """
    Represents minute precision.
    """

    HOUR = _ffi.lib.TimeUnit_Hour
    """
    Represents hour precision.
    """

    DAY = _ffi.lib.TimeUnit_Day
    """
    Represents day precision.
    """

Ancestors

  • enum.IntEnum
  • builtins.int
  • enum.Enum

Class variables

var DAY

Represents day precision.

var HOUR

Represents hour precision.

var MICROSECOND

Represents microsecond precision.

var MILLISECOND

Represents millisecond precision.

var MINUTE

Represents minute precision.

var NANOSECOND

Represents nanosecond precision.

var SECOND

Represents second precision.

class TimespanMetricType (disabled, category, lifetime, name, send_in_pings, time_unit)

This implements the developer facing API for recording timespan 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 timespan API exposes the TimespanMetricType.start(), TimespanMetricType.stop() and TimespanMetricType.cancel() methods.

Expand source code
class TimespanMetricType:
    """
    This implements the developer facing API for recording timespan 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 timespan API exposes the `TimespanMetricType.start`,
    `TimespanMetricType.stop` and `TimespanMetricType.cancel` methods.
    """

    def __init__(
        self,
        disabled: bool,
        category: str,
        lifetime: Lifetime,
        name: str,
        send_in_pings: List[str],
        time_unit: TimeUnit,
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_timespan_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,
            time_unit.value,
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_timespan_metric(self._handle)

    def start(self) -> None:
        """
        Start tracking time for the provided metric.

        This records an error if it’s already tracking time (i.e. `start` was
        already called with no corresponding `stop`): in that case the original
        start time will be preserved.
        """
        if self._disabled:
            return

        start_time = _util.time_ns()

        @Dispatcher.launch
        def start():
            _ffi.lib.glean_timespan_set_start(self._handle, start_time)

    def stop(self) -> None:
        """
        Stop tracking time for the provided metric.

        Sets the metric to the elapsed time, but does not overwrite an already
        existing value.
        This will record an error if no `start` was called or there is an already
        existing value.
        """
        if self._disabled:
            return

        stop_time = _util.time_ns()

        @Dispatcher.launch
        def stop():
            _ffi.lib.glean_timespan_set_stop(self._handle, stop_time)

    def cancel(self) -> None:
        """
        Abort a previous `start` call. No error is recorded if no `start` was called.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def cancel():
            _ffi.lib.glean_timespan_cancel(self._handle)

    def set_raw_nanos(self, elapsed_nanos: int) -> None:
        """
        Explicitly set the timespan value, in nanoseconds.

        This API should only be used if your library or application requires recording
        times in a way that can not make use of [start]/[stop]/[cancel].

        [setRawNanos] does not overwrite a running timer or an already existing value.

        Args:
            elapsed_nanos (int): The elapsed time to record, in nanoseconds.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set_raw_nanos():
            _ffi.lib.glean_timespan_set_raw_nanos(self._handle, elapsed_nanos)

    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_timespan_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 (bool): 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_timespan_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_timespan_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def cancel(self)

Abort a previous start call. No error is recorded if no start was called.

Expand source code
def cancel(self) -> None:
    """
    Abort a previous `start` call. No error is recorded if no `start` was called.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def cancel():
        _ffi.lib.glean_timespan_cancel(self._handle)
def set_raw_nanos(self, elapsed_nanos)

Explicitly set the timespan value, in nanoseconds.

This API should only be used if your library or application requires recording times in a way that can not make use of [start]/[stop]/[cancel].

[setRawNanos] does not overwrite a running timer or an already existing value.

Args

elapsed_nanos : int
The elapsed time to record, in nanoseconds.
Expand source code
def set_raw_nanos(self, elapsed_nanos: int) -> None:
    """
    Explicitly set the timespan value, in nanoseconds.

    This API should only be used if your library or application requires recording
    times in a way that can not make use of [start]/[stop]/[cancel].

    [setRawNanos] does not overwrite a running timer or an already existing value.

    Args:
        elapsed_nanos (int): The elapsed time to record, in nanoseconds.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set_raw_nanos():
        _ffi.lib.glean_timespan_set_raw_nanos(self._handle, elapsed_nanos)
def start(self)

Start tracking time for the provided metric.

This records an error if it’s already tracking time (i.e. start was already called with no corresponding stop): in that case the original start time will be preserved.

Expand source code
def start(self) -> None:
    """
    Start tracking time for the provided metric.

    This records an error if it’s already tracking time (i.e. `start` was
    already called with no corresponding `stop`): in that case the original
    start time will be preserved.
    """
    if self._disabled:
        return

    start_time = _util.time_ns()

    @Dispatcher.launch
    def start():
        _ffi.lib.glean_timespan_set_start(self._handle, start_time)
def stop(self)

Stop tracking time for the provided metric.

Sets the metric to the elapsed time, but does not overwrite an already existing value. This will record an error if no start was called or there is an already existing value.

Expand source code
def stop(self) -> None:
    """
    Stop tracking time for the provided metric.

    Sets the metric to the elapsed time, but does not overwrite an already
    existing value.
    This will record an error if no `start` was called or there is an already
    existing value.
    """
    if self._disabled:
        return

    stop_time = _util.time_ns()

    @Dispatcher.launch
    def stop():
        _ffi.lib.glean_timespan_set_stop(self._handle, stop_time)
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_timespan_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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 : bool
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 (bool): 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_timespan_test_get_value(
        self._handle, _ffi.ffi_encode_string(ping_name)
    )
def test_has_value(self, ping_name=None)

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_timespan_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class TimingDistributionMetricType (disabled, category, lifetime, name, send_in_pings, time_unit)

This implements the developer facing API for recording timing 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 TimingDistributionMetricType:
    """
    This implements the developer facing API for recording timing 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],
        time_unit: TimeUnit,
    ):
        self._disabled = disabled
        self._send_in_pings = send_in_pings

        self._handle = _ffi.lib.glean_new_timing_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,
            time_unit.value,
        )

    def __del__(self):
        if getattr(self, "_handle", 0) != 0:
            _ffi.lib.glean_destroy_timing_distribution_metric(self._handle)

    def start(self) -> Optional[int]:
        """
        Start tracking time. This records an error if it’s already tracking
        time (i.e. start was already called with no corresponding
        `stop_and_ccumulate`): in that case the original start time will be
        preserved.

        Returns:
            timer_id: The object to associate with this timing.
        """
        if self._disabled:
            return None

        # Even though the Rust code for `start` runs synchronously, the Rust
        # code for `stopAndAccumulate` runs asynchronously, and we need to use
        # the same clock for start and stop. Therefore we take the time on the
        # Python side, both here and in `stopAndAccumulate`.
        start_time = _util.time_ns()

        # No dispatcher, we need the return value
        return _ffi.lib.glean_timing_distribution_set_start(self._handle, start_time)

    def stop_and_accumulate(self, timer_id: Optional[int]) -> None:
        """
        Stop tracking time for the provided metric and associated timer id. Add a
        count to the corresponding bucket in the timing distribution.
        This will record an error if no `start` was called.

        Args:
            timer_id: The timer id associated with this timing. This allows for
                concurrent timing of events associated with different ids to
                the same timespan metric.
        """
        # `start` may have returned None.
        # Accepting that means users of this API don't need to do a None check.
        if self._disabled or timer_id is None:
            return

        # The Rust code runs async and might be delayed. We need the time as
        # precisely as possible. We also need the same clock for start and stop
        # (`start` takes the time on the Python side).
        stop_time = _util.time_ns()

        @Dispatcher.launch
        def stop_and_accumulate():
            # CFFI prevents us from passing a negative value to
            # `glean_timing_distribution_set_stop_and_accumulate`. However, to be
            # consistent with the other platforms, we should still pass an invalid
            # value to the Rust side so it can record an error using the Glean
            # error reporting system.
            if timer_id < 0:
                # 64-bit maxint
                corrected_timer_id = 0xFFFFFFFFFFFFFFFF
            else:
                corrected_timer_id = timer_id

            _ffi.lib.glean_timing_distribution_set_stop_and_accumulate(
                self._handle, corrected_timer_id, stop_time
            )

    def cancel(self, timer_id: Optional[int]) -> None:
        """
        Abort a previous `start` call. No error is recorded if no `start` was called.

        Args:
            timer_id: The timer id associated with this timing. This allows for
                concurrent timing of events associated with different ids to
                the same timing distribution metric.
        """
        # `start` may have returned None.
        # Accepting that means users of this API don't need to do a None check.
        if self._disabled or timer_id is None:
            return

        @Dispatcher.launch
        def cancel():
            # CFFI prevents us from passing a negative value to
            # `glean_timing_distribution_cancel`. However, to be consistent
            # with the other platforms, we should still pass an invalid value
            # to the Rust side so it can record an error using the Glean error
            # reporting system.
            if timer_id < 0:
                # 64-bit maxint
                corrected_timer_id = 0xFFFFFFFFFFFFFFFF
            else:
                corrected_timer_id = timer_id

            _ffi.lib.glean_timing_distribution_cancel(self._handle, corrected_timer_id)

    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_timing_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_timing_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_timing_distribution_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def cancel(self, timer_id)

Abort a previous start call. No error is recorded if no start was called.

Args

timer_id
The timer id associated with this timing. This allows for concurrent timing of events associated with different ids to the same timing distribution metric.
Expand source code
def cancel(self, timer_id: Optional[int]) -> None:
    """
    Abort a previous `start` call. No error is recorded if no `start` was called.

    Args:
        timer_id: The timer id associated with this timing. This allows for
            concurrent timing of events associated with different ids to
            the same timing distribution metric.
    """
    # `start` may have returned None.
    # Accepting that means users of this API don't need to do a None check.
    if self._disabled or timer_id is None:
        return

    @Dispatcher.launch
    def cancel():
        # CFFI prevents us from passing a negative value to
        # `glean_timing_distribution_cancel`. However, to be consistent
        # with the other platforms, we should still pass an invalid value
        # to the Rust side so it can record an error using the Glean error
        # reporting system.
        if timer_id < 0:
            # 64-bit maxint
            corrected_timer_id = 0xFFFFFFFFFFFFFFFF
        else:
            corrected_timer_id = timer_id

        _ffi.lib.glean_timing_distribution_cancel(self._handle, corrected_timer_id)
def start(self)

Start tracking time. This records an error if it’s already tracking time (i.e. start was already called with no corresponding stop_and_ccumulate): in that case the original start time will be preserved.

Returns

timer_id
The object to associate with this timing.
Expand source code
def start(self) -> Optional[int]:
    """
    Start tracking time. This records an error if it’s already tracking
    time (i.e. start was already called with no corresponding
    `stop_and_ccumulate`): in that case the original start time will be
    preserved.

    Returns:
        timer_id: The object to associate with this timing.
    """
    if self._disabled:
        return None

    # Even though the Rust code for `start` runs synchronously, the Rust
    # code for `stopAndAccumulate` runs asynchronously, and we need to use
    # the same clock for start and stop. Therefore we take the time on the
    # Python side, both here and in `stopAndAccumulate`.
    start_time = _util.time_ns()

    # No dispatcher, we need the return value
    return _ffi.lib.glean_timing_distribution_set_start(self._handle, start_time)
def stop_and_accumulate(self, timer_id)

Stop tracking time for the provided metric and associated timer id. Add a count to the corresponding bucket in the timing distribution. This will record an error if no start was called.

Args

timer_id
The timer id associated with this timing. This allows for concurrent timing of events associated with different ids to the same timespan metric.
Expand source code
def stop_and_accumulate(self, timer_id: Optional[int]) -> None:
    """
    Stop tracking time for the provided metric and associated timer id. Add a
    count to the corresponding bucket in the timing distribution.
    This will record an error if no `start` was called.

    Args:
        timer_id: The timer id associated with this timing. This allows for
            concurrent timing of events associated with different ids to
            the same timespan metric.
    """
    # `start` may have returned None.
    # Accepting that means users of this API don't need to do a None check.
    if self._disabled or timer_id is None:
        return

    # The Rust code runs async and might be delayed. We need the time as
    # precisely as possible. We also need the same clock for start and stop
    # (`start` takes the time on the Python side).
    stop_time = _util.time_ns()

    @Dispatcher.launch
    def stop_and_accumulate():
        # CFFI prevents us from passing a negative value to
        # `glean_timing_distribution_set_stop_and_accumulate`. However, to be
        # consistent with the other platforms, we should still pass an invalid
        # value to the Rust side so it can record an error using the Glean
        # error reporting system.
        if timer_id < 0:
            # 64-bit maxint
            corrected_timer_id = 0xFFFFFFFFFFFFFFFF
        else:
            corrected_timer_id = timer_id

        _ffi.lib.glean_timing_distribution_set_stop_and_accumulate(
            self._handle, corrected_timer_id, stop_time
        )
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_timing_distribution_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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_timing_distribution_test_get_value_as_json_string(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )
    )
def test_has_value(self, ping_name=None)

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_timing_distribution_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )
class UuidMetricType (disabled, category, lifetime, name, send_in_pings)

This implements the developer facing API for recording UUID 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 UUID API exposes the UuidMetricType.generate_and_set() and UuidMetricType.set() methods.

Expand source code
class UuidMetricType:
    """
    This implements the developer facing API for recording UUID 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 UUID API exposes the `UuidMetricType.generate_and_set` and
    `UuidMetricType.set` methods.
    """

    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_uuid_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_uuid_metric(self._handle)

    def generate_and_set(self) -> Optional[uuid.UUID]:
        """
        Generate a new UUID value and set it in the metric store.
        """
        if self._disabled:
            return None

        id = uuid.uuid4()
        self.set(id)
        return id

    def set(self, value: uuid.UUID) -> None:
        """
        Explicitly set an existing UUID value.

        Args:
            value (uuid.UUID): A valid UUID to set the metric to.
        """
        if self._disabled:
            return

        @Dispatcher.launch
        def set():
            _ffi.lib.glean_uuid_set(self._handle, _ffi.ffi_encode_string(str(value)))

    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_uuid_test_has_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )

    def test_get_value(self, ping_name: Optional[str] = None) -> uuid.UUID:
        """
        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 uuid.UUID(
            "urn:uuid:"
            + _ffi.ffi_decode_string(
                _ffi.lib.glean_uuid_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_uuid_test_get_num_recorded_errors(
            self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
        )

Methods

def generate_and_set(self)

Generate a new UUID value and set it in the metric store.

Expand source code
def generate_and_set(self) -> Optional[uuid.UUID]:
    """
    Generate a new UUID value and set it in the metric store.
    """
    if self._disabled:
        return None

    id = uuid.uuid4()
    self.set(id)
    return id
def set(self, value)

Explicitly set an existing UUID value.

Args

value : uuid.UUID
A valid UUID to set the metric to.
Expand source code
def set(self, value: uuid.UUID) -> None:
    """
    Explicitly set an existing UUID value.

    Args:
        value (uuid.UUID): A valid UUID to set the metric to.
    """
    if self._disabled:
        return

    @Dispatcher.launch
    def set():
        _ffi.lib.glean_uuid_set(self._handle, _ffi.ffi_encode_string(str(value)))
def test_get_num_recorded_errors(self, error_type, ping_name=None)

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_uuid_test_get_num_recorded_errors(
        self._handle, error_type.value, _ffi.ffi_encode_string(ping_name),
    )
def test_get_value(self, ping_name=None)

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) -> uuid.UUID:
    """
    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 uuid.UUID(
        "urn:uuid:"
        + _ffi.ffi_decode_string(
            _ffi.lib.glean_uuid_test_get_value(
                self._handle, _ffi.ffi_encode_string(ping_name)
            )
        )
    )
def test_has_value(self, ping_name=None)

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_uuid_test_has_value(
            self._handle, _ffi.ffi_encode_string(ping_name)
        )
    )