terghalin/metalcheck
1
0
Fork 0
mirror of https://ak-git.vectorsigma.ru/terghalin/metalcheck.git synced 2026-03-12 16:45:17 +09:00
Code Issues Actions Packages Projects Releases 1 Wiki Activity

fix(k8s): serialization

Browse Source
This commit is contained in:
Aleksandr Tcitlionok
2024-12-05 07:00:03 +00:00
parent 1db862e770
commit 092ac307ed
1 changed files with 17 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
app/routes/k8s.py
View File

@@ -5,11 +5,11 @@ import math
router = APIRouter()
def fetch_k8s_data_with_usage():
config.load_incluster_config()
config.load_incluster_config() # Use in-cluster configuration
v1 = client.CoreV1Api()
metrics_client = client.CustomObjectsApi()
# Fetch nodes and namespaces
# Fetch nodes
nodes = [{
"node_name": node.metadata.name,
"cpu": node.status.capacity.get("cpu"),
@@ -17,12 +17,13 @@ def fetch_k8s_data_with_usage():
"pods_allocatable": node.status.allocatable.get("pods"),
} for node in v1.list_node().items]
# Fetch namespaces
namespaces = [ns.metadata.name for ns in v1.list_namespace().items]
# Fetch pod metrics and calculate namespace resource usage
namespace_usage = {}
pod_metrics = metrics_client.list_namespaced_custom_object(
group="metrics.k8s.io", version="v1beta1", namespace="", plural="pods"
pod_metrics = metrics_client.list_cluster_custom_object(
group="metrics.k8s.io", version="v1beta1", plural="pods"
)
for pod in pod_metrics["items"]:
pod_namespace = pod["metadata"]["namespace"]
@@ -37,14 +38,24 @@ def fetch_k8s_data_with_usage():
namespace_usage[pod_namespace]["cpu"] += convert_cpu_to_millicores(cpu_usage)
namespace_usage[pod_namespace]["memory"] += convert_memory_to_mib(memory_usage)
# Convert usage to serializable types
namespace_usage = {
ns: {
"cpu": round(usage["cpu"], 2), # Round to 2 decimal places for readability
"memory": round(usage["memory"], 2),
}
for ns, usage in namespace_usage.items()
}
return {"nodes": nodes, "namespaces": namespaces, "namespace_usage": namespace_usage}
def convert_cpu_to_millicores(cpu):
if "n" in cpu:
return int(cpu.replace("n", "")) / 1e6
elif "m" in cpu:
return int(cpu.replace("m", ""))
return int(cpu) * 1000
return float(cpu) * 1000 # Use float to ensure compatibility
def convert_memory_to_mib(memory):
if "Ki" in memory:
@@ -53,7 +64,7 @@ def convert_memory_to_mib(memory):
return int(memory.replace("Mi", ""))
elif "Gi" in memory:
return int(memory.replace("Gi", "")) * 1024
return int(memory)
return float(memory)
@router.get("/k8s/data")
def get_k8s_data():