fix(diagnostic): Scope Pod CSI topology

Restrict Pod storage diagnostics to the relevant CSI node agent while preserving controller evidence. Allow diagnostic graph loading for cluster-scoped storage resources without sending stale namespaces.

Co-Authored-By: Codex <codex@openai.com>

Author: Colvin-Y

AI_CONTRACT.md

@@ -251,7 +251,7 @@ It should not be used as the only signal for graph facts already present in node
 - this pod mounts PVC X, which is bound to PV Y
 - this PV uses CSI driver X
 - this PV is managed by configured CSI controller components
-- no configured CSI node agent was found for the PV affinity node in the current observed graph slice
+- no configured CSI node agent was found for the PV affinity or consuming Pod node in the current observed graph slice
 - this workload owns the pod through the owner chain
 
 ### Unsafe conclusions

cmd/kubernetes-ontology-viewer/main.go

@@ -157,7 +157,10 @@ func (h *handler) serveDiagnostic(w http.ResponseWriter, r *http.Request) {
 	kind := first(q, "kind", "Pod")
 	namespace := first(q, "namespace", "")
 	name := first(q, "name", "")
-	if namespace == "" && (strings.EqualFold(kind, "Pod") || strings.EqualFold(kind, "Workload")) {
+	if isDiagnosticClusterScopedKind(kind) {
+		namespace = ""
+	}
+	if namespace == "" && isDiagnosticNamespacedKind(kind) {
 		writeJSON(w, map[string]string{"error": "namespace is required"}, http.StatusBadRequest)
 		return
 	}
@@ -189,6 +192,24 @@ func (h *handler) serveDiagnostic(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, data, http.StatusOK)
 }
 
+func isDiagnosticNamespacedKind(kind string) bool {
+	switch strings.ToLower(strings.TrimSpace(kind)) {
+	case "pod", "workload", "pvc", "persistentvolumeclaim":
+		return true
+	default:
+		return false
+	}
+}
+
+func isDiagnosticClusterScopedKind(kind string) bool {
+	switch strings.ToLower(strings.TrimSpace(kind)) {
+	case "pv", "persistentvolume", "storageclass", "csidriver":
+		return true
+	default:
+		return false
+	}
+}
+
 func (h *handler) serveExpand(w http.ResponseWriter, r *http.Request) {
 	q := r.URL.Query()
 	server := first(q, "server", h.defaultServer)

docs/ontology/kubernetes-ontology.owl

@@ -422,7 +422,7 @@
   </owl:ObjectProperty>
   <owl:ObjectProperty rdf:about="#served_by_csi_node_agent">
     <rdfs:label>served_by_csi_node_agent</rdfs:label>
-    <rdfs:comment>Relates a PersistentVolume to the CSI node-agent Pod on its affinity node.</rdfs:comment>
+    <rdfs:comment>Relates a PersistentVolume to a CSI node-agent Pod on a PV affinity node or consuming Pod node.</rdfs:comment>
     <ko:edgeKind>served_by_csi_node_agent</ko:edgeKind>
     <rdfs:domain rdf:resource="#PV"/>
     <rdfs:range rdf:resource="#Pod"/>

internal/graph/builder.go

@@ -1,6 +1,7 @@
 package graph
 
 import (
+	"sort"
 	"strings"
 
 	"github.com/Colvin-Y/kubernetes-ontology/internal/collect/k8s"
@@ -313,12 +314,11 @@ func (b *Builder) Build(snapshot k8s.Snapshot) ([]model.Node, []model.Edge) {
 		if !ok {
 			continue
 		}
-		affinityNodeName, _ := pv.CSI["nodeAffinity"]
 		pvNode, ok := findNode(nodes, pvID)
 		if !ok {
 			continue
 		}
-		correlation := correlator.Correlate(pvNode, affinityNodeName, infraPods)
+		correlation := infer.CorrelatePVToCSIComponents(correlator, pvNode, csiAgentNodeNamesForPV(pv, snapshot), infraPods)
 		edges = append(edges, correlation.Edges...)
 		b.lastEvidence = append(b.lastEvidence, correlation.Evidence...)
 	}
@@ -374,6 +374,44 @@ func storageClassNameForPVC(pvc k8sresources.PVC, pvs []k8sresources.PV) string
 	return ""
 }
 
+func csiAgentNodeNamesForPV(pv k8sresources.PV, snapshot k8s.Snapshot) []string {
+	seen := make(map[string]struct{})
+	addNode := func(nodeName string) {
+		nodeName = strings.TrimSpace(nodeName)
+		if nodeName == "" {
+			return
+		}
+		seen[nodeName] = struct{}{}
+	}
+	addNode(pv.CSI["nodeAffinity"])
+	for _, pvc := range snapshot.PVCs {
+		if pvc.VolumeName != pv.Metadata.Name {
+			continue
+		}
+		for _, pod := range snapshot.Pods {
+			if pod.Metadata.Namespace != pvc.Metadata.Namespace || pod.NodeName == "" || !podReferencesPVC(pod, pvc.Metadata.Name) {
+				continue
+			}
+			addNode(pod.NodeName)
+		}
+	}
+	out := make([]string, 0, len(seen))
+	for nodeName := range seen {
+		out = append(out, nodeName)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func podReferencesPVC(pod k8sresources.Pod, pvcName string) bool {
+	for _, ref := range pod.PVCRefs {
+		if ref == pvcName {
+			return true
+		}
+	}
+	return false
+}
+
 func roleBindingID(cluster string, roleBinding k8sresources.RoleBinding) model.CanonicalID {
 	return model.NewCanonicalID(model.ResourceRef{Cluster: cluster, Group: "rbac.authorization.k8s.io", Kind: "RoleBinding", Namespace: roleBinding.Metadata.Namespace, Name: roleBinding.Metadata.Name, UID: roleBinding.Metadata.UID})
 }

internal/graph/pvc_contract_test.go

@@ -146,6 +146,70 @@ func TestBuilderUsesConfiguredCSIComponentRule(t *testing.T) {
 	}
 }
 
+func TestBuilderScopesPVCSINodeAgentToConsumingPodNode(t *testing.T) {
+	builder := graph.NewBuilder("cluster-a")
+	builder.SetCSIComponentRules(openLocalCSIComponentRules())
+	snapshot := collectk8s.Snapshot{
+		Pods: []resources.Pod{
+			{
+				Metadata: resources.Metadata{UID: "pod-uid", Name: "app-0", Namespace: "default"},
+				NodeName: "node-a",
+				PVCRefs:  []string{"data"},
+			},
+			{
+				Metadata: resources.Metadata{UID: "csi-agent-a-uid", Name: "open-local-agent-node-a", Namespace: "kube-system"},
+				NodeName: "node-a",
+			},
+			{
+				Metadata: resources.Metadata{UID: "csi-agent-b-uid", Name: "open-local-agent-node-b", Namespace: "kube-system"},
+				NodeName: "node-b",
+			},
+		},
+		PVCs: []resources.PVC{{
+			Metadata:         resources.Metadata{UID: "pvc-uid", Name: "data", Namespace: "default"},
+			VolumeName:       "pv-data",
+			StorageClassName: "open-local",
+			Status:           "Bound",
+		}},
+		PVs: []resources.PV{{
+			Metadata:         resources.Metadata{UID: "pv-uid", Name: "pv-data"},
+			StorageClassName: "open-local",
+			Status:           "Bound",
+			CSI:              map[string]string{"driver": "local.csi.aliyun.com", "handle": "vol-123"},
+		}},
+		StorageClasses: []resources.StorageClass{{
+			Metadata:    resources.Metadata{UID: "sc-uid", Name: "open-local"},
+			Provisioner: "local.csi.aliyun.com",
+		}},
+	}
+
+	nodes, edges := builder.Build(snapshot)
+	agentA, ok := nodeIDByName(nodes, model.NodeKindPod, "kube-system", "open-local-agent-node-a")
+	if !ok {
+		t.Fatal("expected node-a CSI agent")
+	}
+	agentB, ok := nodeIDByName(nodes, model.NodeKindPod, "kube-system", "open-local-agent-node-b")
+	if !ok {
+		t.Fatal("expected node-b CSI agent")
+	}
+	servedByEdges := 0
+	for _, edge := range edges {
+		if edge.Kind != model.EdgeKindServedByCSINodeAgent {
+			continue
+		}
+		servedByEdges++
+		if edge.To == agentB {
+			t.Fatal("did not expect PV to be served by off-node CSI agent")
+		}
+		if edge.To != agentA {
+			t.Fatalf("expected PV to be served by node-a CSI agent, got %s", edge.To)
+		}
+	}
+	if servedByEdges != 1 {
+		t.Fatalf("expected one scoped PV node-agent edge, got %d", servedByEdges)
+	}
+}
+
 func storageTopologySnapshot() collectk8s.Snapshot {
 	return collectk8s.Snapshot{
 		Pods: []resources.Pod{
@@ -192,6 +256,15 @@ func containsNodeKind(nodes []model.Node, kind model.NodeKind) bool {
 	return false
 }
 
+func nodeIDByName(nodes []model.Node, kind model.NodeKind, namespace, name string) (model.CanonicalID, bool) {
+	for _, node := range nodes {
+		if node.Kind == kind && node.Namespace == namespace && node.Name == name {
+			return node.ID, true
+		}
+	}
+	return "", false
+}
+
 func openLocalCSIComponentRules() []infer.CSIComponentRule {
 	return []infer.CSIComponentRule{{
 		Driver:                "local.csi.aliyun.com",

internal/model/relation_spec.go

@@ -286,7 +286,7 @@ var relationSpecs = []RelationSpec{
 	},
 	{
 		Kind:              EdgeKindServedByCSINodeAgent,
-		Comment:           "Relates a PersistentVolume to the CSI node-agent Pod on its affinity node.",
+		Comment:           "Relates a PersistentVolume to a CSI node-agent Pod on a PV affinity node or consuming Pod node.",
 		Domain:            string(NodeKindPV),
 		Range:             string(NodeKindPod),
 		DefaultSourceType: EdgeSourceTypeInferenceRule,

internal/reconcile/storage.go

@@ -2,6 +2,8 @@ package reconcile
 
 import (
 	"fmt"
+	"sort"
+	"strings"
 
 	collectk8s "github.com/Colvin-Y/kubernetes-ontology/internal/collect/k8s"
 	"github.com/Colvin-Y/kubernetes-ontology/internal/collect/k8s/resources"
@@ -235,7 +237,7 @@ func (r *StorageReconciler) rebuildStorageEdges(snapshot collectk8s.Snapshot, cu
 		}
 		pvNode := pvNode(r.cluster, pv)
 		pvNode.ID = pvID
-		correlation := correlator.Correlate(pvNode, pv.CSI["nodeAffinity"], infraPods)
+		correlation := infer.CorrelatePVToCSIComponents(correlator, pvNode, csiAgentNodeNamesForPV(pv, snapshot), infraPods)
 		for _, edge := range correlation.Edges {
 			if err := r.kernel.UpsertEdge(edge); err != nil {
 				return upserted, err
@@ -409,6 +411,44 @@ func storageClassNameForPVC(pvc resources.PVC, pvs []resources.PV) string {
 	return ""
 }
 
+func csiAgentNodeNamesForPV(pv resources.PV, snapshot collectk8s.Snapshot) []string {
+	seen := make(map[string]struct{})
+	addNode := func(nodeName string) {
+		nodeName = strings.TrimSpace(nodeName)
+		if nodeName == "" {
+			return
+		}
+		seen[nodeName] = struct{}{}
+	}
+	addNode(pv.CSI["nodeAffinity"])
+	for _, pvc := range snapshot.PVCs {
+		if pvc.VolumeName != pv.Metadata.Name {
+			continue
+		}
+		for _, pod := range snapshot.Pods {
+			if pod.Metadata.Namespace != pvc.Metadata.Namespace || pod.NodeName == "" || !podReferencesPVC(pod, pvc.Metadata.Name) {
+				continue
+			}
+			addNode(pod.NodeName)
+		}
+	}
+	out := make([]string, 0, len(seen))
+	for nodeName := range seen {
+		out = append(out, nodeName)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func podReferencesPVC(pod resources.Pod, pvcName string) bool {
+	for _, ref := range pod.PVCRefs {
+		if ref == pvcName {
+			return true
+		}
+	}
+	return false
+}
+
 func storageInfraPods(cluster string, snapshot collectk8s.Snapshot) []model.Node {
 	out := make([]model.Node, 0)
 	for _, pod := range snapshot.Pods {

internal/resolve/infer/csi.go

@@ -2,6 +2,7 @@ package infer
 
 import (
 	"fmt"
+	"sort"
 	"strings"
 
 	"github.com/Colvin-Y/kubernetes-ontology/internal/model"
@@ -106,6 +107,35 @@ func NewCSIComponentRegistry(rules []CSIComponentRule) *Registry {
 	return NewRegistry(correlators...)
 }
 
+func CorrelatePVToCSIComponents(correlator CSICorrelator, pv model.Node, nodeNames []string, infraPods []model.Node) CorrelationResult {
+	normalizedNodeNames := normalizeNodeNames(nodeNames)
+	if len(normalizedNodeNames) == 0 {
+		return correlator.Correlate(pv, "", infraPods)
+	}
+
+	result := CorrelationResult{Edges: make([]model.Edge, 0), Evidence: make([]string, 0)}
+	seenEdges := make(map[string]struct{})
+	seenEvidence := make(map[string]struct{})
+	for _, nodeName := range normalizedNodeNames {
+		correlation := correlator.Correlate(pv, nodeName, infraPods)
+		for _, edge := range correlation.Edges {
+			if _, seen := seenEdges[edge.Key()]; seen {
+				continue
+			}
+			result.Edges = append(result.Edges, edge)
+			seenEdges[edge.Key()] = struct{}{}
+		}
+		for _, evidence := range correlation.Evidence {
+			if _, seen := seenEvidence[evidence]; seen {
+				continue
+			}
+			result.Evidence = append(result.Evidence, evidence)
+			seenEvidence[evidence] = struct{}{}
+		}
+	}
+	return result
+}
+
 func IsCSIProvisioner(provisioner string, observedCSIDriver bool, rules []CSIComponentRule) bool {
 	if provisioner == "" {
 		return false
@@ -187,9 +217,9 @@ func (c componentRuleCorrelator) Correlate(pv model.Node, affinityNodeName strin
 	}
 	if len(c.rule.NodeAgentPodPrefixes) > 0 {
 		if affinityNodeName == "" {
-			result.Evidence = append(result.Evidence, fmt.Sprintf("csi: PV affinity node missing for driver %s", c.rule.Driver))
+			result.Evidence = append(result.Evidence, fmt.Sprintf("csi: PV affinity or consuming pod node missing for driver %s", c.rule.Driver))
 		} else if !foundAgent {
-			result.Evidence = append(result.Evidence, fmt.Sprintf("csi: no node agent found for driver %s on PV affinity node %s", c.rule.Driver, affinityNodeName))
+			result.Evidence = append(result.Evidence, fmt.Sprintf("csi: no node agent found for driver %s on node %s", c.rule.Driver, affinityNodeName))
 		}
 	}
 	return result
@@ -217,6 +247,23 @@ func hasAnyPrefix(value string, prefixes []string) bool {
 	return false
 }
 
+func normalizeNodeNames(nodeNames []string) []string {
+	seen := make(map[string]struct{}, len(nodeNames))
+	for _, nodeName := range nodeNames {
+		nodeName = strings.TrimSpace(nodeName)
+		if nodeName == "" {
+			continue
+		}
+		seen[nodeName] = struct{}{}
+	}
+	out := make([]string, 0, len(seen))
+	for nodeName := range seen {
+		out = append(out, nodeName)
+	}
+	sort.Strings(out)
+	return out
+}
+
 func looksLikeCSIProvisioner(provisioner string) bool {
 	if strings.HasPrefix(provisioner, "kubernetes.io/") {
 		return false

internal/resolve/infer/csi_test.go

@@ -121,18 +121,35 @@ func TestCSIComponentRuleCorrelatesPVToComponents(t *testing.T) {
 		Namespace:  "storage-system",
 		Attributes: map[string]any{"nodeName": "node-a"},
 	}
+	otherAgent := model.Node{
+		ID:         model.NewCanonicalID(model.ResourceRef{Cluster: "cluster-a", Group: "core", Kind: "Pod", Namespace: "storage-system", Name: "disk-agent-node-b"}),
+		Kind:       model.NodeKindPod,
+		Name:       "disk-agent-node-b",
+		Namespace:  "storage-system",
+		Attributes: map[string]any{"nodeName": "node-b"},
+	}
 
-	result := correlator.Correlate(pv, "node-a", []model.Node{controller, agent})
+	result := correlator.Correlate(pv, "node-a", []model.Node{controller, agent, otherAgent})
 	kinds := map[model.EdgeKind]bool{}
+	servedByAgentEdges := 0
 	for _, edge := range result.Edges {
 		kinds[edge.Kind] = true
+		if edge.Kind == model.EdgeKindServedByCSINodeAgent {
+			servedByAgentEdges++
+			if edge.To != agent.ID {
+				t.Fatalf("expected PV node-agent edge to target same-node agent, got %s", edge.To)
+			}
+		}
 	}
 	if !kinds[model.EdgeKindManagedByCSIController] {
 		t.Fatal("expected PV controller edge")
 	}
 	if !kinds[model.EdgeKindServedByCSINodeAgent] {
 		t.Fatal("expected PV node-agent edge")
 	}
+	if servedByAgentEdges != 1 {
+		t.Fatalf("expected exactly one PV node-agent edge, got %d", servedByAgentEdges)
+	}
 }
 
 func TestParseCSIComponentRules(t *testing.T) {