123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
from fastapi import APIRouter
|
|
from kubernetes import client, config
|
|
from datetime import datetime, timezone
|
|
|
|
router = APIRouter()
|
|
|
|
def fetch_k8s_data_with_usage():
|
|
config.load_incluster_config()
|
|
v1 = client.CoreV1Api()
|
|
metrics_client = client.CustomObjectsApi()
|
|
|
|
# Fetch nodes
|
|
nodes = []
|
|
for node in v1.list_node().items:
|
|
# Extract storage (ephemeral-storage) and instance type
|
|
ephemeral_storage = node.status.capacity.get("ephemeral-storage", "0")
|
|
instance_type = node.metadata.labels.get("beta.kubernetes.io/instance-type", "N/A")
|
|
|
|
# Calculate time on duty
|
|
creation_timestamp = node.metadata.creation_timestamp
|
|
if creation_timestamp:
|
|
time_on_duty = calculate_time_on_duty(creation_timestamp)
|
|
else:
|
|
time_on_duty = "N/A"
|
|
|
|
nodes.append({
|
|
"node_name": node.metadata.name,
|
|
"cpu": node.status.capacity.get("cpu"),
|
|
"memory": round(convert_memory_to_mib(node.status.capacity.get("memory")), 2), # Convert to MiB
|
|
"storage": f"{round(convert_memory_to_gb(ephemeral_storage), 2)} GB",
|
|
"instance_type": instance_type,
|
|
"pods_allocatable": node.status.allocatable.get("pods"),
|
|
"time_on_duty": time_on_duty, # Add time on duty
|
|
})
|
|
|
|
# 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_cluster_custom_object(
|
|
group="metrics.k8s.io", version="v1beta1", plural="pods"
|
|
)
|
|
for pod in pod_metrics["items"]:
|
|
pod_namespace = pod["metadata"]["namespace"]
|
|
if pod_namespace not in namespace_usage:
|
|
namespace_usage[pod_namespace] = {"cpu": 0, "memory": 0}
|
|
|
|
for container in pod["containers"]:
|
|
cpu_usage = container["usage"]["cpu"]
|
|
memory_usage = container["usage"]["memory"]
|
|
|
|
# Convert CPU to cores and memory to MiB
|
|
namespace_usage[pod_namespace]["cpu"] += convert_cpu_to_cores(cpu_usage)
|
|
namespace_usage[pod_namespace]["memory"] += convert_memory_to_mib(memory_usage)
|
|
|
|
# Round and format usage for readability
|
|
namespace_usage = {
|
|
ns: {
|
|
"cpu": round(usage["cpu"], 4), # Round to 4 decimal places
|
|
"memory": round(usage["memory"], 2), # Memory in MiB
|
|
}
|
|
for ns, usage in namespace_usage.items()
|
|
}
|
|
|
|
return {"nodes": nodes, "namespaces": namespaces, "namespace_usage": namespace_usage}
|
|
|
|
|
|
def calculate_time_on_duty(creation_timestamp):
|
|
"""
|
|
Calculate the time on duty in hours, days, or minutes from the creation timestamp.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
delta = now - creation_timestamp
|
|
|
|
# If less than an hour, return minutes
|
|
if delta.days < 1 and delta.seconds < 3600:
|
|
minutes = delta.seconds // 60
|
|
return f"{minutes} minutes" if minutes > 1 else "less than a minute"
|
|
|
|
# If less than a day, return hours
|
|
if delta.days < 1:
|
|
hours = delta.seconds // 3600
|
|
return f"{hours} hours" if hours > 1 else "1 hour"
|
|
|
|
# Otherwise, return days
|
|
return f"{delta.days} days" if delta.days > 1 else "1 day"
|
|
|
|
|
|
def convert_memory_to_gb(memory):
|
|
"""
|
|
Convert memory to GB (gigabytes) for ephemeral-storage.
|
|
"""
|
|
if "Ki" in memory:
|
|
return int(memory.replace("Ki", "")) / (1024 ** 2)
|
|
elif "Mi" in memory:
|
|
return int(memory.replace("Mi", "")) / 1024
|
|
elif "Gi" in memory:
|
|
return int(memory.replace("Gi", ""))
|
|
return float(memory)
|
|
|
|
def convert_cpu_to_cores(cpu):
|
|
if "n" in cpu:
|
|
return round(int(cpu.replace("n", "")) / 1e9, 4)
|
|
elif "u" in cpu:
|
|
return round(int(cpu.replace("u", "")) / 1e6, 4)
|
|
elif "m" in cpu:
|
|
return round(int(cpu.replace("m", "")) / 1000, 4)
|
|
return float(cpu)
|
|
|
|
def convert_memory_to_mib(memory):
|
|
if "Ki" in memory:
|
|
return int(memory.replace("Ki", "")) / 1024
|
|
elif "Mi" in memory:
|
|
return int(memory.replace("Mi", ""))
|
|
elif "Gi" in memory:
|
|
return int(memory.replace("Gi", "")) * 1024
|
|
return float(memory)
|
|
|
|
@router.get("/k8s/data")
|
|
def get_k8s_data():
|
|
return fetch_k8s_data_with_usage()
|