980 lines
46 KiB
Python
980 lines
46 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from datetime import UTC, datetime
|
|
from hashlib import sha256
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from .models import (
|
|
DiscoveryConfidence,
|
|
DiscoveryDiff,
|
|
DiscoveryLifecycle,
|
|
DiscoveryObservation,
|
|
DiscoveryRun,
|
|
DiscoveryRunStatus,
|
|
LogicalRelation,
|
|
RelationEvidence,
|
|
RelationType,
|
|
ServiceSource,
|
|
SuggestionStatus,
|
|
TopologyChange,
|
|
TopologyDiff,
|
|
TopologySnapshot,
|
|
TopologySuggestion,
|
|
)
|
|
from .store import Store
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
return value.isoformat() if value else None
|
|
|
|
|
|
def _json(value: object) -> str:
|
|
return json.dumps(value, separators=(",", ":"), sort_keys=True)
|
|
|
|
|
|
def _obj(value: object) -> dict[str, object]:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _list(value: object) -> list[str]:
|
|
if isinstance(value, list):
|
|
return [str(item) for item in value]
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
return [str(item) for item in parsed] if isinstance(parsed, list) else []
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return []
|
|
|
|
|
|
_CONFIDENCE_RANK = {
|
|
DiscoveryConfidence.LOW: 1,
|
|
DiscoveryConfidence.MEDIUM: 2,
|
|
DiscoveryConfidence.HIGH: 3,
|
|
DiscoveryConfidence.CONFIRMED: 4,
|
|
}
|
|
_CONFIDENCE_RANK_INV = {rank: confidence.value for confidence, rank in _CONFIDENCE_RANK.items()}
|
|
|
|
|
|
def _higher_confidence(*values: DiscoveryConfidence) -> DiscoveryConfidence:
|
|
return max(values, key=lambda value: _CONFIDENCE_RANK[value])
|
|
|
|
|
|
class DiscoveryStore:
|
|
"""Durable discovery observations and review state.
|
|
|
|
The existing Store remains the owner of canonical services and probe state.
|
|
This companion store deliberately keeps observations and suggestions separate
|
|
so a discovery run cannot mutate health topology unless reconciliation or an
|
|
explicit suggestion action asks it to do so.
|
|
"""
|
|
|
|
def __init__(self, store: Store) -> None:
|
|
self.store = store
|
|
self._init_schema()
|
|
self.ensure_baseline_snapshot()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
return self.store._connect()
|
|
|
|
def _init_schema(self) -> None:
|
|
with self._connect() as db:
|
|
db.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS discovery_runs (
|
|
id TEXT PRIMARY KEY,
|
|
status TEXT NOT NULL,
|
|
started_at TEXT NOT NULL,
|
|
completed_at TEXT,
|
|
source_counts TEXT NOT NULL DEFAULT '{}',
|
|
diff TEXT NOT NULL DEFAULT '{}',
|
|
error TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_discovery_runs_status
|
|
ON discovery_runs(status, started_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS discovery_sources (
|
|
id TEXT PRIMARY KEY,
|
|
source_type TEXT NOT NULL,
|
|
source_id TEXT NOT NULL,
|
|
source_name TEXT,
|
|
first_seen_at TEXT NOT NULL,
|
|
last_seen_at TEXT NOT NULL,
|
|
lifecycle TEXT NOT NULL DEFAULT 'active',
|
|
managed INTEGER NOT NULL DEFAULT 1,
|
|
metadata TEXT NOT NULL DEFAULT '{}',
|
|
UNIQUE(source_type, source_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_discovery_sources_identity
|
|
ON discovery_sources(source_type, source_id);
|
|
CREATE INDEX IF NOT EXISTS idx_discovery_sources_last_seen
|
|
ON discovery_sources(last_seen_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS service_sources (
|
|
service_id TEXT NOT NULL,
|
|
source_id TEXT NOT NULL,
|
|
role TEXT NOT NULL DEFAULT 'observed',
|
|
PRIMARY KEY(service_id, source_id),
|
|
FOREIGN KEY(service_id) REFERENCES services(id) ON DELETE CASCADE,
|
|
FOREIGN KEY(source_id) REFERENCES discovery_sources(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_service_sources_source
|
|
ON service_sources(source_id, service_id);
|
|
|
|
CREATE TABLE IF NOT EXISTS discovery_observations (
|
|
id TEXT PRIMARY KEY,
|
|
run_id TEXT NOT NULL,
|
|
source_id TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
source_name TEXT NOT NULL,
|
|
lifecycle TEXT NOT NULL,
|
|
change_type TEXT NOT NULL,
|
|
observed_at TEXT NOT NULL,
|
|
metadata TEXT NOT NULL DEFAULT '{}',
|
|
FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE CASCADE,
|
|
FOREIGN KEY(source_id) REFERENCES discovery_sources(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_discovery_observations_run
|
|
ON discovery_observations(run_id, observed_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_discovery_observations_source
|
|
ON discovery_observations(source_id, observed_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS relation_evidence (
|
|
id TEXT PRIMARY KEY,
|
|
source_service_id TEXT NOT NULL,
|
|
target_service_id TEXT,
|
|
target_ref TEXT,
|
|
relation_type TEXT NOT NULL DEFAULT 'depends_on',
|
|
evidence_type TEXT NOT NULL,
|
|
confidence TEXT NOT NULL,
|
|
explanation TEXT NOT NULL,
|
|
source_ref TEXT,
|
|
locked INTEGER NOT NULL DEFAULT 0,
|
|
active INTEGER NOT NULL DEFAULT 1,
|
|
first_seen_at TEXT NOT NULL,
|
|
last_seen_at TEXT NOT NULL,
|
|
UNIQUE(source_service_id, target_service_id, target_ref, relation_type, evidence_type),
|
|
FOREIGN KEY(source_service_id) REFERENCES services(id) ON DELETE CASCADE,
|
|
FOREIGN KEY(target_service_id) REFERENCES services(id) ON DELETE SET NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_relation_evidence_source_target
|
|
ON relation_evidence(source_service_id, target_service_id);
|
|
CREATE INDEX IF NOT EXISTS idx_relation_evidence_confidence
|
|
ON relation_evidence(confidence, active);
|
|
|
|
CREATE TABLE IF NOT EXISTS topology_suggestions (
|
|
id TEXT PRIMARY KEY,
|
|
run_id TEXT NOT NULL,
|
|
suggestion_type TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'open',
|
|
fingerprint TEXT NOT NULL UNIQUE,
|
|
source_ref TEXT,
|
|
source_service_id TEXT,
|
|
target_ref TEXT,
|
|
target_service_id TEXT,
|
|
proposed_name TEXT,
|
|
proposed_kind TEXT,
|
|
proposed_url TEXT,
|
|
proposed_host TEXT,
|
|
proposed_port INTEGER,
|
|
proposed_category TEXT NOT NULL DEFAULT 'unknown',
|
|
proposed_group TEXT,
|
|
relation_type TEXT,
|
|
confidence TEXT NOT NULL,
|
|
evidence TEXT NOT NULL DEFAULT '[]',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_topology_suggestions_status
|
|
ON topology_suggestions(status, updated_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_topology_suggestions_source
|
|
ON topology_suggestions(source_ref, target_ref);
|
|
|
|
CREATE TABLE IF NOT EXISTS topology_overrides (
|
|
target_type TEXT NOT NULL,
|
|
target_id TEXT NOT NULL,
|
|
field_name TEXT NOT NULL,
|
|
value TEXT NOT NULL,
|
|
locked INTEGER NOT NULL DEFAULT 1,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY(target_type, target_id, field_name)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_topology_overrides_target
|
|
ON topology_overrides(target_type, target_id);
|
|
|
|
CREATE TABLE IF NOT EXISTS topology_change_events (
|
|
id TEXT PRIMARY KEY,
|
|
event_type TEXT NOT NULL,
|
|
service_id TEXT,
|
|
source_service_id TEXT,
|
|
target_service_id TEXT,
|
|
detail TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_topology_change_events_time
|
|
ON topology_change_events(created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS topology_snapshots (
|
|
id TEXT PRIMARY KEY,
|
|
run_id TEXT,
|
|
created_at TEXT NOT NULL,
|
|
revision TEXT NOT NULL UNIQUE,
|
|
service_count INTEGER NOT NULL,
|
|
relationship_count INTEGER NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '{}',
|
|
FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE SET NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_topology_snapshots_time
|
|
ON topology_snapshots(created_at DESC);
|
|
CREATE TABLE IF NOT EXISTS topology_snapshot_services (
|
|
snapshot_id TEXT NOT NULL,
|
|
service_id TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
fingerprint TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
PRIMARY KEY(snapshot_id, service_id),
|
|
FOREIGN KEY(snapshot_id) REFERENCES topology_snapshots(id) ON DELETE CASCADE
|
|
);
|
|
CREATE TABLE IF NOT EXISTS topology_snapshot_relations (
|
|
snapshot_id TEXT NOT NULL,
|
|
relation_id TEXT NOT NULL,
|
|
source_service_id TEXT NOT NULL,
|
|
target_service_id TEXT,
|
|
relation_type TEXT NOT NULL,
|
|
fingerprint TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
PRIMARY KEY(snapshot_id, relation_id),
|
|
FOREIGN KEY(snapshot_id) REFERENCES topology_snapshots(id) ON DELETE CASCADE
|
|
);
|
|
"""
|
|
)
|
|
|
|
@staticmethod
|
|
def _diff(value: object) -> DiscoveryDiff:
|
|
data = _obj(value)
|
|
return DiscoveryDiff(**{key: int(data.get(key, 0) or 0) for key in DiscoveryDiff.model_fields})
|
|
|
|
@classmethod
|
|
def _run(cls, row: sqlite3.Row | None) -> DiscoveryRun | None:
|
|
if row is None:
|
|
return None
|
|
return DiscoveryRun(
|
|
id=row["id"],
|
|
status=DiscoveryRunStatus(row["status"]),
|
|
started_at=datetime.fromisoformat(row["started_at"]),
|
|
completed_at=datetime.fromisoformat(row["completed_at"]) if row["completed_at"] else None,
|
|
source_counts={str(k): int(v) for k, v in _obj(row["source_counts"]).items()},
|
|
diff=cls._diff(row["diff"]),
|
|
error=row["error"],
|
|
)
|
|
|
|
@classmethod
|
|
def _source(cls, row: sqlite3.Row) -> ServiceSource:
|
|
return ServiceSource(
|
|
id=row["id"],
|
|
source_type=row["source_type"],
|
|
source_id=row["source_id"],
|
|
source_name=row["source_name"],
|
|
first_seen_at=datetime.fromisoformat(row["first_seen_at"]),
|
|
last_seen_at=datetime.fromisoformat(row["last_seen_at"]),
|
|
lifecycle=DiscoveryLifecycle(row["lifecycle"]),
|
|
managed=bool(row["managed"]),
|
|
metadata=_obj(row["metadata"]),
|
|
)
|
|
|
|
@classmethod
|
|
def _suggestion(cls, row: sqlite3.Row) -> TopologySuggestion:
|
|
return TopologySuggestion(
|
|
id=row["id"],
|
|
run_id=row["run_id"],
|
|
suggestion_type=row["suggestion_type"],
|
|
status=SuggestionStatus(row["status"]),
|
|
source_ref=row["source_ref"],
|
|
source_service_id=row["source_service_id"],
|
|
target_ref=row["target_ref"],
|
|
target_service_id=row["target_service_id"],
|
|
proposed_name=row["proposed_name"],
|
|
proposed_kind=row["proposed_kind"],
|
|
proposed_url=row["proposed_url"],
|
|
proposed_host=row["proposed_host"],
|
|
proposed_port=row["proposed_port"],
|
|
proposed_category=row["proposed_category"],
|
|
proposed_group=row["proposed_group"],
|
|
relation_type=row["relation_type"],
|
|
confidence=DiscoveryConfidence(row["confidence"]),
|
|
evidence=_list(row["evidence"]),
|
|
created_at=datetime.fromisoformat(row["created_at"]),
|
|
updated_at=datetime.fromisoformat(row["updated_at"]),
|
|
)
|
|
|
|
@classmethod
|
|
def _evidence(cls, row: sqlite3.Row) -> RelationEvidence:
|
|
return RelationEvidence(
|
|
id=row["id"],
|
|
source_service_id=row["source_service_id"],
|
|
target_service_id=row["target_service_id"],
|
|
target_ref=row["target_ref"],
|
|
relation_type=RelationType(row["relation_type"]),
|
|
evidence_type=row["evidence_type"],
|
|
confidence=DiscoveryConfidence(row["confidence"]),
|
|
explanation=row["explanation"],
|
|
source_ref=row["source_ref"],
|
|
locked=bool(row["locked"]),
|
|
active=bool(row["active"]),
|
|
first_seen_at=datetime.fromisoformat(row["first_seen_at"]),
|
|
last_seen_at=datetime.fromisoformat(row["last_seen_at"]),
|
|
)
|
|
|
|
def start_run(self) -> DiscoveryRun:
|
|
run = DiscoveryRun(id=f"disc-{uuid4().hex[:12]}", status=DiscoveryRunStatus.DISCOVERING, started_at=_now())
|
|
with self._connect() as db:
|
|
# A process restart or an unexpected reconcile exception can leave
|
|
# a durable run in an active state. Close that stale marker before
|
|
# creating the next run so status never reports a phantom active
|
|
# discovery forever.
|
|
db.execute(
|
|
"UPDATE discovery_runs SET status=?, completed_at=?, error=COALESCE(error, ?) WHERE status IN (?, ?)",
|
|
(
|
|
DiscoveryRunStatus.FAILED.value,
|
|
run.started_at.isoformat(),
|
|
"Superseded by a newer discovery run",
|
|
DiscoveryRunStatus.DISCOVERING.value,
|
|
DiscoveryRunStatus.RECONCILING.value,
|
|
),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO discovery_runs (id, status, started_at) VALUES (?, ?, ?)",
|
|
(run.id, run.status.value, run.started_at.isoformat()),
|
|
)
|
|
return run
|
|
|
|
def finish_run(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
status: DiscoveryRunStatus,
|
|
source_counts: dict[str, int],
|
|
diff: DiscoveryDiff,
|
|
error: str | None = None,
|
|
) -> DiscoveryRun:
|
|
completed = _now()
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"UPDATE discovery_runs SET status=?, completed_at=?, source_counts=?, diff=?, error=? WHERE id=?",
|
|
(status.value, completed.isoformat(), _json(source_counts), _json(diff.model_dump()), error, run_id),
|
|
)
|
|
return self.get_run(run_id) or DiscoveryRun(
|
|
id=run_id,
|
|
status=status,
|
|
started_at=completed,
|
|
completed_at=completed,
|
|
source_counts=source_counts,
|
|
diff=diff,
|
|
error=error,
|
|
)
|
|
|
|
def get_run(self, run_id: str) -> DiscoveryRun | None:
|
|
with self._connect() as db:
|
|
row = db.execute("SELECT * FROM discovery_runs WHERE id = ?", (run_id,)).fetchone()
|
|
return self._run(row)
|
|
|
|
def last_completed(self) -> DiscoveryRun | None:
|
|
with self._connect() as db:
|
|
row = db.execute(
|
|
"SELECT * FROM discovery_runs WHERE status = 'completed' ORDER BY completed_at DESC LIMIT 1"
|
|
).fetchone()
|
|
return self._run(row)
|
|
|
|
def status(self) -> tuple[DiscoveryRun | None, DiscoveryRun | None]:
|
|
with self._connect() as db:
|
|
current_row = db.execute(
|
|
"SELECT * FROM discovery_runs WHERE status IN ('discovering', 'reconciling') ORDER BY started_at DESC LIMIT 1"
|
|
).fetchone()
|
|
return self._run(current_row), self.last_completed()
|
|
|
|
@staticmethod
|
|
def _snapshot_row(row: sqlite3.Row | None) -> TopologySnapshot | None:
|
|
if row is None:
|
|
return None
|
|
return TopologySnapshot(
|
|
id=row["id"],
|
|
run_id=row["run_id"],
|
|
created_at=datetime.fromisoformat(row["created_at"]),
|
|
revision=row["revision"],
|
|
service_count=int(row["service_count"]),
|
|
relationship_count=int(row["relationship_count"]),
|
|
summary={str(key): int(value) for key, value in _obj(row["summary"]).items()},
|
|
)
|
|
|
|
def ensure_baseline_snapshot(self) -> TopologySnapshot:
|
|
"""Create one migration baseline without manufacturing repeated no-op snapshots."""
|
|
|
|
with self._connect() as db:
|
|
existing = db.execute("SELECT * FROM topology_snapshots ORDER BY created_at DESC LIMIT 1").fetchone()
|
|
return self._snapshot_row(existing) if existing else self.capture_snapshot(run_id=None)
|
|
|
|
def capture_snapshot(self, *, run_id: str | None) -> TopologySnapshot:
|
|
"""Persist a compact canonical topology capture when its revision changes."""
|
|
|
|
services = sorted(self.store.list_services(), key=lambda item: item.id)
|
|
service_rows: list[tuple[str, str, str, dict[str, object]]] = []
|
|
relation_rows: list[tuple[str, str, str | None, str, str, dict[str, object]]] = []
|
|
for service in services:
|
|
payload: dict[str, object] = {
|
|
"id": service.id,
|
|
"name": service.name,
|
|
"kind": service.kind.value,
|
|
"source_ref": service.source_ref,
|
|
"application": service.application,
|
|
"domain": service.domain,
|
|
"component_role": service.component_role.value,
|
|
"category": service.category.value,
|
|
"criticality": service.criticality.value,
|
|
"visibility": service.visibility.value,
|
|
"lifecycle": service.lifecycle.value,
|
|
"monitoring_state": service.monitoring_state.value,
|
|
"depends_on": sorted(service.depends_on),
|
|
}
|
|
service_rows.append((service.id, service.name, sha256(_json(payload).encode()).hexdigest(), payload))
|
|
for target_id in sorted(service.depends_on):
|
|
relation_payload: dict[str, object] = {
|
|
"source_service_id": service.id,
|
|
"target_service_id": target_id,
|
|
"relation_type": RelationType.DEPENDS_ON.value,
|
|
}
|
|
relation_id = sha256(_json(relation_payload).encode()).hexdigest()[:24]
|
|
relation_rows.append(
|
|
(
|
|
relation_id,
|
|
service.id,
|
|
target_id,
|
|
RelationType.DEPENDS_ON.value,
|
|
sha256(_json(relation_payload).encode()).hexdigest(),
|
|
relation_payload,
|
|
)
|
|
)
|
|
revision_payload = {"services": [(item[0], item[2]) for item in service_rows], "relations": [(item[0], item[4]) for item in relation_rows]}
|
|
revision = sha256(_json(revision_payload).encode()).hexdigest()
|
|
with self._connect() as db:
|
|
existing = db.execute("SELECT * FROM topology_snapshots WHERE revision=?", (revision,)).fetchone()
|
|
if existing:
|
|
return self._snapshot_row(existing) # type: ignore[return-value]
|
|
created_at = _now()
|
|
summary = {
|
|
"visible_services": sum(1 for item in services if item.visibility.value == "visible"),
|
|
"active_services": sum(1 for item in services if item.lifecycle.value == "active"),
|
|
}
|
|
snapshot = TopologySnapshot(
|
|
id=f"topo-{uuid4().hex[:14]}",
|
|
run_id=run_id,
|
|
created_at=created_at,
|
|
revision=revision,
|
|
service_count=len(service_rows),
|
|
relationship_count=len(relation_rows),
|
|
summary=summary,
|
|
)
|
|
db.execute(
|
|
"INSERT INTO topology_snapshots VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(snapshot.id, snapshot.run_id, snapshot.created_at.isoformat(), snapshot.revision, snapshot.service_count, snapshot.relationship_count, _json(snapshot.summary)),
|
|
)
|
|
db.executemany(
|
|
"INSERT INTO topology_snapshot_services VALUES (?, ?, ?, ?, ?)",
|
|
[(snapshot.id, service_id, name, fingerprint, _json(payload)) for service_id, name, fingerprint, payload in service_rows],
|
|
)
|
|
db.executemany(
|
|
"INSERT INTO topology_snapshot_relations VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
[(snapshot.id, relation_id, source_id, target_id, relation_type, fingerprint, _json(payload)) for relation_id, source_id, target_id, relation_type, fingerprint, payload in relation_rows],
|
|
)
|
|
self.record_change("topology_snapshot", f"Topology snapshot captured ({snapshot.service_count} services)")
|
|
return snapshot
|
|
|
|
def list_snapshots(self, *, limit: int = 50) -> list[TopologySnapshot]:
|
|
with self._connect() as db:
|
|
rows = db.execute("SELECT * FROM topology_snapshots ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall()
|
|
return [self._snapshot_row(row) for row in rows if self._snapshot_row(row) is not None] # type: ignore[list-item]
|
|
|
|
def topology_diff(self, *, after_snapshot_id: str, before_snapshot_id: str | None = None) -> TopologyDiff:
|
|
with self._connect() as db:
|
|
after = db.execute("SELECT * FROM topology_snapshots WHERE id=?", (after_snapshot_id,)).fetchone()
|
|
if after is None:
|
|
raise ValueError("Topology snapshot not found")
|
|
before = (
|
|
db.execute("SELECT * FROM topology_snapshots WHERE id=?", (before_snapshot_id,)).fetchone()
|
|
if before_snapshot_id
|
|
else db.execute("SELECT * FROM topology_snapshots WHERE created_at < ? ORDER BY created_at DESC LIMIT 1", (after["created_at"],)).fetchone()
|
|
)
|
|
after_services = {
|
|
row["service_id"]: row
|
|
for row in db.execute("SELECT * FROM topology_snapshot_services WHERE snapshot_id=?", (after_snapshot_id,)).fetchall()
|
|
}
|
|
before_services = (
|
|
{row["service_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_services WHERE snapshot_id=?", (before["id"],)).fetchall()}
|
|
if before else {}
|
|
)
|
|
after_relations = {
|
|
row["relation_id"]: row
|
|
for row in db.execute("SELECT * FROM topology_snapshot_relations WHERE snapshot_id=?", (after_snapshot_id,)).fetchall()
|
|
}
|
|
before_relations = (
|
|
{row["relation_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_relations WHERE snapshot_id=?", (before["id"],)).fetchall()}
|
|
if before else {}
|
|
)
|
|
|
|
added: list[TopologyChange] = []
|
|
removed: list[TopologyChange] = []
|
|
changed: list[TopologyChange] = []
|
|
for item_id, row in after_services.items():
|
|
payload = _obj(row["payload"])
|
|
if item_id not in before_services:
|
|
added.append(TopologyChange(change_type="added", entity_type="service", entity_id=item_id, name=row["name"], after=payload))
|
|
elif row["fingerprint"] != before_services[item_id]["fingerprint"]:
|
|
changed.append(TopologyChange(change_type="changed", entity_type="service", entity_id=item_id, name=row["name"], before=_obj(before_services[item_id]["payload"]), after=payload))
|
|
for item_id, row in before_services.items():
|
|
if item_id not in after_services:
|
|
removed.append(TopologyChange(change_type="removed", entity_type="service", entity_id=item_id, name=row["name"], before=_obj(row["payload"])))
|
|
for item_id, row in after_relations.items():
|
|
name = f"{row['source_service_id']} → {row['target_service_id'] or 'unknown'}"
|
|
if item_id not in before_relations:
|
|
added.append(TopologyChange(change_type="added", entity_type="relationship", entity_id=item_id, name=name, after=_obj(row["payload"])))
|
|
elif row["fingerprint"] != before_relations[item_id]["fingerprint"]:
|
|
changed.append(TopologyChange(change_type="changed", entity_type="relationship", entity_id=item_id, name=name, before=_obj(before_relations[item_id]["payload"]), after=_obj(row["payload"])))
|
|
for item_id, row in before_relations.items():
|
|
if item_id not in after_relations:
|
|
removed.append(TopologyChange(change_type="removed", entity_type="relationship", entity_id=item_id, name=f"{row['source_service_id']} → {row['target_service_id'] or 'unknown'}", before=_obj(row["payload"])))
|
|
return TopologyDiff(
|
|
before_snapshot_id=before["id"] if before else None,
|
|
after_snapshot_id=after_snapshot_id,
|
|
added=added,
|
|
removed=removed,
|
|
changed=changed,
|
|
summary={"added": len(added), "removed": len(removed), "changed": len(changed)},
|
|
)
|
|
|
|
def save_source(
|
|
self,
|
|
*,
|
|
source_type: str,
|
|
source_id: str,
|
|
source_name: str | None,
|
|
lifecycle: DiscoveryLifecycle,
|
|
metadata: dict[str, object],
|
|
managed: bool = True,
|
|
observed_at: datetime | None = None,
|
|
) -> ServiceSource:
|
|
at = observed_at or _now()
|
|
with self._connect() as db:
|
|
existing = db.execute(
|
|
"SELECT * FROM discovery_sources WHERE source_type=? AND source_id=?",
|
|
(source_type, source_id),
|
|
).fetchone()
|
|
source_pk = existing["id"] if existing else f"src-{uuid4().hex[:14]}"
|
|
first_seen = existing["first_seen_at"] if existing else at.isoformat()
|
|
db.execute(
|
|
"""
|
|
INSERT INTO discovery_sources
|
|
(id, source_type, source_id, source_name, first_seen_at, last_seen_at, lifecycle, managed, metadata)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(source_type, source_id) DO UPDATE SET
|
|
source_name=excluded.source_name, last_seen_at=excluded.last_seen_at,
|
|
lifecycle=excluded.lifecycle, managed=excluded.managed, metadata=excluded.metadata
|
|
""",
|
|
(source_pk, source_type, source_id, source_name, first_seen, at.isoformat(), lifecycle.value, int(managed), _json(metadata)),
|
|
)
|
|
row = db.execute("SELECT * FROM discovery_sources WHERE id=?", (source_pk,)).fetchone()
|
|
return self._source(row)
|
|
|
|
def attach_source(self, service_id: str, source_id: str, role: str = "observed") -> None:
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"INSERT OR IGNORE INTO service_sources (service_id, source_id, role) VALUES (?, ?, ?)",
|
|
(service_id, source_id, role),
|
|
)
|
|
|
|
def get_source(self, source_type: str, source_id: str) -> ServiceSource | None:
|
|
with self._connect() as db:
|
|
row = db.execute(
|
|
"SELECT * FROM discovery_sources WHERE source_type=? AND source_id=?",
|
|
(source_type, source_id),
|
|
).fetchone()
|
|
return self._source(row) if row else None
|
|
|
|
def save_observation(
|
|
self,
|
|
*,
|
|
run_id: str,
|
|
source: ServiceSource,
|
|
source_name: str,
|
|
lifecycle: DiscoveryLifecycle,
|
|
change_type: str,
|
|
metadata: dict[str, object],
|
|
observed_at: datetime | None = None,
|
|
) -> DiscoveryObservation:
|
|
at = observed_at or _now()
|
|
observation = DiscoveryObservation(
|
|
id=f"obs-{uuid4().hex[:14]}",
|
|
run_id=run_id,
|
|
source_id=source.id,
|
|
source_type=source.source_type,
|
|
source_name=source_name,
|
|
lifecycle=lifecycle,
|
|
change_type=change_type,
|
|
observed_at=at,
|
|
metadata=metadata,
|
|
)
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"INSERT INTO discovery_observations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(observation.id, run_id, source.id, source.source_type, source_name, lifecycle.value, change_type, at.isoformat(), _json(metadata)),
|
|
)
|
|
return observation
|
|
|
|
def save_suggestion(self, suggestion: TopologySuggestion, fingerprint: str) -> TopologySuggestion:
|
|
with self._connect() as db:
|
|
existing = db.execute(
|
|
"SELECT evidence, confidence FROM topology_suggestions WHERE fingerprint=?", (fingerprint,)
|
|
).fetchone()
|
|
merged_evidence = list(dict.fromkeys([
|
|
*(_list(existing["evidence"]) if existing else []),
|
|
*suggestion.evidence,
|
|
]))
|
|
confidence = _higher_confidence(
|
|
suggestion.confidence,
|
|
DiscoveryConfidence(existing["confidence"]) if existing else suggestion.confidence,
|
|
)
|
|
db.execute(
|
|
"""
|
|
INSERT INTO topology_suggestions
|
|
(id, run_id, suggestion_type, status, fingerprint, source_ref, source_service_id, target_ref,
|
|
target_service_id, proposed_name, proposed_kind, proposed_url, proposed_host, proposed_port,
|
|
proposed_category, proposed_group, relation_type, confidence, evidence, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(fingerprint) DO UPDATE SET
|
|
run_id=excluded.run_id,
|
|
source_ref=excluded.source_ref,
|
|
source_service_id=excluded.source_service_id,
|
|
target_ref=excluded.target_ref,
|
|
target_service_id=excluded.target_service_id,
|
|
proposed_name=excluded.proposed_name,
|
|
proposed_kind=excluded.proposed_kind,
|
|
proposed_url=excluded.proposed_url,
|
|
proposed_host=excluded.proposed_host,
|
|
proposed_port=excluded.proposed_port,
|
|
proposed_category=excluded.proposed_category,
|
|
proposed_group=excluded.proposed_group,
|
|
relation_type=excluded.relation_type,
|
|
confidence=excluded.confidence,
|
|
evidence=excluded.evidence,
|
|
updated_at=excluded.updated_at
|
|
""",
|
|
(
|
|
suggestion.id,
|
|
suggestion.run_id,
|
|
suggestion.suggestion_type,
|
|
suggestion.status.value,
|
|
fingerprint,
|
|
suggestion.source_ref,
|
|
suggestion.source_service_id,
|
|
suggestion.target_ref,
|
|
suggestion.target_service_id,
|
|
suggestion.proposed_name,
|
|
suggestion.proposed_kind.value if suggestion.proposed_kind else None,
|
|
suggestion.proposed_url,
|
|
suggestion.proposed_host,
|
|
suggestion.proposed_port,
|
|
suggestion.proposed_category.value,
|
|
suggestion.proposed_group,
|
|
suggestion.relation_type.value if suggestion.relation_type else None,
|
|
confidence.value,
|
|
_json(merged_evidence),
|
|
suggestion.created_at.isoformat(),
|
|
suggestion.updated_at.isoformat(),
|
|
),
|
|
)
|
|
row = db.execute("SELECT * FROM topology_suggestions WHERE fingerprint=?", (fingerprint,)).fetchone()
|
|
return self._suggestion(row)
|
|
|
|
def list_suggestions(
|
|
self,
|
|
status: SuggestionStatus | None = None,
|
|
limit: int = 500,
|
|
run_id: str | None = None,
|
|
) -> list[TopologySuggestion]:
|
|
query = "SELECT * FROM topology_suggestions"
|
|
params: list[Any] = []
|
|
filters: list[str] = []
|
|
if status:
|
|
filters.append("status = ?")
|
|
params.append(status.value)
|
|
if run_id:
|
|
filters.append("run_id = ?")
|
|
params.append(run_id)
|
|
if filters:
|
|
query += " WHERE " + " AND ".join(filters)
|
|
query += " ORDER BY updated_at DESC LIMIT ?"
|
|
params.append(limit)
|
|
with self._connect() as db:
|
|
rows = db.execute(query, params).fetchall()
|
|
return [self._suggestion(row) for row in rows]
|
|
|
|
def get_suggestion(self, suggestion_id: str) -> TopologySuggestion | None:
|
|
with self._connect() as db:
|
|
row = db.execute("SELECT * FROM topology_suggestions WHERE id=?", (suggestion_id,)).fetchone()
|
|
return self._suggestion(row) if row else None
|
|
|
|
def set_suggestion_status(self, suggestion_id: str, status: SuggestionStatus) -> TopologySuggestion | None:
|
|
now = _now()
|
|
with self._connect() as db:
|
|
db.execute("UPDATE topology_suggestions SET status=?, updated_at=? WHERE id=?", (status.value, now.isoformat(), suggestion_id))
|
|
row = db.execute("SELECT * FROM topology_suggestions WHERE id=?", (suggestion_id,)).fetchone()
|
|
return self._suggestion(row) if row else None
|
|
|
|
def save_evidence(self, evidence: RelationEvidence) -> RelationEvidence:
|
|
with self._connect() as db:
|
|
existing = db.execute(
|
|
"""
|
|
SELECT * FROM relation_evidence
|
|
WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ?
|
|
AND relation_type=? AND evidence_type=?
|
|
""",
|
|
(evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type),
|
|
).fetchone()
|
|
if existing:
|
|
previous = self._evidence(existing)
|
|
evidence = evidence.model_copy(update={
|
|
"confidence": _higher_confidence(previous.confidence, evidence.confidence),
|
|
"explanation": "; ".join(dict.fromkeys(part for part in (previous.explanation, evidence.explanation) if part)),
|
|
"source_ref": evidence.source_ref or previous.source_ref,
|
|
"locked": previous.locked or evidence.locked,
|
|
"active": evidence.active,
|
|
"first_seen_at": min(previous.first_seen_at, evidence.first_seen_at),
|
|
})
|
|
db.execute(
|
|
"""
|
|
INSERT INTO relation_evidence
|
|
(id, source_service_id, target_service_id, target_ref, relation_type, evidence_type,
|
|
confidence, explanation, source_ref, locked, active, first_seen_at, last_seen_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(source_service_id, target_service_id, target_ref, relation_type, evidence_type) DO UPDATE SET
|
|
target_service_id=excluded.target_service_id, confidence=excluded.confidence,
|
|
explanation=excluded.explanation, source_ref=excluded.source_ref,
|
|
locked=MAX(relation_evidence.locked, excluded.locked), active=excluded.active,
|
|
last_seen_at=excluded.last_seen_at
|
|
""",
|
|
(evidence.id, evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type, evidence.confidence.value, evidence.explanation, evidence.source_ref, int(evidence.locked), int(evidence.active), evidence.first_seen_at.isoformat(), evidence.last_seen_at.isoformat()),
|
|
)
|
|
row = db.execute(
|
|
"SELECT * FROM relation_evidence WHERE id=?",
|
|
(evidence.id,),
|
|
).fetchone()
|
|
if row is None:
|
|
row = db.execute(
|
|
"SELECT * FROM relation_evidence WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ? AND relation_type=? AND evidence_type=?",
|
|
(evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type),
|
|
).fetchone()
|
|
return self._evidence(row)
|
|
|
|
def list_logical_relations(self, service_id: str | None = None, active_only: bool = True) -> list[LogicalRelation]:
|
|
"""Return one canonical graph edge while retaining every evidence row.
|
|
|
|
`relation_evidence` is an append/merge audit record. Rendering every
|
|
row as an edge made a relation appear duplicated whenever independent
|
|
discovery sources agreed. The aggregation is deliberately query-time
|
|
so existing v1.3 evidence is preserved without a destructive rebuild.
|
|
"""
|
|
|
|
grouped: dict[tuple[str, str | None, str | None, RelationType], list[RelationEvidence]] = {}
|
|
for evidence in self.list_evidence(service_id=service_id, active_only=active_only):
|
|
key = (evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type)
|
|
grouped.setdefault(key, []).append(evidence)
|
|
result: list[LogicalRelation] = []
|
|
for (source_id, target_id, target_ref, relation_type), evidence_rows in grouped.items():
|
|
confidence = _higher_confidence(*(item.confidence for item in evidence_rows))
|
|
independent_sources = {f"{item.evidence_type}:{item.source_ref or item.explanation}" for item in evidence_rows}
|
|
if len(independent_sources) > 1 and confidence != DiscoveryConfidence.CONFIRMED:
|
|
confidence = DiscoveryConfidence(_CONFIDENCE_RANK_INV[_CONFIDENCE_RANK[confidence] + 1])
|
|
first_seen = min(item.first_seen_at for item in evidence_rows)
|
|
last_seen = max(item.last_seen_at for item in evidence_rows)
|
|
digest = sha256("|".join((source_id, target_id or "", target_ref or "", relation_type.value)).encode("utf-8")).hexdigest()[:16]
|
|
result.append(LogicalRelation(
|
|
id=f"relation-{digest}",
|
|
source_service_id=source_id,
|
|
target_service_id=target_id,
|
|
target_ref=target_ref,
|
|
relation_type=relation_type,
|
|
confidence=confidence,
|
|
active=any(item.active for item in evidence_rows),
|
|
locked=any(item.locked for item in evidence_rows),
|
|
first_seen_at=first_seen,
|
|
last_seen_at=last_seen,
|
|
evidence=sorted(evidence_rows, key=lambda item: item.last_seen_at, reverse=True),
|
|
))
|
|
return sorted(result, key=lambda item: item.last_seen_at, reverse=True)
|
|
|
|
def list_evidence(self, service_id: str | None = None, active_only: bool = True) -> list[RelationEvidence]:
|
|
query = "SELECT * FROM relation_evidence WHERE 1=1"
|
|
params: list[Any] = []
|
|
if service_id:
|
|
query += " AND (source_service_id=? OR target_service_id=?)"
|
|
params.extend([service_id, service_id])
|
|
if active_only:
|
|
query += " AND active=1"
|
|
query += " ORDER BY last_seen_at DESC"
|
|
with self._connect() as db:
|
|
rows = db.execute(query, params).fetchall()
|
|
return [self._evidence(row) for row in rows]
|
|
|
|
def get_evidence(self, evidence_id: str) -> RelationEvidence | None:
|
|
with self._connect() as db:
|
|
row = db.execute("SELECT * FROM relation_evidence WHERE id=?", (evidence_id,)).fetchone()
|
|
return self._evidence(row) if row else None
|
|
|
|
def get_evidence_for_relation(
|
|
self,
|
|
source_service_id: str,
|
|
target_service_id: str | None,
|
|
target_ref: str | None,
|
|
relation_type: RelationType,
|
|
evidence_type: str = "discovery",
|
|
) -> RelationEvidence | None:
|
|
with self._connect() as db:
|
|
row = db.execute(
|
|
"""
|
|
SELECT * FROM relation_evidence
|
|
WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ?
|
|
AND relation_type=? AND evidence_type=?
|
|
""",
|
|
(source_service_id, target_service_id, target_ref, relation_type.value, evidence_type),
|
|
).fetchone()
|
|
return self._evidence(row) if row else None
|
|
|
|
def set_evidence_state(
|
|
self,
|
|
evidence_id: str,
|
|
*,
|
|
active: bool | None = None,
|
|
locked: bool | None = None,
|
|
) -> RelationEvidence | None:
|
|
current = self.get_evidence(evidence_id)
|
|
if current is None:
|
|
return None
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"UPDATE relation_evidence SET active=?, locked=?, last_seen_at=? WHERE id=?",
|
|
(int(current.active if active is None else active), int(current.locked if locked is None else locked), _now().isoformat(), evidence_id),
|
|
)
|
|
return self.get_evidence(evidence_id)
|
|
|
|
def evidence_for_service(self, service_id: str) -> list[RelationEvidence]:
|
|
return self.list_evidence(service_id, active_only=True)
|
|
|
|
def sources_for_service(self, service_id: str) -> list[ServiceSource]:
|
|
with self._connect() as db:
|
|
rows = db.execute(
|
|
"SELECT s.* FROM discovery_sources s JOIN service_sources ss ON ss.source_id=s.id WHERE ss.service_id=? ORDER BY s.source_type, s.source_name",
|
|
(service_id,),
|
|
).fetchall()
|
|
return [self._source(row) for row in rows]
|
|
|
|
def observations_for_run(self, run_id: str, limit: int = 1000) -> list[DiscoveryObservation]:
|
|
with self._connect() as db:
|
|
rows = db.execute(
|
|
"SELECT * FROM discovery_observations WHERE run_id=? ORDER BY source_name COLLATE NOCASE LIMIT ?",
|
|
(run_id, limit),
|
|
).fetchall()
|
|
return [
|
|
DiscoveryObservation(
|
|
id=row["id"],
|
|
run_id=row["run_id"],
|
|
source_id=row["source_id"],
|
|
source_type=row["source_type"],
|
|
source_name=row["source_name"],
|
|
lifecycle=DiscoveryLifecycle(row["lifecycle"]),
|
|
change_type=row["change_type"],
|
|
observed_at=datetime.fromisoformat(row["observed_at"]),
|
|
metadata=_obj(row["metadata"]),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def candidates_for_run(self, run_id: str, limit: int = 1000) -> list[dict[str, object]]:
|
|
candidates: list[dict[str, object]] = []
|
|
for observation in self.observations_for_run(run_id, limit):
|
|
candidate = observation.metadata.get("candidate")
|
|
if isinstance(candidate, dict):
|
|
candidates.append(candidate)
|
|
return candidates
|
|
|
|
def set_run_status(self, run_id: str, status: DiscoveryRunStatus) -> None:
|
|
with self._connect() as db:
|
|
db.execute("UPDATE discovery_runs SET status=? WHERE id=?", (status.value, run_id))
|
|
|
|
def record_change(
|
|
self,
|
|
event_type: str,
|
|
detail: str,
|
|
*,
|
|
service_id: str | None = None,
|
|
source_service_id: str | None = None,
|
|
target_service_id: str | None = None,
|
|
) -> None:
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"INSERT INTO topology_change_events (id, event_type, service_id, source_service_id, target_service_id, detail, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(f"change-{uuid4().hex[:14]}", event_type, service_id, source_service_id, target_service_id, detail, _now().isoformat()),
|
|
)
|
|
|
|
def save_override(self, target_type: str, target_id: str, field_name: str, value: object, locked: bool = True) -> None:
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"INSERT INTO topology_overrides (target_type, target_id, field_name, value, locked, updated_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(target_type, target_id, field_name) DO UPDATE SET value=excluded.value, locked=excluded.locked, updated_at=excluded.updated_at",
|
|
(target_type, target_id, field_name, _json(value), int(locked), _now().isoformat()),
|
|
)
|
|
|
|
def has_override(self, target_type: str, target_id: str, field_name: str) -> bool:
|
|
with self._connect() as db:
|
|
row = db.execute(
|
|
"SELECT 1 FROM topology_overrides WHERE target_type=? AND target_id=? AND field_name=? AND locked=1",
|
|
(target_type, target_id, field_name),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
def list_overrides(self, target_type: str | None = None, target_id: str | None = None) -> list[dict[str, object]]:
|
|
query = "SELECT target_type, target_id, field_name, value, locked, updated_at FROM topology_overrides WHERE 1=1"
|
|
params: list[Any] = []
|
|
if target_type:
|
|
query += " AND target_type=?"
|
|
params.append(target_type)
|
|
if target_id:
|
|
query += " AND target_id=?"
|
|
params.append(target_id)
|
|
with self._connect() as db:
|
|
rows = db.execute(query, params).fetchall()
|
|
return [
|
|
{
|
|
"target_type": row["target_type"],
|
|
"target_id": row["target_id"],
|
|
"field": row["field_name"],
|
|
"value": _obj(row["value"]) if row["value"].startswith("{") else row["value"],
|
|
"locked": bool(row["locked"]),
|
|
"updated_at": row["updated_at"],
|
|
}
|
|
for row in rows
|
|
]
|