orchestrata_ultimate.kernel.performance.sla

Reference › orchestrata_ultimate › kernel › performance › sla

The service-level targets of the kernel, SPEC-001 story US-010 (REQ-001-US-010-001 to -010). SLATarget is an enum of the targets: routing latency p95 and p99, policy latency p95 and p99, throughput, memory per session, startup time, concurrent volks, uptime and graceful degradation. The comments beside the members record the intended figures: routing under 50 ms at p95 and 100 ms at p99, policy under 10 ms and 20 ms, 10,000 messages per second, under 5 MB per session, startup under 2 seconds, 1000 or more concurrent volks and 99.9% uptime. SLAThreshold pairs a target with a threshold value, a unit and a comparison (lt, lte, gt, gte); is_met applies the comparison to an actual value and returns False for an unknown comparison. manager uses them in check_sla_compliance.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SLATarget (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-010', reqs=['REQ-001-US-010-001', 'REQ-001-US-010-002', 'REQ-001-US-010-003', 'REQ-001-US-… · Line 26

SLA target types.

SLAThreshold (dataclass)

Decorators: dataclass · Line 42

SLA threshold configuration.

Source

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

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

from dataclasses import dataclass
from enum import Enum

from orchestrata_ultimate.core.base.traces import traces


@traces(
    spec="001",
    story="US-010",
    reqs=[
        "REQ-001-US-010-001",
        "REQ-001-US-010-002",
        "REQ-001-US-010-003",
        "REQ-001-US-010-004",
        "REQ-001-US-010-005",
        "REQ-001-US-010-006",
        "REQ-001-US-010-007",
        "REQ-001-US-010-008",
        "REQ-001-US-010-009",
        "REQ-001-US-010-010",
    ],
)
class SLATarget(Enum):
    """SLA target types."""

    ROUTING_LATENCY_P95 = "routing_latency_p95"  # REQ-001-US-010-001: <50ms
    ROUTING_LATENCY_P99 = "routing_latency_p99"  # REQ-001-US-010-002: <100ms
    POLICY_LATENCY_P95 = "policy_latency_p95"  # REQ-001-US-010-003: <10ms
    POLICY_LATENCY_P99 = "policy_latency_p99"  # REQ-001-US-010-004: <20ms
    THROUGHPUT = "throughput"  # REQ-001-US-010-005: 10k msg/sec
    MEMORY_PER_SESSION = "memory_per_session"  # REQ-001-US-010-006: <5MB
    STARTUP_TIME = "startup_time"  # REQ-001-US-010-007: <2s
    CONCURRENT_VOLKS = "concurrent_volks"  # REQ-001-US-010-008: 1000+
    UPTIME = "uptime"  # REQ-001-US-010-009: 99.9%
    GRACEFUL_DEGRADATION = "graceful_degradation"  # REQ-001-US-010-010


@dataclass
class SLAThreshold:
    """SLA threshold configuration."""

    target: SLATarget
    threshold_value: float
    unit: str
    comparison: str  # "lt", "gt", "lte", "gte"

    def is_met(self, actual_value: float) -> bool:
        """Check if SLA is met."""
        if self.comparison == "lt":
            return actual_value < self.threshold_value
        elif self.comparison == "lte":
            return actual_value <= self.threshold_value
        elif self.comparison == "gt":
            return actual_value > self.threshold_value
        elif self.comparison == "gte":
            return actual_value >= self.threshold_value
        return False


DEFAULT_SLA_THRESHOLDS: dict[SLATarget, SLAThreshold] = {
    SLATarget.ROUTING_LATENCY_P95: SLAThreshold(
        SLATarget.ROUTING_LATENCY_P95, 50.0, "ms", "lt"
    ),
    SLATarget.ROUTING_LATENCY_P99: SLAThreshold(
        SLATarget.ROUTING_LATENCY_P99, 100.0, "ms", "lt"
    ),
    SLATarget.POLICY_LATENCY_P95: SLAThreshold(
        SLATarget.POLICY_LATENCY_P95, 10.0, "ms", "lt"
    ),
    SLATarget.POLICY_LATENCY_P99: SLAThreshold(
        SLATarget.POLICY_LATENCY_P99, 20.0, "ms", "lt"
    ),
    SLATarget.THROUGHPUT: SLAThreshold(SLATarget.THROUGHPUT, 10000.0, "msg/sec", "gte"),
    SLATarget.MEMORY_PER_SESSION: SLAThreshold(
        SLATarget.MEMORY_PER_SESSION, 5.0, "MB", "lt"
    ),
    SLATarget.STARTUP_TIME: SLAThreshold(SLATarget.STARTUP_TIME, 2.0, "sec", "lt"),
    SLATarget.CONCURRENT_VOLKS: SLAThreshold(
        SLATarget.CONCURRENT_VOLKS, 1000.0, "count", "gte"
    ),
    SLATarget.UPTIME: SLAThreshold(SLATarget.UPTIME, 99.9, "percent", "gte"),
}