orchestrata_ultimate.kernel.performance.chaos

Reference › orchestrata_ultimate › kernel › performance › chaos

Fault injection for resilience testing, SPEC-001 requirement REQ-001-US-010-030. ChaosController is disabled by default; when enabled, callers ask it whether to inject a fault at a given point. The manager owns one and exposes it through enable_chaos_testing, add_chaos_fault and maybe_inject_chaos.

The nested FaultType enum and FaultConfig dataclass describe a fault. add_fault and remove_fault register faults by id, enable and disable switch the controller. maybe_inject returns False at once when the controller is disabled or no registered fault matches the requested type and component (_select_fault). When one matches, _apply_fault acts outside the lock: a latency fault sleeps for the configured milliseconds, an error fault raises a RuntimeError naming the fault id, and a timeout fault sleeps and then raises a TimeoutError. get_stats reports the controller's counters.

At a glance

Class diagram

Dependency diagram

API

Classes

ChaosController

Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-030']) · Line 15

Chaos engineering controller (REQ-001-US-010-030).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/performance/chaos.py

# mypy: ignore-errors
"""Performance subsystem: chaos."""

import asyncio
import collections
import threading
from dataclasses import dataclass
from enum import Enum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-030"])
class ChaosController:
    """Chaos engineering controller (REQ-001-US-010-030).

    Injects failures for testing system resilience.
    """

    class FaultType(Enum):
        """Types of faults to inject."""

        LATENCY = "latency"
        ERROR = "error"
        TIMEOUT = "timeout"
        PARTITION = "partition"
        RESOURCE_EXHAUSTION = "resource_exhaustion"

    @dataclass
    class FaultConfig:
        """Configuration for a fault."""

        fault_type: ChaosController.FaultType
        probability: float  # 0-1
        duration_ms: float
        target: str | None  # Specific component or None for all
        enabled: bool = True

    def __init__(self, enabled: bool = False) -> None:
        """Initialize chaos controller.

        Args:
            enabled: Whether chaos testing is enabled
        """
        self._enabled = enabled
        self._faults: dict[str, ChaosController.FaultConfig] = {}
        self._injected_count: dict[str, int] = collections.defaultdict(int)
        self._lock = threading.Lock()

    def enable(self) -> None:
        """Enable chaos testing."""
        self._enabled = True

    def disable(self) -> None:
        """Disable chaos testing."""
        self._enabled = False

    def add_fault(self, fault_id: str, config: ChaosController.FaultConfig) -> None:
        """Add fault configuration.

        Args:
            fault_id: Unique fault identifier
            config: Fault configuration
        """
        with self._lock:
            self._faults[fault_id] = config

    def remove_fault(self, fault_id: str) -> None:
        """Remove fault configuration.

        Args:
            fault_id: Fault identifier to remove
        """
        with self._lock:
            self._faults.pop(fault_id, None)

    async def maybe_inject(
        self,
        fault_type: ChaosController.FaultType,
        component: str | None = None,
    ) -> bool:
        """Possibly inject a fault.

        Args:
            fault_type: Type of fault to potentially inject
            component: Component requesting (for targeted faults)

        Returns:
            True if fault was injected
        """
        if not self._enabled:
            return False

        # Capture fault config under lock, then release before any await/raise
        matched_fault_id, matched_config = self._select_fault(fault_type, component)
        if matched_config is None:
            return False

        # Act on matched fault OUTSIDE the lock — safe to await/raise here
        await self._apply_fault(matched_fault_id, matched_config)
        return True

    def _select_fault(
        self,
        fault_type: ChaosController.FaultType,
        component: str | None,
    ) -> tuple[str | None, ChaosController.FaultConfig | None]:
        """Pick a matching, probability-triggered fault under the lock.

        Returns ``(fault_id, config)`` for a triggered fault, or
        ``(None, None)`` if nothing matched. Increments the injected counter
        for the selected fault.
        """
        import random

        with self._lock:
            for fault_id, config in self._faults.items():
                if not config.enabled:
                    continue
                if config.fault_type != fault_type:
                    continue
                if config.target and config.target != component:
                    continue
                if random.random() < config.probability:
                    self._injected_count[fault_id] += 1
                    return fault_id, config
        return None, None

    @staticmethod
    async def _apply_fault(
        fault_id: str | None,
        config: ChaosController.FaultConfig,
    ) -> None:
        """Act on a selected fault outside the lock (sleep and/or raise)."""
        if config.fault_type == ChaosController.FaultType.LATENCY:
            await asyncio.sleep(config.duration_ms / 1000.0)
        elif config.fault_type == ChaosController.FaultType.ERROR:
            raise RuntimeError(f"Chaos: Injected error ({fault_id})")
        elif config.fault_type == ChaosController.FaultType.TIMEOUT:
            await asyncio.sleep(config.duration_ms / 1000.0)
            raise TimeoutError(f"Chaos: Injected timeout ({fault_id})")

    def get_stats(self) -> dict[str, Any]:
        """Get chaos testing statistics."""
        with self._lock:
            return {
                "enabled": self._enabled,
                "fault_count": len(self._faults),
                "injected": dict(self._injected_count),
            }