74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from kubernetes import client, config
|
|
from typing import List, Dict
|
|
import requests
|
|
import os
|
|
|
|
router = APIRouter()
|
|
|
|
def fetch_ai_summary(cluster_name: str, nodes: List[Dict], api_url: str, auth_token: str) -> str:
|
|
"""
|
|
Sends node data to an external AI API endpoint and retrieves a summary response.
|
|
"""
|
|
payload = {
|
|
"data": {
|
|
"cluster_name": cluster_name,
|
|
"nodes": nodes
|
|
}
|
|
}
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {auth_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(api_url, json=payload, headers=headers)
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
if result.get("success"):
|
|
return result.get("result", "No result provided by the API.")
|
|
else:
|
|
return "API responded with success=false."
|
|
else:
|
|
return f"API request failed with status code {response.status_code}: {response.text}"
|
|
except requests.RequestException as e:
|
|
return f"An error occurred while contacting the API: {str(e)}"
|
|
|
|
@router.get("/think/k8s")
|
|
def think_k8s():
|
|
"""
|
|
Fetch Kubernetes data, send it to the AI API, and return the AI-generated summary.
|
|
"""
|
|
config.load_incluster_config()
|
|
v1 = client.CoreV1Api()
|
|
|
|
cluster_name = "eks-staging"
|
|
|
|
nodes = []
|
|
for node in v1.list_node().items:
|
|
# Fetch 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")
|
|
|
|
nodes.append({
|
|
"name": node.metadata.name,
|
|
"cpu": f"{node.status.capacity.get('cpu')} cores",
|
|
"memory": f"{round(int(node.status.capacity.get('memory').replace('Ki', '')) / 1024 / 1024, 2)} GB",
|
|
"storage": f"{round(int(ephemeral_storage.replace('Ki', '')) / (1024 ** 2), 2)} GB",
|
|
"type": instance_type,
|
|
"namespaces": [ns.metadata.name for ns in v1.list_namespace().items]
|
|
})
|
|
|
|
api_url = os.getenv("AI_API_URL")
|
|
auth_token = os.getenv("AI_API_TOKEN")
|
|
|
|
if not api_url or not auth_token:
|
|
raise HTTPException(status_code=500, detail="AI API URL or token is not set. Please set AI_API_URL and AI_API_TOKEN environment variables.")
|
|
|
|
# Call AI API
|
|
summary = fetch_ai_summary(cluster_name, nodes, api_url, auth_token)
|
|
|
|
return {"summary": summary}
|
|
|