pygx.instrument.monitoring

Pluggable metric systems for monitoring.

monitoring

Pluggable metric systems for monitoring.

This module allows PyGX to plug in metric systems to monitor the execution of programs. There are three common kinds of metrics:

• Counters track the number of times an event occurs. Their values increase monotonically over time.

• Scalars track a single value at a given time, for example, available memory size. They do not accumulate over time like counters.

• Distributions track the distribution of a numerical value. For example, the latency of an operation.

Metric

Metric(
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any
)

Bases: Generic[MetricValueType]

Base class for metrics.

Initializes the metric.

Parameters:

Name	Type	Description	Default
namespace	str	The namespace of the metric.	required
name	str	The name of the metric.	required
description	str	The description of the metric.	required
parameter_definitions	dict[str, type[int | str | bool]]	The definitions of the parameters for the metric.	required
**additional_flags	Any	Additional flags for the metric.	{}

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any,
) -> None:
    """Initializes the metric.

    Args:
      namespace: The namespace of the metric.
      name: The name of the metric.
      description: The description of the metric.
      parameter_definitions: The definitions of the parameters for the
        metric.
      **additional_flags: Additional flags for the metric.
    """
    self._namespace = namespace
    self._name = name
    self._description = description
    self._parameter_definitions = parameter_definitions
    self._flags = additional_flags

namespace property

namespace: str

Returns the namespace of the metric.

name property

name: str

Returns the name of the metric.

full_name property

full_name: str

Returns the full name of the metric.

description property

description: str

Returns the description of the metric.

parameter_definitions property

parameter_definitions: dict[str, type[int | str | bool]]

Returns the parameter definitions of the metric.

flags property

flags: dict[str, Any]

Returns the flags of the metric.

value abstractmethod

value(**parameters: Any) -> MetricValueType

Returns the value of the metric for the given parameters.

Parameters:

Name	Type	Description	Default
**parameters	Any	Parameters for parameterized counters.	{}

Returns:

Type	Description
MetricValueType	The value of the metric.

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def value(self, **parameters: Any) -> MetricValueType:
    """Returns the value of the metric for the given parameters.

    Args:
      **parameters: Parameters for parameterized counters.

    Returns:
      The value of the metric.
    """

Counter

Counter(
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any
)

Bases: Metric[int]

Base class for counters.

Counters are metrics that track the number of times an event occurs. Their values increase monotonically over time.

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any,
) -> None:
    """Initializes the metric.

    Args:
      namespace: The namespace of the metric.
      name: The name of the metric.
      description: The description of the metric.
      parameter_definitions: The definitions of the parameters for the
        metric.
      **additional_flags: Additional flags for the metric.
    """
    self._namespace = namespace
    self._name = name
    self._description = description
    self._parameter_definitions = parameter_definitions
    self._flags = additional_flags

increment abstractmethod

increment(delta: int = 1, **parameters: Any) -> int

Increments the counter by delta and returns the new value.

Parameters:

Name	Type	Description	Default
delta	int	The amount to increment the counter by.	1
**parameters	Any	Parameters for parameterized counters.	{}

Returns:

Type	Description
int	The new value of the counter.

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def increment(self, delta: int = 1, **parameters: Any) -> int:
    """Increments the counter by delta and returns the new value.

    Args:
      delta: The amount to increment the counter by.
      **parameters: Parameters for parameterized counters.

    Returns:
      The new value of the counter.
    """

Scalar

Scalar(
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any
)

Bases: Metric[MetricValueType]

Base class for scalar values.

Scalar values are metrics that track a single value at a given time, for example, available memory size. They do not accumulate over time like counters.

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any,
) -> None:
    """Initializes the metric.

    Args:
      namespace: The namespace of the metric.
      name: The name of the metric.
      description: The description of the metric.
      parameter_definitions: The definitions of the parameters for the
        metric.
      **additional_flags: Additional flags for the metric.
    """
    self._namespace = namespace
    self._name = name
    self._description = description
    self._parameter_definitions = parameter_definitions
    self._flags = additional_flags

set abstractmethod

set(value: MetricValueType, **parameters: Any) -> None

Sets the value of the scalar.

Parameters:

Name	Type	Description	Default
value	MetricValueType	The value to record.	required
**parameters	Any	Parameters for parameterized scalars.	{}

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def set(self, value: MetricValueType, **parameters: Any) -> None:
    """Sets the value of the scalar.

    Args:
      value: The value to record.
      **parameters: Parameters for parameterized scalars.
    """

increment abstractmethod

increment(delta: MetricValueType = 1, **parameters: Any) -> MetricValueType

Increments the scalar by delta and returns the new value.

Parameters:

Name	Type	Description	Default
delta	MetricValueType	The amount to increment the scalar by.	1
**parameters	Any	Parameters for parameterized scalars.	{}

Returns:

Type	Description
MetricValueType	The new value of the scalar.

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def increment(
    self, delta: MetricValueType = 1, **parameters: Any
) -> MetricValueType:
    """Increments the scalar by delta and returns the new value.

    Args:
      delta: The amount to increment the scalar by.
      **parameters: Parameters for parameterized scalars.

    Returns:
      The new value of the scalar.
    """

DistributionValue

Base for distribution value.

count abstractmethod property

count: int

Returns the number of samples in the distribution.

sum abstractmethod property

sum: float

Returns the sum of the distribution.

mean abstractmethod property

mean: float

Returns the mean of the distribution.

stddev abstractmethod property

stddev: float

Returns the standard deviation of the distribution.

median property

median: float

Returns the median of the distribution.

variance abstractmethod property

variance: float

Returns the variance of the distribution.

percentile abstractmethod

percentile(n: float) -> float

Returns the n-th percentile of the distribution.

Parameters:

Name	Type	Description	Default
n	float	The percentile to return. Should be in the range [0, 100].	required

Returns:

Type	Description
float	The n-th percentile of the distribution.

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def percentile(self, n: float) -> float:
    """Returns the n-th percentile of the distribution.

    Args:
      n: The percentile to return. Should be in the range [0, 100].

    Returns:
      The n-th percentile of the distribution.
    """

fraction_less_than abstractmethod

fraction_less_than(value: float) -> float

Returns the fraction of distribution values less than value.

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def fraction_less_than(self, value: float) -> float:
    """Returns the fraction of distribution values less than ``value``."""

Distribution

Distribution(
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any
)

Bases: Metric[DistributionValue]

Base class for distributional metrics.

Distributions are metrics that track the distribution of a numerical value. For example, the latency of an operation.

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    name: str,
    description: str,
    parameter_definitions: dict[str, type[int | str | bool]],
    **additional_flags: Any,
) -> None:
    """Initializes the metric.

    Args:
      namespace: The namespace of the metric.
      name: The name of the metric.
      description: The description of the metric.
      parameter_definitions: The definitions of the parameters for the
        metric.
      **additional_flags: Additional flags for the metric.
    """
    self._namespace = namespace
    self._name = name
    self._description = description
    self._parameter_definitions = parameter_definitions
    self._flags = additional_flags

record abstractmethod

record(value: float, **parameters: Any) -> None

Records a value to the distribution.

Parameters:

Name	Type	Description	Default
value	float	The value to record.	required
**parameters	Any	Parameters for parameterized distributions.	{}

Source code in pygx/instrument/_monitoring.py

@abc.abstractmethod
def record(self, value: float, **parameters: Any) -> None:
    """Records a value to the distribution.

    Args:
      value: The value to record.
      **parameters: Parameters for parameterized distributions.
    """

record_duration

record_duration(
    *, scale: int = 1000, error_parameter: str = "error", **parameters: Any
) -> Iterator[None]

Context manager that records the duration of a code block.

Parameters:

Name	Type	Description	Default
scale	int	Multiplier applied to the elapsed seconds before recording (e.g. the default 1000 records milliseconds).	1000
error_parameter	str	The parameter name for recording the error. If the name is not defined as a parameter for the distribution, the error tag will not be recorded.	'error'
**parameters	Any	Parameters for parameterized distributions.	{}

Source code in pygx/instrument/_monitoring.py

@contextlib.contextmanager
def record_duration(
    self,
    *,
    scale: int = 1000,
    error_parameter: str = 'error',
    **parameters: Any,
) -> Iterator[None]:
    """Context manager that records the duration of a code block.

    Args:
      scale: Multiplier applied to the elapsed seconds before recording
        (e.g. the default 1000 records milliseconds).
      error_parameter: The parameter name for recording the error. If
        the name is not defined as a parameter for the distribution, the
        error tag will not be recorded.
      **parameters: Parameters for parameterized distributions.
    """
    start_time = time.time()
    error = None
    try:
        yield
    except BaseException as e:
        error = e
        raise e
    finally:
        duration = (time.time() - start_time) * scale
        if error_parameter in self._parameter_definitions:
            parameters[error_parameter] = (
                error_info.ErrorInfo.from_exception(error).tag
                if error is not None
                else ''
            )
        self.record(duration, **parameters)

MetricCollection

MetricCollection(
    namespace: str,
    default_parameters: None | dict[str, type[int | str | bool]] = None,
)

Base class for metric collections.

A metric collection creates and caches metrics (counters, scalars and distributions) under a shared namespace; get_* methods return the existing metric when one with the same name was already created.

Initializes the metric collection.

Parameters:

Name	Type	Description	Default
namespace	str	The namespace of the metric collection.	required
default_parameters	None | dict[str, type[int | str | bool]]	The default parameters used to create metrics if not specified.	None

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    default_parameters: None | (dict[str, type[int | str | bool]]) = None,
):
    """Initializes the metric collection.

    Args:
      namespace: The namespace of the metric collection.
      default_parameters: The default parameters used to create metrics
        if not specified.
    """
    self._namespace = namespace
    self._default_parameter_definitions = default_parameters or {}
    self._metrics = self._metric_container()

namespace property

namespace: str

Returns the namespace of the metric collection.

metrics

metrics() -> list[Metric]

Returns the metrics created under this collection's namespace.

Source code in pygx/instrument/_monitoring.py

def metrics(self) -> list[Metric]:
    """Returns the metrics created under this collection's namespace."""
    return [
        m for m in self._metrics.values() if m.namespace == self._namespace
    ]

get_counter

get_counter(
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    **additional_flags: Any
) -> Counter

Gets or creates a counter with the given name.

Parameters:

Name	Type	Description	Default
name	str	The name of the counter.	required
description	str	The description of the counter.	required
parameters	dict[str, type[int | str | bool]] | None	The definitions of the parameters for the counter. default_parameters from the collection will be used if not specified.	None
**additional_flags	Any	Additional arguments for creating the counter. Subclasses can use these arguments to provide additional information for creating the counter.	{}

Returns:

Type	Description
Counter	The counter with the given name.

Source code in pygx/instrument/_monitoring.py

def get_counter(
    self,
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    **additional_flags: Any,
) -> Counter:
    """Gets or creates a counter with the given name.

    Args:
      name: The name of the counter.
      description: The description of the counter.
      parameters: The definitions of the parameters for the counter.
        ``default_parameters`` from the collection will be used if not
        specified.
      **additional_flags: Additional arguments for creating the counter.
        Subclasses can use these arguments to provide additional
        information for creating the counter.

    Returns:
      The counter with the given name.
    """
    if parameters is None:
        parameters = self._default_parameter_definitions
    return typing.cast(
        Counter,
        self._get_or_create_metric(
            self._COUNTER_CLASS,
            name,
            description,
            parameters,
            **additional_flags,
        ),
    )

get_scalar

get_scalar(
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    value_type: type[int | float] = int,
    **additional_flags: Any
) -> Scalar

Gets or creates a scalar with the given name.

Parameters:

Name	Type	Description	Default
name	str	The name of the scalar.	required
description	str	The description of the scalar.	required
parameters	dict[str, type[int | str | bool]] | None	The definitions of the parameters for the scalar. default_parameters from the collection will be used if not specified.	None
value_type	type[int | float]	The type of the value for the scalar.	int
**additional_flags	Any	Additional arguments for creating the scalar.	{}

Returns:

Type	Description
Scalar	The scalar with the given name.

Source code in pygx/instrument/_monitoring.py

def get_scalar(
    self,
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    value_type: type[int | float] = int,
    **additional_flags: Any,
) -> Scalar:
    """Gets or creates a scalar with the given name.

    Args:
      name: The name of the scalar.
      description: The description of the scalar.
      parameters: The definitions of the parameters for the scalar.
        ``default_parameters`` from the collection will be used if not
        specified.
      value_type: The type of the value for the scalar.
      **additional_flags: Additional arguments for creating the scalar.

    Returns:
      The scalar with the given name.
    """
    if parameters is None:
        parameters = self._default_parameter_definitions
    return typing.cast(
        Scalar,
        self._get_or_create_metric(
            self._SCALAR_CLASS,
            name,
            description,
            parameters,
            value_type=value_type,
            **additional_flags,
        ),
    )

get_distribution

get_distribution(
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    **additional_flags: Any
) -> Distribution

Gets or creates a distribution with the given name.

Parameters:

Name	Type	Description	Default
name	str	The name of the distribution.	required
description	str	The description of the distribution.	required
parameters	dict[str, type[int | str | bool]] | None	The definitions of the parameters for the distribution. default_parameters from the collection will be used if not specified.	None
**additional_flags	Any	Additional arguments for creating the distribution.	{}

Returns:

Type	Description
Distribution	The distribution with the given name.

Source code in pygx/instrument/_monitoring.py

def get_distribution(
    self,
    name: str,
    description: str,
    parameters: dict[str, type[int | str | bool]] | None = None,
    **additional_flags: Any,
) -> Distribution:
    """Gets or creates a distribution with the given name.

    Args:
      name: The name of the distribution.
      description: The description of the distribution.
      parameters: The definitions of the parameters for the distribution.
        ``default_parameters`` from the collection will be used if not
        specified.
      **additional_flags: Additional arguments for creating the
        distribution.

    Returns:
      The distribution with the given name.
    """
    if parameters is None:
        parameters = self._default_parameter_definitions
    return typing.cast(
        Distribution,
        self._get_or_create_metric(
            self._DISTRIBUTION_CLASS,
            name,
            description,
            parameters,
            **additional_flags,
        ),
    )

InMemoryMetricCollection

InMemoryMetricCollection(
    namespace: str,
    default_parameters: None | dict[str, type[int | str | bool]] = None,
)

Bases: MetricCollection

In-memory metric collection.

Source code in pygx/instrument/_monitoring.py

def __init__(
    self,
    namespace: str,
    default_parameters: None | (dict[str, type[int | str | bool]]) = None,
):
    """Initializes the metric collection.

    Args:
      namespace: The namespace of the metric collection.
      default_parameters: The default parameters used to create metrics
        if not specified.
    """
    self._namespace = namespace
    self._default_parameter_definitions = default_parameters or {}
    self._metrics = self._metric_container()

metric_collection

metric_collection(namespace: str, **kwargs) -> MetricCollection

Creates a metric collection using the default collection class.

Source code in pygx/instrument/_monitoring.py

def metric_collection(namespace: str, **kwargs) -> MetricCollection:
    """Creates a metric collection using the default collection class."""
    return _METRIC_COLLECTION_CLS(namespace, **kwargs)

set_default_metric_collection_cls

set_default_metric_collection_cls(cls: type[MetricCollection]) -> None

Sets the default metric collection class.

Source code in pygx/instrument/_monitoring.py

def set_default_metric_collection_cls(cls: type[MetricCollection]) -> None:
    """Sets the default metric collection class."""
    global _METRIC_COLLECTION_CLS  # pylint: disable=global-statement
    _METRIC_COLLECTION_CLS = cls

default_metric_collection_cls

default_metric_collection_cls() -> type[MetricCollection]

Returns the default metric collection class.

Source code in pygx/instrument/_monitoring.py

def default_metric_collection_cls() -> type[MetricCollection]:
    """Returns the default metric collection class."""
    return _METRIC_COLLECTION_CLS

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2