[Python] Dependency graph of a kubernetes cluster

i'm trying to build a dependency graph of a kubernetes cluster, with resources fetched through APIs, and using networkx to build it, the nodes are captured and well represented, however not all the relations are captured, i don't have previous knowledge of kubernetes but , all the resources are fetched, these are the two functions of fetch and build:

import os
import networkx as nx
from rdflib import Graph as RDFGraph, Namespace, URIRef, Literal, RDF
from rdflib.namespace import RDFS, RDF, OWL
from rdflib.plugins.sparql import prepareQuery
from kubernetes import client,config
from collections import Counter,defaultdict



def fetch_and_enrich(api_func, kind, api_version):
try:
return [dict(item.to_dict(), kind=kind, apiVersion=api_version) for item in api_func().items]
except Exception as e:
print(f"Error fetching {kind}: {e}")
return []

def parse_k8s_manifests():

config.load_kube_config()

#predefined Endpoints to fetch resources
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()
batch_v1 = client.BatchV1Api()
networking_v1= client.NetworkingV1Api()
storage_v1 = client.StorageV1Api()
autoscaling_v2 = client.AutoscalingV2Api()
apiextensions_v1 = client.ApiextensionsV1Api()
rbac_v1 = client.RbacAuthorizationV1Api()
policy_v1 = client.PolicyV1Api()
manifests = []

# CoreV1 resources
manifests += fetch_and_enrich(v1.list_namespace, "Namespace", "v1")
manifests += fetch_and_enrich(v1.list_pod_for_all_namespaces, "Pod", "v1")
manifests += fetch_and_enrich(v1.list_service_for_all_namespaces, "Service", "v1")
manifests += fetch_and_enrich(v1.list_config_map_for_all_namespaces, "ConfigMap", "v1")
manifests += fetch_and_enrich(v1.list_secret_for_all_namespaces, "Secret", "v1")
manifests += fetch_and_enrich(v1.list_node, "Node", "v1")
manifests += fetch_and_enrich(v1.list_resource_quota_for_all_namespaces, "ResourceQuota", "v1")
manifests += fetch_and_enrich(v1.list_limit_range_for_all_namespaces, "LimitRange", "v1")

# AppsV1 resources
manifests += fetch_and_enrich(apps_v1.list_deployment_for_all_namespaces, "Deployment", "apps/v1")
manifests += fetch_and_enrich(apps_v1.list_stateful_set_for_all_namespaces, "StatefulSet", "apps/v1")
manifests += fetch_and_enrich(apps_v1.list_daemon_set_for_all_namespaces, "DaemonSet", "apps/v1")
manifests += fetch_and_enrich(apps_v1.list_replica_set_for_all_namespaces, "ReplicaSet", "apps/v1")

#NetworkingV1Api resources
manifests += fetch_and_enrich(networking_v1.list_ingress_for_all_namespaces, "Ingress", "networking.k8s.io/v1")
manifests += fetch_and_enrich(networking_v1.list_network_policy_for_all_namespaces, "NetworkPolicy", "networking.k8s.io/v1")

# BatchV1 resources
manifests += fetch_and_enrich(batch_v1.list_job_for_all_namespaces, "Job", "batch/v1")
manifests += fetch_and_enrich(batch_v1.list_cron_job_for_all_namespaces, "CronJob", "batch/v1")

# Storage
manifests += fetch_and_enrich(storage_v1.list_storage_class, "StorageClass", "storage.k8s.io/v1")

# AutoscalingV2
manifests += fetch_and_enrich(autoscaling_v2.list_horizontal_pod_autoscaler_for_all_namespaces, "HorizontalPodAutoscaler", "autoscaling/v2")

# APIExtensionsV1
manifests += fetch_and_enrich(apiextensions_v1.list_custom_resource_definition, "CustomResourceDefinition", "apiextensions.k8s.io/v1")

# RBAC Authorization
manifests += fetch_and_enrich(rbac_v1.list_cluster_role, "ClusterRole", "rbac.authorization.k8s.io/v1")
manifests += fetch_and_enrich(rbac_v1.list_cluster_role_binding, "ClusterRoleBinding", "rbac.authorization.k8s.io/v1")
manifests += fetch_and_enrich(rbac_v1.list_role_for_all_namespaces, "Role", "rbac.authorization.k8s.io/v1")
manifests += fetch_and_enrich(rbac_v1.list_role_binding_for_all_namespaces, "RoleBinding", "rbac.authorization.k8s.io/v1")

# Policy
manifests += fetch_and_enrich(policy_v1.list_pod_disruption_budget_for_all_namespaces, "PodDisruptionBudget", "policy/v1")



kinds = Counter(m.get('kind') for m in manifests)
print("Manifest counts by kind:", kinds)

return manifests


def build_dependency_graph(manifests):
G = nx.DiGraph()

# Group manifests by kind for fast lookups
kind_map = defaultdict(list)
for m in manifests:
kind_map[m.get('kind')].append(m)

# Precompute node metadata for each manifest
node_info = []
for m in manifests:
kind = m.get('kind')
meta = m.get('metadata', {}) or {}
name = meta.get('name')
namespace = meta.get('namespace') or 'default'
if not kind or not name:
continue
node_id = f"{kind}:{namespace}:{name}"
spec = m.get('spec') or {}
node_info.append((node_id, kind, meta, spec))

# Add the node with its attributes
attrs = {'kind': kind, 'name': name, 'namespace': namespace}
labels = meta.get('labels')
if isinstance(labels, dict):
attrs['labels'] = labels
ann = meta.get('annotations')
if isinstance(ann, dict):
attrs['annotations'] = ann
for k, v in spec.items():
if isinstance(v, (str, int, float, bool)):
attrs[f'spec:{k}'] = v
G.add_node(node_id, **attrs)

# Now build edges using the precomputed node_info and kind_map
for node_id, kind, meta, spec in node_info:
# inNamespace
namespace = meta.get('namespace') or 'default'
G.add_edge(node_id, f"Namespace:{namespace}", relation='inNamespace')

# ownerReferences → owns
for ref in meta.get('ownerReferences', []):
owner_id = f"{ref['kind']}:{namespace}:{ref['name']}"
if G.has_node(owner_id):
G.add_edge(owner_id, node_id, relation='owns')

# Per-kind relationships
if kind == 'Pod':
# runsOn node
node_name = spec.get('nodeName')
if node_name:
G.add_edge(node_id, f"Node:{node_name}", relation='runsOn')
# serviceAccount
sa = spec.get('serviceAccountName')
if sa:
G.add_edge(node_id, f"ServiceAccount:{namespace}:{sa}", relation='uses')
# PVC mounts
for vol in spec.get('volumes', []) or []:
pvc = (vol.get('persistentVolumeClaim') or {}).get('claimName')
if pvc:
G.add_edge(node_id, f"PersistentVolumeClaim:{namespace}:{pvc}", relation='mounts')

elif kind == 'Service':
selector = spec.get('selector') or {}
if selector:
for pod in kind_map['Pod']:
pl = pod.get('metadata', {}).get('labels', {}) or {}
if all(pl.get(k) == v for k, v in selector.items()):
G.add_edge(node_id,
f"Pod:{pod['metadata'].get('namespace','default')}:{pod['metadata']['name']}",
relation='routes-to')

elif kind == 'Ingress':
for rule in spec.get('rules', []) or []:
for path in (rule.get('http') or {}).get('paths', []) or []:
svc = (path.get('backend', {}).get('service') or {}).get('name')
if svc:
G.add_edge(node_id, f"Service:{namespace}:{svc}", relation='routesTo')

elif kind == 'Deployment':
for rs in kind_map['ReplicaSet']:
for ref in rs.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'Deployment' and ref['name'] == meta['name']:
G.add_edge(node_id, f"ReplicaSet:{namespace}:{rs['metadata']['name']}", relation='controls')

elif kind == 'ReplicaSet':
for pod in kind_map['Pod']:
for ref in pod.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'ReplicaSet' and ref['name'] == meta['name']:
G.add_edge(node_id, f"Pod:{namespace}:{pod['metadata']['name']}", relation='controls')

elif kind == 'StatefulSet':
# controls Pods
for pod in kind_map['Pod']:
for ref in pod.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'StatefulSet' and ref['name'] == meta['name']:
G.add_edge(node_id, f"Pod:{namespace}:{pod['metadata']['name']}", relation='controls')
# creates PVCs
for claim in spec.get('volumeClaimTemplates', []) or []:
pvc = claim.get('metadata', {}).get('name')
if pvc:
G.add_edge(node_id, f"PersistentVolumeClaim:{namespace}:{pvc}", relation='creates')

elif kind == 'DaemonSet':
for pod in kind_map['Pod']:
for ref in pod.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'DaemonSet' and ref['name'] == meta['name']:
G.add_edge(node_id, f"Pod:{namespace}:{pod['metadata']['name']}", relation='controls')

elif kind in ('Job', 'CronJob'):
if kind == 'CronJob':
for job in kind_map['Job']:
for ref in job.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'CronJob' and ref['name'] == meta['name']:
G.add_edge(node_id, f"Job:{namespace}:{job['metadata']['name']}", relation='schedules')
for pod in kind_map['Pod']:
for ref in pod.get('metadata', {}).get('ownerReferences', []):
if ref['kind'] == 'Job' and ref['name'] == meta['name']:
G.add_edge(node_id, f"Pod:{namespace}:{pod['metadata']['name']}", relation='controls')

elif kind == 'HorizontalPodAutoscaler':
tr = spec.get('scaleTargetRef', {}) or {}
tk, tn = tr.get('kind'), tr.get('name')
if tk and tn:
G.add_edge(node_id, f"{tk}:{namespace}:{tn}", relation='scales')

elif kind == 'PersistentVolumeClaim':
pv = spec.get('volumeName')
if pv:
G.add_edge(node_id, f"PersistentVolume:{pv}", relation='binds')

elif kind == 'PersistentVolume':
sc = spec.get('storageClassName')
if sc:
G.add_edge(node_id, f"StorageClass:{sc}", relation='provisionedBy')

elif kind == 'NetworkPolicy':
sel = spec.get('podSelector', {}).get('matchLabels', {}) or {}
for pod in kind_map['Pod']:
pl = pod.get('metadata', {}).get('labels', {}) or {}
if all(pl.get(k) == v for k, v in sel.items()):
G.add_edge(node_id, f"Pod:{pod['metadata']['namespace']}:{pod['metadata']['name']}", relation='applies-to')

elif kind == 'PodDisruptionBudget':
sel = spec.get('selector', {}).get('matchLabels', {}) or {}
for pod in kind_map['Pod']:
pl = pod.get('metadata', {}).get('labels', {}) or {}
if all(pl.get(k) == v for k, v in sel.items()):
G.add_edge(node_id, f"Pod:{pod['metadata']['namespace']}:{pod['metadata']['name']}", relation='protects')

elif kind in ('RoleBinding', 'ClusterRoleBinding'):
rr = m.get('roleRef', {}) or {}
rk, rn = rr.get('kind'), rr.get('name')
if rk and rn:
ns = namespace if rk == 'Role' else ''
G.add_edge(node_id, f"{rk}:{ns}:{rn}", relation='binds')

return G


the captured relations are: applies-to, inNamespace, protects, routes-to, routesTo. so please if anyone could help fix this urgent matter, cuz i already spent a long time trying to fix but this is the closest i could get to what i need.

Continue reading...