package services import ( "context" "encoding/json" "fmt" "strconv" "strings" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "lokistack/models" ) // ResourceUsage 是页面上「资源」弹框的数据。 // // 三种部署模式都能采集: // - k8s-yaml : StatefulSet resources + metrics-server(实时用量) // - host-docker : docker stats / docker inspect(容器限额未设则按宿主机内存做参考) // - host-binary : ps 采集进程 CPU%/RSS(无 cgroup 限额) // // 权限兼容:集群 token 常是最小权限,可能没有 metrics.k8s.io。 // 这时**不报错**,照样返回 limits/就绪/重启,并把原因写进 MetricsNote 让页面提示。 type ResourceUsage struct { Mode string `json:"mode"` Source string `json:"source"` Namespace string `json:"namespace,omitempty"` Pod string `json:"pod,omitempty"` Node string `json:"node,omitempty"` Ready bool `json:"ready"` Restarts int32 `json:"restarts"` CPUCurrentMilli int64 `json:"cpuCurrentMilli"` CPURequestMilli int64 `json:"cpuRequestMilli"` CPULimitMilli int64 `json:"cpuLimitMilli"` MemCurrentBytes int64 `json:"memCurrentBytes"` MemRequestBytes int64 `json:"memRequestBytes"` MemLimitBytes int64 `json:"memLimitBytes"` // 限额是否真的配置了(docker 未加 --memory 时为 false,页面不画百分比) LimitsConfigured bool `json:"limitsConfigured"` // 采集受限说明(如缺 metrics.k8s.io 权限、未装 metrics-server、docker 无历史数据) MetricsNote string `json:"metricsNote,omitempty"` CPUSuggest string `json:"cpuSuggest,omitempty"` MemSuggest string `json:"memSuggest,omitempty"` Summary string `json:"summary"` } type podMetricsResponse struct { Containers []struct { Name string `json:"name"` Usage struct { CPU string `json:"cpu"` Memory string `json:"memory"` } `json:"usage"` } `json:"containers"` } // GetUsage 采集某个 Loki 部署的资源用量(按部署模式分派)。 func (s *DeploymentService) GetUsage(ctx context.Context, id uint) (*ResourceUsage, error) { d, err := s.Get(id) if err != nil { return nil, err } switch d.Mode { case "k8s-yaml": return s.k8sUsage(ctx, d) case "host-docker": return s.dockerUsage(ctx, d) case "host-binary": return s.processUsage(ctx, d) default: return nil, fmt.Errorf("不支持的部署模式:%s", d.Mode) } } // ---------- Kubernetes ---------- func (s *DeploymentService) k8sUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Cluster == nil { return nil, fmt.Errorf("部署未关联 Kubernetes 集群") } client, err := clusterClient(d.Cluster, 20*time.Second) if err != nil { return nil, err } ctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() ns := d.Namespace if ns == "" { ns = "log" } usage := &ResourceUsage{Mode: d.Mode, Source: "kubernetes", Namespace: ns, Pod: "loki-0"} // 1) limits/requests 取线上 StatefulSet(以运行态为准,不信部署记录) if sts, err := client.AppsV1().StatefulSets(ns).Get(ctx, "loki", metav1.GetOptions{}); err == nil && len(sts.Spec.Template.Spec.Containers) > 0 { res := sts.Spec.Template.Spec.Containers[0].Resources usage.CPULimitMilli = parseQuantityMilli(res.Limits.Cpu().String()) usage.CPURequestMilli = parseQuantityMilli(res.Requests.Cpu().String()) usage.MemLimitBytes = parseQuantityBytes(res.Limits.Memory().String()) usage.MemRequestBytes = parseQuantityBytes(res.Requests.Memory().String()) usage.LimitsConfigured = usage.CPULimitMilli > 0 && usage.MemLimitBytes > 0 } // 2) Pod 状态 if pod, err := client.CoreV1().Pods(ns).Get(ctx, usage.Pod, metav1.GetOptions{}); err == nil { usage.Node = pod.Spec.NodeName for _, cs := range pod.Status.ContainerStatuses { if cs.Name == "loki" { usage.Ready = cs.Ready usage.Restarts = cs.RestartCount } } } // 3) 实时用量:metrics-server(最小权限 token 常常没有这个 API 组,要降级而不是报错) raw, err := client.Discovery().RESTClient().Get(). AbsPath("/apis/metrics.k8s.io/v1beta1/namespaces/" + ns + "/pods/" + usage.Pod). DoRaw(ctx) switch { case err != nil: usage.MetricsNote = explainMetricsError(err) case json.Unmarshal(raw, &podMetricsResponse{}) == nil: var pm podMetricsResponse if json.Unmarshal(raw, &pm) == nil { for _, c := range pm.Containers { if c.Name == "loki" { usage.CPUCurrentMilli = parseQuantityMilli(c.Usage.CPU) usage.MemCurrentBytes = parseQuantityBytes(c.Usage.Memory) } } } default: usage.MetricsNote = "无法解析 metrics-server 返回" } usage.buildHints() return usage, nil } // explainMetricsError 把 RBAC/安装问题翻译成可执行的提示。 func explainMetricsError(err error) string { msg := strings.ToLower(err.Error()) switch { case strings.Contains(msg, "forbidden") || strings.Contains(msg, "cannot get resource"): return "当前集群凭证缺少 metrics.k8s.io 权限,看不到实时用量。需要给该 ServiceAccount 增加:" + "apiGroups=[\"metrics.k8s.io\"] resources=[\"pods\"] verbs=[\"get\",\"list\"](其余限额/就绪数据仍可正常显示)" case strings.Contains(msg, "not found") || strings.Contains(msg, "could not find the requested resource"): return "集群未安装 metrics-server,无法获取实时 CPU/内存用量(限额与就绪状态仍可显示)" default: return "读取实时用量失败:" + err.Error() } } // ---------- host-docker ---------- func (s *DeploymentService) dockerUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Host == nil { return nil, fmt.Errorf("部署未关联主机") } client, err := NewSSHClientWithAuth(HostSSHAuth(d.Host)) if err != nil { return nil, fmt.Errorf("连接部署主机失败:%w", err) } defer client.Close() ctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() name := dockerContainerName(d.ID) usage := &ResourceUsage{Mode: d.Mode, Source: "docker", Pod: name, Node: d.Host.HostIP} // 容器状态 + 限额(docker run 未指定 --memory/--cpus 时限额为 0 = 不限制) inspectFmt := "{{.State.Status}}|{{.RestartCount}}|{{.HostConfig.NanoCpus}}|{{.HostConfig.Memory}}" out, stderr, err := client.Execute(ctx, fmt.Sprintf("docker inspect -f '%s' %s 2>&1", inspectFmt, name)) if err != nil || strings.Contains(out, "No such object") { return nil, fmt.Errorf("容器 %s 不存在或主机无法访问:%s", name, strings.TrimSpace(stderr)) } parts := strings.Split(strings.TrimSpace(out), "|") if len(parts) >= 4 { usage.Ready = parts[0] == "running" if n, err := strconv.Atoi(parts[1]); err == nil { usage.Restarts = int32(n) } if nano, err := strconv.ParseInt(parts[2], 10, 64); err == nil && nano > 0 { usage.CPULimitMilli = nano / 1e6 } if mem, err := strconv.ParseInt(parts[3], 10, 64); err == nil && mem > 0 { usage.MemLimitBytes = mem } usage.LimitsConfigured = usage.CPULimitMilli > 0 && usage.MemLimitBytes > 0 } // 实时用量:docker stats(一次采样) statsFmt := "{{.CPUPerc}}|{{.MemUsage}}" out, _, err = client.Execute(ctx, fmt.Sprintf("docker stats --no-stream --format '%s' %s 2>/dev/null", statsFmt, name)) if err == nil { if fields := strings.Split(strings.TrimSpace(out), "|"); len(fields) >= 2 { usage.CPUCurrentMilli = int64(parsePercent(fields[0]) * 10) // 1% = 10m used, limit := parseDockerMemUsage(fields[1]) usage.MemCurrentBytes = used if usage.MemLimitBytes == 0 { usage.MemLimitBytes = limit // 未设限额时 docker 报宿主机内存,作为参考上限 } } } else { usage.MetricsNote = "docker stats 读取失败:" + err.Error() } if usage.MetricsNote == "" { usage.MetricsNote = "未接入 Prometheus:暂无峰值/限流数据" } usage.buildHints() return usage, nil } // ---------- host-binary ---------- func (s *DeploymentService) processUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Host == nil { return nil, fmt.Errorf("部署未关联主机") } client, err := NewSSHClientWithAuth(HostSSHAuth(d.Host)) if err != nil { return nil, fmt.Errorf("连接部署主机失败:%w", err) } defer client.Close() ctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() usage := &ResourceUsage{Mode: d.Mode, Source: "binary", Pod: "loki", Node: d.Host.HostIP} out, _, err := client.Execute(ctx, "ps -eo pcpu,rss,comm --no-headers | awk '$3==\"loki\"{print $1\"|\"$2; exit}'") if err != nil || strings.TrimSpace(out) == "" { return nil, fmt.Errorf("主机上未找到 loki 进程") } if fields := strings.Split(strings.TrimSpace(out), "|"); len(fields) >= 2 { usage.CPUCurrentMilli = int64(parseFloatSafe(fields[0]) * 10) // %CPU → millicores(单核百分比) rssKB, _ := strconv.ParseInt(strings.TrimSpace(fields[1]), 10, 64) usage.MemCurrentBytes = rssKB * 1024 usage.Ready = true } usage.MetricsNote = "裸进程部署没有 cgroup 限额,只能看实时占用(无峰值/限额对比)" usage.buildHints() return usage, nil } // ---------- 共用:扩容建议 ---------- func (u *ResourceUsage) buildHints() { // 峰值需 Prometheus,平台不接入:扩容建议基于 metrics-server 实时值 cpuRef := u.CPUCurrentMilli memRef := u.MemCurrentBytes var notes []string if u.CPULimitMilli > 0 { ratio := float64(cpuRef) / float64(u.CPULimitMilli) switch { case ratio >= 0.9: u.CPUSuggest = fmt.Sprintf("CPU 峰值已达 limit 的 %.0f%%,建议把 CPU limit 提升到 %s 核", ratio*100, formatMilli(nextCPU(u.CPULimitMilli))) notes = append(notes, "CPU 打满") case ratio >= 0.7: u.CPUSuggest = fmt.Sprintf("CPU 峰值占 limit 的 %.0f%%,接近上限,可考虑提前扩容", ratio*100) notes = append(notes, "CPU 偏高") } } if u.MemLimitBytes > 0 { ratio := float64(memRef) / float64(u.MemLimitBytes) switch { case ratio >= 0.9: u.MemSuggest = fmt.Sprintf("内存峰值已达 limit 的 %.0f%%,有 OOMKilled 风险,建议提升到 %s", ratio*100, formatBytesShort(nextMem(u.MemLimitBytes))) notes = append(notes, "内存吃紧") case ratio >= 0.75: u.MemSuggest = fmt.Sprintf("内存峰值占 limit 的 %.0f%%,接近上限,可考虑提前扩容", ratio*100) notes = append(notes, "内存偏高") } } if !u.Ready { notes = append(notes, "实例未就绪") } if len(notes) == 0 { if u.LimitsConfigured { u.Summary = "资源充足:CPU / 内存均低于 limit 的 70%" } else { u.Summary = "未配置资源限额(建议设置 CPU / 内存上限,便于判断扩容时机)" } } else { u.Summary = "需要关注:" + strings.Join(notes, "、") } } // ---------- 单位解析 ---------- // parseQuantityMilli 解析 CPU 用量/限额为毫核。 // // 注意 metrics-server 返回的 CPU 常常是**纳核**(如 "194533645n" = 194.5 毫核), // 早先只处理了 "m" 后缀,导致 "194533645" 被当成 194 亿核, // 页面出现「CPU 已达 limit 的 4863341125%」这种荒谬结论。 func parseQuantityMilli(s string) int64 { s = strings.TrimSpace(s) if s == "" || s == "0" { return 0 } switch { case strings.HasSuffix(s, "n"): // 纳核 return int64(parseFloatSafe(strings.TrimSuffix(s, "n")) / 1e6) case strings.HasSuffix(s, "u"): // 微核 return int64(parseFloatSafe(strings.TrimSuffix(s, "u")) / 1e3) case strings.HasSuffix(s, "m"): // 毫核 return int64(parseFloatSafe(strings.TrimSuffix(s, "m"))) default: // 整核 return int64(parseFloatSafe(s) * 1000) } } func parseQuantityBytes(s string) int64 { s = strings.TrimSpace(s) if s == "" { return 0 } units := []struct { suffix string mul float64 }{ {"Gi", 1024 * 1024 * 1024}, {"Mi", 1024 * 1024}, {"Ki", 1024}, {"G", 1e9}, {"M", 1e6}, {"K", 1e3}, } for _, u := range units { if strings.HasSuffix(s, u.suffix) { return int64(parseFloatSafe(strings.TrimSuffix(s, u.suffix)) * u.mul) } } return int64(parseFloatSafe(s)) } func parseFloatSafe(s string) float64 { var f float64 if _, err := fmt.Sscanf(strings.TrimSpace(s), "%g", &f); err != nil { return 0 } return f } // parsePercent 解析 "0.15%" / "12.5%" → 数值百分比 func parsePercent(s string) float64 { return parseFloatSafe(strings.TrimSuffix(strings.TrimSpace(s), "%")) } // parseDockerMemUsage 解析 docker stats 的 "120MiB / 7.6GiB" → (used, limit) func parseDockerMemUsage(s string) (int64, int64) { parts := strings.Split(s, "/") if len(parts) != 2 { return 0, 0 } return parseDockerSize(parts[0]), parseDockerSize(parts[1]) } func parseDockerSize(s string) int64 { s = strings.TrimSpace(s) units := []struct { suffix string mul float64 }{ {"GiB", 1024 * 1024 * 1024}, {"MiB", 1024 * 1024}, {"KiB", 1024}, {"GB", 1e9}, {"MB", 1e6}, {"KB", 1e3}, {"B", 1}, } for _, u := range units { if strings.HasSuffix(s, u.suffix) { return int64(parseFloatSafe(strings.TrimSuffix(s, u.suffix)) * u.mul) } } return int64(parseFloatSafe(s)) } func nextCPU(currentMilli int64) int64 { for _, c := range []int64{1000, 2000, 4000, 6000, 8000, 16000} { if c > currentMilli { return c } } return currentMilli * 2 } func nextMem(currentBytes int64) int64 { for _, g := range []int64{2, 4, 8, 16, 32, 64} { if g*1024*1024*1024 > currentBytes { return g * 1024 * 1024 * 1024 } } return currentBytes * 2 } func formatMilli(milli int64) string { if milli%1000 == 0 { return fmt.Sprintf("%d", milli/1000) } return fmt.Sprintf("%.1f", float64(milli)/1000) } func formatBytesShort(b int64) string { if b >= 1024*1024*1024 { return fmt.Sprintf("%dGi", b/(1024*1024*1024)) } return fmt.Sprintf("%dMi", b/(1024*1024)) }