997 lines
51 KiB
Python
997 lines
51 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import threading
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from .discovery_store import DiscoveryStore
|
|
from .docker_discovery import discover_containers
|
|
from .evidence import appops_health_status
|
|
from .health import validate_dependencies
|
|
from .models import (
|
|
Criticality,
|
|
DiscoveryCategory,
|
|
DiscoveryConfidence,
|
|
DiscoveryDiff,
|
|
DiscoveryLifecycle,
|
|
DiscoveryReconcileRequest,
|
|
DiscoveryResult,
|
|
DiscoveryRun,
|
|
DiscoveryRunRequest,
|
|
DiscoveryRunStatus,
|
|
DockerCandidate,
|
|
EdgeMutationRequest,
|
|
EvidenceProvider,
|
|
EvidenceScope,
|
|
HealthEvidence,
|
|
HealthStatus,
|
|
MonitoringState,
|
|
ProbeEndpointSource,
|
|
ProbeReachability,
|
|
ProbeRepairChange,
|
|
ProbeRepairReport,
|
|
ProviderHealth,
|
|
ProviderStatus,
|
|
ReconcileResult,
|
|
RelationEvidence,
|
|
RelationType,
|
|
Service,
|
|
ServiceCreate,
|
|
ServiceKind,
|
|
ServiceUpdate,
|
|
SuggestionStatus,
|
|
TopologySuggestion,
|
|
)
|
|
from .observation import observation_ttl
|
|
from .store import Store
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _evidence_ttl():
|
|
try:
|
|
interval = max(10, int(os.getenv("SERVICEMAP_OBSERVATION_INTERVAL_SECONDS", "60")))
|
|
except ValueError:
|
|
interval = 60
|
|
return observation_ttl(interval)
|
|
|
|
|
|
def _fingerprint(*parts: object) -> str:
|
|
value = "|".join(str(part or "") for part in parts)
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _confidence(value: str | DiscoveryConfidence) -> DiscoveryConfidence:
|
|
try:
|
|
return value if isinstance(value, DiscoveryConfidence) else DiscoveryConfidence(value)
|
|
except ValueError:
|
|
return DiscoveryConfidence.LOW
|
|
|
|
|
|
def _criticality(value: str | None) -> Criticality:
|
|
normalized = (value or "medium").casefold()
|
|
if normalized in {item.value for item in Criticality}:
|
|
return Criticality(normalized)
|
|
if normalized in {"urgent", "very_high"}:
|
|
return Criticality.CRITICAL
|
|
return Criticality.MEDIUM
|
|
|
|
|
|
def _application_name(candidate: DockerCandidate) -> str:
|
|
explicit = candidate.labels.get("servicemap.application", "").strip()
|
|
if explicit:
|
|
return explicit
|
|
if candidate.compose_project:
|
|
return candidate.compose_project.replace("-", " ").replace("_", " ").title()
|
|
if candidate.appops_app_id:
|
|
return f"AppOps · {candidate.appops_app_id}"
|
|
return candidate.category.value.replace("_", " ").title()
|
|
|
|
|
|
def _metadata(candidate: DockerCandidate) -> dict[str, object]:
|
|
return {
|
|
"candidate": candidate.model_dump(mode="json"),
|
|
"stable_source_ref": candidate.stable_source_ref,
|
|
"container_id": candidate.container_id,
|
|
"image": candidate.image,
|
|
"state": candidate.state,
|
|
"health": candidate.health,
|
|
"networks": candidate.networks,
|
|
"ports": [item.model_dump(mode="json") for item in candidate.ports],
|
|
"labels": candidate.labels,
|
|
"compose_project": candidate.compose_project,
|
|
"compose_service": candidate.compose_service,
|
|
"category": candidate.category.value,
|
|
"visibility": candidate.visibility.value,
|
|
"environment_evidence": candidate.environment_evidence,
|
|
"proxy_route_evidence": candidate.proxy_route_evidence,
|
|
"appops_app_id": candidate.appops_app_id,
|
|
}
|
|
|
|
|
|
def _same_observation(previous: dict[str, object], current: dict[str, object]) -> bool:
|
|
ignored = {"state", "health", "container_id"}
|
|
return {key: value for key, value in previous.items() if key not in ignored} == {
|
|
key: value for key, value in current.items() if key not in ignored
|
|
}
|
|
|
|
|
|
class DiscoveryManager:
|
|
def __init__(self, store: Store, discovery_store: DiscoveryStore) -> None:
|
|
self.store = store
|
|
self.discovery_store = discovery_store
|
|
self._lock = threading.Lock()
|
|
|
|
def _docker_host(self, requested: str | None) -> str:
|
|
return requested or os.getenv("SERVICEMAP_DOCKER_HOST", "unix:///var/run/docker.sock")
|
|
|
|
def _appops_url(self, requested: str | None) -> str | None:
|
|
return requested or os.getenv("SERVICEMAP_APPOPS_URL")
|
|
|
|
def _existing_service_for_ref(self, services: list[Service], source_ref: str | None) -> Service | None:
|
|
if not source_ref:
|
|
return None
|
|
return next((item for item in services if item.source_ref == source_ref), None)
|
|
|
|
def _candidate_suggestion(self, candidate: DockerCandidate):
|
|
return candidate.suggestions[0] if candidate.suggestions else None
|
|
|
|
def _relation_suggestions(
|
|
self,
|
|
run: DiscoveryRun,
|
|
candidates: list[DockerCandidate],
|
|
service_by_ref: dict[str, Service],
|
|
) -> list[TopologySuggestion]:
|
|
result: list[TopologySuggestion] = []
|
|
|
|
def aliases(candidate: DockerCandidate) -> set[str]:
|
|
return {item.casefold() for item in (candidate.name, candidate.compose_service or "") if item}
|
|
|
|
def resolve(source: DockerCandidate, reference: str, *, compose_only: bool = False) -> list[DockerCandidate]:
|
|
"""Resolve only inside a defensible compose/network scope.
|
|
|
|
A duplicate service name must remain a review suggestion. Choosing
|
|
the last name seen in a host-wide index is not topology evidence.
|
|
"""
|
|
token = reference.casefold().strip()
|
|
matches = [item for item in candidates if token in aliases(item) or token == (item.stable_source_ref or "").casefold()]
|
|
same_compose = [item for item in matches if source.compose_project and item.compose_project and item.compose_project.casefold() == source.compose_project.casefold()]
|
|
if compose_only:
|
|
return same_compose
|
|
if same_compose:
|
|
return same_compose
|
|
shared_network = [item for item in matches if set(source.networks).intersection(item.networks)]
|
|
return shared_network or matches
|
|
|
|
def add(source: DockerCandidate, target: DockerCandidate, confidence: DiscoveryConfidence, evidence: str, relation_type: RelationType = RelationType.DEPENDS_ON) -> None:
|
|
if source.stable_source_ref is None or target.stable_source_ref is None or source is target:
|
|
return
|
|
# Evidence changes as discovery learns more. It must enrich the
|
|
# same relationship rather than create a second suggestion/edge.
|
|
fingerprint = _fingerprint("relation", source.stable_source_ref, target.stable_source_ref, relation_type.value)
|
|
now = _now()
|
|
result.append(
|
|
self.discovery_store.save_suggestion(
|
|
TopologySuggestion(
|
|
id=f"suggest-{fingerprint[:14]}",
|
|
run_id=run.id,
|
|
suggestion_type="relation",
|
|
source_ref=source.stable_source_ref,
|
|
source_service_id=service_by_ref.get(source.stable_source_ref).id if source.stable_source_ref in service_by_ref else None,
|
|
target_ref=target.stable_source_ref,
|
|
target_service_id=service_by_ref.get(target.stable_source_ref).id if target.stable_source_ref in service_by_ref else None,
|
|
proposed_name=f"{source.name} depends on {target.name}",
|
|
relation_type=relation_type,
|
|
confidence=confidence,
|
|
evidence=[evidence],
|
|
created_at=now,
|
|
updated_at=now,
|
|
),
|
|
fingerprint,
|
|
)
|
|
)
|
|
|
|
def resolve_and_add(
|
|
source: DockerCandidate,
|
|
reference: str,
|
|
confidence: DiscoveryConfidence,
|
|
evidence: str,
|
|
*,
|
|
compose_only: bool = False,
|
|
relation_type: RelationType = RelationType.DEPENDS_ON,
|
|
) -> None:
|
|
matches = resolve(source, reference, compose_only=compose_only)
|
|
if len(matches) == 1:
|
|
add(source, matches[0], confidence, evidence, relation_type)
|
|
return
|
|
if not matches or not source.stable_source_ref:
|
|
return
|
|
candidates_label = ", ".join(sorted(item.stable_source_ref or item.name for item in matches)[:6])
|
|
fingerprint = _fingerprint("ambiguous-relation", source.stable_source_ref, reference.casefold(), RelationType.DEPENDS_ON.value)
|
|
now = _now()
|
|
result.append(
|
|
self.discovery_store.save_suggestion(
|
|
TopologySuggestion(
|
|
id=f"suggest-{fingerprint[:14]}",
|
|
run_id=run.id,
|
|
suggestion_type="relation",
|
|
source_ref=source.stable_source_ref,
|
|
source_service_id=service_by_ref.get(source.stable_source_ref).id if source.stable_source_ref in service_by_ref else None,
|
|
proposed_name=f"{source.name} has an ambiguous reference to {reference}",
|
|
relation_type=RelationType.DEPENDS_ON,
|
|
confidence=DiscoveryConfidence.LOW,
|
|
evidence=[f"{evidence}; candidates: {candidates_label}", "Manual review required; no canonical dependency was created"],
|
|
created_at=now,
|
|
updated_at=now,
|
|
),
|
|
fingerprint,
|
|
)
|
|
)
|
|
|
|
for source in candidates:
|
|
if not source.stable_source_ref:
|
|
continue
|
|
labels = source.labels
|
|
explicit = [item.strip() for item in labels.get("servicemap.depends_on", "").split(",") if item.strip()]
|
|
compose = [item.split(":", 1)[0].strip() for item in labels.get("com.docker.compose.depends_on", "").split(",") if item.strip()]
|
|
for name in sorted(set(explicit)):
|
|
resolve_and_add(source, name, DiscoveryConfidence.CONFIRMED, "Explicit servicemap.depends_on label")
|
|
for name in sorted(set(compose)):
|
|
resolve_and_add(source, name, DiscoveryConfidence.HIGH, "Docker Compose com.docker.compose.depends_on label", compose_only=True)
|
|
for item in source.environment_evidence:
|
|
if "hostname=" not in item:
|
|
continue
|
|
hostname = item.split("hostname=", 1)[1].split(":", 1)[0].casefold()
|
|
resolve_and_add(source, hostname, DiscoveryConfidence.MEDIUM, f"Environment endpoint reference: {item}")
|
|
for item in source.proxy_route_evidence:
|
|
proxies = [proxy for proxy in candidates if proxy.category == DiscoveryCategory.PROXY and set(proxy.networks).intersection(source.networks)]
|
|
if len(proxies) == 1:
|
|
add(proxies[0], source, DiscoveryConfidence.MEDIUM, item, RelationType.ROUTES_TO)
|
|
# Shared networks are review-only hints. Cap them to avoid flooding
|
|
# a large host with O(n²) low-confidence suggestions.
|
|
for name in source.dependency_suggestions[:8]:
|
|
resolve_and_add(source, name, DiscoveryConfidence.LOW, f"Shared Docker network hint: {', '.join(source.networks)}")
|
|
return result
|
|
|
|
def run(self, request: DiscoveryRunRequest) -> DiscoveryResult:
|
|
if not self._lock.acquire(blocking=False):
|
|
raise ValueError("A discovery run is already in progress")
|
|
run = self.discovery_store.start_run()
|
|
try:
|
|
docker_host = self._docker_host(request.docker_host)
|
|
result = discover_containers(
|
|
docker_host,
|
|
self.store.list_services(),
|
|
include_stopped=request.include_stopped,
|
|
appops_url=self._appops_url(request.appops_url),
|
|
)
|
|
observed_at = _now()
|
|
evidence_ttl = _evidence_ttl()
|
|
self.store.set_provider_status(
|
|
ProviderStatus(
|
|
provider=EvidenceProvider.DOCKER,
|
|
status=ProviderHealth.HEALTHY,
|
|
observed_at=observed_at,
|
|
valid_until=observed_at + evidence_ttl,
|
|
detail=f"Docker inventory returned {len(result.candidates)} candidates",
|
|
metadata={"docker_host": docker_host},
|
|
)
|
|
)
|
|
if self._appops_url(request.appops_url):
|
|
self.store.set_provider_status(
|
|
ProviderStatus(
|
|
provider=EvidenceProvider.APPOPS,
|
|
status=ProviderHealth.HEALTHY if result.appops_available else ProviderHealth.UNAVAILABLE,
|
|
observed_at=observed_at,
|
|
valid_until=observed_at + evidence_ttl,
|
|
detail=result.appops_detail,
|
|
metadata={"mappings": sum(1 for item in result.candidates if item.appops_app_id)},
|
|
)
|
|
)
|
|
existing = self.store.list_services()
|
|
by_ref = {service.source_ref: service for service in existing if service.source_ref}
|
|
diff = DiscoveryDiff()
|
|
observations = []
|
|
source_counts: dict[str, int] = {"docker": len(result.candidates), "running": 0, "stopped": 0, "appops": 0, "proxy_routes": result.proxy_routes}
|
|
for candidate in result.candidates:
|
|
if candidate.state == "running":
|
|
source_counts["running"] += 1
|
|
else:
|
|
source_counts["stopped"] += 1
|
|
if candidate.appops_app_id:
|
|
source_counts["appops"] += 1
|
|
stable_ref = candidate.stable_source_ref or f"docker:name:{candidate.name.casefold()}"
|
|
previous = self.discovery_store.get_source("docker", stable_ref)
|
|
previous_meta = previous.metadata if previous else {}
|
|
missing_cycles = int(previous_meta.get("missing_cycles", 0) or 0) if previous else 0
|
|
if candidate.lifecycle == DiscoveryLifecycle.MISSING:
|
|
missing_cycles += 1
|
|
if missing_cycles >= 3:
|
|
candidate.lifecycle = DiscoveryLifecycle.RETIRED
|
|
elif missing_cycles >= 2:
|
|
candidate.lifecycle = DiscoveryLifecycle.STALE
|
|
else:
|
|
missing_cycles = 0
|
|
current_meta = _metadata(candidate)
|
|
current_meta["missing_cycles"] = missing_cycles
|
|
if candidate.lifecycle == DiscoveryLifecycle.MISSING:
|
|
change = "missing"
|
|
diff.missing += 1
|
|
elif candidate.lifecycle == DiscoveryLifecycle.STALE:
|
|
change = "stale"
|
|
diff.stale += 1
|
|
elif candidate.lifecycle == DiscoveryLifecycle.RETIRED:
|
|
change = "retired"
|
|
diff.stale += 1
|
|
elif previous is None:
|
|
change = "new"
|
|
diff.new += 1
|
|
elif _same_observation(previous_meta, current_meta):
|
|
change = "unchanged"
|
|
diff.unchanged += 1
|
|
else:
|
|
change = "changed"
|
|
diff.changed += 1
|
|
source = self.discovery_store.save_source(
|
|
source_type="docker",
|
|
source_id=stable_ref,
|
|
source_name=candidate.name,
|
|
lifecycle=candidate.lifecycle,
|
|
metadata=current_meta,
|
|
)
|
|
observations.append(
|
|
self.discovery_store.save_observation(
|
|
run_id=run.id,
|
|
source=source,
|
|
source_name=candidate.name,
|
|
lifecycle=candidate.lifecycle,
|
|
change_type=change,
|
|
metadata=current_meta,
|
|
)
|
|
)
|
|
suggestion = self._candidate_suggestion(candidate)
|
|
if suggestion and candidate.lifecycle != DiscoveryLifecycle.MISSING and candidate.visibility.value != "ignored":
|
|
fingerprint = _fingerprint("service", stable_ref)
|
|
now = _now()
|
|
saved = self.discovery_store.save_suggestion(
|
|
TopologySuggestion(
|
|
id=f"suggest-{fingerprint[:14]}",
|
|
run_id=run.id,
|
|
suggestion_type="service",
|
|
source_ref=stable_ref,
|
|
source_service_id=by_ref.get(stable_ref).id if stable_ref in by_ref else None,
|
|
proposed_name=suggestion.name,
|
|
proposed_kind=suggestion.kind,
|
|
proposed_url=suggestion.url,
|
|
proposed_host=suggestion.host,
|
|
proposed_port=suggestion.port,
|
|
proposed_category=candidate.category,
|
|
proposed_group=suggestion.group,
|
|
confidence=_confidence(suggestion.confidence),
|
|
evidence=suggestion.evidence + candidate.environment_evidence,
|
|
created_at=now,
|
|
updated_at=now,
|
|
),
|
|
fingerprint,
|
|
)
|
|
if saved.status == SuggestionStatus.OPEN and change in {"new", "changed"}:
|
|
diff.new_suggestions += 1
|
|
self._relation_suggestions(run, result.candidates, by_ref)
|
|
completed = self.discovery_store.finish_run(
|
|
run.id,
|
|
status=DiscoveryRunStatus.COMPLETED,
|
|
source_counts=source_counts,
|
|
diff=diff,
|
|
)
|
|
suggestions = self.discovery_store.list_suggestions(run_id=run.id)
|
|
return DiscoveryResult(
|
|
docker_host=docker_host,
|
|
candidates=result.candidates,
|
|
run=completed,
|
|
suggestions=suggestions,
|
|
observations=observations,
|
|
appops_available=result.appops_available,
|
|
appops_mappings=source_counts["appops"],
|
|
proxy_routes=result.proxy_routes,
|
|
)
|
|
except Exception as exc:
|
|
self.discovery_store.finish_run(
|
|
run.id,
|
|
status=DiscoveryRunStatus.FAILED,
|
|
source_counts={},
|
|
diff=DiscoveryDiff(),
|
|
error=str(exc)[:500],
|
|
)
|
|
raise
|
|
finally:
|
|
self._lock.release()
|
|
|
|
def status(self) -> dict[str, object]:
|
|
current, completed = self.discovery_store.status()
|
|
return {"current": current, "last_completed": completed, "active": current is not None}
|
|
|
|
def _payload_for_candidate(self, candidate: DockerCandidate, name: str | None = None) -> ServiceCreate:
|
|
suggestion = self._candidate_suggestion(candidate)
|
|
primary = next((item for item in candidate.probe_candidates if item.primary and item.reachability in {ProbeReachability.VERIFIED, ProbeReachability.REACHABLE}), None)
|
|
kind = primary.kind if primary else (suggestion.kind if suggestion else ServiceKind.TCP)
|
|
# Suggestions are discovery evidence, not authorization to probe. A
|
|
# published endpoint is only promoted after a bounded preflight marks
|
|
# it as the primary candidate.
|
|
url = primary.url if primary else None
|
|
host = primary.host if primary else None
|
|
port = primary.port if primary else None
|
|
# HTTP probing requires a complete URL. A container can expose only
|
|
# an internal HTTP-like port, which is useful classification evidence
|
|
# but is not a safe externally reachable endpoint for ServiceMap.
|
|
if kind == ServiceKind.HTTP:
|
|
# Normalize stale/legacy candidate suggestions as well as fresh
|
|
# Docker observations to the canonical URL-only representation.
|
|
host = None
|
|
port = None
|
|
monitoring = bool(url) if kind == ServiceKind.HTTP else bool(url or (host and port))
|
|
if not monitoring:
|
|
url = None
|
|
host = None
|
|
port = None
|
|
group = suggestion.group if suggestion else (f"Compose · {candidate.compose_project}" if candidate.compose_project else candidate.category.value.title())
|
|
if monitoring:
|
|
monitoring_state = MonitoringState.ACTIVE
|
|
probe_source = primary.source if primary else ProbeEndpointSource.KNOWN_URL
|
|
probe_reachability = primary.reachability if primary else ProbeReachability.REACHABLE
|
|
probe_confidence = primary.confidence if primary else DiscoveryConfidence.HIGH
|
|
elif candidate.healthcheck_present:
|
|
monitoring_state = MonitoringState.DOCKER_HEALTH
|
|
probe_source = ProbeEndpointSource.DOCKER_HEALTHCHECK
|
|
probe_reachability = ProbeReachability.VERIFIED
|
|
probe_confidence = DiscoveryConfidence.HIGH
|
|
elif candidate.state == "running":
|
|
monitoring_state = MonitoringState.OBSERVED
|
|
probe_source = (
|
|
ProbeEndpointSource.CONTAINER_INTERNAL
|
|
if candidate.probe_candidates
|
|
else ProbeEndpointSource.DISCOVERED_ENVIRONMENT
|
|
)
|
|
probe_reachability = (
|
|
ProbeReachability.INTERNAL_ONLY
|
|
if candidate.probe_candidates
|
|
else ProbeReachability.NOT_CONFIGURED
|
|
)
|
|
probe_confidence = DiscoveryConfidence.LOW
|
|
else:
|
|
monitoring_state = MonitoringState.UNMONITORED
|
|
probe_source = ProbeEndpointSource.CONTAINER_INTERNAL if candidate.probe_candidates else ProbeEndpointSource.DISCOVERED_ENVIRONMENT
|
|
probe_reachability = ProbeReachability.INTERNAL_ONLY if candidate.probe_candidates else ProbeReachability.NOT_CONFIGURED
|
|
probe_confidence = DiscoveryConfidence.LOW
|
|
return ServiceCreate(
|
|
name=name or (suggestion.name if suggestion else candidate.name),
|
|
kind=kind,
|
|
url=url,
|
|
host=host,
|
|
port=port,
|
|
description=f"Discovered Docker service from {candidate.image}",
|
|
tags=["discovered", "docker"],
|
|
group=group,
|
|
environment="production",
|
|
criticality=_criticality(candidate.appops_criticality),
|
|
category=candidate.category,
|
|
visibility=candidate.visibility,
|
|
lifecycle=candidate.lifecycle,
|
|
source_type="docker",
|
|
source_ref=candidate.stable_source_ref,
|
|
first_seen_at=_now(),
|
|
last_seen_at=_now(),
|
|
last_discovery_status=candidate.state,
|
|
container_image=candidate.image,
|
|
compose_project=candidate.compose_project,
|
|
monitoring_enabled=monitoring,
|
|
monitoring_state=monitoring_state,
|
|
probe_source=probe_source,
|
|
probe_reachability=probe_reachability,
|
|
probe_confidence=probe_confidence,
|
|
runtime_state=candidate.state,
|
|
runtime_health=candidate.health,
|
|
application=_application_name(candidate),
|
|
enabled=True,
|
|
)
|
|
|
|
def _update_discovered_service(self, service: Service, candidate: DockerCandidate) -> Service:
|
|
payload = self._payload_for_candidate(candidate, service.name)
|
|
updates: dict[str, Any] = payload.model_dump()
|
|
overrides = {
|
|
"name": self.discovery_store.has_override("service", service.id, "name"),
|
|
"group": self.discovery_store.has_override("service", service.id, "group"),
|
|
"category": self.discovery_store.has_override("service", service.id, "category"),
|
|
"criticality": self.discovery_store.has_override("service", service.id, "criticality"),
|
|
"environment": self.discovery_store.has_override("service", service.id, "environment"),
|
|
"probe": self.discovery_store.has_override("service", service.id, "probe"),
|
|
"visibility": self.discovery_store.has_override("service", service.id, "visibility"),
|
|
}
|
|
current = service.model_dump()
|
|
for field, locked in overrides.items():
|
|
if locked:
|
|
if field == "probe":
|
|
for endpoint_field in ("kind", "url", "host", "port", "monitoring_enabled", "monitoring_state", "probe_source", "probe_reachability", "probe_confidence"):
|
|
updates[endpoint_field] = current[endpoint_field]
|
|
else:
|
|
updates[field] = current[field]
|
|
updates["depends_on"] = service.depends_on
|
|
# Layout is operator-owned state; rediscovery must not snap a manually
|
|
# positioned node back to an inferred default.
|
|
for field in ("layout_x", "layout_y", "layout_pinned"):
|
|
updates[field] = current[field]
|
|
updates["first_seen_at"] = service.first_seen_at or payload.first_seen_at
|
|
updates["last_seen_at"] = payload.last_seen_at
|
|
updates["lifecycle"] = candidate.lifecycle
|
|
updates["last_discovery_status"] = candidate.state
|
|
return self.store.update_service(service.id, ServiceUpdate(**updates)) or service
|
|
|
|
def _record_candidate_evidence(self, candidate: DockerCandidate, service: Service) -> None:
|
|
"""Persist source-scoped discovery claims without flattening their scope."""
|
|
|
|
now = _now()
|
|
evidence_ttl = _evidence_ttl()
|
|
health = (candidate.health or "").casefold()
|
|
runtime = candidate.state.casefold()
|
|
component_status: HealthStatus | None = None
|
|
if health in {"unhealthy", "down"} or runtime in {"exited", "dead", "stopped"}:
|
|
component_status = HealthStatus.DOWN
|
|
elif health == "degraded" or runtime == "restarting":
|
|
component_status = HealthStatus.DEGRADED
|
|
elif health == "healthy":
|
|
component_status = HealthStatus.HEALTHY
|
|
if component_status is not None:
|
|
self.store.record_health_evidence(
|
|
HealthEvidence(
|
|
id=str(uuid4()),
|
|
subject_type="service",
|
|
subject_id=service.id,
|
|
service_id=service.id,
|
|
provider=EvidenceProvider.DOCKER,
|
|
provider_ref=candidate.stable_source_ref,
|
|
status=component_status,
|
|
scope=EvidenceScope.COMPONENT,
|
|
confidence=0.95 if candidate.healthcheck_present else 0.8,
|
|
observed_at=now,
|
|
valid_until=now + evidence_ttl,
|
|
received_at=now,
|
|
summary="Docker healthcheck" if candidate.healthcheck_present else "Docker runtime state",
|
|
metadata={"container_id": candidate.container_id, "state": candidate.state, "health": candidate.health},
|
|
)
|
|
)
|
|
normalized_appops = appops_health_status(candidate.appops_health)
|
|
if candidate.appops_app_id and normalized_appops is not None:
|
|
self.store.record_health_evidence(
|
|
HealthEvidence(
|
|
id=str(uuid4()),
|
|
subject_type="application",
|
|
subject_id=f"appops:{candidate.appops_app_id}",
|
|
service_id=None,
|
|
provider=EvidenceProvider.APPOPS,
|
|
provider_ref=candidate.appops_app_id,
|
|
status=normalized_appops,
|
|
scope=EvidenceScope.APPLICATION,
|
|
confidence=candidate.appops_discovery_confidence if candidate.appops_discovery_confidence is not None else 0.5,
|
|
observed_at=now,
|
|
valid_until=now + evidence_ttl,
|
|
received_at=now,
|
|
summary="AppOps application health; not used as component direct health",
|
|
metadata={
|
|
"app_id": candidate.appops_app_id,
|
|
"category": candidate.appops_category,
|
|
"policy_profile": candidate.appops_policy_profile,
|
|
"app_type": candidate.appops_app_type,
|
|
"discovery_confidence": candidate.appops_discovery_confidence,
|
|
},
|
|
),
|
|
service_ids=[service.id],
|
|
)
|
|
|
|
def _service_by_ref(self) -> dict[str, Service]:
|
|
return {service.source_ref: service for service in self.store.list_services() if service.source_ref}
|
|
|
|
def reconcile(self, request: DiscoveryReconcileRequest) -> ReconcileResult:
|
|
run = self.discovery_store.get_run(request.run_id) if request.run_id else self.discovery_store.last_completed()
|
|
if run is None or run.status != DiscoveryRunStatus.COMPLETED:
|
|
raise ValueError("A completed discovery run is required before reconciliation")
|
|
self.discovery_store.set_run_status(run.id, DiscoveryRunStatus.RECONCILING)
|
|
candidates = [DockerCandidate.model_validate(item.get("candidate", item)) for item in self.discovery_store.candidates_for_run(run.id)]
|
|
by_ref = self._service_by_ref()
|
|
created = updated = unchanged = missing = retired = 0
|
|
service_ids: list[str] = []
|
|
for candidate in candidates:
|
|
if not candidate.stable_source_ref:
|
|
continue
|
|
existing = by_ref.get(candidate.stable_source_ref)
|
|
if candidate.lifecycle in {DiscoveryLifecycle.MISSING, DiscoveryLifecycle.STALE, DiscoveryLifecycle.RETIRED}:
|
|
if existing:
|
|
if candidate.lifecycle == DiscoveryLifecycle.MISSING:
|
|
missing += 1
|
|
else:
|
|
retired += 1
|
|
payload = ServiceUpdate.model_validate(existing.model_dump())
|
|
payload.lifecycle = candidate.lifecycle
|
|
payload.last_discovery_status = "missing"
|
|
payload.monitoring_enabled = False
|
|
self.store.update_service(existing.id, payload)
|
|
continue
|
|
if candidate.visibility.value == "ignored":
|
|
continue
|
|
if existing is None and candidate.state == "running" and request.apply_high_confidence:
|
|
payload = self._payload_for_candidate(candidate)
|
|
name_conflict = self.store.get_service_by_name(payload.name)
|
|
if name_conflict:
|
|
payload.name = f"{payload.name} [{candidate.container_id[:6]}]"
|
|
existing = self.store.create_service(payload)
|
|
by_ref[candidate.stable_source_ref] = existing
|
|
created += 1
|
|
self.discovery_store.record_change("discovered", f"Discovered {existing.name}", service_id=existing.id)
|
|
elif existing is not None:
|
|
current = existing.model_dump()
|
|
updated_service = self._update_discovered_service(existing, candidate)
|
|
unchanged += 1 if updated_service.model_dump() == current else 0
|
|
updated += 0 if updated_service.model_dump() == current else 1
|
|
existing = updated_service
|
|
by_ref[candidate.stable_source_ref] = existing
|
|
if existing:
|
|
service_ids.append(existing.id)
|
|
source = self.discovery_store.get_source("docker", candidate.stable_source_ref)
|
|
if source:
|
|
self.discovery_store.attach_source(existing.id, source.id)
|
|
self._record_candidate_evidence(candidate, existing)
|
|
|
|
# The v1.1 generic baseline is preserved for history but no longer
|
|
# appears as the primary production topology after explicit bootstrap.
|
|
if request.apply_high_confidence and created:
|
|
discovered_ids = set(service_ids)
|
|
for service in self.store.list_services():
|
|
if service.id in discovered_ids or service.source_type:
|
|
continue
|
|
payload = ServiceUpdate.model_validate(service.model_dump())
|
|
payload.visibility = "hidden"
|
|
payload.lifecycle = DiscoveryLifecycle.RETIRED
|
|
payload.monitoring_enabled = False
|
|
self.store.update_service(service.id, payload)
|
|
retired += 1
|
|
|
|
accepted_relations = 0
|
|
suggestions = self.discovery_store.list_suggestions(run_id=run.id)
|
|
by_ref = self._service_by_ref()
|
|
for suggestion in suggestions:
|
|
if suggestion.suggestion_type != "relation" or not suggestion.source_ref or not suggestion.target_ref:
|
|
continue
|
|
source = by_ref.get(suggestion.source_ref)
|
|
target = by_ref.get(suggestion.target_ref)
|
|
if not source or not target:
|
|
continue
|
|
relation_type = suggestion.relation_type or RelationType.DEPENDS_ON
|
|
previous_evidence = self.discovery_store.get_evidence_for_relation(
|
|
source.id,
|
|
target.id,
|
|
suggestion.target_ref,
|
|
relation_type,
|
|
)
|
|
confidence_rank = {
|
|
DiscoveryConfidence.LOW: 1,
|
|
DiscoveryConfidence.MEDIUM: 2,
|
|
DiscoveryConfidence.HIGH: 3,
|
|
DiscoveryConfidence.CONFIRMED: 4,
|
|
}
|
|
evidence_parts = list(suggestion.evidence)
|
|
if previous_evidence:
|
|
evidence_parts = [previous_evidence.explanation, *evidence_parts]
|
|
evidence_text = "; ".join(dict.fromkeys(part for part in evidence_parts if part))
|
|
evidence = RelationEvidence(
|
|
id=suggestion.id,
|
|
source_service_id=source.id,
|
|
target_service_id=target.id,
|
|
target_ref=suggestion.target_ref,
|
|
relation_type=relation_type,
|
|
evidence_type="discovery",
|
|
confidence=(
|
|
previous_evidence.confidence
|
|
if previous_evidence and confidence_rank[previous_evidence.confidence] >= confidence_rank[suggestion.confidence]
|
|
else suggestion.confidence
|
|
),
|
|
explanation=evidence_text,
|
|
source_ref=suggestion.source_ref,
|
|
locked=previous_evidence.locked if previous_evidence else suggestion.status == SuggestionStatus.ACCEPTED,
|
|
active=previous_evidence.active if previous_evidence else True,
|
|
first_seen_at=previous_evidence.first_seen_at if previous_evidence else suggestion.created_at,
|
|
last_seen_at=suggestion.updated_at,
|
|
)
|
|
self.discovery_store.save_evidence(evidence)
|
|
if previous_evidence and not previous_evidence.active:
|
|
# An operator rejected/removed this inferred edge. Keep the
|
|
# durable evidence for audit, but do not resurrect its health
|
|
# propagation on a later discovery cycle.
|
|
continue
|
|
should_accept = suggestion.status == SuggestionStatus.ACCEPTED or (
|
|
request.apply_high_confidence and suggestion.confidence in {DiscoveryConfidence.CONFIRMED, DiscoveryConfidence.HIGH}
|
|
)
|
|
if not should_accept or suggestion.relation_type != RelationType.DEPENDS_ON:
|
|
continue
|
|
if target.id in source.depends_on:
|
|
if suggestion.status != SuggestionStatus.ACCEPTED:
|
|
self.discovery_store.set_suggestion_status(suggestion.id, SuggestionStatus.ACCEPTED)
|
|
continue
|
|
try:
|
|
validate_dependencies([*source.depends_on, target.id], self.store, service_id=source.id)
|
|
except ValueError:
|
|
continue
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [*source.depends_on, target.id]
|
|
self.store.update_service(source.id, payload)
|
|
self.discovery_store.set_suggestion_status(suggestion.id, SuggestionStatus.ACCEPTED)
|
|
self.discovery_store.record_change("dependency_added", f"{source.name} depends on {target.name}", source_service_id=source.id, target_service_id=target.id)
|
|
accepted_relations += 1
|
|
|
|
final_run = self.discovery_store.finish_run(
|
|
run.id,
|
|
status=DiscoveryRunStatus.COMPLETED,
|
|
source_counts=run.source_counts,
|
|
diff=run.diff,
|
|
)
|
|
self.discovery_store.capture_snapshot(run_id=final_run.id)
|
|
return ReconcileResult(
|
|
run=final_run,
|
|
created=created,
|
|
updated=updated,
|
|
unchanged=unchanged,
|
|
missing=missing,
|
|
retired=retired,
|
|
accepted_relations=accepted_relations,
|
|
open_suggestions=len(self.discovery_store.list_suggestions(SuggestionStatus.OPEN)),
|
|
service_ids=sorted(set(service_ids)),
|
|
)
|
|
|
|
@staticmethod
|
|
def _endpoint_label(service: Service | ServiceCreate) -> str | None:
|
|
if service.url:
|
|
return service.url
|
|
if service.host:
|
|
return f"{service.host}{f':{service.port}' if service.port else ''}"
|
|
return None
|
|
|
|
def repair_probe_reachability(self, *, run_id: str | None = None, dry_run: bool = True) -> ProbeRepairReport:
|
|
"""Safely retire unsafe discovery endpoints without touching probe history.
|
|
|
|
Only Docker-managed services are considered. User probe overrides remain
|
|
authoritative and the operation intentionally never deletes a service,
|
|
relation, probe result, or discovery observation.
|
|
"""
|
|
|
|
run = self.discovery_store.get_run(run_id) if run_id else self.discovery_store.last_completed()
|
|
if run is None or run.status != DiscoveryRunStatus.COMPLETED:
|
|
raise ValueError("A completed discovery run is required before repairing probe reachability")
|
|
candidates = [DockerCandidate.model_validate(item.get("candidate", item)) for item in self.discovery_store.candidates_for_run(run.id)]
|
|
services = self._service_by_ref()
|
|
changes: list[ProbeRepairChange] = []
|
|
disabled_unsafe = enabled_published = docker_health_observed = 0
|
|
appops_observed = runtime_observed = unmonitored = 0
|
|
for candidate in candidates:
|
|
if not candidate.stable_source_ref or candidate.lifecycle != DiscoveryLifecycle.ACTIVE:
|
|
continue
|
|
current = services.get(candidate.stable_source_ref)
|
|
if current is None:
|
|
continue
|
|
desired = self._payload_for_candidate(candidate, current.name)
|
|
is_override = self.discovery_store.has_override("service", current.id, "probe")
|
|
next_endpoint = self._endpoint_label(desired)
|
|
previous_endpoint = self._endpoint_label(current)
|
|
if is_override:
|
|
next_endpoint = previous_endpoint
|
|
changed = (not is_override) and any(
|
|
getattr(current, field) != getattr(desired, field)
|
|
for field in ("kind", "url", "host", "port", "monitoring_enabled", "monitoring_state", "probe_source", "probe_reachability", "probe_confidence")
|
|
)
|
|
if not changed:
|
|
if not dry_run:
|
|
self.store.set_runtime_observation(
|
|
current.id,
|
|
runtime_state=candidate.state,
|
|
runtime_health=candidate.health,
|
|
monitoring_state=current.monitoring_state.value,
|
|
)
|
|
continue
|
|
if current.monitoring_enabled and not desired.monitoring_enabled:
|
|
reason = "Internal container endpoint is not reachable from ServiceMap; active probe disabled"
|
|
disabled_unsafe += 1
|
|
elif desired.monitoring_enabled:
|
|
reason = "Host-published endpoint selected using the configured probe host"
|
|
enabled_published += 1
|
|
elif candidate.healthcheck_present:
|
|
reason = "Docker healthcheck retained as runtime observation without a network probe"
|
|
docker_health_observed += 1
|
|
elif desired.monitoring_state == MonitoringState.OBSERVED:
|
|
reason = "No verified reachable endpoint; Docker runtime observation is retained without a network probe"
|
|
runtime_observed += 1
|
|
else:
|
|
reason = "No verified reachable endpoint; service remains observed as unmonitored"
|
|
unmonitored += 1
|
|
changes.append(
|
|
ProbeRepairChange(
|
|
service_id=current.id,
|
|
service_name=current.name,
|
|
previous_monitoring_state=current.monitoring_state,
|
|
next_monitoring_state=desired.monitoring_state,
|
|
previous_endpoint=previous_endpoint,
|
|
next_endpoint=next_endpoint,
|
|
next_probe_source=desired.probe_source,
|
|
next_probe_confidence=desired.probe_confidence,
|
|
reason=reason,
|
|
)
|
|
)
|
|
|
|
if not dry_run:
|
|
updated = self._update_discovered_service(current, candidate)
|
|
monitoring_state = updated.monitoring_state.value
|
|
self.store.set_runtime_observation(
|
|
updated.id,
|
|
runtime_state=candidate.state,
|
|
runtime_health=candidate.health,
|
|
monitoring_state=monitoring_state,
|
|
)
|
|
self.discovery_store.record_change("probe_endpoint_superseded", reason, service_id=updated.id)
|
|
resolved = 0
|
|
if not dry_run:
|
|
from .incidents import sync_incidents
|
|
|
|
before = len([item for item in self.store.list_incidents(active_only=True)])
|
|
sync_incidents(self.store, resolution_reason="Monitoring endpoint superseded; active probe was replaced by reachability-safe observation")
|
|
after = len([item for item in self.store.list_incidents(active_only=True)])
|
|
resolved = max(0, before - after)
|
|
return ProbeRepairReport(
|
|
run_id=run.id,
|
|
dry_run=dry_run,
|
|
inspected=sum(1 for candidate in candidates if candidate.stable_source_ref in services),
|
|
changed=len(changes),
|
|
disabled_unsafe=disabled_unsafe,
|
|
enabled_published=enabled_published,
|
|
docker_health_observed=docker_health_observed,
|
|
appops_observed=appops_observed,
|
|
runtime_observed=runtime_observed,
|
|
unmonitored=unmonitored,
|
|
resolved_incidents=resolved,
|
|
changes=changes,
|
|
)
|
|
|
|
def apply_edge(self, request: EdgeMutationRequest) -> RelationEvidence:
|
|
source = self.store.get_service(request.source_service_id)
|
|
target = self.store.get_service(request.target_service_id)
|
|
if source is None or target is None:
|
|
raise ValueError("Both edge services must exist")
|
|
if request.relation_type == RelationType.DEPENDS_ON and target.id not in source.depends_on:
|
|
validate_dependencies([*source.depends_on, target.id], self.store, service_id=source.id)
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [*source.depends_on, target.id]
|
|
self.store.update_service(source.id, payload)
|
|
now = _now()
|
|
evidence = RelationEvidence(
|
|
id=f"manual-edge-{uuid4().hex[:14]}",
|
|
source_service_id=source.id,
|
|
target_service_id=target.id,
|
|
relation_type=request.relation_type,
|
|
evidence_type="manual",
|
|
confidence=DiscoveryConfidence.CONFIRMED,
|
|
explanation=request.explanation,
|
|
source_ref=f"manual:{source.id}->{target.id}",
|
|
locked=request.lock,
|
|
first_seen_at=now,
|
|
last_seen_at=now,
|
|
)
|
|
saved = self.discovery_store.save_evidence(evidence)
|
|
self.discovery_store.save_override("relation", saved.id, "locked", request.lock, request.lock)
|
|
self.discovery_store.record_change("dependency_added", request.explanation, source_service_id=source.id, target_service_id=target.id)
|
|
return saved
|
|
|
|
def suggestion_action(self, suggestion_id: str, action: str) -> TopologySuggestion:
|
|
suggestion = self.discovery_store.get_suggestion(suggestion_id)
|
|
if suggestion is None:
|
|
raise ValueError("Suggestion not found")
|
|
if action == "reject":
|
|
if suggestion.suggestion_type == "relation" and suggestion.relation_type == RelationType.DEPENDS_ON:
|
|
source = self._service_by_ref().get(suggestion.source_ref or "")
|
|
target = self._service_by_ref().get(suggestion.target_ref or "")
|
|
existing_evidence = self.discovery_store.get_evidence(suggestion.id)
|
|
if source and target and (existing_evidence is None or not existing_evidence.locked):
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [item for item in source.depends_on if item != target.id]
|
|
self.store.update_service(source.id, payload)
|
|
if existing_evidence:
|
|
self.discovery_store.set_evidence_state(suggestion.id, active=False)
|
|
return self.discovery_store.set_suggestion_status(suggestion_id, SuggestionStatus.REJECTED) or suggestion
|
|
if action in {"confirm", "lock"}:
|
|
if suggestion.suggestion_type == "service":
|
|
if not suggestion.source_ref:
|
|
raise ValueError("Service suggestion has no stable source reference")
|
|
existing = self._service_by_ref().get(suggestion.source_ref)
|
|
if existing is None:
|
|
candidate = next(
|
|
(
|
|
DockerCandidate.model_validate(item.get("candidate", item))
|
|
for item in self.discovery_store.candidates_for_run(suggestion.run_id)
|
|
if item.get("candidate", {}).get("stable_source_ref") == suggestion.source_ref
|
|
),
|
|
None,
|
|
)
|
|
if candidate is None:
|
|
raise ValueError("Discovery observation for service suggestion is no longer available")
|
|
existing = self.store.create_service(self._payload_for_candidate(candidate))
|
|
source = self.discovery_store.get_source("docker", suggestion.source_ref)
|
|
if source:
|
|
self.discovery_store.attach_source(existing.id, source.id)
|
|
self.discovery_store.record_change("accepted", f"Accepted discovered service {existing.name}", service_id=existing.id)
|
|
self.discovery_store.save_override("service", existing.id, "source", suggestion.source_ref, True)
|
|
elif suggestion.suggestion_type == "relation":
|
|
source = self._service_by_ref().get(suggestion.source_ref or "")
|
|
target = self._service_by_ref().get(suggestion.target_ref or "")
|
|
if source is None or target is None:
|
|
raise ValueError("Accept the related services before accepting this relationship")
|
|
if suggestion.relation_type == RelationType.DEPENDS_ON and target.id not in source.depends_on:
|
|
validate_dependencies([*source.depends_on, target.id], self.store, service_id=source.id)
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [*source.depends_on, target.id]
|
|
self.store.update_service(source.id, payload)
|
|
now = _now()
|
|
self.discovery_store.save_evidence(
|
|
RelationEvidence(
|
|
id=suggestion.id,
|
|
source_service_id=source.id,
|
|
target_service_id=target.id,
|
|
target_ref=suggestion.target_ref,
|
|
relation_type=suggestion.relation_type or RelationType.DEPENDS_ON,
|
|
evidence_type="discovery",
|
|
confidence=suggestion.confidence,
|
|
explanation="; ".join(suggestion.evidence),
|
|
source_ref=suggestion.source_ref,
|
|
locked=action == "lock",
|
|
first_seen_at=suggestion.created_at,
|
|
last_seen_at=now,
|
|
)
|
|
)
|
|
self.discovery_store.record_change("dependency_added", f"Accepted relationship {source.name} → {target.name}", source_service_id=source.id, target_service_id=target.id)
|
|
return self.discovery_store.set_suggestion_status(suggestion_id, SuggestionStatus.ACCEPTED) or suggestion
|
|
raise ValueError("Unsupported suggestion action")
|
|
|
|
def record_service_override(self, before: Service, after: Service) -> None:
|
|
for field in ("name", "group", "category", "criticality", "environment", "visibility"):
|
|
if getattr(before, field) != getattr(after, field):
|
|
value = getattr(after, field)
|
|
self.discovery_store.save_override("service", after.id, field, value.value if hasattr(value, "value") else value)
|
|
if (before.kind, before.url, before.host, before.port, before.monitoring_enabled) != (
|
|
after.kind,
|
|
after.url,
|
|
after.host,
|
|
after.port,
|
|
after.monitoring_enabled,
|
|
):
|
|
self.discovery_store.save_override("service", after.id, "probe", after.model_dump(include={"kind", "url", "host", "port", "monitoring_enabled"}))
|
|
|
|
def edge_action(self, evidence_id: str, action: str) -> RelationEvidence:
|
|
evidence = self.discovery_store.get_evidence(evidence_id)
|
|
if evidence is None or evidence.target_service_id is None:
|
|
raise ValueError("Edge evidence not found")
|
|
source = self.store.get_service(evidence.source_service_id)
|
|
if source is None:
|
|
raise ValueError("Edge source service not found")
|
|
target_id = evidence.target_service_id
|
|
if action in {"remove", "reject"} and evidence.relation_type == RelationType.DEPENDS_ON:
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [item for item in source.depends_on if item != target_id]
|
|
self.store.update_service(source.id, payload)
|
|
self.discovery_store.record_change("dependency_removed", f"Removed relationship from {source.name}", source_service_id=source.id, target_service_id=target_id)
|
|
return self.discovery_store.set_evidence_state(evidence_id, active=False) or evidence
|
|
if action == "confirm" and evidence.relation_type == RelationType.DEPENDS_ON and target_id not in source.depends_on:
|
|
validate_dependencies([*source.depends_on, target_id], self.store, service_id=source.id)
|
|
payload = ServiceUpdate.model_validate(source.model_dump())
|
|
payload.depends_on = [*source.depends_on, target_id]
|
|
self.store.update_service(source.id, payload)
|
|
if action == "lock":
|
|
self.discovery_store.save_override("relation", evidence_id, "locked", True, True)
|
|
return self.discovery_store.set_evidence_state(evidence_id, active=True, locked=action == "lock" or evidence.locked) or evidence
|