Get leaked clusters

find_leaked.py

#You need to be logged in to your google sdk account and have gcloud set to cilium-ci project

import re
import subprocess
import json

import requests
from bs4 import BeautifulSoup


jenkinsUrl = 'https://jenkins.cilium.io'
url = f'{jenkinsUrl}/computer/api/xml?tree=computer[executors[currentExecutable[url]],oneOffExecutors[currentExecutable[url]]]&xpath=//url&wrapper=builds'
response = requests.get(url)
soup = BeautifulSoup(response.text, 'xml')

urls = set([x.text for x in soup.find_all("url") if "GKE" in x.text])

def get_cluster_name(url):
    rootUrl = url + "wfapi/describe"
    root = requests.get(rootUrl).json()
    selectRootNodeUrl = [a for a in root['stages'] if "Select" in a["name"]][0]['_links']['self']['href']
    selectRootNode = requests.get(f'{jenkinsUrl}{selectRootNodeUrl}').json()
    shNodeUrl = [x for x in selectRootNode['stageFlowNodes'] if x['parameterDescription'] == './select-cluster.sh'][0]['_links']['log']['href']
    shNodeLog = requests.get(f'{jenkinsUrl}{shNodeUrl}').json()['text']

    lock_search = re.search("lock acquired on cluster cilium-ci-[0-9]+", shNodeLog)
    if lock_search is not None:
        return re.search("cilium-ci-[0-9]+", lock_search[0])[0]

used_clusters = {x for x in [get_cluster_name(url) for url in urls] if x is not None}

clusters = json.loads(
    subprocess.run(
        "gcloud container clusters list --format json".split(),
        capture_output=True
    ).stdout
)

scaled_clusters = {c["name"] for c in clusters if "initialNodeCount" in c["nodePools"][0]}

print("scaled but not used clusters, possible leaks:")
for c in scaled_clusters.difference(used_clusters):
    print(c)