orchestrata_ultimate.kernel.performance.scaling

Reference › orchestrata_ultimate › kernel › performance › scaling

The autoscaler of the performance package, SPEC-001 requirements REQ-001-US-010-028 and -029. AutoScaler decides, from a metrics dictionary, whether the number of kernel instances should go up, down or stay. manager calls it through evaluate_scaling. It only produces decisions and a count; it does not start or stop any process.

evaluate takes the metrics and returns a ScalingDecision with a timestamp, an action of scale_up, scale_down or none, the current and target instance counts, a reason and the metrics. _check_cooldown returns a no-op decision while the cooldown after the last action is running. Otherwise _compute_scaling_decision reads avg_load (0.5 if absent): at or above the scale-up threshold and below the maximum it adds one instance, and at or below the scale-down threshold and above the minimum it removes one. The minimum, maximum, thresholds and cooldown are constructor arguments with defaults from _common. get_current_instances, set_current_instances and get_recent_decisions read and set the state.

At a glance

Class diagram

Dependency diagram

API

Classes

ScalingDecision (dataclass)

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

Autoscaling decision (REQ-001-US-010-028).

AutoScaler

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

Autoscaling controller (REQ-001-US-010-028, REQ-001-US-010-029).

Source

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

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

import collections
import threading
from dataclasses import dataclass
from datetime import datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel.performance._common import (
    AUTOSCALER_DECISIONS_HISTORY_SIZE,
    AUTOSCALER_DEFAULT_COOLDOWN_SECONDS,
    AUTOSCALER_DEFAULT_MAX_INSTANCES,
    AUTOSCALER_DEFAULT_SCALE_DOWN_THRESHOLD,
    AUTOSCALER_DEFAULT_SCALE_UP_THRESHOLD,
    AUTOSCALER_RECENT_DECISIONS_DEFAULT_COUNT,
    _utc_now,
)


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-028"])
@dataclass
class ScalingDecision:
    """Autoscaling decision (REQ-001-US-010-028)."""

    timestamp: datetime
    action: str  # "scale_up", "scale_down", "none"
    current_instances: int
    target_instances: int
    reason: str
    metrics: dict[str, float]


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-028", "REQ-001-US-010-029"])
class AutoScaler:
    """Autoscaling controller (REQ-001-US-010-028, REQ-001-US-010-029).

    Makes scaling decisions based on metrics.
    """

    def __init__(
        self,
        min_instances: int = 1,
        max_instances: int = AUTOSCALER_DEFAULT_MAX_INSTANCES,
        scale_up_threshold: float = AUTOSCALER_DEFAULT_SCALE_UP_THRESHOLD,
        scale_down_threshold: float = AUTOSCALER_DEFAULT_SCALE_DOWN_THRESHOLD,
        cooldown_seconds: float = AUTOSCALER_DEFAULT_COOLDOWN_SECONDS,
    ) -> None:
        """Initialize autoscaler.

        Args:
            min_instances: Minimum number of instances
            max_instances: Maximum number of instances
            scale_up_threshold: Load threshold for scaling up
            scale_down_threshold: Load threshold for scaling down
            cooldown_seconds: Cooldown between scaling actions
        """
        self._min_instances = min_instances
        self._max_instances = max_instances
        self._scale_up_threshold = scale_up_threshold
        self._scale_down_threshold = scale_down_threshold
        self._cooldown_seconds = cooldown_seconds

        self._current_instances = min_instances
        self._last_scaling_action: datetime | None = None
        self._decisions: collections.deque[ScalingDecision] = collections.deque(
            maxlen=AUTOSCALER_DECISIONS_HISTORY_SIZE
        )
        self._lock = threading.Lock()

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

        Args:
            metrics: Current system metrics

        Returns:
            Scaling decision
        """
        with self._lock:
            now = _utc_now()

            cooldown_decision = self._check_cooldown(now, metrics)
            if cooldown_decision is not None:
                self._decisions.append(cooldown_decision)
                return cooldown_decision

            decision = self._compute_scaling_decision(now, metrics)
            self._decisions.append(decision)
            return decision

    @traces(
        spec="001", story="US-010", reqs=["REQ-001-US-010-028", "REQ-001-US-010-029"]
    )
    def _check_cooldown(
        self, now: datetime, metrics: dict[str, float]
    ) -> ScalingDecision | None:
        """Return a no-op decision if still in cooldown, else None."""
        if not self._last_scaling_action:
            return None
        elapsed = (now - self._last_scaling_action).total_seconds()
        if elapsed < self._cooldown_seconds:
            return ScalingDecision(
                timestamp=now,
                action="none",
                current_instances=self._current_instances,
                target_instances=self._current_instances,
                reason=f"In cooldown ({self._cooldown_seconds - elapsed:.0f}s remaining)",
                metrics=metrics,
            )
        return None

    @traces(
        spec="001", story="US-010", reqs=["REQ-001-US-010-028", "REQ-001-US-010-029"]
    )
    def _compute_scaling_decision(
        self, now: datetime, metrics: dict[str, float]
    ) -> ScalingDecision:
        """Determine scale-up, scale-down, or no-op based on avg_load.

        Mutates _current_instances and _last_scaling_action when scaling occurs.
        Caller must hold self._lock.
        """
        avg_load = metrics.get("avg_load", 0.5)

        if (
            avg_load >= self._scale_up_threshold
            and self._current_instances < self._max_instances
        ):
            previous = self._current_instances
            target = min(self._current_instances + 1, self._max_instances)
            self._current_instances = target
            self._last_scaling_action = now
            return ScalingDecision(
                timestamp=now,
                action="scale_up",
                current_instances=previous,
                target_instances=target,
                reason=f"Load {avg_load:.1%} exceeds threshold {self._scale_up_threshold:.1%}",
                metrics=metrics,
            )

        if (
            avg_load <= self._scale_down_threshold
            and self._current_instances > self._min_instances
        ):
            previous = self._current_instances
            target = max(self._current_instances - 1, self._min_instances)
            self._current_instances = target
            self._last_scaling_action = now
            return ScalingDecision(
                timestamp=now,
                action="scale_down",
                current_instances=previous,
                target_instances=target,
                reason=f"Load {avg_load:.1%} below threshold {self._scale_down_threshold:.1%}",
                metrics=metrics,
            )

        return ScalingDecision(
            timestamp=now,
            action="none",
            current_instances=self._current_instances,
            target_instances=self._current_instances,
            reason=f"Load {avg_load:.1%} within acceptable range",
            metrics=metrics,
        )

    def get_current_instances(self) -> int:
        """Get current instance count."""
        return self._current_instances

    def set_current_instances(self, count: int) -> None:
        """Set current instance count."""
        with self._lock:
            self._current_instances = max(
                self._min_instances, min(count, self._max_instances)
            )

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