54 lines
2.3 KiB
Python
54 lines
2.3 KiB
Python
import json
|
|
import subprocess
|
|
|
|
def get_kubectl_output(command):
|
|
result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
if result.returncode != 0:
|
|
raise Exception(f"Error running command: {result.stderr}")
|
|
return json.loads(result.stdout)
|
|
|
|
def filter_namespaces(items):
|
|
system_namespaces = ['kube-system', 'default', 'kube-node-lease']
|
|
return [item for item in items if 'd8-' not in item['metadata']['namespace'] and item['metadata']['namespace'] not in system_namespaces]
|
|
|
|
def main():
|
|
services = get_kubectl_output("kubectl get services --all-namespaces -o json")
|
|
pods = get_kubectl_output("kubectl get pods --all-namespaces -o json")
|
|
|
|
filtered_services = filter_namespaces(services['items'])
|
|
filtered_pods = filter_namespaces(pods['items'])
|
|
|
|
service_dict = {}
|
|
for svc in filtered_services:
|
|
svc_namespace = svc['metadata']['namespace']
|
|
svc_name = svc['metadata']['name']
|
|
selectors = svc['spec'].get('selector', {})
|
|
service_dict[(svc_namespace, svc_name)] = selectors
|
|
|
|
pod_dict = {}
|
|
for pod in filtered_pods:
|
|
pod_namespace = pod['metadata']['namespace']
|
|
pod_labels = pod['metadata'].get('labels', {})
|
|
pod_name = pod['metadata']['name']
|
|
pod_dict[(pod_namespace, pod_name)] = pod_labels
|
|
|
|
cross_namespace_communication = []
|
|
|
|
for (svc_namespace, svc_name), selectors in service_dict.items():
|
|
for (pod_namespace, pod_name), labels in pod_dict.items():
|
|
if svc_namespace != pod_namespace and all(labels.get(key) == value for key, value in selectors.items()):
|
|
cross_namespace_communication.append({
|
|
"service_namespace": svc_namespace,
|
|
"service_name": svc_name,
|
|
"service_selectors": selectors,
|
|
"pod_namespace": pod_namespace,
|
|
"pod_name": pod_name,
|
|
"pod_labels": labels
|
|
})
|
|
|
|
for comm in cross_namespace_communication:
|
|
print(f"Service {comm['service_namespace']}/{comm['service_name']} (selectors: {comm['service_selectors']}) communicates with Pod {comm['pod_namespace']}/{comm['pod_name']} (labels: {comm['pod_labels']}) in different namespace")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|