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()
network_api = client.NetworkingV1Api()

# diagram setup containers
diagram = D2Diagram()
cluster = D2Shape(name="kubernetes_cluster", label="cluster name: staging-gcp-env-gke")
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)

try:
  # ingress = v1.list_namespaced_deployment(args.namespace)
  ingress = network_api.list_namespaced_ingress(args.namespace)

  for i in ingress.items:
    # pp.pprint(i)
    print("ingress {}\t{}".format(i.metadata.name, i.metadata.namespace))
    namespace.add_shape(D2Shape(name="ingress."+i.metadata.name,label="ingress: " + i.metadata.name))
    if i.spec.default_backend.service.name:
      print ( "{}".format(i.spec.default_backend.service.name))
      namespace.add_connection(D2Connection(shape_1="ingress."+i.metadata.name, shape_2="service."+i.spec.default_backend.service.name))
except Exception as e:
  print("Exception when calling network_api.list_namespaced_ingress(args.namespace) : {}".format(e))

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