diff --git a/dns_sync/orchestrator.py b/dns_sync/orchestrator.py new file mode 100644 index 0000000..e3684d9 --- /dev/null +++ b/dns_sync/orchestrator.py @@ -0,0 +1,224 @@ +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Protocol + +from .adapters import ( + DNS_RECORDS, + PROXY_HOSTS, + Capability, + CapabilityOperation, + DiscoveredObject, + Endpoint, + EndpointPlan, + NormalizedObject, + Observation, +) +from .engine import hostname_matches_domain +from .object_normalizer import normalize_observations +from .persistence import SQLiteStateStore +from .projections import project_proxy_hosts_to_dns +from .reconciliation import ( + build_endpoint_plans, + reconcile_observations, +) + + +class ReconciliationAdapter(Protocol): + endpoint: Endpoint + capabilities: frozenset[Capability] + + def connect(self) -> None: ... + + def discover( + self, + object_type: str, + ) -> list[DiscoveredObject]: ... + + +@dataclass(frozen=True) +class DiscoveryFailure: + endpoint_id: str + object_type: str + error: str + + +@dataclass(frozen=True) +class ReconciliationRun: + observations: tuple[Observation, ...] + desired: tuple[NormalizedObject, ...] + plans: tuple[EndpointPlan, ...] + failures: tuple[DiscoveryFailure, ...] + + +def _discovery_types( + adapter: ReconciliationAdapter, +) -> tuple[str, ...]: + return tuple( + sorted( + capability.object_type + for capability in adapter.capabilities + if capability.operation == CapabilityOperation.DISCOVER + ) + ) + + +def _configured_observations( + store: SQLiteStateStore, + endpoint_ids: set[str], +) -> list[Observation]: + return [ + observation + for observation in store.load_observations() + if observation.endpoint_id in endpoint_ids + ] + + +def _filter_projected_dns( + observations: Iterable[Observation], + *, + domain_matching: bool, + domain: str | None, +) -> list[Observation]: + rows = list(observations) + if not domain_matching: + return rows + if not domain: + raise RuntimeError( + "Domain filtering enabled without a configured DOMAIN" + ) + + return [ + observation + for observation in rows + if hostname_matches_domain( + observation.object.logical_key, + domain, + ) + ] + + +def discover_reconcile_plan( + adapters: Iterable[ReconciliationAdapter], + store: SQLiteStateStore, + *, + target_ip: str, + include_disabled: bool = False, + domain_matching: bool = False, + domain: str | None = None, + run_at: datetime | None = None, +) -> ReconciliationRun: + timestamp = run_at or datetime.now(timezone.utc) + if timestamp.tzinfo is None: + raise RuntimeError( + "Reconciliation run timestamp must be timezone-aware" + ) + timestamp = timestamp.astimezone(timezone.utc) + + adapter_list = list(adapters) + adapters_by_endpoint: dict[str, ReconciliationAdapter] = {} + for adapter in adapter_list: + endpoint_id = adapter.endpoint.identity + if endpoint_id in adapters_by_endpoint: + raise RuntimeError( + f"Duplicate adapter endpoint: {endpoint_id}" + ) + adapters_by_endpoint[endpoint_id] = adapter + + failures: list[DiscoveryFailure] = [] + + for endpoint_id, adapter in sorted( + adapters_by_endpoint.items() + ): + object_types = _discovery_types(adapter) + if not object_types: + continue + + try: + adapter.connect() + except Exception as exc: + for object_type in object_types: + error = str(exc) + store.record_discovery_failure( + endpoint_id, + object_type, + error, + timestamp, + ) + failures.append( + DiscoveryFailure( + endpoint_id, + object_type, + error, + ) + ) + continue + + for object_type in object_types: + try: + discovered = adapter.discover(object_type) + observations = normalize_observations( + endpoint_id, + discovered, + timestamp, + ) + store.replace_observations( + endpoint_id, + object_type, + observations, + timestamp, + ) + except Exception as exc: + error = str(exc) + store.record_discovery_failure( + endpoint_id, + object_type, + error, + timestamp, + ) + failures.append( + DiscoveryFailure( + endpoint_id, + object_type, + error, + ) + ) + + observations = _configured_observations( + store, + set(adapters_by_endpoint), + ) + proxy_observations = [ + observation + for observation in observations + if observation.object.object_type == PROXY_HOSTS + ] + projected_dns = project_proxy_hosts_to_dns( + proxy_observations, + target_ip, + include_disabled=include_disabled, + ) + projected_dns = _filter_projected_dns( + projected_dns, + domain_matching=domain_matching, + domain=domain, + ) + + desired = reconcile_observations(projected_dns) + store.replace_desired_state( + DNS_RECORDS, + desired, + timestamp, + ) + plans = build_endpoint_plans( + adapter_list, + desired, + observations, + ) + + return ReconciliationRun( + observations=tuple(observations), + desired=tuple(desired), + plans=plans, + failures=tuple(failures), + ) diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 0000000..19efcaf --- /dev/null +++ b/tests/test_orchestrator.py @@ -0,0 +1,353 @@ +import unittest +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from dns_sync.adapters import ( + DNS_RECORDS, + PROXY_HOSTS, + Capability, + CapabilityOperation, + DiscoveredObject, + Endpoint, + PlanOperation, +) +from dns_sync.object_normalizer import normalize_observations +from dns_sync.orchestrator import discover_reconcile_plan +from dns_sync.persistence import SQLiteStateStore + + +T1 = datetime(2026, 9, 19, 22, 0, tzinfo=timezone.utc) +T2 = datetime(2026, 9, 19, 22, 5, tzinfo=timezone.utc) + + +def capabilities( + object_type: str, + *operations: CapabilityOperation, +) -> frozenset[Capability]: + return frozenset( + Capability(object_type, operation) + for operation in operations + ) + + +def proxy_host( + local_key: str, + domains: list[str], + backend_ip: str = "192.168.1.50", + enabled: bool = True, +) -> DiscoveredObject: + return DiscoveredObject( + object_type=PROXY_HOSTS, + local_key=local_key, + payload={ + "domain_names": domains, + "forward_scheme": "http", + "forward_host": backend_ip, + "forward_port": 8080, + "enabled": enabled, + }, + ) + + +def dns_record( + local_key: str, + hostname: str, + ip: str, +) -> DiscoveredObject: + return DiscoveredObject( + object_type=DNS_RECORDS, + local_key=local_key, + payload={ + "hostname": hostname, + "ip": ip, + "record_type": "A", + }, + ) + + +@dataclass +class FakeAdapter: + endpoint: Endpoint + capabilities: frozenset[Capability] + discovered: dict[str, list[DiscoveredObject]] = field( + default_factory=dict + ) + connect_error: Exception | None = None + connect_count: int = 0 + discover_calls: list[str] = field(default_factory=list) + apply_count: int = 0 + verify_count: int = 0 + + def connect(self) -> None: + self.connect_count += 1 + if self.connect_error is not None: + raise self.connect_error + + def discover( + self, + object_type: str, + ) -> list[DiscoveredObject]: + self.discover_calls.append(object_type) + return list(self.discovered.get(object_type, [])) + + def apply(self, action): + self.apply_count += 1 + raise AssertionError("read-only orchestration called apply") + + def verify(self, object): + self.verify_count += 1 + raise AssertionError("read-only orchestration called verify") + + +class ReadOnlyOrchestrationTests(unittest.TestCase): + def test_discovers_persists_reconciles_and_plans_vip_update(self): + npm = FakeAdapter( + endpoint=Endpoint("npm", "192.168.1.3:81"), + capabilities=capabilities( + PROXY_HOSTS, + CapabilityOperation.DISCOVER, + ), + discovered={ + PROXY_HOSTS: [ + proxy_host( + "12", + ["app.example.com"], + backend_ip="192.168.1.50", + ) + ] + }, + ) + verizon = FakeAdapter( + endpoint=Endpoint("verizon", "192.168.1.1:443"), + capabilities=capabilities( + DNS_RECORDS, + CapabilityOperation.DISCOVER, + CapabilityOperation.APPLY, + CapabilityOperation.VERIFY, + ), + discovered={ + DNS_RECORDS: [ + dns_record( + "41", + "app.example.com", + "192.168.1.50", + ) + ] + }, + ) + + with SQLiteStateStore(":memory:") as store: + result = discover_reconcile_plan( + [npm, verizon], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + persisted_desired = store.load_desired_state( + DNS_RECORDS + ) + + self.assertEqual(len(result.desired), 1) + self.assertEqual( + result.desired[0].payload["ip"], + "192.168.1.101", + ) + self.assertEqual(len(result.plans), 1) + self.assertEqual( + result.plans[0].endpoint_id, + verizon.endpoint.identity, + ) + self.assertEqual( + result.plans[0].actions[0].operation, + PlanOperation.UPDATE, + ) + self.assertEqual( + result.plans[0].actions[0].desired.payload["ip"], + "192.168.1.101", + ) + self.assertEqual( + persisted_desired[0].object.payload["ip"], + "192.168.1.101", + ) + self.assertEqual(npm.connect_count, 1) + self.assertEqual(verizon.connect_count, 1) + self.assertEqual(npm.apply_count, 0) + self.assertEqual(verizon.apply_count, 0) + self.assertEqual(npm.verify_count, 0) + self.assertEqual(verizon.verify_count, 0) + + def test_failed_node_remains_represented_by_cached_observation(self): + npm_endpoint = Endpoint("npm", "192.168.1.3:81") + cached = normalize_observations( + npm_endpoint.identity, + [proxy_host("12", ["cached.example.com"])], + T1, + ) + npm = FakeAdapter( + endpoint=npm_endpoint, + capabilities=capabilities( + PROXY_HOSTS, + CapabilityOperation.DISCOVER, + ), + connect_error=RuntimeError("temporarily unavailable"), + ) + verizon = FakeAdapter( + endpoint=Endpoint("verizon", "192.168.1.1:443"), + capabilities=capabilities( + DNS_RECORDS, + CapabilityOperation.DISCOVER, + CapabilityOperation.APPLY, + ), + discovered={DNS_RECORDS: []}, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + npm_endpoint.identity, + PROXY_HOSTS, + cached, + T1, + ) + result = discover_reconcile_plan( + [npm, verizon], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + status = store.observation_status( + npm_endpoint.identity, + PROXY_HOSTS, + ) + + self.assertEqual( + tuple(item.logical_key for item in result.desired), + ("cached.example.com",), + ) + self.assertEqual( + result.plans[0].actions[0].operation, + PlanOperation.CREATE, + ) + self.assertFalse(status.is_live) + self.assertEqual(status.last_success_at, T1) + self.assertEqual(status.last_attempt_at, T2) + self.assertEqual( + status.last_error, + "temporarily unavailable", + ) + self.assertEqual(len(result.failures), 1) + + def test_live_empty_snapshot_supersedes_cached_state(self): + npm_endpoint = Endpoint("npm", "192.168.1.3:81") + npm = FakeAdapter( + endpoint=npm_endpoint, + capabilities=capabilities( + PROXY_HOSTS, + CapabilityOperation.DISCOVER, + ), + discovered={PROXY_HOSTS: []}, + ) + cached = normalize_observations( + npm_endpoint.identity, + [proxy_host("12", ["removed.example.com"])], + T1, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + npm_endpoint.identity, + PROXY_HOSTS, + cached, + T1, + ) + result = discover_reconcile_plan( + [npm], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + persisted_desired = store.load_desired_state( + DNS_RECORDS + ) + + self.assertEqual(result.desired, ()) + self.assertEqual(result.plans, ()) + self.assertEqual(persisted_desired, []) + + def test_domain_filtering_occurs_after_projection(self): + npm = FakeAdapter( + endpoint=Endpoint("npm", "192.168.1.3:81"), + capabilities=capabilities( + PROXY_HOSTS, + CapabilityOperation.DISCOVER, + ), + discovered={ + PROXY_HOSTS: [ + proxy_host( + "12", + [ + "app.example.com", + "outside.example.net", + ], + ) + ] + }, + ) + + with SQLiteStateStore(":memory:") as store: + result = discover_reconcile_plan( + [npm], + store, + target_ip="192.168.1.101", + domain_matching=True, + domain="example.com", + run_at=T2, + ) + + self.assertEqual( + tuple(item.logical_key for item in result.desired), + ("app.example.com",), + ) + + def test_removed_endpoints_do_not_contribute_cached_state(self): + removed_endpoint = Endpoint("npm", "192.168.1.4:81") + active_endpoint = Endpoint("npm", "192.168.1.3:81") + removed_cache = normalize_observations( + removed_endpoint.identity, + [proxy_host("12", ["removed.example.com"])], + T1, + ) + active = FakeAdapter( + endpoint=active_endpoint, + capabilities=capabilities( + PROXY_HOSTS, + CapabilityOperation.DISCOVER, + ), + discovered={ + PROXY_HOSTS: [ + proxy_host("13", ["active.example.com"]) + ] + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + removed_endpoint.identity, + PROXY_HOSTS, + removed_cache, + T1, + ) + result = discover_reconcile_plan( + [active], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + + self.assertEqual( + tuple(item.logical_key for item in result.desired), + ("active.example.com",), + ) + + +if __name__ == "__main__": + unittest.main()