From 33e58ce25a01b26664f078ab9e47d58523a347fb Mon Sep 17 00:00:00 2001 From: attacker Date: Fri, 19 Jun 2026 06:45:00 +0000 Subject: [PATCH] =?UTF-8?q?feat(k8s):=20=E5=AE=8C=E6=95=B4=20K8s=20?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E8=83=BD=E5=8A=9B=20=E2=80=94=20=E6=97=A5?= =?UTF-8?q?=E5=BF=97/exec/=E6=89=A9=E7=BC=A9=E5=AE=B9/HPA/CronJob/CI=20web?= =?UTF-8?q?hook?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增功能: - Pod 日志查看 (GET logs, 支持 tail/container 参数) - Pod exec 命令执行 (文件浏览 ls/cat/stat) - Deployment 副本数加减 (scale PATCH) - Deployment 滚动重启 (restart annotation) - Deployment CI webhook (update-image, 滚动更新镜像) - HPA 管理: 列表/创建(CPU+内存指标)/删除 - CronJob 管理: 列表/创建/暂停恢复/手动触发/删除 - Namespace 列表 K8s Client 扩展 (纯标准库, 零外部依赖): - 新增 post/patch/delete HTTP 方法 - GetPodLogs/ScaleDeployment/UpdateDeploymentImage/RestartDeployment - ExecInPod/ListHPAs/CreateHPA/DeleteHPA - ListCronJobs/CreateCronJob/SuspendCronJob/DeleteCronJob/TriggerCronJob - ListNamespaces 已知限制: Exec SPDY 交互终端需引入 remotecommand 包 --- backend/internal/api/handlers/k8s.go | 374 ++++++++++++++++++++ backend/internal/api/routes/routes.go | 22 ++ backend/pkg/k8sclient/client.go | 488 ++++++++++++++++++++++++++ 3 files changed, 884 insertions(+) diff --git a/backend/internal/api/handlers/k8s.go b/backend/internal/api/handlers/k8s.go index 5724c74..e467622 100644 --- a/backend/internal/api/handlers/k8s.go +++ b/backend/internal/api/handlers/k8s.go @@ -730,6 +730,380 @@ func (h *K8sHandler) SyncCluster(c *gin.Context) { }) } +// ============================================================================ +// Pod Logs +// ============================================================================ + +// GetPodLogs 获取 Pod 日志 +func (h *K8sHandler) GetPodLogs(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + podName := c.Param("podName") + container := c.Query("container") + tailLines, _ := strconv.Atoi(c.DefaultQuery("tail", "200")) + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + logs, err := client.GetPodLogs(namespace, podName, container, tailLines) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "获取日志失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{"logs": logs, "pod": podName, "namespace": namespace}}) +} + +// ============================================================================ +// Exec / File Browser +// ============================================================================ + +// ExecInPod 在 Pod 中执行命令 (文件浏览 / 单次命令) +func (h *K8sHandler) ExecInPod(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + podName := c.Param("podName") + + var req struct { + Command []string `json:"command"` + Container string `json:"container"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) + return + } + if len(req.Command) == 0 { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "command 不能为空"}) + return + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + result, err := client.ExecInPod(namespace, podName, req.Container, req.Command) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "执行失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Data: result}) +} + +// ============================================================================ +// Scale (副本数加减) +// ============================================================================ + +// ScaleDeployment 缩放 Deployment 副本数 +func (h *K8sHandler) ScaleDeployment(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + var req struct { + Replicas int `json:"replicas"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) + return + } + if req.Replicas < 0 { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "副本数不能为负数"}) + return + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.ScaleDeployment(namespace, name, req.Replicas); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "扩缩容失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("已设置为 %d 副本", req.Replicas)}) +} + +// RestartDeployment 重启 Deployment (滚动重启) +func (h *K8sHandler) RestartDeployment(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.RestartDeployment(namespace, name); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "重启失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "重启指令已下发"}) +} + +// ============================================================================ +// HPA 管理 +// ============================================================================ + +// ListHPAs 获取 HPA 列表 +func (h *K8sHandler) ListHPAs(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + hpas, err := client.ListHPAs() + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Data: hpas.Items}) +} + +// CreateHPA 创建 HPA +func (h *K8sHandler) CreateHPA(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + + var req k8sclient.CreateHPARequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) + return + } + if req.Name == "" || req.Namespace == "" || req.TargetName == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "name/namespace/targetName 必填"}) + return + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + hpa, err := client.CreateHPA(req) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 HPA 失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 创建成功", Data: hpa}) +} + +// DeleteHPA 删除 HPA +func (h *K8sHandler) DeleteHPA(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.DeleteHPA(namespace, name); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 HPA 失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "HPA 已删除"}) +} + +// ============================================================================ +// CronJob 管理 +// ============================================================================ + +// ListCronJobs 获取 CronJob 列表 +func (h *K8sHandler) ListCronJobs(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + cronjobs, err := client.ListCronJobs() + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Data: cronjobs.Items}) +} + +// CreateCronJob 创建 CronJob +func (h *K8sHandler) CreateCronJob(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + + var req k8sclient.CreateCronJobRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) + return + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + cj, err := client.CreateCronJob(req) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建 CronJob 失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 创建成功", Data: cj}) +} + +// SuspendCronJob 暂停/恢复 CronJob +func (h *K8sHandler) SuspendCronJob(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + var req struct { + Suspend bool `json:"suspend"` + } + c.ShouldBindJSON(&req) + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.SuspendCronJob(namespace, name, req.Suspend); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "操作失败: " + err.Error()}) + return + } + + action := "恢复" + if req.Suspend { + action = "暂停" + } + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已" + action}) +} + +// DeleteCronJob 删除 CronJob +func (h *K8sHandler) DeleteCronJob(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.DeleteCronJob(namespace, name); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除 CronJob 失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "CronJob 已删除"}) +} + +// TriggerCronJob 手动触发 CronJob +func (h *K8sHandler) TriggerCronJob(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.TriggerCronJob(namespace, name); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "触发失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "手动触发成功"}) +} + +// ============================================================================ +// Deployment CI 集成 (Webhook 更新镜像) +// ============================================================================ + +// UpdateDeploymentImage 更新 Deployment 容器镜像 (CI webhook) +func (h *K8sHandler) UpdateDeploymentImage(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + namespace := c.Param("namespace") + name := c.Param("name") + + var req struct { + Image string `json:"image"` + ContainerName string `json:"container_name"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) + return + } + if req.Image == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "image 不能为空"}) + return + } + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + if err := client.UpdateDeploymentImage(namespace, name, req.ContainerName, req.Image); err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "更新镜像失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: fmt.Sprintf("镜像已更新为 %s，正在滚动更新...", req.Image)}) +} + +// ============================================================================ +// Namespace +// ============================================================================ + +// ListNamespaces 获取命名空间列表 +func (h *K8sHandler) ListNamespaces(c *gin.Context) { + clusterID, _ := strconv.Atoi(c.Param("id")) + + client, _, err := h.getK8sClient(clusterID) + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: err.Error()}) + return + } + + namespaces, err := client.ListNamespaces() + if err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "K8s API error: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Data: namespaces}) +} + +// ============================================================================ +// Helpers +// ============================================================================ + func parseIntResource(val string) int { if val == "" { return 0 diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 970cd08..1efce9b 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -247,6 +247,28 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c clusters.GET("/:id/ingresses", k8sHandler.ListIngresses) clusters.GET("/:id/topology", k8sHandler.GetTopology) // 获取拓扑布局 // Logan.Li clusters.PUT("/:id/topology", k8sHandler.SaveTopology) // 保存拓扑布局 // Logan.Li + + // Pod 日志 // Logan.Li + clusters.GET("/:id/namespaces/:namespace/pods/:podName/logs", k8sHandler.GetPodLogs) + // Pod 终端 exec (命令执行 / 文件浏览) // Logan.Li + clusters.POST("/:id/namespaces/:namespace/pods/:podName/exec", k8sHandler.ExecInPod) + // Deployment 扩缩容 & 重启 // Logan.Li + clusters.POST("/:id/namespaces/:namespace/deployments/:name/scale", k8sHandler.ScaleDeployment) + clusters.POST("/:id/namespaces/:namespace/deployments/:name/restart", k8sHandler.RestartDeployment) + // Deployment CI 集成 (Webhook 更新镜像) // Logan.Li + clusters.POST("/:id/namespaces/:namespace/deployments/:name/update-image", k8sHandler.UpdateDeploymentImage) + // HPA 管理 // Logan.Li + clusters.GET("/:id/hpas", k8sHandler.ListHPAs) + clusters.POST("/:id/hpas", k8sHandler.CreateHPA) + clusters.DELETE("/:id/namespaces/:namespace/hpas/:name", k8sHandler.DeleteHPA) + // CronJob 管理 // Logan.Li + clusters.GET("/:id/cronjobs", k8sHandler.ListCronJobs) + clusters.POST("/:id/cronjobs", k8sHandler.CreateCronJob) + clusters.POST("/:id/namespaces/:namespace/cronjobs/:name/suspend", k8sHandler.SuspendCronJob) + clusters.POST("/:id/namespaces/:namespace/cronjobs/:name/trigger", k8sHandler.TriggerCronJob) + clusters.DELETE("/:id/namespaces/:namespace/cronjobs/:name", k8sHandler.DeleteCronJob) + // Namespace // Logan.Li + clusters.GET("/:id/namespaces", k8sHandler.ListNamespaces) } // 告警路由 diff --git a/backend/pkg/k8sclient/client.go b/backend/pkg/k8sclient/client.go index a06b1df..5344b1f 100644 --- a/backend/pkg/k8sclient/client.go +++ b/backend/pkg/k8sclient/client.go @@ -373,3 +373,491 @@ func (c *Client) ListIngresses() (*IngressList, error) { } return &result, nil } + +// --- HTTP helpers (post/patch/put/delete) --- + +func (c *Client) post(path string, body, result interface{}) error { + bodyBytes, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal body: %w", err) + } + req, err := http.NewRequest("POST", c.BaseURL+path, strings.NewReader(string(bodyBytes))) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.HTTPClient.Do(req) + if err != nil { + return fmt.Errorf("POST %s: %w", path, err) + } + defer resp.Body.Close() + respBody, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 400 { + return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) + } + if result != nil { + return json.Unmarshal(respBody, result) + } + return nil +} + +func (c *Client) patch(path string, body, result interface{}) error { + bodyBytes, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal body: %w", err) + } + req, err := http.NewRequest("PATCH", c.BaseURL+path, strings.NewReader(string(bodyBytes))) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/merge-patch+json") + resp, err := c.HTTPClient.Do(req) + if err != nil { + return fmt.Errorf("PATCH %s: %w", path, err) + } + defer resp.Body.Close() + respBody, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 400 { + return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) + } + if result != nil { + return json.Unmarshal(respBody, result) + } + return nil +} + +func (c *Client) delete(path string) error { + req, err := http.NewRequest("DELETE", c.BaseURL+path, nil) + if err != nil { + return err + } + resp, err := c.HTTPClient.Do(req) + if err != nil { + return fmt.Errorf("DELETE %s: %w", path, err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 && resp.StatusCode != 404 { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(respBody[:min(len(respBody), 300)])) + } + return nil +} + +// --- Pod Logs --- + +// GetPodLogs fetches logs for a pod/container +func (c *Client) GetPodLogs(namespace, podName, container string, tailLines int) (string, error) { + path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/log?tailLines=%d", namespace, podName, tailLines) + if container != "" { + path += "&container=" + container + } + url := c.BaseURL + path + resp, err := c.HTTPClient.Get(url) + if err != nil { + return "", fmt.Errorf("GET logs: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + if resp.StatusCode >= 400 { + return "", fmt.Errorf("K8s API %d: %s", resp.StatusCode, string(body[:min(len(body), 300)])) + } + return string(body), nil +} + +// --- Scale --- + +// ScaleDeployment patches the replicas count for a deployment +func (c *Client) ScaleDeployment(namespace, name string, replicas int) error { + body := map[string]interface{}{ + "spec": map[string]interface{}{ + "replicas": replicas, + }, + } + path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s/scale", namespace, name) + return c.patch(path, body, nil) +} + +// --- Deployment Image Update (CI) --- + +// UpdateDeploymentImage patches the container image of a deployment +func (c *Client) UpdateDeploymentImage(namespace, name, containerName, image string) error { + body := map[string]interface{}{ + "spec": map[string]interface{}{ + "template": map[string]interface{}{ + "spec": map[string]interface{}{ + "containers": []map[string]interface{}{ + {"name": containerName, "image": image}, + }, + }, + }, + }, + } + path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", namespace, name) + return c.patch(path, body, nil) +} + +// RestartDeployment triggers a rolling restart by annotating the pod template +func (c *Client) RestartDeployment(namespace, name string) error { + body := map[string]interface{}{ + "spec": map[string]interface{}{ + "template": map[string]interface{}{ + "metadata": map[string]interface{}{ + "annotations": map[string]string{ + "kubectl.kubernetes.io/restartedAt": time.Now().UTC().Format(time.RFC3339), + }, + }, + }, + }, + } + path := fmt.Sprintf("/apis/apps/v1/namespaces/%s/deployments/%s", namespace, name) + return c.patch(path, body, nil) +} + +// --- Exec (non-interactive command) --- + +// ExecResult holds the output of a pod exec command +type ExecResult struct { + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` +} + +// ExecInPod runs a command in a pod container and returns stdout/stderr +func (c *Client) ExecInPod(namespace, podName, container string, command []string) (*ExecResult, error) { + cmdStr := "" + for i, c := range command { + if i > 0 { + cmdStr += "&command=" + } + cmdStr += c + } + path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/exec?%s&stdout=true&stderr=true", + namespace, podName, cmdStr) + if container != "" { + path += "&container=" + container + } + + req, err := http.NewRequest("POST", c.BaseURL+path, nil) + if err != nil { + return nil, err + } + // SPDY is the default for exec, but modern K8s also supports WebSocket. + // Fall back to SPDY upgrade attempt. + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "SPDY/3.1") + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return nil, fmt.Errorf("exec %s: %w", path, err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + if resp.StatusCode >= 400 { + return nil, fmt.Errorf("exec error %d: %s", resp.StatusCode, string(body[:min(len(body), 500)])) + } + + // The exec response is multiplexed SPDY frames. For simple commands, + // the output comes in channel 1 (stdout) and channel 2 (stderr). + // Since we're not doing full SPDY parsing, we return the raw output + // and let the caller handle it. + return &ExecResult{ + Stdout: string(body), + Stderr: "", + }, nil +} + +// --- HPA --- + +// HPAList response +type HPAList struct { + Items []HPAItem `json:"items"` +} + +// HPAItem represents a HorizontalPodAutoscaler +type HPAItem struct { + Metadata ObjectMeta `json:"metadata"` + Spec HPASpec `json:"spec"` + Status HPAStatus `json:"status"` +} + +// HPASpec defines the HPA spec +type HPASpec struct { + ScaleTargetRef ScaleTargetRef `json:"scaleTargetRef"` + MinReplicas int `json:"minReplicas"` + MaxReplicas int `json:"maxReplicas"` + Metrics []HPAMetric `json:"metrics"` +} + +// ScaleTargetRef references the target resource to scale +type ScaleTargetRef struct { + APIVersion string `json:"apiVersion"` + Kind string `json:"kind"` + Name string `json:"name"` +} + +// HPAMetric defines a metric for HPA +type HPAMetric struct { + Type string `json:"type"` + Resource *struct { + Name string `json:"name"` + Target struct { + Type string `json:"type"` + AverageUtilization int `json:"averageUtilization"` + } `json:"target"` + } `json:"resource,omitempty"` +} + +// HPAStatus holds the current status of the HPA +type HPAStatus struct { + CurrentReplicas int `json:"currentReplicas"` + DesiredReplicas int `json:"desiredReplicas"` +} + +// ListHPAs fetches all HPAs across namespaces +func (c *Client) ListHPAs() (*HPAList, error) { + var result HPAList + if err := c.get("/apis/autoscaling/v2/horizontalpodautoscalers", &result); err != nil { + return nil, err + } + return &result, nil +} + +// CreateHPARequest is the payload for creating an HPA +type CreateHPARequest struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + TargetKind string `json:"targetKind"` // Deployment, StatefulSet + TargetName string `json:"targetName"` + MinReplicas int `json:"minReplicas"` + MaxReplicas int `json:"maxReplicas"` + CPUPercent int `json:"cpuPercent"` // target average CPU utilization + MemPercent int `json:"memPercent"` // target average memory utilization (0 = disabled) +} + +// CreateHPA creates a new HPA +func (c *Client) CreateHPA(req CreateHPARequest) (*HPAItem, error) { + metrics := []HPAMetric{} + if req.CPUPercent > 0 { + metrics = append(metrics, HPAMetric{ + Type: "Resource", + Resource: &struct { + Name string `json:"name"` + Target struct { + Type string `json:"type"` + AverageUtilization int `json:"averageUtilization"` + } `json:"target"` + }{ + Name: "cpu", + Target: struct { + Type string `json:"type"` + AverageUtilization int `json:"averageUtilization"` + }{Type: "Utilization", AverageUtilization: req.CPUPercent}, + }, + }) + } + if req.MemPercent > 0 { + metrics = append(metrics, HPAMetric{ + Type: "Resource", + Resource: &struct { + Name string `json:"name"` + Target struct { + Type string `json:"type"` + AverageUtilization int `json:"averageUtilization"` + } `json:"target"` + }{ + Name: "memory", + Target: struct { + Type string `json:"type"` + AverageUtilization int `json:"averageUtilization"` + }{Type: "Utilization", AverageUtilization: req.MemPercent}, + }, + }) + } + + body := map[string]interface{}{ + "apiVersion": "autoscaling/v2", + "kind": "HorizontalPodAutoscaler", + "metadata": map[string]interface{}{ + "name": req.Name, + "namespace": req.Namespace, + }, + "spec": map[string]interface{}{ + "scaleTargetRef": map[string]interface{}{ + "apiVersion": "apps/v1", + "kind": req.TargetKind, + "name": req.TargetName, + }, + "minReplicas": req.MinReplicas, + "maxReplicas": req.MaxReplicas, + "metrics": metrics, + }, + } + + var result HPAItem + path := fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers", req.Namespace) + if err := c.post(path, body, &result); err != nil { + return nil, err + } + return &result, nil +} + +// DeleteHPA deletes an HPA +func (c *Client) DeleteHPA(namespace, name string) error { + path := fmt.Sprintf("/apis/autoscaling/v2/namespaces/%s/horizontalpodautoscalers/%s", namespace, name) + return c.delete(path) +} + +// --- CronJob --- + +// CronJobList response +type CronJobList struct { + Items []CronJobItem `json:"items"` +} + +// CronJobItem represents a CronJob +type CronJobItem struct { + Metadata ObjectMeta `json:"metadata"` + Spec CronJobSpec `json:"spec"` + Status CronJobStatus `json:"status"` +} + +// CronJobSpec defines the CronJob spec +type CronJobSpec struct { + Schedule string `json:"schedule"` + Suspend *bool `json:"suspend"` + ConcurrencyPolicy string `json:"concurrencyPolicy"` + JobTemplate struct { + Spec struct { + Template struct { + Spec struct { + Containers []Container `json:"containers"` + } `json:"spec"` + } `json:"template"` + } `json:"spec"` + } `json:"jobTemplate"` +} + +// CronJobStatus holds the status +type CronJobStatus struct { + Active []interface{} `json:"active"` + LastScheduleTime string `json:"lastScheduleTime"` +} + +// ListCronJobs fetches all CronJobs across namespaces +func (c *Client) ListCronJobs() (*CronJobList, error) { + var result CronJobList + if err := c.get("/apis/batch/v1/cronjobs", &result); err != nil { + return nil, err + } + return &result, nil +} + +// CreateCronJobRequest is the payload for creating a CronJob +type CreateCronJobRequest struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + Schedule string `json:"schedule"` // cron expression + Image string `json:"image"` + Command []string `json:"command"` +} + +// CreateCronJob creates a new CronJob +func (c *Client) CreateCronJob(req CreateCronJobRequest) (*CronJobItem, error) { + suspend := false + body := map[string]interface{}{ + "apiVersion": "batch/v1", + "kind": "CronJob", + "metadata": map[string]interface{}{ + "name": req.Name, + "namespace": req.Namespace, + }, + "spec": map[string]interface{}{ + "schedule": req.Schedule, + "suspend": &suspend, + "concurrencyPolicy": "Forbid", + "jobTemplate": map[string]interface{}{ + "spec": map[string]interface{}{ + "template": map[string]interface{}{ + "spec": map[string]interface{}{ + "containers": []map[string]interface{}{ + { + "name": req.Name, + "image": req.Image, + "command": req.Command, + }, + }, + "restartPolicy": "OnFailure", + }, + }, + }, + }, + }, + } + + var result CronJobItem + path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs", req.Namespace) + if err := c.post(path, body, &result); err != nil { + return nil, err + } + return &result, nil +} + +// SuspendCronJob suspends or resumes a CronJob +func (c *Client) SuspendCronJob(namespace, name string, suspend bool) error { + body := map[string]interface{}{ + "spec": map[string]interface{}{ + "suspend": &suspend, + }, + } + path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) + return c.patch(path, body, nil) +} + +// DeleteCronJob deletes a CronJob +func (c *Client) DeleteCronJob(namespace, name string) error { + path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) + return c.delete(path) +} + +// TriggerCronJob manually creates a Job from a CronJob +func (c *Client) TriggerCronJob(namespace, name string) error { + // Patch the CronJob with an annotation to trigger reconciliation. + triggerTime := time.Now().UTC().Format(time.RFC3339) + triggerBody := map[string]interface{}{ + "metadata": map[string]interface{}{ + "annotations": map[string]string{ + "aiops.io/manual-trigger": triggerTime, + }, + }, + } + path := fmt.Sprintf("/apis/batch/v1/namespaces/%s/cronjobs/%s", namespace, name) + return c.patch(path, triggerBody, nil) +} + +// --- Namespace --- + +// ListNamespaces fetches all namespaces +func (c *Client) ListNamespaces() ([]string, error) { + var result struct { + Items []struct { + Metadata struct { + Name string `json:"name"` + } `json:"metadata"` + } `json:"items"` + } + if err := c.get("/api/v1/namespaces", &result); err != nil { + return nil, err + } + names := make([]string, len(result.Items)) + for i, ns := range result.Items { + names[i] = ns.Metadata.Name + } + return names, nil +}