honeyhive.tracer.processing.otlp_session

Optimized HTTP session factory for OpenTelemetry OTLP exports.

This module provides utilities for creating high-performance HTTP sessions specifically optimized for OTLP (OpenTelemetry Protocol) span exports. The sessions feature enhanced connection pooling, intelligent retry strategies, and configurations tuned for telemetry workloads.

Key optimizations: - Connection pooling with configurable pool sizes - Retry strategies for transient network failures - Non-blocking pool behavior for high throughput - Optimized timeouts for telemetry data

OTLPSessionConfig

Bases: BaseModel

Configuration for optimized OTLP HTTP sessions.

This class encapsulates all configuration options for creating high-performance HTTP sessions for OTLP exports.

Source code in src/honeyhive/tracer/processing/otlp_session.py

class OTLPSessionConfig(BaseModel):
    """Configuration for optimized OTLP HTTP sessions.

    This class encapsulates all configuration options for creating
    high-performance HTTP sessions for OTLP exports.
    """

    model_config = ConfigDict(
        validate_assignment=True,
        extra="forbid",
        frozen=False,  # Allow modifications during dynamic adjustments
    )

    pool_connections: int = Field(
        default=10,
        description="Number of connection pools to cache",
        ge=1,
        le=100,
    )

    pool_maxsize: int = Field(
        default=20,
        description="Maximum connections per pool",
        ge=1,
        le=200,
    )

    max_retries: int = Field(
        default=3,
        description="Maximum retry attempts for failed requests",
        ge=0,
        le=20,
    )

    pool_block: bool = Field(
        default=False,
        description="Whether to block when pool is full",
    )

    timeout: Optional[float] = Field(
        default=30.0,
        description="Request timeout in seconds",
        gt=0.0,
        le=600.0,
    )

    backoff_factor: float = Field(
        default=0.5,
        description="Backoff factor for retry delays",
        ge=0.0,
        le=10.0,
    )

    retry_status_codes: List[int] = Field(
        default_factory=lambda: [429, 500, 502, 503, 504],
        description="HTTP status codes to retry",
    )

    @field_validator("pool_maxsize")
    @classmethod
    def validate_pool_maxsize(cls, v: int, info: Any) -> int:  # pylint: disable=duplicate-code
        """Ensure pool_maxsize is at least as large as pool_connections.

        Note: This validation logic is intentionally duplicated between
        OTLPSessionConfig and EnvironmentProfile classes as both need
        the same pool size validation constraints.
        """
        if hasattr(info, "data") and "pool_connections" in info.data:
            pool_connections = info.data["pool_connections"]
            if v < pool_connections:
                return int(max(pool_connections, v))
        return v

    @field_validator("retry_status_codes")
    @classmethod
    def validate_retry_status_codes(cls, v: List[int]) -> List[int]:
        """Validate HTTP status codes are in valid range."""
        if not v:
            return [429, 500, 502, 503, 504]  # Default codes

        # Filter to valid HTTP status codes (100-599)
        valid_codes = [code for code in v if 100 <= code <= 599]
        return valid_codes if valid_codes else [429, 500, 502, 503, 504]

    def to_dict(self) -> Dict[str, Any]:
        """Convert configuration to dictionary for logging."""
        return self.model_dump()

pool_connections class-attribute instance-attribute

pool_connections: int = Field(
    default=10,
    description="Number of connection pools to cache",
    ge=1,
    le=100,
)

pool_maxsize class-attribute instance-attribute

pool_maxsize: int = Field(
    default=20,
    description="Maximum connections per pool",
    ge=1,
    le=200,
)

max_retries class-attribute instance-attribute

max_retries: int = Field(
    default=3,
    description="Maximum retry attempts for failed requests",
    ge=0,
    le=20,
)

pool_block class-attribute instance-attribute

pool_block: bool = Field(
    default=False,
    description="Whether to block when pool is full",
)

timeout class-attribute instance-attribute

timeout: Optional[float] = Field(
    default=30.0,
    description="Request timeout in seconds",
    gt=0.0,
    le=600.0,
)

backoff_factor class-attribute instance-attribute

backoff_factor: float = Field(
    default=0.5,
    description="Backoff factor for retry delays",
    ge=0.0,
    le=10.0,
)

retry_status_codes class-attribute instance-attribute

retry_status_codes: List[int] = Field(
    default_factory=lambda: [429, 500, 502, 503, 504],
    description="HTTP status codes to retry",
)

validate_pool_maxsize classmethod

validate_pool_maxsize(v: int, info: Any) -> int

Ensure pool_maxsize is at least as large as pool_connections.

Note: This validation logic is intentionally duplicated between OTLPSessionConfig and EnvironmentProfile classes as both need the same pool size validation constraints.

Source code in src/honeyhive/tracer/processing/otlp_session.py

@field_validator("pool_maxsize")
@classmethod
def validate_pool_maxsize(cls, v: int, info: Any) -> int:  # pylint: disable=duplicate-code
    """Ensure pool_maxsize is at least as large as pool_connections.

    Note: This validation logic is intentionally duplicated between
    OTLPSessionConfig and EnvironmentProfile classes as both need
    the same pool size validation constraints.
    """
    if hasattr(info, "data") and "pool_connections" in info.data:
        pool_connections = info.data["pool_connections"]
        if v < pool_connections:
            return int(max(pool_connections, v))
    return v

validate_retry_status_codes classmethod

validate_retry_status_codes(v: List[int]) -> List[int]

Validate HTTP status codes are in valid range.

Source code in src/honeyhive/tracer/processing/otlp_session.py

@field_validator("retry_status_codes")
@classmethod
def validate_retry_status_codes(cls, v: List[int]) -> List[int]:
    """Validate HTTP status codes are in valid range."""
    if not v:
        return [429, 500, 502, 503, 504]  # Default codes

    # Filter to valid HTTP status codes (100-599)
    valid_codes = [code for code in v if 100 <= code <= 599]
    return valid_codes if valid_codes else [429, 500, 502, 503, 504]

to_dict

to_dict() -> Dict[str, Any]

Convert configuration to dictionary for logging.

Source code in src/honeyhive/tracer/processing/otlp_session.py

def to_dict(self) -> Dict[str, Any]:
    """Convert configuration to dictionary for logging."""
    return self.model_dump()

create_optimized_otlp_session

create_optimized_otlp_session(
    config: Optional[OTLPSessionConfig] = None,
    tracer_instance: Optional[Any] = None,
) -> Session

Create optimized requests.Session for OTLP exports.

This function creates a high-performance HTTP session specifically optimized for OpenTelemetry OTLP span exports. The session features:

• Enhanced connection pooling for reduced connection overhead
• Intelligent retry strategy for transient network failures
• Optimized timeouts and backoff strategies for telemetry workloads
• Non-blocking pool behavior for high-throughput scenarios

Parameters:

Name	Type	Description	Default
config	Optional[OTLPSessionConfig]	Optional session configuration (uses defaults if None)	None
tracer_instance	Optional[Any]	Optional tracer instance for logging context	None

Returns:

Type	Description
Session	Optimized requests.Session configured for OTLP exports

Example

config = OTLPSessionConfig(pool_maxsize=30, max_retries=5) session = create_optimized_otlp_session(config, tracer_instance)

Use session with OTLPSpanExporter

exporter = OTLPSpanExporter(endpoint="...", session=session)

Source code in src/honeyhive/tracer/processing/otlp_session.py

def create_optimized_otlp_session(
    config: Optional[OTLPSessionConfig] = None,
    tracer_instance: Optional[Any] = None,
) -> requests.Session:
    """Create optimized requests.Session for OTLP exports.

    This function creates a high-performance HTTP session specifically
    optimized for OpenTelemetry OTLP span exports. The session features:

    - Enhanced connection pooling for reduced connection overhead
    - Intelligent retry strategy for transient network failures
    - Optimized timeouts and backoff strategies for telemetry workloads
    - Non-blocking pool behavior for high-throughput scenarios

    Args:
        config: Optional session configuration (uses defaults if None)
        tracer_instance: Optional tracer instance for logging context

    Returns:
        Optimized requests.Session configured for OTLP exports

    Example:
        >>> config = OTLPSessionConfig(pool_maxsize=30, max_retries=5)
        >>> session = create_optimized_otlp_session(config, tracer_instance)
        >>> # Use session with OTLPSpanExporter
        >>> exporter = OTLPSpanExporter(endpoint="...", session=session)
    """
    if config is None:
        config = OTLPSessionConfig()

    safe_log(
        tracer_instance,
        "debug",
        "Creating optimized OTLP session with connection pooling",
        honeyhive_data={"session_config": config.to_dict()},
    )

    try:
        # Create new session
        session = requests.Session()

        # Configure retry strategy optimized for telemetry
        retry_strategy = Retry(
            total=config.max_retries,
            status_forcelist=config.retry_status_codes,
            backoff_factor=config.backoff_factor,
            raise_on_status=False,  # Don't raise on retry-able status codes
            respect_retry_after_header=True,  # Honor server retry-after headers
        )

        # Create high-performance HTTP adapter
        adapter = HTTPAdapter(
            pool_connections=config.pool_connections,
            pool_maxsize=config.pool_maxsize,
            max_retries=retry_strategy,
            pool_block=config.pool_block,
        )

        # Mount adapter for both HTTP and HTTPS
        session.mount("http://", adapter)
        session.mount("https://", adapter)

        # Set default timeout if specified
        if config.timeout:
            # Note: This sets a default, but can be overridden per request
            session.timeout = config.timeout  # type: ignore[attr-defined]

        safe_log(
            tracer_instance,
            "info",
            "Successfully created optimized OTLP session",
            honeyhive_data={
                "pool_connections": config.pool_connections,
                "pool_maxsize": config.pool_maxsize,
                "max_retries": config.max_retries,
                "timeout": config.timeout,
            },
        )

        return session

    except Exception as e:
        safe_log(
            tracer_instance,
            "error",
            f"Failed to create optimized OTLP session: {e}",
            honeyhive_data={
                "error_type": type(e).__name__,
                "session_config": config.to_dict(),
            },
        )

        # Fallback to basic session
        safe_log(
            tracer_instance,
            "warning",
            "Falling back to basic requests.Session for OTLP exports",
        )
        return requests.Session()

get_session_stats

get_session_stats(session: Session) -> Dict[str, Any]

Get connection pool statistics from a requests session.

Parameters:

Name	Type	Description	Default
session	Session	The requests.Session to analyze	required

Returns:

Type	Description
Dict[str, Any]	Dictionary containing pool statistics and configuration

Source code in src/honeyhive/tracer/processing/otlp_session.py

def get_session_stats(session: requests.Session) -> Dict[str, Any]:
    """Get connection pool statistics from a requests session.

    Args:
        session: The requests.Session to analyze

    Returns:
        Dictionary containing pool statistics and configuration
    """
    stats: Dict[str, Any] = {
        "adapters": {},
        "total_pools": 0,
    }

    try:
        for prefix, adapter in session.adapters.items():
            adapter_stats: Dict[str, Any] = {
                "type": type(adapter).__name__,
                "pools": 0,
            }

            # Get pool manager stats if available
            if hasattr(adapter, "poolmanager") and adapter.poolmanager:
                pool_manager = adapter.poolmanager
                adapter_stats.update(
                    {
                        "pools": len(getattr(pool_manager, "pools", {})),
                        "pool_connections": getattr(adapter, "config", {}).get(
                            "pool_connections", "default"
                        ),
                        "pool_maxsize": getattr(adapter, "config", {}).get(
                            "pool_maxsize", "default"
                        ),
                    }
                )
                pools_count = adapter_stats["pools"]
                if isinstance(pools_count, int):
                    stats["total_pools"] += pools_count

            stats["adapters"][prefix] = adapter_stats

    except Exception as e:
        stats["error"] = f"Failed to get session stats: {e}"

    return stats

create_dynamic_otlp_config

create_dynamic_otlp_config(
    tracer_instance: Optional[Any] = None,
    scenario: str = "default",
    **overrides: Any
) -> OTLPSessionConfig

Create fully dynamic OTLP session configuration based on environment analysis.

This function uses pure dynamic logic to determine optimal configuration values based on actual environment conditions, resource constraints, and tracer settings. NO hardcoded values - everything calculated from real conditions.

Parameters:

Name	Type	Description	Default
tracer_instance	Optional[Any]	Optional tracer instance for dynamic configuration	None
scenario	str	Configuration scenario hint (used for dynamic adjustments)	'default'
**overrides	Any	Explicit configuration overrides	{}

Returns:

Type	Description
OTLPSessionConfig	Dynamically configured OTLPSessionConfig based on actual environment

Source code in src/honeyhive/tracer/processing/otlp_session.py

def create_dynamic_otlp_config(
    tracer_instance: Optional[Any] = None, scenario: str = "default", **overrides: Any
) -> OTLPSessionConfig:
    """Create fully dynamic OTLP session configuration based on environment analysis.

    This function uses pure dynamic logic to determine optimal configuration values
    based on actual environment conditions, resource constraints, and tracer settings.
    NO hardcoded values - everything calculated from real conditions.

    Args:
        tracer_instance: Optional tracer instance for dynamic configuration
        scenario: Configuration scenario hint (used for dynamic adjustments)
        **overrides: Explicit configuration overrides

    Returns:
        Dynamically configured OTLPSessionConfig based on actual environment
    """
    # Get comprehensive environment analysis
    try:
        # Import at top level to avoid inline imports
        env_analysis = _get_comprehensive_environment_analysis(tracer_instance)
    except Exception:
        # Fallback to basic analysis if environment module not available
        env_analysis = _get_basic_environment_analysis(tracer_instance)

    # Extract dynamic factors from environment analysis
    resource_constraints = env_analysis.get("resource_constraints", {})
    performance_chars = env_analysis.get("performance_characteristics", {})

    # Calculate base configuration dynamically from environment
    base_config = _calculate_base_config_from_environment(
        resource_constraints, performance_chars, tracer_instance
    )

    # Apply dynamic tracer-specific adjustments
    if tracer_instance:
        base_config = _apply_tracer_dynamic_adjustments(base_config, tracer_instance)

    # Apply dynamic scenario adjustments (using environment multipliers)
    base_config = _apply_scenario_dynamic_adjustments(
        base_config, scenario, performance_chars
    )

    # Apply explicit overrides
    base_config.update(overrides)

    # Log the fully dynamic configuration
    safe_log(
        tracer_instance,
        "debug",
        "Created fully dynamic OTLP config from environment analysis",
        honeyhive_data={
            "scenario": scenario,
            "dynamic_config": base_config,
            "environment_type": env_analysis.get("environment_type"),
            "resource_scaling_factors": {
                "memory": resource_constraints.get("memory_constraint_factor"),
                "cpu": resource_constraints.get("cpu_scaling_factor"),
                "network": resource_constraints.get("network_scaling_factor"),
            },
            "performance_multipliers": {
                "timeout": performance_chars.get("timeout_multiplier"),
                "retry": performance_chars.get("retry_multiplier"),
                "concurrency": performance_chars.get("concurrency_multiplier"),
            },
        },
    )

    return OTLPSessionConfig(**base_config)

get_default_otlp_config

get_default_otlp_config(
    tracer_instance: Optional[Any] = None,
) -> OTLPSessionConfig

Get default OTLP configuration with dynamic adjustments.

Source code in src/honeyhive/tracer/processing/otlp_session.py

def get_default_otlp_config(tracer_instance: Optional[Any] = None) -> OTLPSessionConfig:
    """Get default OTLP configuration with dynamic adjustments."""
    return create_dynamic_otlp_config(tracer_instance, "default")

get_high_volume_otlp_config

get_high_volume_otlp_config(
    tracer_instance: Optional[Any] = None,
) -> OTLPSessionConfig

Get high-volume OTLP configuration with dynamic adjustments.

Source code in src/honeyhive/tracer/processing/otlp_session.py

def get_high_volume_otlp_config(
    tracer_instance: Optional[Any] = None,
) -> OTLPSessionConfig:
    """Get high-volume OTLP configuration with dynamic adjustments."""
    return create_dynamic_otlp_config(tracer_instance, "high_volume")

get_low_latency_otlp_config

get_low_latency_otlp_config(
    tracer_instance: Optional[Any] = None,
) -> OTLPSessionConfig

Get low-latency OTLP configuration with dynamic adjustments.

Source code in src/honeyhive/tracer/processing/otlp_session.py

def get_low_latency_otlp_config(
    tracer_instance: Optional[Any] = None,
) -> OTLPSessionConfig:
    """Get low-latency OTLP configuration with dynamic adjustments."""
    return create_dynamic_otlp_config(tracer_instance, "low_latency")