flaskpaste/app/metrics.py

"""Custom Prometheus metrics for FlaskPaste."""

from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from flask import Flask

# Metrics are only initialized when not testing
_paste_created_counter = None
_paste_accessed_counter = None
_paste_deleted_counter = None
_rate_limit_counter = None
_pow_counter = None
_dedup_counter = None
_url_created_counter = None
_url_accessed_counter = None
_url_deleted_counter = None
_request_duration_histogram = None


def setup_custom_metrics(app: Flask) -> None:
    """Initialize custom Prometheus metrics.

    Should be called after prometheus_flask_exporter is set up.
    """
    global _paste_created_counter, _paste_accessed_counter, _paste_deleted_counter
    global _rate_limit_counter, _pow_counter, _dedup_counter
    global _url_created_counter, _url_accessed_counter, _url_deleted_counter
    global _request_duration_histogram

    if app.config.get("TESTING"):
        return

    try:
        from prometheus_client import Counter, Histogram
    except ImportError:
        app.logger.warning("prometheus_client not available, custom metrics disabled")
        return

    _paste_created_counter = Counter(
        "flaskpaste_paste_created_total",
        "Total number of paste creation attempts",
        ["auth_type", "outcome"],
    )

    _paste_accessed_counter = Counter(
        "flaskpaste_paste_accessed_total",
        "Total number of paste accesses",
        ["auth_type", "burn"],
    )

    _paste_deleted_counter = Counter(
        "flaskpaste_paste_deleted_total",
        "Total number of paste deletion attempts",
        ["auth_type", "outcome"],
    )

    _rate_limit_counter = Counter(
        "flaskpaste_rate_limit_total",
        "Total number of rate limit events",
        ["outcome"],
    )

    _pow_counter = Counter(
        "flaskpaste_pow_total",
        "Total number of proof-of-work validations",
        ["outcome"],
    )

    _dedup_counter = Counter(
        "flaskpaste_dedup_total",
        "Total number of content deduplication events",
        ["outcome"],
    )

    _url_created_counter = Counter(
        "flaskpaste_url_created_total",
        "Total number of short URL creation attempts",
        ["auth_type", "outcome"],
    )

    _url_accessed_counter = Counter(
        "flaskpaste_url_accessed_total",
        "Total number of short URL accesses",
        ["auth_type"],
    )

    _url_deleted_counter = Counter(
        "flaskpaste_url_deleted_total",
        "Total number of short URL deletion attempts",
        ["auth_type", "outcome"],
    )

    _request_duration_histogram = Histogram(
        "flaskpaste_request_duration_seconds",
        "Request duration in seconds",
        ["method", "endpoint", "status"],
        buckets=(0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
    )

    app.logger.info("Custom Prometheus metrics initialized")


def record_paste_created(auth_type: str, outcome: str) -> None:
    """Record a paste creation attempt.

    Args:
        auth_type: "authenticated" or "anonymous"
        outcome: "success", "failure", or "blocked"
    """
    if _paste_created_counter:
        _paste_created_counter.labels(auth_type=auth_type, outcome=outcome).inc()


def record_paste_accessed(auth_type: str, burn: bool) -> None:
    """Record a paste access.

    Args:
        auth_type: "authenticated" or "anonymous"
        burn: True if paste was burn-after-read
    """
    if _paste_accessed_counter:
        _paste_accessed_counter.labels(auth_type=auth_type, burn=str(burn).lower()).inc()


def record_paste_deleted(auth_type: str, outcome: str) -> None:
    """Record a paste deletion attempt.

    Args:
        auth_type: "authenticated" or "anonymous"
        outcome: "success" or "failure"
    """
    if _paste_deleted_counter:
        _paste_deleted_counter.labels(auth_type=auth_type, outcome=outcome).inc()


def record_rate_limit(outcome: str) -> None:
    """Record a rate limit event.

    Args:
        outcome: "allowed" or "blocked"
    """
    if _rate_limit_counter:
        _rate_limit_counter.labels(outcome=outcome).inc()


def record_pow(outcome: str) -> None:
    """Record a proof-of-work validation.

    Args:
        outcome: "success" or "failure"
    """
    if _pow_counter:
        _pow_counter.labels(outcome=outcome).inc()


def record_dedup(outcome: str) -> None:
    """Record a content deduplication event.

    Args:
        outcome: "allowed" or "blocked"
    """
    if _dedup_counter:
        _dedup_counter.labels(outcome=outcome).inc()


def record_url_created(auth_type: str, outcome: str) -> None:
    """Record a short URL creation attempt."""
    if _url_created_counter:
        _url_created_counter.labels(auth_type=auth_type, outcome=outcome).inc()


def record_url_accessed(auth_type: str) -> None:
    """Record a short URL access."""
    if _url_accessed_counter:
        _url_accessed_counter.labels(auth_type=auth_type).inc()


def record_url_deleted(auth_type: str, outcome: str) -> None:
    """Record a short URL deletion attempt."""
    if _url_deleted_counter:
        _url_deleted_counter.labels(auth_type=auth_type, outcome=outcome).inc()


def observe_request_duration(method: str, endpoint: str, status: int, duration: float) -> None:
    """Record request duration.

    Args:
        method: HTTP method (GET, POST, etc.)
        endpoint: Request endpoint path
        status: HTTP status code
        duration: Request duration in seconds
    """
    if _request_duration_histogram:
        _request_duration_histogram.labels(
            method=method, endpoint=endpoint, status=str(status)
        ).observe(duration)