Refine runtime entity taxonomy

railiance_fabric/scanner.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import ipaddress
 import re
 import subprocess
 import tomllib
@@ -1118,10 +1119,11 @@ def _add_runtime_endpoint(
     if not host or port_number is None:
         return ""
 
-    server_key = discovery_stable_key(context.repo_slug, "Server", host)
+    target_kind = _runtime_target_kind(host, server_type)
+    target_key = discovery_stable_key(context.repo_slug, target_kind, host)
     context.accumulator.add_node(
-        stable_key=server_key,
-        kind="Server",
+        stable_key=target_key,
+        kind=target_kind,
         label=host,
         replacement_scope=scope,
         provenance=provenance,
@@ -1129,7 +1131,7 @@ def _add_runtime_endpoint(
         aliases=[host],
         attributes={
             "host": host,
-            "server_type": server_type,
+            "runtime_target_type": server_type,
             **(attributes or {}),
         },
         confidence=confidence,
@@ -1154,8 +1156,8 @@ def _add_runtime_endpoint(
         confidence=confidence,
     )
     context.accumulator.add_edge(
-        edge_type="opens_port",
-        source_key=server_key,
+        edge_type="opens_port" if target_kind == "Server" else "listens_on",
+        source_key=target_key,
         target_key=port_key,
         replacement_scope=scope,
         provenance=provenance,
@@ -1174,9 +1176,31 @@ def _add_runtime_endpoint(
         )
     route_domain = _normalize_domain(domain)
     if route_domain:
-        _add_domain_route(context, scope, provenance, anchor, route_domain, port_key, host, confidence=confidence)
+        _add_domain_route(
+            context,
+            scope,
+            provenance,
+            anchor,
+            route_domain,
+            port_key,
+            host,
+            runtime_target_key=target_key,
+            runtime_target_kind=target_kind,
+            confidence=confidence,
+        )
     elif _looks_like_domain(host):
-        _add_domain_route(context, scope, provenance, anchor, host, port_key, host, confidence=confidence)
+        _add_domain_route(
+            context,
+            scope,
+            provenance,
+            anchor,
+            host,
+            port_key,
+            host,
+            runtime_target_key=target_key,
+            runtime_target_kind=target_kind,
+            confidence=confidence,
+        )
     return port_key
 
 
@@ -1225,13 +1249,14 @@ def _add_domain_route(
     port_key: str,
     server_host: str,
     *,
+    runtime_target_key: str = "",
+    runtime_target_kind: str = "",
     confidence: float,
 ) -> None:
     domain_value = _normalize_domain(domain)
     if not domain_value:
         return
     domain_key = discovery_stable_key(context.repo_slug, "DomainName", domain_value)
-    server_key = discovery_stable_key(context.repo_slug, "Server", _normalize_host(server_host))
     context.accumulator.add_node(
         stable_key=domain_key,
         kind="DomainName",
@@ -1253,7 +1278,23 @@ def _add_domain_route(
             source_anchor=anchor,
             confidence=confidence,
         )
-    if server_host:
+    if runtime_target_key:
+        edge_type = {
+            "ApplicationEndpoint": "names_endpoint",
+            "RuntimeService": "routes_to_service",
+            "Server": "resolves_to",
+        }.get(runtime_target_kind, "routes_to")
+        context.accumulator.add_edge(
+            edge_type=edge_type,
+            source_key=domain_key,
+            target_key=runtime_target_key,
+            replacement_scope=scope,
+            provenance=provenance,
+            source_anchor=anchor,
+            confidence=confidence,
+        )
+    elif server_host:
+        server_key = discovery_stable_key(context.repo_slug, "Server", _normalize_host(server_host))
         context.accumulator.add_edge(
             edge_type="resolves_to",
             source_key=domain_key,
@@ -1509,6 +1550,28 @@ def _looks_like_domain(host: str) -> bool:
     return "." in value
 
 
+def _runtime_target_kind(host: str, runtime_target_type: str) -> str:
+    value = _normalize_host(host)
+    if _looks_like_machine_address(value):
+        return "Server"
+    if runtime_target_type == "kubernetes-service-dns" or value.endswith(".svc.cluster.local"):
+        return "RuntimeService"
+    if runtime_target_type == "ingress-host" or _looks_like_domain(value):
+        return "ApplicationEndpoint"
+    return "RuntimeService"
+
+
+def _looks_like_machine_address(host: str) -> bool:
+    value = _normalize_host(host)
+    if value in {"localhost", "127.0.0.1"}:
+        return True
+    try:
+        ipaddress.ip_address(value)
+    except ValueError:
+        return False
+    return True
+
+
 def _int_value(value: object) -> int | None:
     if isinstance(value, bool):
         return None