Source code for stamina.instrumentation._prometheus

# SPDX-FileCopyrightText: 2022 Hynek Schlawack <hs@ox.cx>
#
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import TYPE_CHECKING

from ._data import RetryDetails, RetryHook, RetryHookFactory, guess_name


if TYPE_CHECKING:
    from prometheus_client import Counter


RETRIES_TOTAL = None


def init_prometheus() -> RetryHook:
    """
    Initialize Prometheus instrumentation.
    """
    from prometheus_client import Counter

    global RETRIES_TOTAL  # noqa: PLW0603

    # Mostly for testing so we can call init_prometheus more than once.
    if RETRIES_TOTAL is None:
        RETRIES_TOTAL = Counter(
            "stamina_retries_total",
            "Total number of retries.",
            ("callable", "retry_num", "error_type"),
        )

    def count_retries(details: RetryDetails) -> None:
        """
        Count and log retries for callable *name*.
        """
        RETRIES_TOTAL.labels(
            callable=details.name,
            retry_num=details.retry_num,
            error_type=guess_name(details.caused_by.__class__),
        ).inc()

    return count_retries



[docs]
def get_prometheus_counter() -> Counter | None:
    """
    Return the Prometheus counter for the number of retries.

    Returns:
        If active, the Prometheus `counter
        <https://github.com/prometheus/client_python>`_ for the number of
        retries. None otherwise.

    .. versionadded:: 23.2.0
    """
    from . import get_on_retry_hooks

    # Finalize the hooks if not done yet.
    get_on_retry_hooks()

    return RETRIES_TOTAL



PrometheusOnRetryHook = RetryHookFactory(init_prometheus)