orchestrata_ultimate.kernel.performance.manager

Reference › orchestrata_ultimate › kernel › performance › manager

The facade of the performance package. PerformanceManager gives the kernel one object through which latency, throughput, memory, uptime, caching, backpressure, clustering, autoscaling, chaos testing and SLA checks are reached. The package initialiser re-exports it, and the shim performance_manager keeps the old path. It serves SPEC-001 story US-010.

The constructor takes a node id (a UUID when omitted) and a configuration dictionary, and builds the parts: routing and policy LatencyTrackers, a ThroughputTracker, MemoryTracker and UptimeTracker from metrics; an LRUCache and BackpressureController from resources; a ClusterManager and LeaderElection from cluster; an AutoScaler from scaling; and a ChaosController from chaos. Configuration keys such as the cache size and time-to-live, heartbeat interval, leader time-to-live, instance limits and a chaos switch override the defaults.

start records the start time, starts the uptime tracker and the cluster manager, tries to acquire leadership and returns the startup time in seconds (REQ-001-US-010-007); stop shuts the parts down. record_routing_latency and record_policy_latency feed the trackers, and the p95 and p99 getters read them. record_message and get_throughput, the session memory methods, mark_system_down, mark_system_up and get_uptime_percent, and cache_get, cache_put and cache_invalidate delegate to their parts. update_load, should_accept_request and get_backpressure_state drive load shedding. evaluate_scaling and get_scaling_decisions use the autoscaler. check_sla_compliance compares the measured values with the SLA targets in two helper methods, and get_all_stats collects everything.

At a glance

Class diagram

Dependency diagram

API

Classes

PerformanceManager

Line 38

Central performance management for the kernel.

Source

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

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

import time
import uuid
from datetime import datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel.performance._common import (
    AUTOSCALER_RECENT_DECISIONS_DEFAULT_COUNT,
    _logger,
    _utc_now,
)
from orchestrata_ultimate.kernel.performance.chaos import ChaosController
from orchestrata_ultimate.kernel.performance.cluster import (
    ClusterManager,
    LeaderElection,
)
from orchestrata_ultimate.kernel.performance.metrics import (
    LatencyTracker,
    MemoryTracker,
    ThroughputTracker,
    UptimeTracker,
)
from orchestrata_ultimate.kernel.performance.resources import (
    BackpressureController,
    ConnectionPool,
    LRUCache,
)
from orchestrata_ultimate.kernel.performance.scaling import AutoScaler, ScalingDecision
from orchestrata_ultimate.kernel.performance.sla import (
    DEFAULT_SLA_THRESHOLDS,
    SLATarget,
)


class PerformanceManager:
    """Central performance management for the kernel.

    Coordinates all performance monitoring, optimization, and scaling features.
    Implements US-010: Performance and Scalability (REQ-001-US-010-001 to REQ-001-US-010-030).
    """

    def __init__(
        self,
        node_id: str | None = None,
        config: dict[str, Any] | None = None,
    ) -> None:
        """Initialize performance manager.

        Args:
            node_id: This kernel instance's identifier
            config: Optional configuration overrides
        """
        self._node_id = node_id or str(uuid.uuid4())
        self._config = config or {}
        self._started_at: datetime | None = None

        # Performance tracking (REQ-001-US-010-001 to REQ-001-US-010-010)
        self._routing_latency = LatencyTracker("routing")
        self._policy_latency = LatencyTracker("policy")
        self._throughput = ThroughputTracker()
        self._memory = MemoryTracker()
        self._uptime = UptimeTracker()

        # Optimization components (REQ-001-US-010-011 to REQ-001-US-010-020)
        self._query_cache: LRUCache[Any] = LRUCache(
            max_size=self._config.get("cache_size", 1000),
            default_ttl=self._config.get("cache_ttl", 300.0),
        )
        self._backpressure = BackpressureController()

        # Scalability components (REQ-010-021 to REQ-001-US-010-030)
        self._cluster = ClusterManager(
            self._node_id,
            heartbeat_interval=self._config.get("heartbeat_interval", 5.0),
        )
        self._leader_election = LeaderElection(
            self._node_id,
            ttl_seconds=self._config.get("leader_ttl", 30.0),
        )
        self._autoscaler = AutoScaler(
            min_instances=self._config.get("min_instances", 1),
            max_instances=self._config.get("max_instances", 10),
        )
        self._chaos = ChaosController(enabled=self._config.get("chaos_enabled", False))

        # Connection pools will be initialized on demand
        self._redis_pool: ConnectionPool | None = None
        self._neo4j_pool: ConnectionPool | None = None

        _logger.info(f"PerformanceManager initialized for node {self._node_id}")

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-007"])
    async def start(self) -> float:
        """Start performance manager and return startup time (REQ-001-US-010-007).

        Returns:
            Startup time in seconds
        """
        start_time = time.monotonic()

        self._started_at = _utc_now()
        self._uptime.start()

        await self._cluster.start()
        await self._leader_election.try_acquire_leadership()

        startup_seconds = time.monotonic() - start_time
        _logger.info(f"PerformanceManager started in {startup_seconds:.3f}s")

        return startup_seconds

    async def stop(self) -> None:
        """Stop performance manager."""
        await self._cluster.stop()
        await self._leader_election.release_leadership()

        if self._redis_pool:
            await self._redis_pool.close()
        if self._neo4j_pool:
            await self._neo4j_pool.close()

        _logger.info("PerformanceManager stopped")

    # -------------------------------------------------------------------------
    # Latency Tracking (REQ-001-US-010-001 to REQ-001-US-010-004)
    # -------------------------------------------------------------------------

    def record_routing_latency(self, duration_ms: float, success: bool = True) -> None:
        """Record routing operation latency.

        Args:
            duration_ms: Duration in milliseconds
            success: Whether operation succeeded
        """
        self._routing_latency.record(duration_ms, success)

    def record_policy_latency(self, duration_ms: float, success: bool = True) -> None:
        """Record policy query latency.

        Args:
            duration_ms: Duration in milliseconds
            success: Whether operation succeeded
        """
        self._policy_latency.record(duration_ms, success)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-001"])
    def get_routing_p95(self) -> float | None:
        """Get routing p95 latency (REQ-001-US-010-001)."""
        return self._routing_latency.get_p95()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-002"])
    def get_routing_p99(self) -> float | None:
        """Get routing p99 latency (REQ-001-US-010-002)."""
        return self._routing_latency.get_p99()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-003"])
    def get_policy_p95(self) -> float | None:
        """Get policy p95 latency (REQ-001-US-010-003)."""
        return self._policy_latency.get_p95()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-004"])
    def get_policy_p99(self) -> float | None:
        """Get policy p99 latency (REQ-001-US-010-004)."""
        return self._policy_latency.get_p99()

    # -------------------------------------------------------------------------
    # Throughput Tracking (REQ-001-US-010-005)
    # -------------------------------------------------------------------------

    def record_message(self, count: int = 1) -> None:
        """Record message(s) processed.

        Args:
            count: Number of messages
        """
        self._throughput.record(count)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-005"])
    def get_throughput(self) -> float:
        """Get current throughput (REQ-001-US-010-005).

        Returns:
            Messages per second
        """
        return self._throughput.get_rate()

    # -------------------------------------------------------------------------
    # Memory Tracking (REQ-001-US-010-006)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-006"])
    def record_session_memory(self, session_id: str, bytes_used: int) -> None:
        """Record session memory usage.

        Args:
            session_id: Session identifier
            bytes_used: Bytes used by session
        """
        self._memory.record_session(session_id, bytes_used)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-006"])
    def get_session_memory_mb(self, session_id: str) -> float | None:
        """Get session memory in MB (REQ-001-US-010-006)."""
        return self._memory.get_session_mb(session_id)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-006"])
    def get_average_session_memory_mb(self) -> float | None:
        """Get average session memory in MB."""
        return self._memory.get_average_mb()

    # -------------------------------------------------------------------------
    # Uptime Tracking (REQ-001-US-010-009)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-009"])
    def mark_system_down(self) -> None:
        """Mark system as down."""
        self._uptime.mark_down()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-009"])
    def mark_system_up(self) -> None:
        """Mark system as up."""
        self._uptime.mark_up()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-009"])
    def get_uptime_percent(self) -> float:
        """Get uptime percentage (REQ-001-US-010-009)."""
        return self._uptime.get_uptime_percent()

    # -------------------------------------------------------------------------
    # Caching (REQ-001-US-010-011)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-011"])
    def cache_get(self, key: str) -> Any | None:
        """Get value from query cache.

        Args:
            key: Cache key

        Returns:
            Cached value or None
        """
        return self._query_cache.get(key)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-011"])
    def cache_put(self, key: str, value: Any, ttl: float | None = None) -> None:
        """Put value in query cache.

        Args:
            key: Cache key
            value: Value to cache
            ttl: Optional TTL in seconds
        """
        self._query_cache.put(key, value, ttl)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-011"])
    def cache_invalidate(self, key: str) -> bool:
        """Invalidate cache entry.

        Args:
            key: Cache key

        Returns:
            True if entry was found and removed
        """
        return self._query_cache.invalidate(key)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-011"])
    def get_cache_stats(self) -> dict[str, Any]:
        """Get cache statistics."""
        return self._query_cache.get_stats()

    # -------------------------------------------------------------------------
    # Backpressure (REQ-001-US-010-020)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-020"])
    def update_load(self, load: float) -> None:
        """Update current system load.

        Args:
            load: Load level (0-1)
        """
        self._backpressure.update_load(load)

    @traces(
        spec="001", story="US-010", reqs=["REQ-001-US-010-010", "REQ-001-US-010-020"]
    )
    def should_accept_request(self, priority: int = 0) -> bool:
        """Check if request is accepted (REQ-001-US-010-010, REQ-001-US-010-020).

        Args:
            priority: Request priority

        Returns:
            True if request is accepted
        """
        return self._backpressure.should_accept(priority)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-020"])
    def get_backpressure_state(self) -> BackpressureController.State:
        """Get current backpressure state."""
        return self._backpressure.get_state()

    # -------------------------------------------------------------------------
    # Cluster Management (REQ-010-021 to REQ-010-027)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-022"])
    async def get_node_for_session(self, session_id: str) -> str | None:
        """Get node for session (REQ-001-US-010-022).

        Args:
            session_id: Session identifier

        Returns:
            Node identifier
        """
        return await self._cluster.get_node_for_session(session_id)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-023"])
    async def get_node_for_volk(self, volk_alias: str) -> str | None:
        """Get node for Volk (REQ-001-US-010-023).

        Args:
            volk_alias: Volk alias

        Returns:
            Node identifier
        """
        return await self._cluster.get_node_for_volk(volk_alias)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-025"])
    def is_leader(self) -> bool:
        """Check if this node is leader (REQ-001-US-010-025)."""
        return self._leader_election.is_leader()

    async def get_cluster_stats(self) -> dict[str, Any]:
        """Get cluster statistics."""
        return await self._cluster.get_cluster_stats()

    # -------------------------------------------------------------------------
    # Autoscaling (REQ-001-US-010-028, REQ-001-US-010-029)
    # -------------------------------------------------------------------------

    @traces(
        spec="001", story="US-010", reqs=["REQ-001-US-010-028", "REQ-001-US-010-029"]
    )
    def evaluate_scaling(self) -> ScalingDecision:
        """Evaluate and make scaling decision (REQ-001-US-010-028, REQ-001-US-010-029).

        Returns:
            Scaling decision
        """
        metrics = {
            "avg_load": self._backpressure.get_load(),
            "throughput": self._throughput.get_rate(),
            "routing_p95": self._routing_latency.get_p95() or 0,
            "session_count": self._memory.get_session_count(),
        }
        return self._autoscaler.evaluate(metrics)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-029"])
    def get_scaling_decisions(
        self, count: int = AUTOSCALER_RECENT_DECISIONS_DEFAULT_COUNT
    ) -> list[ScalingDecision]:
        """Get recent scaling decisions."""
        return self._autoscaler.get_recent_decisions(count)

    # -------------------------------------------------------------------------
    # Chaos Testing (REQ-001-US-010-030)
    # -------------------------------------------------------------------------

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-030"])
    def enable_chaos_testing(self) -> None:
        """Enable chaos testing."""
        self._chaos.enable()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-030"])
    def disable_chaos_testing(self) -> None:
        """Disable chaos testing."""
        self._chaos.disable()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-030"])
    def add_chaos_fault(
        self, fault_id: str, config: ChaosController.FaultConfig
    ) -> None:
        """Add chaos fault configuration.

        Args:
            fault_id: Fault identifier
            config: Fault configuration
        """
        self._chaos.add_fault(fault_id, config)

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-030"])
    async def maybe_inject_chaos(
        self,
        fault_type: ChaosController.FaultType,
        component: str | None = None,
    ) -> bool:
        """Possibly inject chaos fault.

        Args:
            fault_type: Type of fault
            component: Target component

        Returns:
            True if fault was injected
        """
        return await self._chaos.maybe_inject(fault_type, component)

    # -------------------------------------------------------------------------
    # SLA Compliance
    # -------------------------------------------------------------------------

    def check_sla_compliance(self) -> dict[SLATarget, bool]:
        """Check SLA compliance for all targets.

        Returns:
            Dict of target to compliance status
        """
        results: dict[SLATarget, bool] = {}
        self._check_latency_slas(results)
        self._check_throughput_and_resource_slas(results)
        return results

    def _check_latency_slas(self, results: dict[SLATarget, bool]) -> None:
        """Populate results with routing and policy latency SLA checks."""
        p95 = self._routing_latency.get_p95()
        if p95 is not None:
            results[SLATarget.ROUTING_LATENCY_P95] = DEFAULT_SLA_THRESHOLDS[
                SLATarget.ROUTING_LATENCY_P95
            ].is_met(p95)

        p99 = self._routing_latency.get_p99()
        if p99 is not None:
            results[SLATarget.ROUTING_LATENCY_P99] = DEFAULT_SLA_THRESHOLDS[
                SLATarget.ROUTING_LATENCY_P99
            ].is_met(p99)

        policy_p95 = self._policy_latency.get_p95()
        if policy_p95 is not None:
            results[SLATarget.POLICY_LATENCY_P95] = DEFAULT_SLA_THRESHOLDS[
                SLATarget.POLICY_LATENCY_P95
            ].is_met(policy_p95)

        policy_p99 = self._policy_latency.get_p99()
        if policy_p99 is not None:
            results[SLATarget.POLICY_LATENCY_P99] = DEFAULT_SLA_THRESHOLDS[
                SLATarget.POLICY_LATENCY_P99
            ].is_met(policy_p99)

    def _check_throughput_and_resource_slas(
        self, results: dict[SLATarget, bool]
    ) -> None:
        """Populate results with throughput, memory, and uptime SLA checks."""
        throughput = self._throughput.get_rate()
        results[SLATarget.THROUGHPUT] = DEFAULT_SLA_THRESHOLDS[
            SLATarget.THROUGHPUT
        ].is_met(throughput)

        avg_memory = self._memory.get_average_mb()
        if avg_memory is not None:
            results[SLATarget.MEMORY_PER_SESSION] = DEFAULT_SLA_THRESHOLDS[
                SLATarget.MEMORY_PER_SESSION
            ].is_met(avg_memory)

        uptime = self._uptime.get_uptime_percent()
        results[SLATarget.UPTIME] = DEFAULT_SLA_THRESHOLDS[SLATarget.UPTIME].is_met(
            uptime
        )

    # -------------------------------------------------------------------------
    # Comprehensive Stats
    # -------------------------------------------------------------------------

    def get_all_stats(self) -> dict[str, Any]:
        """Get comprehensive performance statistics.

        Returns:
            Dict with all performance metrics
        """
        return {
            "node_id": self._node_id,
            "started_at": self._started_at.isoformat() if self._started_at else None,
            "is_leader": self._leader_election.is_leader(),
            "routing_latency": self._routing_latency.get_stats(),
            "policy_latency": self._policy_latency.get_stats(),
            "throughput": {
                "rate": self._throughput.get_rate(),
                "total": self._throughput.get_total(),
            },
            "memory": {
                "total_mb": self._memory.get_total_mb(),
                "average_mb": self._memory.get_average_mb(),
                "session_count": self._memory.get_session_count(),
            },
            "uptime_percent": self._uptime.get_uptime_percent(),
            "cache": self._query_cache.get_stats(),
            "backpressure": self._backpressure.get_stats(),
            "chaos": self._chaos.get_stats(),
        }