diff --git a/CLAUDE.md b/CLAUDE.md index 8ffc0e5..5a9a5df 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -454,7 +454,7 @@ of the three may veto a candidate destination. The cache is invalidated each engine cycle. -Future: NUMA, CPU feature flags, flavor extra specs, soft-rule +Future: NUMA, flavor extra specs, soft-rule penalties in the planner - https://docs.openstack.org/nova/latest/user/server-groups.html @@ -798,3 +798,28 @@ Engine notification listener updates cooldown/quarantine state 4. Hard timeout on migration polling 5. Post-flight verification (host + status) 6. Idempotent: pre-flight catches duplicate or stale tasks + + +## CPU compatibility gate + +`[engine] require_cpu_compatibility` defaults to false and is independent +of the Placement claims gate. `PlacementClient.fetch_cpu_traits(hosts)` +reads `HW_CPU_*` traits using Placement microversion 1.6. The engine +resolves aggregate/AZ scope before fetching the union once per cycle. + +`ConstraintChecker` requires source traits to be a subset of destination +traits after availability and claims checks, before server-group checks. +All four movers inherit the check. Pair verdicts and trait data reset +each cycle. Missing data fails closed. A known empty source set warns +and passes when the destination also has known data. + +Both snapshot entry points capture `placement/cpu_traits.json` when +enabled. `ReplayPlacementClient` reads it without network access. +Missing or invalid snapshots fail closed. Claims remain unrecorded, +so CPU replay requires `enforce_placement_claims = false`. + +The host-level rule is approximate. `HW_CPU_*` includes hyperthreading +as well as mapped instruction flags. Current host configuration can +differ from a running guest's retained CPU model. Nova remains the +authoritative migration compatibility check. See the configuration +reference for behavior and limitations. diff --git a/docs/configuration/index.rst b/docs/configuration/index.rst index 06db788..6a1293c 100644 --- a/docs/configuration/index.rst +++ b/docs/configuration/index.rst @@ -43,3 +43,56 @@ commented example. Load-time invariants: - ``imbalance_query`` must return per-host values in [0, 1] - enforced at runtime by the scorer, which skips the policy for the cycle (with an error logged) on out-of-range data. + + +CPU compatibility +----------------- + +Set ``[engine] require_cpu_compatibility = true`` to require a +destination's Placement ``HW_CPU_*`` traits to contain every CPU trait +reported for the source host. The default is false. This option is +independent of ``enforce_placement_claims``. + +The shared constraint checker applies this rule to spread, pack, +disabled-host evacuation, and affinity repair. Checks run after host +availability and capacity claims, before server-group checks. A DEBUG +message names the rejected host pair and missing traits. + +The engine lists Placement providers and reads traits once per in-scope +host per cycle, deduplicating overlapping aggregates and filtering to +the configured availability zone. The service user needs +``placement:resource_providers:list`` and +``placement:resource_providers:traits:list`` access. The traits endpoint +uses Placement microversion 1.6. Provider names must match the Nova host +names used by the engine. Missing source or destination data blocks a +move. Any traits fetch failure blocks all moves for that cycle. + +A successful response containing no ``HW_CPU_*`` traits is different +from unavailable data. An empty source set permits moves to destinations +with known data and logs a warning once per cycle that the check is +blind for that source. + +This is a host-level approximation, not a per-instance CPU model check. +Nova's libvirt driver reports host features for host-model and +host-passthrough, and a union of configured models for custom mode. +The prefix also includes ``HW_CPU_HYPERTHREADING``, so the check can +reject moves for more than instruction-set differences. It can reject +moves Nova would accept and miss incompatibilities involving unmapped +flags or guests retaining an older CPU model after reconfiguration. +Nova's migration pre-check remains authoritative. + +Producer source: +`Nova libvirt CPU trait reporting `_. +API source: +`Placement provider traits `_. + +When the option is enabled, both ``kronos-record`` and engine SIGUSR1 +snapshots write ``placement/cpu_traits.json``, a mapping from host names +to sorted trait lists. Failed collection writes an empty mapping. +Replay reads this file without contacting Placement. Missing files in +older snapshots fail closed when CPU checking is enabled. + +For CPU compatibility replay, enable ``require_cpu_compatibility`` and +set ``enforce_placement_claims = false``. Capacity claims are not +recorded by the existing snapshot format. Leaving that gate enabled +pauses replay planning with an explicit error in the logs. diff --git a/docs/configuration/kronos.conf.sample b/docs/configuration/kronos.conf.sample index 5caa5e8..be77ceb 100644 --- a/docs/configuration/kronos.conf.sample +++ b/docs/configuration/kronos.conf.sample @@ -292,6 +292,11 @@ # aggregate. (boolean value) #enforce_placement_disk = false +# Require destination Placement HW_CPU_* traits to contain all source traits. +# Missing data blocks moves. Empty source sets warn and pass. Independent of +# enforce_placement_claims. Nova still checks migrations. (boolean value) +#require_cpu_compatibility = false + [executor] @@ -387,6 +392,21 @@ # Log requests to multiple loggers. (boolean value) #split_loggers = false +# An OpenSSL cipher string to set the allowed ciphers for TLS connections. If +# not specified, the default ciphers from the underlying OpenSSL library are +# used. (string value) +#tls_ciphers = + +# The minimum TLS version to require for TLS connections. When not specified, +# the behavior is undefined and defers to the system cryptographic policy. +# (string value) +# Possible values: +# - None - the absence of a setting, defers to the system cryptographic +# policy +# 1.2 - TLS 1.2 - enforce minimum TLS version 1.2 +# 1.3 - TLS 1.3 - enforce minimum TLS version 1.3 +#tls_min_version = + # The default service_type for endpoint URL discovery. (string value) #service_type = @@ -584,8 +604,7 @@ # (boolean value) # This option is deprecated for removal. # Its value may be silently ignored in the future. -# Reason: Hostname verification should remain enabled once operators have -# completed the migration. +# Reason: Verification is always enabled now #ssl_enforce_hostname_verification = true # DEPRECATED: Global toggle for enforcing the OpenSSL FIPS mode. This feature diff --git a/etc/kronos/kronos.conf.sample b/etc/kronos/kronos.conf.sample index f5b2050..3ee88d5 100644 --- a/etc/kronos/kronos.conf.sample +++ b/etc/kronos/kronos.conf.sample @@ -2,6 +2,7 @@ # debug = false [engine] +require_cpu_compatibility = false # Seconds between policy evaluation cycles. # evaluation_interval = 60 diff --git a/kronos/clients/placement.py b/kronos/clients/placement.py index 859a18f..1fb6178 100644 --- a/kronos/clients/placement.py +++ b/kronos/clients/placement.py @@ -206,3 +206,29 @@ def _fetch_usages(self, rp_id: str) -> dict[str, int]: for rc, used in raw.items() if rc in TRACKED_RESOURCE_CLASSES } + + def fetch_cpu_traits(self, hosts: set[str]) -> dict[str, frozenset[str]]: + if not hosts: + return {} + try: + traits: dict[str, frozenset[str]] = {} + for rp in self._placement.resource_providers(): + if rp.name not in hosts: + continue + response = self._placement.get( + f"/resource_providers/{rp.id}/traits", microversion="1.6", + ) + response.raise_for_status() + raw = response.json()["traits"] + if not isinstance(raw, list) or not all( + isinstance(trait, str) for trait in raw + ): + raise ValueError(f"Invalid CPU traits response for {rp.name}") + traits[rp.name] = frozenset( + trait for trait in raw if trait.startswith("HW_CPU_") + ) + return traits + except Exception as exc: + raise PlacementClientError( + reason=f"Failed to fetch CPU traits: {exc}", + ) from exc diff --git a/kronos/cmd/record.py b/kronos/cmd/record.py index 2306b7c..fd94ba4 100644 --- a/kronos/cmd/record.py +++ b/kronos/cmd/record.py @@ -15,6 +15,7 @@ from oslo_log import log as logging from kronos.clients.nova import NovaClient +from kronos.clients.placement import PlacementClient from kronos.clients.prometheus import PrometheusClient from kronos.common.config import register_opts from kronos.common.snapshot import write_snapshot @@ -72,8 +73,12 @@ def main() -> int: nova = NovaClient(CONF) prometheus = PrometheusClient(CONF) + placement = ( + PlacementClient(CONF) if CONF.engine.require_cpu_compatibility else None + ) target = write_snapshot( parent_dir, nova, prometheus, policies, aggregate_names, + placement=placement, ) LOG.info("Snapshot written to %s", target) diff --git a/kronos/cmd/replay.py b/kronos/cmd/replay.py index 64d3ec7..9f63b58 100644 --- a/kronos/cmd/replay.py +++ b/kronos/cmd/replay.py @@ -22,9 +22,10 @@ from oslo_log import log as logging from kronos.clients.nova import ComputeService, Instance +from kronos.clients.placement import PlacementClient, ProviderSnapshot from kronos.clients.prometheus import PrometheusHealth, QueryResult from kronos.common.config import register_opts -from kronos.common.exceptions import AggregateNotFound +from kronos.common.exceptions import AggregateNotFound, PlacementClientError from kronos.common.messaging import UNASSIGNED_TOPIC_MARKER from kronos.engine.cooldown import CooldownTracker from kronos.engine.loop import EngineLoop @@ -295,6 +296,7 @@ def main() -> int: prometheus=prometheus, # type: ignore[arg-type] cooldown=cooldown, timings=timings, + placement=ReplayPlacementClient(snapshot_dir), ) started = time.perf_counter() @@ -307,3 +309,32 @@ def main() -> int: _log_timings(timings, total) return 0 + + +class ReplayPlacementClient(PlacementClient): + def __init__(self, snapshot_dir: Path) -> None: + self._traits_path = snapshot_dir / "placement" / "cpu_traits.json" + + def fetch_cpu_traits(self, hosts: set[str]) -> dict[str, frozenset[str]]: + if not self._traits_path.exists(): + raise PlacementClientError(reason="Snapshot has no CPU traits") + raw = json.loads(self._traits_path.read_text()) + if not isinstance(raw, dict): + raise PlacementClientError(reason="Invalid CPU traits snapshot") + traits: dict[str, frozenset[str]] = {} + for host, flags in raw.items(): + if not isinstance(flags, list) or not all( + isinstance(flag, str) for flag in flags + ): + raise PlacementClientError(reason=f"Invalid CPU traits for {host}") + if host in hosts: + traits[host] = frozenset( + flag for flag in flags if flag.startswith("HW_CPU_") + ) + return traits + + def fetch_snapshots(self) -> dict[str, ProviderSnapshot]: + raise PlacementClientError( + reason="Placement claims are not recorded. Disable " + "enforce_placement_claims for offline CPU compatibility replay.", + ) diff --git a/kronos/common/config.py b/kronos/common/config.py index 6a8a631..81b88d4 100644 --- a/kronos/common/config.py +++ b/kronos/common/config.py @@ -250,6 +250,15 @@ "identical across all hosts in the aggregate." ), ), + cfg.BoolOpt( + "require_cpu_compatibility", + default=False, + help=( + "Require destination Placement HW_CPU_* traits to contain all source " + "traits. Missing data blocks moves. Empty source sets warn and pass. " + "Independent of enforce_placement_claims. Nova still checks migrations." + ), + ), ] prometheus_opts: list[cfg.Opt] = [ diff --git a/kronos/common/snapshot.py b/kronos/common/snapshot.py index fc935c7..28c3e84 100644 --- a/kronos/common/snapshot.py +++ b/kronos/common/snapshot.py @@ -27,6 +27,7 @@ from oslo_log import log as logging from kronos.clients.nova import NovaClient +from kronos.clients.placement import PlacementClient from kronos.clients.prometheus import PrometheusClient from kronos.common.messaging import UNASSIGNED_TOPIC_MARKER from kronos.policies.models import PoliciesConfig @@ -43,6 +44,8 @@ def write_snapshot( prometheus: PrometheusClient, policies: PoliciesConfig, aggregate_names: list[str | None], + *, + placement: PlacementClient | None = None, ) -> Path: """Write a complete snapshot under ``parent_dir`` and return its path. @@ -77,7 +80,9 @@ def write_snapshot( ), ) - _write_nova(nova, aggregate_names, target) + hosts = _write_nova(nova, aggregate_names, target) + if placement is not None: + _write_cpu_traits(placement, hosts, target) _write_prometheus(prometheus, policies, target) (target / "cooldowns.json").write_text( @@ -98,7 +103,7 @@ def _write_nova( nova: NovaClient, aggregate_names: list[str | None], output_dir: Path, -) -> None: +) -> set[str]: nova_dir = output_dir / "nova" nova_dir.mkdir() @@ -163,6 +168,8 @@ def _write_nova( LOG.error("Failed to list compute services", exc_info=True) (nova_dir / "services.json").write_text("[]") + return all_hosts + def _write_prometheus( prometheus: PrometheusClient, @@ -234,3 +241,18 @@ def _write_prometheus( policy.name, exc_info=True, ) + + +def _write_cpu_traits( + placement: PlacementClient, hosts: set[str], output_dir: Path, +) -> None: + directory = output_dir / "placement" + directory.mkdir() + try: + traits = placement.fetch_cpu_traits(hosts) + except Exception: + LOG.error("Failed to record CPU traits", exc_info=True) + traits = {} + (directory / "cpu_traits.json").write_text( + json.dumps({host: sorted(flags) for host, flags in traits.items()}, indent=2), + ) diff --git a/kronos/engine/constraints.py b/kronos/engine/constraints.py index dffe346..f304064 100644 --- a/kronos/engine/constraints.py +++ b/kronos/engine/constraints.py @@ -104,13 +104,17 @@ class ConstraintChecker: Future: - NUMA topology - - CPU feature flags - Flavor extra specs / traits - Promote soft rules to planner-side penalties instead of vetoes """ - def __init__(self, nova: NovaClient) -> None: + def __init__( + self, nova: NovaClient, *, require_cpu_compatibility: bool = False, + ) -> None: self._nova = nova + self._require_cpu_compatibility = require_cpu_compatibility + self._cpu_traits: dict[str, frozenset[str]] = {} + self._cpu_compatibility: dict[tuple[str, str], bool] = {} # Lazy-loaded cache; populated on first check() in a cycle. self._groups: list[ServerGroup] | None = None # Per-cycle nova-compute service map (host -> ComputeService). @@ -209,6 +213,9 @@ def check( ): return False + if not self._check_cpu_compatibility(vm.host, dest_host): + return False + groups = self._get_groups() if not groups: return True @@ -382,5 +389,42 @@ def invalidate_cache(self) -> None: """ self._groups = None self._services = None + self._cpu_traits = {} + self._cpu_compatibility.clear() if self._placement_gate is not None: self._placement_gate.invalidate() + + def set_cpu_traits(self, traits: dict[str, frozenset[str]]) -> None: + self._cpu_traits = dict(traits) + self._cpu_compatibility.clear() + for host, features in traits.items(): + if not features: + LOG.warning( + "Host %s reports no HW_CPU_* traits. CPU compatibility " + "checking is blind for moves from this host.", host, + ) + + def _check_cpu_compatibility(self, source: str, destination: str) -> bool: + if not self._require_cpu_compatibility: + return True + pair = (source, destination) + if pair in self._cpu_compatibility: + return self._cpu_compatibility[pair] + source_traits = self._cpu_traits.get(source) + destination_traits = self._cpu_traits.get(destination) + if source_traits is None or destination_traits is None: + compatible = False + LOG.debug( + "CPU compatibility: %s -> %s rejected (missing host traits).", + source, destination, + ) + else: + missing = source_traits - destination_traits + compatible = not missing + if missing: + LOG.debug( + "CPU compatibility: %s -> %s rejected (missing traits: %s).", + source, destination, ", ".join(sorted(missing)), + ) + self._cpu_compatibility[pair] = compatible + return compatible diff --git a/kronos/engine/loop.py b/kronos/engine/loop.py index 2a99841..7748ae7 100644 --- a/kronos/engine/loop.py +++ b/kronos/engine/loop.py @@ -83,7 +83,10 @@ def __init__( self._nova = nova if nova is not None else NovaClient(conf) self._scorer = PolicyScorer(self._prometheus) self._profiler = VmProfiler(self._prometheus, self._nova) - self._constraints = ConstraintChecker(self._nova) + self._cpu_compatibility_enabled = bool(conf.engine.require_cpu_compatibility) + self._constraints = ConstraintChecker( + self._nova, require_cpu_compatibility=self._cpu_compatibility_enabled, + ) # Pluggable placement claims gate. When enabled, every # candidate destination is intersected with the placement # headroom (cpu/ram/disk allocation ratios applied) before @@ -94,7 +97,9 @@ def __init__( # affinity enforcer. self._placement_enabled = bool(conf.engine.enforce_placement_claims) self._placement: PlacementClient | None = placement - if self._placement_enabled and self._placement is None: + if ( + self._placement_enabled or self._cpu_compatibility_enabled + ) and self._placement is None: self._placement = PlacementClient(conf) self._placement_gate = PlacementGate( enabled=self._placement_enabled, @@ -470,11 +475,20 @@ def _run_cycle( ) enabled_policies = [p for p in policies.policies if p.enabled] + scoped_hosts = None + if self._cpu_compatibility_enabled: + scoped_hosts = self._resolve_cpu_scope(aggregates, services, report) + self._constraints.set_cpu_traits(self._fetch_cpu_traits({ + host for hosts in scoped_hosts.values() for host in hosts + })) for aggregate in aggregates: + if scoped_hosts is not None and aggregate not in scoped_hosts: + continue try: result = self._evaluate_aggregate( aggregate, enabled_policies, dry_run, services, + hosts=scoped_hosts.get(aggregate) if scoped_hosts is not None else None, ) report.aggregate_results.append(result) except Exception as exc: @@ -531,10 +545,12 @@ def _evaluate_aggregate( policies: list[PolicyConfig], dry_run: bool, services: dict[str, ComputeService], + hosts: list[str] | None = None, ) -> AggregateResult: """Run all policies against one aggregate and plan migrations.""" name = aggregate if aggregate is not None else "" - hosts = self._nova.get_hosts_in_aggregate(aggregate) + if hosts is None: + hosts = self._nova.get_hosts_in_aggregate(aggregate) result = AggregateResult(aggregate=name) @@ -911,6 +927,7 @@ def _write_engine_snapshot( self._prometheus, policies, aggregates, + placement=self._placement if self._cpu_compatibility_enabled else None, ) LOG.info("Engine snapshot written to %s", target) except Exception: @@ -919,3 +936,35 @@ def _write_engine_snapshot( configured, exc_info=True, ) + + def _fetch_cpu_traits(self, hosts: set[str]) -> dict[str, frozenset[str]]: + if self._placement is None: + return {} + try: + return self._placement.fetch_cpu_traits(hosts) + except Exception: + LOG.warning( + "Failed to fetch CPU traits. CPU compatibility gate will " + "reject every destination this cycle.", + exc_info=True, + ) + return {} + + def _resolve_cpu_scope( + self, + aggregates: list[str | None], + services: dict[str, ComputeService], + report: CycleReport, + ) -> dict[str | None, list[str]]: + scoped_hosts: dict[str | None, list[str]] = {} + for aggregate in aggregates: + name = aggregate if aggregate is not None else "" + try: + scoped_hosts[aggregate] = self._filter_hosts_to_az( + self._nova.get_hosts_in_aggregate(aggregate), services, name, + ) + except Exception as exc: + error_msg = f"Aggregate '{name}': {exc}" + LOG.error("Evaluation error: %s", error_msg) + report.errors.append(error_msg) + return scoped_hosts diff --git a/tests/unit/engine/test_cpu_compatibility.py b/tests/unit/engine/test_cpu_compatibility.py new file mode 100644 index 0000000..2c11b74 --- /dev/null +++ b/tests/unit/engine/test_cpu_compatibility.py @@ -0,0 +1,382 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import cast +from unittest.mock import MagicMock, patch + +import pytest +from oslo_config import cfg + +from kronos.clients.nova import ComputeService, Instance, NovaClient +from kronos.clients.placement import PlacementClient +from kronos.clients.prometheus import PrometheusClient, PrometheusHealth, QueryResult +from kronos.cmd.replay import ReplayNovaClient, ReplayPlacementClient, ReplayPrometheusClient +from kronos.common.config import register_opts +from kronos.common.exceptions import PlacementClientError +from kronos.common.snapshot import write_snapshot +from kronos.engine.constraints import ConstraintChecker +from kronos.engine.loop import EngineLoop +from kronos.engine.placement import PlacementGate +from kronos.engine.types import CycleReport, MigrationPlan, VmProfile +from kronos.policies.models import PoliciesConfig, PolicyConfig, PolicyMode, VmProfileLabelType + +AVX = "HW_CPU_X86_AVX" +AVX512 = "HW_CPU_X86_AVX512F" + + +def _services() -> dict[str, ComputeService]: + return { + host: ComputeService( + host=host, binary="nova-compute", state="up", + status="enabled", zone="nova", + ) + for host in ("h1", "h2") + } + + +def _checker(enabled: bool = True) -> ConstraintChecker: + nova = MagicMock() + nova.list_server_groups.return_value = [] + checker = ConstraintChecker(nova, require_cpu_compatibility=enabled) + checker.set_services(_services()) + return checker + + +def _vm() -> VmProfile: + return VmProfile(instance_uuid="v1", instance_name="vm", host="h1") + + +@pytest.mark.parametrize(("source", "destination", "expected"), [ + ({AVX}, {AVX, AVX512}, True), + ({AVX}, {AVX}, True), + ({AVX, AVX512}, {AVX}, False), + ({"HW_CPU_X86_AMD_SVM"}, {"HW_CPU_X86_INTEL_VMX"}, False), + (set(), {AVX}, True), + (set(), set(), True), + ({AVX}, set(), False), + (None, {AVX}, False), + ({AVX}, None, False), + (set(), None, False), + (None, None, False), +]) +def test_cpu_truth_table(source, destination, expected) -> None: + checker = _checker() + checker.set_cpu_traits({ + host: frozenset(flags) + for host, flags in (("h1", source), ("h2", destination)) + if flags is not None + }) + assert checker.check(_vm(), "h2", {}) is expected + + +def test_disabled_cpu_gate_accepts_missing_traits() -> None: + assert _checker(enabled=False).check(_vm(), "h2", {}) + + +def test_cpu_gate_fails_closed_before_install_and_after_invalidation() -> None: + checker = _checker() + assert not checker.check(_vm(), "h2", {}) + checker.set_cpu_traits({"h1": frozenset({AVX}), "h2": frozenset({AVX})}) + assert checker.check(_vm(), "h2", {}) + checker.invalidate_cache() + checker.set_services(_services()) + assert not checker.check(_vm(), "h2", {}) + checker.set_cpu_traits({"h1": frozenset({AVX}), "h2": frozenset({AVX})}) + assert checker.check(_vm(), "h2", {}) + + +def test_rejections_are_memoized_and_refresh_with_traits(caplog) -> None: + checker = _checker() + traits = {"h1": frozenset({AVX512}), "h2": frozenset({AVX})} + checker.set_cpu_traits(traits) + with caplog.at_level("DEBUG"): + assert not checker.check(_vm(), "h2", {}) + assert not checker.check(_vm(), "h2", {}) + assert sum(AVX512 in r.message for r in caplog.records) == 1 + traits["h2"] = traits["h1"] + checker.set_cpu_traits(traits) + assert checker.check(_vm(), "h2", {}) + + +def test_empty_host_warns_once_at_install(caplog) -> None: + checker = _checker() + with caplog.at_level("WARNING"): + checker.set_cpu_traits({"h1": frozenset(), "h2": frozenset({AVX})}) + assert checker.check(_vm(), "h2", {}) + assert checker.check(_vm(), "h2", {}) + assert sum("blind" in r.message for r in caplog.records) == 1 + + +def test_gate_order() -> None: + checker = _checker() + gate = MagicMock(spec=PlacementGate) + gate.is_destination_ok.return_value = False + checker.set_placement_gate(gate) + checker.set_services({}) + assert not checker.check(_vm(), "h2", {}) + gate.is_destination_ok.assert_not_called() + checker.set_services(_services()) + assert not checker.check(_vm(), "h2", {}) + assert not checker._cpu_compatibility + gate.is_destination_ok.return_value = True + with patch.object(checker, "_get_groups") as groups: + assert not checker.check(_vm(), "h2", {}) + groups.assert_not_called() + checker.set_cpu_traits({"h1": frozenset({AVX}), "h2": frozenset({AVX})}) + groups.return_value = [] + assert checker.check(_vm(), "h2", {}) + groups.assert_called_once() + + +def _scenario(phase: str): + conf = cfg.ConfigOpts() + register_opts(conf) + for name, value in { + "require_cpu_compatibility": True, + "enforce_placement_claims": False, + "cooldown": 0, + "instance_cooldown": 0, + "evacuate_disabled_hosts": phase == "evacuate", + "enforce_hard_affinity": phase == "affinity", + }.items(): + conf.set_override(name, value, group="engine") + nova = MagicMock(spec=NovaClient) + services = _services() + if phase == "evacuate": + services["h1"].status = "disabled" + nova.list_compute_services.return_value = list(services.values()) + nova.get_hosts_in_aggregate.return_value = ["h1", "h2"] + nova.list_compute_hosts.return_value = [] + nova.list_server_groups.return_value = ( + [{"id": "g1", "policies": ["anti-affinity"], "members": ["v1", "v2"]}] + if phase == "affinity" else [] + ) + instances = [ + Instance( + uuid=vm, name=vm, internal_name=vm, host="h1", + flavor_vcpus=1, flavor_ram_mb=1024, status="ACTIVE", + ) + for vm in (["v1", "v2"] if phase == "affinity" else ["v1"]) + ] + nova.list_instances_on_host.side_effect = lambda host: ( + instances if host == "h1" else [] + ) + policy = PolicyConfig( + name="cpu", mode=PolicyMode.PACK if phase == "pack" else PolicyMode.SPREAD, + weight=1.0, imbalance_query="load", vm_profile_query="vms", + vm_profile_label_type=VmProfileLabelType.NOVA_INSTANCE_UUID, + threshold=0.15, capacity_query="capacity" if phase == "pack" else None, + capacity_threshold=0.9, max_migrations_per_cycle=1, + ) + prometheus = MagicMock(spec=PrometheusClient) + series = { + "load": {"h1": 0.2, "h2": 0.6} if phase == "pack" else {"h1": 0.8, "h2": 0.2}, + "vms": {"v1": 0.2 if phase == "pack" else 0.3, "v2": 0.1}, + } + prometheus.instant_query.side_effect = lambda query, **kwargs: QueryResult( + query=query, timestamp=datetime.now(tz=UTC), health=PrometheusHealth.HEALTHY, + series=series[query], + ) + placement = MagicMock(spec=PlacementClient) + placement.fetch_cpu_traits.return_value = { + "h1": frozenset({AVX, AVX512}), "h2": frozenset({AVX}), + } + return conf, nova, prometheus, placement, PoliciesConfig(policies=[policy]) + + +@pytest.mark.parametrize("phase", ["spread", "pack", "evacuate", "affinity"]) +def test_all_movers_reject_incompatible_destination_and_recover(phase) -> None: + conf, nova, prometheus, placement, policies = _scenario(phase) + engine = EngineLoop(conf, nova=nova, prometheus=prometheus, placement=placement) + blocked = engine.run_once(policies=policies, aggregates=["a"]) + assert not blocked.errors + assert _plan(blocked).migration_count == 0 + placement.fetch_cpu_traits.assert_called_once_with({"h1", "h2"}) + placement.fetch_cpu_traits.return_value["h2"] = frozenset({AVX, AVX512}) + allowed = engine.run_once(policies=policies, aggregates=["a"]) + steps = _plan(allowed).steps + assert len(steps) == 1 + assert (steps[0].from_host, steps[0].to_host, steps[0].phase.value) == ("h1", "h2", phase) + + +def test_traits_failure_discards_previous_cycle_data() -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + placement.fetch_cpu_traits.return_value["h2"] = frozenset({AVX, AVX512}) + engine = EngineLoop(conf, nova=nova, prometheus=prometheus, placement=placement) + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 1 + placement.fetch_cpu_traits.side_effect = PlacementClientError(reason="unavailable") + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 0 + + +def test_disabled_option_skips_placement_entirely() -> None: + conf, nova, prometheus, _placement, policies = _scenario("spread") + conf.set_override("require_cpu_compatibility", False, group="engine") + with patch("kronos.engine.loop.PlacementClient") as constructor: + engine = EngineLoop(conf, nova=nova, prometheus=prometheus) + constructor.assert_not_called() + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 1 + + +def test_record_replay_cpu_traits_round_trip(tmp_path: Path) -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + target = write_snapshot( + tmp_path, nova, prometheus, policies, ["a"], placement=placement, + ) + data = json.loads((target / "placement" / "cpu_traits.json").read_text()) + assert data == {"h1": sorted([AVX, AVX512]), "h2": [AVX]} + engine = EngineLoop( + conf, + nova=cast(NovaClient, ReplayNovaClient(target)), + prometheus=cast(PrometheusClient, ReplayPrometheusClient(target)), + placement=ReplayPlacementClient(target), + ) + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 0 + (target / "placement" / "cpu_traits.json").write_text( + json.dumps({"h1": [AVX, AVX512], "h2": [AVX, AVX512]}), + ) + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 1 + + +@pytest.mark.parametrize("flags", [None, {}, [], {"h1": None}, {"h1": [1]}]) +def test_old_or_invalid_snapshot_does_not_allow_moves(tmp_path: Path, flags) -> None: + if flags is not None: + (tmp_path / "placement").mkdir() + (tmp_path / "placement" / "cpu_traits.json").write_text(json.dumps(flags)) + conf, nova, prometheus, _placement, policies = _scenario("spread") + engine = EngineLoop( + conf, nova=nova, prometheus=prometheus, placement=ReplayPlacementClient(tmp_path), + ) + assert _plan(engine.run_once( + policies=policies, aggregates=["a"], + )).migration_count == 0 + + +def _plan(report: CycleReport) -> MigrationPlan: + assert not report.errors + plan = report.aggregate_results[0].migration_plan + assert plan is not None + return plan + + +@pytest.mark.parametrize("payload", [ + {"traits": [AVX, AVX512, "CUSTOM_CPU", "COMPUTE_STATUS_DISABLED"]}, + {"traits": []}, +]) +def test_client_filters_cpu_traits_and_scope(payload) -> None: + client = object.__new__(PlacementClient) + client._conn = MagicMock() + proxy = client._conn.placement + providers = [] + for host in ("h1", "outside", "sharing-pool"): + provider = MagicMock() + provider.name = host + provider.id = f"rp-{host}" + providers.append(provider) + proxy.resource_providers.return_value = providers + proxy.get.return_value.json.return_value = payload + assert client.fetch_cpu_traits({"h1", "missing"}) == { + "h1": frozenset(t for t in payload["traits"] if t.startswith("HW_CPU_")), + } + proxy.get.assert_called_once_with("/resource_providers/rp-h1/traits", microversion="1.6") + proxy.get.return_value.raise_for_status.assert_called_once() + + +@pytest.mark.parametrize("payload", [{}, {"traits": None}, {"traits": AVX}, {"traits": [1]}]) +def test_client_rejects_malformed_traits(payload) -> None: + client = object.__new__(PlacementClient) + client._conn = MagicMock() + provider = MagicMock() + provider.name = "h1" + client._conn.placement.resource_providers.return_value = [provider] + client._conn.placement.get.return_value.json.return_value = payload + with pytest.raises(PlacementClientError): + client.fetch_cpu_traits({"h1"}) + + +@pytest.mark.parametrize("failure", ["list", "get", "http"]) +def test_client_wraps_traits_fetch_failures(failure) -> None: + client = object.__new__(PlacementClient) + client._conn = MagicMock() + proxy = client._conn.placement + provider = MagicMock() + provider.name = "h1" + proxy.resource_providers.return_value = [provider] + operation = { + "list": proxy.resource_providers, + "get": proxy.get, + "http": proxy.get.return_value.raise_for_status, + }[failure] + operation.side_effect = RuntimeError("Placement unavailable") + with pytest.raises(PlacementClientError): + client.fetch_cpu_traits({"h1"}) + + +def test_no_scope_performs_no_placement_requests() -> None: + client = object.__new__(PlacementClient) + client._conn = MagicMock() + assert client.fetch_cpu_traits(set()) == {} + assert not client._conn.mock_calls + + +def test_cpu_scope_is_fetched_once_for_overlapping_aggregates_and_az() -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + outside = ComputeService( + host="h3", binary="nova-compute", state="up", status="enabled", zone="other", + ) + nova.list_compute_services.return_value.append(outside) + scopes = {"a": ["h1", "h2", "h3"], "b": ["h2"], None: ["h3"]} + nova.get_hosts_in_aggregate.side_effect = scopes.__getitem__ + engine = EngineLoop(conf, nova=nova, prometheus=prometheus, placement=placement) + report = engine.run_once(policies=policies, aggregates=["a", "b", None, "missing"]) + assert len(report.errors) == 1 + assert "missing" in report.errors[0] + placement.fetch_cpu_traits.assert_called_once_with({"h1", "h2"}) + assert nova.get_hosts_in_aggregate.call_count == 4 + + +def test_cpu_and_claims_gates_are_independent() -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + conf.set_override("enforce_placement_claims", True, group="engine") + placement.fetch_snapshots.return_value = {} + placement.fetch_cpu_traits.return_value["h2"] = frozenset({AVX, AVX512}) + engine = EngineLoop(conf, nova=nova, prometheus=prometheus, placement=placement) + assert _plan(engine.run_once(policies=policies, aggregates=["a"])).migration_count == 0 + placement.fetch_snapshots.assert_called_once() + placement.fetch_cpu_traits.assert_called_once() + + +def test_cpu_only_engine_constructs_placement_client() -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + with patch("kronos.engine.loop.PlacementClient", return_value=placement) as constructor: + engine = EngineLoop(conf, nova=nova, prometheus=prometheus) + constructor.assert_called_once_with(conf) + assert _plan(engine.run_once(policies=policies, aggregates=["a"])).migration_count == 0 + + +def test_engine_snapshot_contains_cpu_traits(tmp_path: Path) -> None: + conf, nova, prometheus, placement, policies = _scenario("spread") + conf.set_override("snapshot_dir", str(tmp_path), group="engine") + engine = EngineLoop(conf, nova=nova, prometheus=prometheus, placement=placement) + engine._write_engine_snapshot(policies, ["a"]) + path, = tmp_path.glob("*/placement/cpu_traits.json") + assert json.loads(path.read_text())["h1"] == sorted([AVX, AVX512]) + + +def test_failed_traits_snapshot_is_missing_data_not_empty_features(tmp_path: Path) -> None: + _conf, nova, prometheus, placement, policies = _scenario("spread") + placement.fetch_cpu_traits.side_effect = PlacementClientError(reason="unavailable") + target = write_snapshot(tmp_path, nova, prometheus, policies, ["a"], placement=placement) + assert ReplayPlacementClient(target).fetch_cpu_traits({"h1", "h2"}) == {} + assert (target / "nova" / "instances.json").exists() diff --git a/tests/unit/engine/test_loop.py b/tests/unit/engine/test_loop.py index 9c4490c..81220a0 100644 --- a/tests/unit/engine/test_loop.py +++ b/tests/unit/engine/test_loop.py @@ -134,6 +134,7 @@ def mock_engine(): # Disable the placement gate for these tests; covered in # tests/unit/engine/test_placement.py. conf.engine.enforce_placement_claims = False + conf.engine.require_cpu_compatibility = False engine = EngineLoop(conf) engine._nova = mock_nova_cls.return_value @@ -605,6 +606,7 @@ def test_injected_clients_replace_defaults(self) -> None: conf.engine.enforce_soft_affinity = False conf.engine.evacuate_disabled_hosts = False conf.engine.enforce_placement_claims = False + conf.engine.require_cpu_compatibility = False nova = MagicMock() prom = MagicMock() @@ -645,6 +647,7 @@ def test_timings_dict_starts_empty_when_provided(self) -> None: conf.engine.enforce_soft_affinity = False conf.engine.evacuate_disabled_hosts = False conf.engine.enforce_placement_claims = False + conf.engine.require_cpu_compatibility = False timings: dict[str, float] = {} with ( @@ -662,6 +665,7 @@ def test_no_timings_dict_means_none(self) -> None: conf.engine.enforce_soft_affinity = False conf.engine.evacuate_disabled_hosts = False conf.engine.enforce_placement_claims = False + conf.engine.require_cpu_compatibility = False with ( patch("kronos.engine.loop.NovaClient"), patch("kronos.engine.loop.PrometheusClient"),