Skip to content

Otel

provide.foundation.metrics.otel

TODO: Add module docstring.

Classes

Functions

setup_opentelemetry_metrics 

setup_opentelemetry_metrics(
    config: TelemetryConfig,
) -> None

Setup OpenTelemetry metrics with configuration.

Parameters:

Name Type Description Default
config TelemetryConfig

Telemetry configuration

 required
Source code in provide/foundation/metrics/otel.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def setup_opentelemetry_metrics(config: TelemetryConfig) -> None:
    """Setup OpenTelemetry metrics with configuration.

    Args:
        config: Telemetry configuration

    """
    # Check if metrics are disabled first, before checking dependencies
    if not config.metrics_enabled or config.globally_disabled:
        slog.debug("📊 OpenTelemetry metrics disabled")
        return

    # Check if OpenTelemetry metrics are available
    if not _HAS_OTEL_METRICS:
        slog.debug("📊 OpenTelemetry metrics not available (dependencies not installed)")
        return

    slog.debug("📊🚀 Setting up OpenTelemetry metrics")

    # Create resource with service information
    resource_attrs = {}
    if config.service_name:
        resource_attrs["service.name"] = config.service_name
    if config.service_version:
        resource_attrs["service.version"] = config.service_version

    resource = Resource.create(resource_attrs)

    # Setup metric readers with OTLP exporters if configured
    readers = []

    if config.otlp_endpoint:
        endpoint = config.otlp_endpoint
        headers = config.get_otlp_headers_dict()

        slog.debug(f"📊📤 Configuring OTLP metrics exporter: {endpoint}")

        # Choose exporter based on protocol
        if config.otlp_protocol == "grpc":
            exporter: OTLPGrpcMetricExporter | OTLPHttpMetricExporter = OTLPGrpcMetricExporter(
                endpoint=endpoint,
                headers=headers,
            )
        else:  # http/protobuf
            exporter = OTLPHttpMetricExporter(
                endpoint=endpoint,
                headers=headers,
            )

        # Create periodic reader (exports every 60 seconds by default)
        reader = PeriodicExportingMetricReader(exporter, export_interval_millis=60000)
        readers.append(reader)

    # Create meter provider
    meter_provider = MeterProvider(resource=resource, metric_readers=readers)

    # Set the global meter provider (only if not already set)
    try:
        current_provider = otel_metrics.get_meter_provider()
        provider_type = type(current_provider).__name__

        # Always allow setup if:
        # 1. It's a default/no-op provider
        # 2. It's a mock (for testing)
        # 3. It's our own MeterProvider type (allow re-configuration)
        should_setup = (
            provider_type in ["NoOpMeterProvider", "ProxyMeterProvider", "Mock", "MagicMock"]
            or not hasattr(current_provider, "get_meter")
            or current_provider.__class__.__module__.startswith("unittest.mock")
        )

        if should_setup:
            otel_metrics.set_meter_provider(meter_provider)

            # Set the global meter for our metrics module
            from provide.foundation.metrics import _set_meter

            meter = otel_metrics.get_meter(__name__)
            _set_meter(meter)

        else:
            slog.debug("📊 OpenTelemetry meter provider already configured")
    except Exception:
        # Broad catch intentional: get_meter_provider() may fail in various OTEL environments
        # Proceed with setup if provider check fails
        otel_metrics.set_meter_provider(meter_provider)

        # Set the global meter for our metrics module
        from provide.foundation.metrics import _set_meter

        meter = otel_metrics.get_meter(__name__)
        _set_meter(meter)

shutdown_opentelemetry_metrics 

shutdown_opentelemetry_metrics() -> None

Shutdown OpenTelemetry metrics.

Source code in provide/foundation/metrics/otel.py 
150
151
152
153
154
155
156
157
158
159
160
161
def shutdown_opentelemetry_metrics() -> None:
    """Shutdown OpenTelemetry metrics."""
    if not _HAS_OTEL_METRICS:
        return

    try:
        meter_provider = otel_metrics.get_meter_provider()
        if hasattr(meter_provider, "shutdown"):
            meter_provider.shutdown()
            slog.debug("📊🛑 OpenTelemetry meter provider shutdown")
    except Exception as e:
        slog.warning(f"⚠️ Error shutting down OpenTelemetry metrics: {e}")