Index

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.logger.otlp

Generic OTLP (OpenTelemetry Protocol) support for Foundation logger.

This package provides generic OpenTelemetry Protocol support that can be used with any OTLP-compatible backend (OpenObserve, Datadog, Honeycomb, etc.).

Key Components: - OTLPLogClient: Generic client for sending logs via OTLP - OTLPCircuitBreaker: Reliability pattern for handling endpoint failures - Helper functions for resource creation, trace context, severity mapping

Example

from provide.foundation.logger.otlp import OTLPLogClient

client = OTLPLogClient( ... endpoint="https://api.example.com/v1/logs", ... headers={"Authorization": "Bearer token"}, ... service_name="my-service", ... )

client.send_log("Hello from OTLP", level="INFO")

Classes

OTLPCircuitBreaker

OTLPCircuitBreaker(
    failure_threshold: int = 5,
    timeout: float = 60.0,
    half_open_timeout: float = 10.0,
)

Circuit breaker for OTLP connections with exponential backoff.

States

• closed: Normal operation, requests allowed
• open: Too many failures, requests blocked
• half_open: Testing if service recovered

Examples:

>>> breaker = OTLPCircuitBreaker(failure_threshold=3, timeout=60.0)
>>> if breaker.can_attempt():
...     success = send_otlp_log()
...     if success:
...         breaker.record_success()
...     else:
...         breaker.record_failure()

Initialize circuit breaker.

Parameters:

Name	Type	Description	Default
failure_threshold	int	Number of failures before opening circuit	5
timeout	float	Seconds to wait before attempting half-open (doubles each time)	60.0
half_open_timeout	float	Seconds to wait in half-open before trying again	10.0

Source code in provide/foundation/logger/otlp/circuit.py

def __init__(
    self,
    failure_threshold: int = 5,
    timeout: float = 60.0,
    half_open_timeout: float = 10.0,
) -> None:
    """Initialize circuit breaker.

    Args:
        failure_threshold: Number of failures before opening circuit
        timeout: Seconds to wait before attempting half-open (doubles each time)
        half_open_timeout: Seconds to wait in half-open before trying again
    """
    self.failure_threshold = failure_threshold
    self.base_timeout = timeout
    self.half_open_timeout = half_open_timeout

    self._state: CircuitState = "closed"
    self._failure_count = 0
    self._last_failure_time: float | None = None
    self._last_attempt_time: float | None = None
    self._open_count = 0  # Track how many times we've opened
    self._lock = threading.Lock()

Attributes

state property

state: CircuitState

Get current circuit state.

Functions

can_attempt

can_attempt() -> bool

Check if we can attempt an OTLP operation.

Returns:

Type	Description
bool	True if operation should be attempted, False if circuit is open

Source code in provide/foundation/logger/otlp/circuit.py

def can_attempt(self) -> bool:
    """Check if we can attempt an OTLP operation.

    Returns:
        True if operation should be attempted, False if circuit is open
    """
    with self._lock:
        now = time.time()

        if self._state == "closed":
            return True

        if self._state == "open":
            # Check if enough time has passed to try half-open
            if self._last_failure_time is None:
                return False

            # Exponential backoff: timeout doubles each time circuit opens
            current_timeout = self.base_timeout * (2 ** min(self._open_count, 10))
            if now - self._last_failure_time >= current_timeout:
                self._state = "half_open"
                self._last_attempt_time = now
                return True

            return False

        if self._state == "half_open":
            # Only allow one attempt in half-open state within timeout window
            if self._last_attempt_time is None:
                return True

            if now - self._last_attempt_time >= self.half_open_timeout:
                self._last_attempt_time = now
                return True

            return False

        return False

get_stats

get_stats() -> dict[str, Any]

Get circuit breaker statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary with current state and statistics

Source code in provide/foundation/logger/otlp/circuit.py

def get_stats(self) -> dict[str, Any]:
    """Get circuit breaker statistics.

    Returns:
        Dictionary with current state and statistics
    """
    with self._lock:
        return {
            "state": self._state,
            "failure_count": self._failure_count,
            "open_count": self._open_count,
            "last_failure_time": self._last_failure_time,
            "last_attempt_time": self._last_attempt_time,
            "current_timeout": self.base_timeout * (2 ** min(self._open_count, 10)),
        }

record_failure

record_failure(error: Exception | None = None) -> None

Record a failed operation.

Parameters:

Name	Type	Description	Default
error	Exception | None	Optional exception that caused the failure	None

Source code in provide/foundation/logger/otlp/circuit.py

def record_failure(self, error: Exception | None = None) -> None:
    """Record a failed operation.

    Args:
        error: Optional exception that caused the failure
    """
    with self._lock:
        self._failure_count += 1
        self._last_failure_time = time.time()

        if self._state == "half_open":
            # Failed during recovery attempt, go back to open
            self._state = "open"
            self._open_count += 1
        elif self._failure_count >= self.failure_threshold:
            # Too many failures, open the circuit
            self._state = "open"
            self._open_count += 1

record_success

record_success() -> None

Record a successful operation.

Source code in provide/foundation/logger/otlp/circuit.py

def record_success(self) -> None:
    """Record a successful operation."""
    with self._lock:
        self._state = "closed"
        self._failure_count = 0
        self._last_failure_time = None
        self._last_attempt_time = None
        # Don't reset _open_count completely, but decay it
        if self._open_count > 0:
            self._open_count = max(0, self._open_count - 1)

reset

reset() -> None

Manually reset the circuit breaker to closed state.

Source code in provide/foundation/logger/otlp/circuit.py

def reset(self) -> None:
    """Manually reset the circuit breaker to closed state."""
    with self._lock:
        self._state = "closed"
        self._failure_count = 0
        self._last_failure_time = None
        self._last_attempt_time = None
        self._open_count = 0

OTLPLogClient

OTLPLogClient(
    endpoint: str,
    headers: dict[str, str] | None = None,
    service_name: str = "foundation",
    service_version: str | None = None,
    environment: str | None = None,
    timeout: float = 30.0,
    use_circuit_breaker: bool = True,
)

Generic OTLP client for any OpenTelemetry-compatible backend.

This client works with any OTLP-compatible backend and provides: - Single log sending with automatic flushing - Persistent LoggerProvider for continuous logging - Circuit breaker pattern for reliability - Automatic trace context extraction - Attribute normalization for OTLP compatibility

Examples:

>>> client = OTLPLogClient(
...     endpoint="https://api.honeycomb.io/v1/logs",
...     headers={"x-honeycomb-team": "YOUR_API_KEY"},
...     service_name="my-service",
... )
>>> client.send_log("Hello OTLP!", level="INFO")
True

>>> # Use with persistent logger provider
>>> provider = client.create_logger_provider()
>>> # Configure structlog to use provider

Initialize OTLP client.

Parameters:

Name	Type	Description	Default
endpoint	str	OTLP endpoint URL (e.g., "https://api.example.com/v1/logs")	required
headers	dict[str, str] | None	Optional custom headers (auth, organization, etc.)	None
service_name	str	Service name for resource attributes	'foundation'
service_version	str | None	Optional service version	None
environment	str | None	Optional environment (dev, staging, prod)	None
timeout	float	Request timeout in seconds	30.0
use_circuit_breaker	bool	Enable circuit breaker pattern	True

Source code in provide/foundation/logger/otlp/client.py

def __init__(
    self,
    endpoint: str,
    headers: dict[str, str] | None = None,
    service_name: str = "foundation",
    service_version: str | None = None,
    environment: str | None = None,
    timeout: float = 30.0,
    use_circuit_breaker: bool = True,
) -> None:
    """Initialize OTLP client.

    Args:
        endpoint: OTLP endpoint URL (e.g., "https://api.example.com/v1/logs")
        headers: Optional custom headers (auth, organization, etc.)
        service_name: Service name for resource attributes
        service_version: Optional service version
        environment: Optional environment (dev, staging, prod)
        timeout: Request timeout in seconds
        use_circuit_breaker: Enable circuit breaker pattern
    """
    self.endpoint = build_otlp_endpoint(endpoint, signal_type="logs")
    self.headers = headers or {}
    self.service_name = service_name
    self.service_version = service_version
    self.environment = environment
    self.timeout = timeout
    self.use_circuit_breaker = use_circuit_breaker

    # Check if OpenTelemetry SDK is available
    self._otlp_available = self._check_otlp_availability()

Functions

create_logger_provider

create_logger_provider() -> Any | None

Create persistent LoggerProvider for continuous logging.

Returns:

Type	Description
Any | None	LoggerProvider if OpenTelemetry SDK available, None otherwise

Use this for long-running applications that need persistent OTLP logging. The provider can be used with structlog processors for automatic OTLP export.

Circuit breaker: - Returns None if circuit is open - Records success if provider created - Records failure if exception occurs

Examples:

>>> provider = client.create_logger_provider()
>>> if provider:
...     # Configure structlog with provider
...     pass

Source code in provide/foundation/logger/otlp/client.py

def create_logger_provider(self) -> Any | None:
    """Create persistent LoggerProvider for continuous logging.

    Returns:
        LoggerProvider if OpenTelemetry SDK available, None otherwise

    Use this for long-running applications that need persistent OTLP logging.
    The provider can be used with structlog processors for automatic OTLP export.

    Circuit breaker:
    - Returns None if circuit is open
    - Records success if provider created
    - Records failure if exception occurs

    Examples:
        >>> provider = client.create_logger_provider()
        >>> if provider:
        ...     # Configure structlog with provider
        ...     pass
    """
    if not self._otlp_available:
        return None

    # Check circuit breaker
    if self.use_circuit_breaker:
        breaker = get_otlp_circuit_breaker()
        if not breaker.can_attempt():
            return None

    try:
        provider = self._create_logger_provider_internal()
        if provider and self.use_circuit_breaker:
            breaker.record_success()
        return provider
    except Exception:
        if self.use_circuit_breaker:
            breaker.record_failure()
        return None

from_config classmethod

from_config(
    config: Any,
    additional_headers: dict[str, str] | None = None,
) -> OTLPLogClient

Create client from TelemetryConfig.

Parameters:

Name	Type	Description	Default
config	Any	TelemetryConfig instance	required
additional_headers	dict[str, str] | None	Additional headers to merge with config headers	None

Returns:

Type	Description
OTLPLogClient	Configured OTLPLogClient instance

Raises:

Type	Description
ValueError	If config.otlp_endpoint is not set

Examples:

>>> from provide.foundation.logger.config.telemetry import TelemetryConfig
>>> config = TelemetryConfig.from_env()
>>> client = OTLPLogClient.from_config(config)

Source code in provide/foundation/logger/otlp/client.py

@classmethod
def from_config(
    cls,
    config: Any,
    additional_headers: dict[str, str] | None = None,
) -> OTLPLogClient:
    """Create client from TelemetryConfig.

    Args:
        config: TelemetryConfig instance
        additional_headers: Additional headers to merge with config headers

    Returns:
        Configured OTLPLogClient instance

    Raises:
        ValueError: If config.otlp_endpoint is not set

    Examples:
        >>> from provide.foundation.logger.config.telemetry import TelemetryConfig
        >>> config = TelemetryConfig.from_env()
        >>> client = OTLPLogClient.from_config(config)
    """
    if not config.otlp_endpoint:
        msg = "otlp_endpoint must be set in TelemetryConfig"
        raise ValueError(msg)

    headers = dict(config.otlp_headers)
    if additional_headers:
        headers.update(additional_headers)

    return cls(
        endpoint=config.otlp_endpoint,
        headers=headers,
        service_name=config.service_name or "foundation",
        service_version=config.service_version,
        environment=None,  # TODO: Add environment to TelemetryConfig
    )

get_stats

get_stats() -> dict[str, Any]

Get client statistics including circuit breaker state.

Returns:

Type	Description
dict[str, Any]	Dictionary with client and circuit breaker statistics

Examples:

>>> stats = client.get_stats()
>>> print(stats["otlp_available"])
True

Source code in provide/foundation/logger/otlp/client.py

def get_stats(self) -> dict[str, Any]:
    """Get client statistics including circuit breaker state.

    Returns:
        Dictionary with client and circuit breaker statistics

    Examples:
        >>> stats = client.get_stats()
        >>> print(stats["otlp_available"])
        True
    """
    stats: dict[str, Any] = {
        "otlp_available": self._otlp_available,
        "endpoint": self.endpoint,
        "service_name": self.service_name,
    }

    if self.use_circuit_breaker:
        breaker = get_otlp_circuit_breaker()
        stats["circuit_breaker"] = breaker.get_stats()

    return stats

is_available

is_available() -> bool

Check if OTLP is available (SDK installed and circuit not open).

Returns:

Type	Description
bool	True if OTLP is available and circuit is closed

Examples:

>>> if client.is_available():
...     client.send_log("Message")

Source code in provide/foundation/logger/otlp/client.py

def is_available(self) -> bool:
    """Check if OTLP is available (SDK installed and circuit not open).

    Returns:
        True if OTLP is available and circuit is closed

    Examples:
        >>> if client.is_available():
        ...     client.send_log("Message")
    """
    if not self._otlp_available:
        return False

    if self.use_circuit_breaker:
        breaker = get_otlp_circuit_breaker()
        return breaker.can_attempt()

    return True

send_log

send_log(
    message: str,
    level: str = "INFO",
    attributes: dict[str, Any] | None = None,
) -> bool

Send single log via OTLP.

Creates a temporary LoggerProvider, sends the log, and flushes immediately. This ensures delivery for single log sends but is less efficient for bulk logging.

Parameters:

Name	Type	Description	Default
message	str	Log message	required
level	str	Log level (DEBUG, INFO, WARN, ERROR, FATAL)	'INFO'
attributes	dict[str, Any] | None	Optional log attributes	None

Returns:

Type	Description
bool	True if sent successfully, False otherwise

Circuit breaker pattern: - Checks circuit before attempting - Records success/failure - Automatically disables after threshold failures - Auto-recovers with exponential backoff

Examples:

>>> client.send_log("User logged in", level="INFO", attributes={"user_id": 123})
True

Source code in provide/foundation/logger/otlp/client.py

def send_log(
    self,
    message: str,
    level: str = "INFO",
    attributes: dict[str, Any] | None = None,
) -> bool:
    """Send single log via OTLP.

    Creates a temporary LoggerProvider, sends the log, and flushes immediately.
    This ensures delivery for single log sends but is less efficient for bulk logging.

    Args:
        message: Log message
        level: Log level (DEBUG, INFO, WARN, ERROR, FATAL)
        attributes: Optional log attributes

    Returns:
        True if sent successfully, False otherwise

    Circuit breaker pattern:
    - Checks circuit before attempting
    - Records success/failure
    - Automatically disables after threshold failures
    - Auto-recovers with exponential backoff

    Examples:
        >>> client.send_log("User logged in", level="INFO", attributes={"user_id": 123})
        True
    """
    if not self._otlp_available:
        return False

    # Check circuit breaker
    if self.use_circuit_breaker:
        breaker = get_otlp_circuit_breaker()
        if not breaker.can_attempt():
            return False

    try:
        # Create temporary logger provider
        provider = self._create_logger_provider_internal()
        if not provider:
            if self.use_circuit_breaker:
                breaker.record_failure()
            return False

        # Get logger from provider
        logger = provider.get_logger(__name__)

        # Prepare attributes
        log_attrs = attributes.copy() if attributes else {}
        add_trace_context_to_attributes(log_attrs)
        normalized_attrs = normalize_attributes(log_attrs)

        # Map level to severity
        severity_number = map_level_to_severity(level)

        # Emit log record
        logger.emit(
            {
                "body": message,
                "severity_number": severity_number,
                "severity_text": level.upper(),
                "attributes": normalized_attrs,
                "timestamp": int(time.time_ns()),
            }
        )

        # Force flush to ensure delivery
        provider.force_flush()

        # Shutdown provider
        provider.shutdown()

        if self.use_circuit_breaker:
            breaker.record_success()

        return True

    except Exception:
        if self.use_circuit_breaker:
            breaker.record_failure()
        return False

Functions

add_trace_context_to_attributes

add_trace_context_to_attributes(
    attributes: dict[str, Any],
) -> None

Add trace context to attributes dict (modifies in place).

Extracts trace context and adds trace_id/span_id to attributes. Safe to call even if no trace context is available (no-op).

Parameters:

Name	Type	Description	Default
attributes	dict[str, Any]	Dictionary to add trace context to (modified in place)	required

Examples:

>>> attrs = {"key": "value"}
>>> add_trace_context_to_attributes(attrs)
>>> # attrs may now include 'trace_id' and 'span_id' if context available

Source code in provide/foundation/logger/otlp/helpers.py

def add_trace_context_to_attributes(attributes: dict[str, Any]) -> None:
    """Add trace context to attributes dict (modifies in place).

    Extracts trace context and adds trace_id/span_id to attributes.
    Safe to call even if no trace context is available (no-op).

    Args:
        attributes: Dictionary to add trace context to (modified in place)

    Examples:
        >>> attrs = {"key": "value"}
        >>> add_trace_context_to_attributes(attrs)
        >>> # attrs may now include 'trace_id' and 'span_id' if context available
    """
    trace_context = extract_trace_context()
    if trace_context:
        attributes.update(trace_context)

build_otlp_endpoint

build_otlp_endpoint(
    base_endpoint: str, signal_type: str = "logs"
) -> str

Build OTLP endpoint URL for specific signal type.

Constructs the full OTLP endpoint URL for the given signal type. Handles trailing slashes and is idempotent (won't double-add paths).

Parameters:

Name	Type	Description	Default
base_endpoint	str	Base OTLP endpoint (e.g., "https://api.example.com")	required
signal_type	str	"logs", "traces", or "metrics"	'logs'

Returns:

Type	Description
str	Full endpoint URL (e.g., "https://api.example.com/v1/logs")

Examples:

>>> build_otlp_endpoint("https://api.example.com")
'https://api.example.com/v1/logs'

>>> build_otlp_endpoint("https://api.example.com/", "traces")
'https://api.example.com/v1/traces'

>>> build_otlp_endpoint("https://api.example.com/v1/logs")
'https://api.example.com/v1/logs'

Source code in provide/foundation/logger/otlp/helpers.py

def build_otlp_endpoint(
    base_endpoint: str,
    signal_type: str = "logs",
) -> str:
    """Build OTLP endpoint URL for specific signal type.

    Constructs the full OTLP endpoint URL for the given signal type.
    Handles trailing slashes and is idempotent (won't double-add paths).

    Args:
        base_endpoint: Base OTLP endpoint (e.g., "https://api.example.com")
        signal_type: "logs", "traces", or "metrics"

    Returns:
        Full endpoint URL (e.g., "https://api.example.com/v1/logs")

    Examples:
        >>> build_otlp_endpoint("https://api.example.com")
        'https://api.example.com/v1/logs'

        >>> build_otlp_endpoint("https://api.example.com/", "traces")
        'https://api.example.com/v1/traces'

        >>> build_otlp_endpoint("https://api.example.com/v1/logs")
        'https://api.example.com/v1/logs'
    """
    # Remove trailing slash
    endpoint = base_endpoint.rstrip("/")

    # Check if already has /v1/{signal} path (idempotent)
    expected_suffix = f"/v1/{signal_type}"
    if endpoint.endswith(expected_suffix):
        return endpoint

    # Build full endpoint
    return f"{endpoint}/v1/{signal_type}"

build_otlp_headers

build_otlp_headers(
    base_headers: dict[str, str] | None = None,
    auth_token: str | None = None,
) -> dict[str, str]

Build OTLP headers with optional authentication.

Creates headers dictionary with OTLP-required headers and optional auth.

Parameters:

Name	Type	Description	Default
base_headers	dict[str, str] | None	Base headers to include	None
auth_token	str | None	Optional bearer token for authentication	None

Returns:

Type	Description
dict[str, str]	Complete headers dict with Content-Type and auth

Examples:

>>> build_otlp_headers()
{'Content-Type': 'application/x-protobuf'}

>>> build_otlp_headers(auth_token="secret123")
{'Content-Type': 'application/x-protobuf', 'Authorization': 'Bearer secret123'}

>>> build_otlp_headers(base_headers={"X-Custom": "value"})
{'X-Custom': 'value', 'Content-Type': 'application/x-protobuf'}

Source code in provide/foundation/logger/otlp/helpers.py

def build_otlp_headers(
    base_headers: dict[str, str] | None = None,
    auth_token: str | None = None,
) -> dict[str, str]:
    """Build OTLP headers with optional authentication.

    Creates headers dictionary with OTLP-required headers and optional auth.

    Args:
        base_headers: Base headers to include
        auth_token: Optional bearer token for authentication

    Returns:
        Complete headers dict with Content-Type and auth

    Examples:
        >>> build_otlp_headers()
        {'Content-Type': 'application/x-protobuf'}

        >>> build_otlp_headers(auth_token="secret123")
        {'Content-Type': 'application/x-protobuf', 'Authorization': 'Bearer secret123'}

        >>> build_otlp_headers(base_headers={"X-Custom": "value"})
        {'X-Custom': 'value', 'Content-Type': 'application/x-protobuf'}
    """
    headers: dict[str, str] = {}

    if base_headers:
        headers.update(base_headers)

    # Add OTLP content type
    headers.setdefault("Content-Type", "application/x-protobuf")

    # Add auth token if provided
    if auth_token:
        headers["Authorization"] = f"Bearer {auth_token}"

    return headers

build_resource_attributes

build_resource_attributes(
    service_name: str,
    service_version: str | None = None,
    environment: str | None = None,
    additional_attrs: dict[str, Any] | None = None,
) -> dict[str, Any]

Build resource attributes dictionary.

Creates a dictionary with standard OpenTelemetry resource attributes: - service.name (required) - service.version (optional) - deployment.environment (optional) - Any additional custom attributes

Parameters:

Name	Type	Description	Default
service_name	str	Service name (required)	required
service_version	str | None	Service version (optional)	None
environment	str | None	Deployment environment (dev, staging, prod, etc.)	None
additional_attrs	dict[str, Any] | None	Additional custom resource attributes	None

Returns:

Type	Description
dict[str, Any]	Dictionary of resource attributes

Examples:

>>> build_resource_attributes("my-service")
{'service.name': 'my-service'}

>>> build_resource_attributes(
...     "my-service",
...     service_version="1.2.3",
...     environment="production",
... )
{'service.name': 'my-service', 'service.version': '1.2.3', 'deployment.environment': 'production'}

Source code in provide/foundation/logger/otlp/resource.py

def build_resource_attributes(
    service_name: str,
    service_version: str | None = None,
    environment: str | None = None,
    additional_attrs: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build resource attributes dictionary.

    Creates a dictionary with standard OpenTelemetry resource attributes:
    - service.name (required)
    - service.version (optional)
    - deployment.environment (optional)
    - Any additional custom attributes

    Args:
        service_name: Service name (required)
        service_version: Service version (optional)
        environment: Deployment environment (dev, staging, prod, etc.)
        additional_attrs: Additional custom resource attributes

    Returns:
        Dictionary of resource attributes

    Examples:
        >>> build_resource_attributes("my-service")
        {'service.name': 'my-service'}

        >>> build_resource_attributes(
        ...     "my-service",
        ...     service_version="1.2.3",
        ...     environment="production",
        ... )
        {'service.name': 'my-service', 'service.version': '1.2.3', 'deployment.environment': 'production'}
    """
    attrs: dict[str, Any] = {
        "service.name": service_name,
    }

    if service_version:
        attrs["service.version"] = service_version

    if environment:
        attrs["deployment.environment"] = environment

    if additional_attrs:
        attrs.update(additional_attrs)

    return attrs

create_otlp_resource

create_otlp_resource(
    service_name: str,
    service_version: str | None = None,
    environment: str | None = None,
    additional_attrs: dict[str, Any] | None = None,
) -> Any | None

Create OpenTelemetry Resource instance.

Attempts to create an OpenTelemetry SDK Resource with the provided attributes. Returns None if the OpenTelemetry SDK is not available (optional dependency).

Parameters:

Name	Type	Description	Default
service_name	str	Service name (required)	required
service_version	str | None	Service version (optional)	None
environment	str | None	Deployment environment (optional)	None
additional_attrs	dict[str, Any] | None	Additional custom resource attributes	None

Returns:

Type	Description
Any | None	Resource instance if OpenTelemetry SDK available, None otherwise

Examples:

>>> resource = create_otlp_resource("my-service", service_version="1.0.0")
>>> # Returns Resource instance or None if SDK not installed

>>> resource = create_otlp_resource(
...     "my-service",
...     environment="production",
...     additional_attrs={"team": "platform"},
... )

Source code in provide/foundation/logger/otlp/resource.py

def create_otlp_resource(
    service_name: str,
    service_version: str | None = None,
    environment: str | None = None,
    additional_attrs: dict[str, Any] | None = None,
) -> Any | None:
    """Create OpenTelemetry Resource instance.

    Attempts to create an OpenTelemetry SDK Resource with the provided attributes.
    Returns None if the OpenTelemetry SDK is not available (optional dependency).

    Args:
        service_name: Service name (required)
        service_version: Service version (optional)
        environment: Deployment environment (optional)
        additional_attrs: Additional custom resource attributes

    Returns:
        Resource instance if OpenTelemetry SDK available, None otherwise

    Examples:
        >>> resource = create_otlp_resource("my-service", service_version="1.0.0")
        >>> # Returns Resource instance or None if SDK not installed

        >>> resource = create_otlp_resource(
        ...     "my-service",
        ...     environment="production",
        ...     additional_attrs={"team": "platform"},
        ... )
    """
    try:
        from opentelemetry.sdk.resources import Resource
    except ImportError:
        return None

    attrs = build_resource_attributes(
        service_name=service_name,
        service_version=service_version,
        environment=environment,
        additional_attrs=additional_attrs,
    )

    return Resource.create(attrs)

extract_trace_context

extract_trace_context() -> dict[str, str] | None

Extract current trace context from OpenTelemetry.

Extracts trace context from OpenTelemetry if SDK is available and a valid span is recording.

Returns:

Type	Description
dict[str, str] | None	Dict with 'trace_id' and 'span_id', or None if not available

Examples:

>>> context = extract_trace_context()
>>> # Returns {'trace_id': '...', 'span_id': '...'} or None

Source code in provide/foundation/logger/otlp/helpers.py

def extract_trace_context() -> dict[str, str] | None:
    """Extract current trace context from OpenTelemetry.

    Extracts trace context from OpenTelemetry if SDK is available
    and a valid span is recording.

    Returns:
        Dict with 'trace_id' and 'span_id', or None if not available

    Examples:
        >>> context = extract_trace_context()
        >>> # Returns {'trace_id': '...', 'span_id': '...'} or None
    """
    try:
        from opentelemetry import trace

        span = trace.get_current_span()
        if span and span.is_recording():
            span_context = span.get_span_context()
            if span_context.is_valid:
                return {
                    "trace_id": format(span_context.trace_id, "032x"),
                    "span_id": format(span_context.span_id, "016x"),
                }
    except ImportError:
        pass

    return None

get_otlp_circuit_breaker

get_otlp_circuit_breaker() -> OTLPCircuitBreaker

Get the global OTLP circuit breaker instance.

Returns:

Type	Description
OTLPCircuitBreaker	Shared OTLPCircuitBreaker instance

Source code in provide/foundation/logger/otlp/circuit.py

def get_otlp_circuit_breaker() -> OTLPCircuitBreaker:
    """Get the global OTLP circuit breaker instance.

    Returns:
        Shared OTLPCircuitBreaker instance
    """
    global _otlp_circuit_breaker

    if _otlp_circuit_breaker is None:
        with _circuit_breaker_lock:
            if _otlp_circuit_breaker is None:
                _otlp_circuit_breaker = OTLPCircuitBreaker(
                    failure_threshold=5,  # Open after 5 failures
                    timeout=30.0,  # Start with 30s timeout
                    half_open_timeout=10.0,  # Wait 10s between half-open attempts
                )

    return _otlp_circuit_breaker

map_level_to_severity

map_level_to_severity(level: str) -> int

Map log level string to OTLP severity number.

Parameters:

Name	Type	Description	Default
level	str	Log level string (e.g., "INFO", "ERROR", "WARN")	required

Returns:

Type	Description
int	OTLP severity number (1-24)
int	Falls back to 9 (INFO) for unknown levels

Examples:

>>> map_level_to_severity("INFO")
9
>>> map_level_to_severity("ERROR")
17
>>> map_level_to_severity("warning")
13
>>> map_level_to_severity("unknown")
9

Source code in provide/foundation/logger/otlp/severity.py

def map_level_to_severity(level: str) -> int:
    """Map log level string to OTLP severity number.

    Args:
        level: Log level string (e.g., "INFO", "ERROR", "WARN")

    Returns:
        OTLP severity number (1-24)
        Falls back to 9 (INFO) for unknown levels

    Examples:
        >>> map_level_to_severity("INFO")
        9
        >>> map_level_to_severity("ERROR")
        17
        >>> map_level_to_severity("warning")
        13
        >>> map_level_to_severity("unknown")
        9
    """
    return _LEVEL_TO_SEVERITY.get(level.upper(), 9)

map_severity_to_level

map_severity_to_level(severity: int) -> str

Map OTLP severity number to log level string.

Parameters:

Name	Type	Description	Default
severity	int	OTLP severity number (1-24)	required

Returns:

Type	Description
str	Log level string (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)
str	Falls back to "INFO" for unknown severity numbers

Examples:

>>> map_severity_to_level(9)
'INFO'
>>> map_severity_to_level(17)
'ERROR'
>>> map_severity_to_level(100)
'INFO'

Source code in provide/foundation/logger/otlp/severity.py

def map_severity_to_level(severity: int) -> str:
    """Map OTLP severity number to log level string.

    Args:
        severity: OTLP severity number (1-24)

    Returns:
        Log level string (TRACE, DEBUG, INFO, WARN, ERROR, FATAL)
        Falls back to "INFO" for unknown severity numbers

    Examples:
        >>> map_severity_to_level(9)
        'INFO'
        >>> map_severity_to_level(17)
        'ERROR'
        >>> map_severity_to_level(100)
        'INFO'
    """
    return _SEVERITY_TO_LEVEL.get(severity, "INFO")

normalize_attributes

normalize_attributes(
    attributes: dict[str, Any],
) -> dict[str, Any]

Normalize attribute values for OTLP compatibility.

Converts non-serializable types to OTLP-compatible values: - Non-serializable types → strings - Nested dicts → JSON strings - Lists → JSON strings - None values → empty strings

Returns new dict (doesn't modify input).

Parameters:

Name	Type	Description	Default
attributes	dict[str, Any]	Dictionary of attributes to normalize	required

Returns:

Type	Description
dict[str, Any]	New dictionary with normalized values

Examples:

>>> normalize_attributes({"key": "value"})
{'key': 'value'}

>>> normalize_attributes({"num": 42, "list": [1, 2, 3]})
{'num': 42, 'list': '[1, 2, 3]'}

>>> normalize_attributes({"nested": {"a": 1}})
{'nested': '{"a": 1}'}

Source code in provide/foundation/logger/otlp/helpers.py

def normalize_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
    """Normalize attribute values for OTLP compatibility.

    Converts non-serializable types to OTLP-compatible values:
    - Non-serializable types → strings
    - Nested dicts → JSON strings
    - Lists → JSON strings
    - None values → empty strings

    Returns new dict (doesn't modify input).

    Args:
        attributes: Dictionary of attributes to normalize

    Returns:
        New dictionary with normalized values

    Examples:
        >>> normalize_attributes({"key": "value"})
        {'key': 'value'}

        >>> normalize_attributes({"num": 42, "list": [1, 2, 3]})
        {'num': 42, 'list': '[1, 2, 3]'}

        >>> normalize_attributes({"nested": {"a": 1}})
        {'nested': '{"a": 1}'}
    """
    normalized: dict[str, Any] = {}

    for key, value in attributes.items():
        if value is None:
            normalized[key] = ""
        elif isinstance(value, (str, int, float, bool)):
            normalized[key] = value
        elif isinstance(value, (dict, list)):
            try:
                normalized[key] = json.dumps(value)
            except (TypeError, ValueError):
                normalized[key] = str(value)
        else:
            normalized[key] = str(value)

    return normalized

reset_otlp_circuit_breaker

reset_otlp_circuit_breaker() -> None

Reset the global circuit breaker (primarily for testing).

Source code in provide/foundation/logger/otlp/circuit.py

def reset_otlp_circuit_breaker() -> None:
    """Reset the global circuit breaker (primarily for testing)."""
    global _otlp_circuit_breaker

    with _circuit_breaker_lock:
        if _otlp_circuit_breaker is not None:
            _otlp_circuit_breaker.reset()