Kubernetes/k8s2d2.py - Diagram with python and D2

This python script will read your kubernetes and write a D2 file from which you can make a diagram.

So far: service -> deployment -> hpa

Missing: sts, cron, ingress, gateway, configmaps, secrets, mounts, pvc.

#!/usr/bin/env python3

from kubernetes import client, config
import argparse
import pprint
from py_d2.connection import D2Connection
from py_d2.diagram    import D2Diagram
from py_d2.shape      import D2Shape

pp = pprint.PrettyPrinter(indent=4)
config.load_kube_config() 
parser = argparse.ArgumentParser()
parser.add_argument('--namespace', type=str, help='kubernetes namespace', default="default");
args = parser.parse_args()

# k8s apis clients, why three?
core = client.CoreV1Api()
v1 = client.AppsV1Api()
autoscaleapi = client.AutoscalingV1Api()

# diagram setup containers
diagram = D2Diagram()
cluster = D2Shape(name="kubernetes_cluster", label="cluster name: mycluster")
namespace = D2Shape(name="namespace", label="namespace:"+args.namespace)
cluster.add_shape(namespace)
diagram.add_shape(cluster)

try:
  deployments = v1.list_namespaced_deployment(args.namespace)
  for i in deployments.items:
    # pp.pprint(i)
    print("deployment {}\t{}".format(i.metadata.name, i.metadata.namespace))
    namespace.add_shape(D2Shape(name="deploy_"+i.metadata.name,label="deploy: " + i.metadata.name))
except Exception as e:
  print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)

try:
  hpas = autoscaleapi.list_namespaced_horizontal_pod_autoscaler(args.namespace)
  for i in hpas.items:
    print("hpa {}\t{}".format(i.metadata.name, i.metadata.namespace))
    namespace.add_shape(D2Shape(name="hpa_"+i.metadata.name,label="hpa:"+i.metadata.name))
    namespace.add_connection(D2Connection(shape_2="hpa_"+i.metadata.name, shape_1="deploy_"+i.spec.scale_target_ref.name))
except Exception as e:
  print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)

try:
  services = core.list_namespaced_service(args.namespace)
  for i in services.items:
    # print("service {}\t{}".format(i.metadata.name, i.metadata.namespace))
    namespace.add_shape(D2Shape(name="service_"+i.metadata.name,label="service:"+i.metadata.name))
    # find the deploy this service points at.
    for deploy in deployments.items:
      if deploy.spec.selector.match_labels == i.spec.selector:
        #print("selectors match")
        #pp.pprint(deploy.spec.selector.match_labels)
        #print("---")
        #pp.pprint(i.spec.selector)
        #print("---")
        namespace.add_connection(D2Connection(shape_1="service_"+i.metadata.name, shape_2="deploy_"+deploy.metadata.name))
except Exception as e:
  print("Exception when calling core.list_namespaced_service: %s\n" % e)

with open("namespace.d2", "w", encoding="utf-8") as f:
    f.write(str(diagram))