Kubernetes/k8s2d2.py - Diagram with python and D2: Difference between revisions

From Federal Burro of Information
Jump to navigationJump to search
(Created page with " 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. <pre> #!/usr/bin/env python3 from kubernetes import client, config import argparse import pprint from py_d2.connection import D2Connection, D2Diagram, D2Shape pp = pprint.PrettyPrinter(indent=4) config.load_kube_config() parser = argparse.ArgumentParser() pa...")
 
No edit summary
 
(3 intermediate revisions by the same user not shown)
Line 8: Line 8:
<pre>
<pre>
#!/usr/bin/env python3
#!/usr/bin/env python3
# connects to you k8s cluster create d2 source file.
# https://kubernetes.io/
# https://d2lang.com/
# https://github.com/MrBlenny/py-d2


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


pp = pprint.PrettyPrinter(indent=4)
pp = pprint.PrettyPrinter(indent=4)
Line 24: Line 30:
v1 = client.AppsV1Api()
v1 = client.AppsV1Api()
autoscaleapi = client.AutoscalingV1Api()
autoscaleapi = client.AutoscalingV1Api()
network_api = client.NetworkingV1Api()


# diagram setup containers
# diagram setup containers
diagram = D2Diagram()
diagram = D2Diagram()
cluster = D2Shape(name="kubernetes_cluster", label="cluster name: mycluster")
cluster = D2Shape(name="kubernetes_cluster", label="cluster name: staging-gcp-env-gke")
namespace = D2Shape(name="namespace", label="namespace:"+args.namespace)
namespace = D2Shape(name="namespace", label="namespace:"+args.namespace)
cluster.add_shape(namespace)
cluster.add_shape(namespace)
Line 37: Line 44:
     # pp.pprint(i)
     # pp.pprint(i)
     print("deployment {}\t{}".format(i.metadata.name, i.metadata.namespace))
     print("deployment {}\t{}".format(i.metadata.name, i.metadata.namespace))
     namespace.add_shape(D2Shape(name="deploy_"+i.metadata.name,label="deploy: " + i.metadata.name))
     namespace.add_shape(D2Shape(name="deploy."+i.metadata.name,label="deploy: " + i.metadata.name))
except Exception as e:
except Exception as e:
   print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)
   print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)
Line 45: Line 52:
   for i in hpas.items:
   for i in hpas.items:
     print("hpa {}\t{}".format(i.metadata.name, i.metadata.namespace))
     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_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))
     namespace.add_connection(D2Connection(shape_2="hpa."+i.metadata.name, shape_1="deploy."+i.spec.scale_target_ref.name))
except Exception as e:
except Exception as e:
   print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)
   print("Exception when calling v1.list_namespaced_horizontal_pod_autoscaler: %s\n" % e)
Line 54: Line 61:
   for i in services.items:
   for i in services.items:
     # print("service {}\t{}".format(i.metadata.name, i.metadata.namespace))
     # print("service {}\t{}".format(i.metadata.name, i.metadata.namespace))
     namespace.add_shape(D2Shape(name="service_"+i.metadata.name,label="service:"+i.metadata.name))
     namespace.add_shape(D2Shape(name="service."+i.metadata.name,label="service: "+i.metadata.name))
     # find the deploy this service points at.
     # find the deploy this service points at.
     for deploy in deployments.items:
     for deploy in deployments.items:
Line 63: Line 70:
         #pp.pprint(i.spec.selector)
         #pp.pprint(i.spec.selector)
         #print("---")
         #print("---")
         namespace.add_connection(D2Connection(shape_1="service_"+i.metadata.name, shape_2="deploy_"+deploy.metadata.name))
         namespace.add_connection(D2Connection(shape_1="service."+i.metadata.name, shape_2="deploy."+deploy.metadata.name))
except Exception as e:
except Exception as e:
   print("Exception when calling core.list_namespaced_service: %s\n" % 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:
with open("namespace.d2", "w", encoding="utf-8") as f:
Line 71: Line 92:


</pre>
</pre>
[[Category:Script]]

Latest revision as of 20:39, 25 March 2024

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
# connects to you k8s cluster create d2 source file.
# https://kubernetes.io/
# https://d2lang.com/
# https://github.com/MrBlenny/py-d2

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))