Files
LokiStack/backend/services/deployment_usage.go
T
opsx bad2a1e624 fix(hosts,collectors,usage): 主机验证结果失真 + 主机模式系统日志解析 + metrics 单位
=== 一、主机「测试连接」永远显示旧错误(真实 bug)===
HostService.Validate 的失败分支也写 status=active,且从不写 statusMessage:
    if info, err := client.GetHostInfo(ctx); err == nil { 更新 os/arch }
    else { s.Update(id, {status:"active"}) }   // ← 失败也标记成功
结果:连不上也显示正常,页面永远停在几分钟前的旧错误文案,
用户改了 IP 再点测试,还以为"改了还是不行"。

改为区分成功/失败并写清原因:
    成功 → status=active + os/arch + statusMessage="连接成功"
    失败 → status=error + statusMessage=FormatSSHError(err)

=== 二、回环地址在平台里必然连不上 ===
平台跑在容器/Pod 内,127.0.0.1 指向平台自身。
- 新建主机时拦截:hostIP 为 127.0.0.1/localhost/::1 直接报错并解释原因
- 前端表单同步校验 + 文案
- SSH 错误映射补充:连接被拒绝时提示可能是地址填了回环

=== 三、主机模式采集系统日志(两处会导致采集失败)===
1. pipeline 一律套 cri 阶段:cri 只认容器运行时格式,
   套在 /var/log/messages 这类 syslog 上会取不到时间戳、整行被丢。
   → appPipelineStages() 按路径判断:容器日志用 cri,普通文本日志只做限流。
2. promtailExternalLabels 写死 8 空格缩进:主机模板 external_labels 只有 4 空格,
   一旦给采集器配置了标签就生成非法 YAML,Promtail 装不上。
   → 增加 indent 参数,主机 6 空格 / DaemonSet 8 空格。
3. 新增 system 采集任务:
   /var/log/{messages,secure,cron,maillog,syslog,dmesg}
   正则解析 syslog 时间戳(Jan _2 15:04:05,Asia/Shanghai)+ 限流

测试:TestHostPromtailConfigValid / CRIPathKeepsCRIStage / PlainFileHasNoCRIStage
      TestPromtailManifestParses 全部通过

=== 四、资源用量 CPU 解析(nanocores)===
metrics-server 返回的 CPU 常是纳核(如 "194533645n"=194.5m)。
原解析只处理 "m",导致 194 亿核、页面出现"CPU 已达 limit 的 4863341125%"。
→ parseQuantityMilli 支持 n/u/m/整核
实测:CPU 当前=178m,峰值=3300m,建议"峰值占 limit 82%,可考虑提前扩容"

=== 五、其他 ===
- 添加集群默认改用 Token 认证(最小权限、可随时吊销)
- RBAC 片段补 metrics.k8s.io(资源面板需要,独立 API 组常被漏授)
- 文案去掉 Grafana 需求描述(平台自身不部署 Grafana)

镜像:lghost/lokistack:latest + v0.1 = sha256:7bddd0f4d1f1
2026-10-28 17:58:30 +00:00

488 lines
16 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package services
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"lokistack/models"
)
// ResourceUsage 是页面上「资源」弹框的数据。
//
// 三种部署模式都能采集:
// - k8s-yaml : StatefulSet resources + metrics-server(实时用量)+ 可选 Prometheus(峰值/限流)
// - 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"`
// 可选(需 Prometheus)
CPUPeakMilli *int64 `json:"cpuPeakMilli,omitempty"`
MemPeakBytes *int64 `json:"memPeakBytes,omitempty"`
ThrottlePercent *float64 `json:"throttlePercent,omitempty"`
// 采集受限说明(如缺 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(id uint) (*ResourceUsage, error) {
d, err := s.Get(id)
if err != nil {
return nil, err
}
switch d.Mode {
case "k8s-yaml":
return s.k8sUsage(d)
case "host-docker":
return s.dockerUsage(d)
case "host-binary":
return s.processUsage(d)
default:
return nil, fmt.Errorf("不支持的部署模式:%s", d.Mode)
}
}
// ---------- Kubernetes ----------
func (s *DeploymentService) k8sUsage(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(context.Background(), 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 返回"
}
s.applyPrometheus(usage)
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()
}
}
// applyPrometheus 补 24h 峰值与 CPU 限流率(未配置 Prometheus 地址则跳过)。
func (s *DeploymentService) applyPrometheus(u *ResourceUsage) {
base := strings.TrimSpace(os.Getenv("LOKI_USAGE_PROMETHEUS_URL"))
if base == "" || u.Pod == "" {
if u.MetricsNote == "" {
u.MetricsNote = "未配置 Prometheus,暂无 24h 峰值与限流率"
}
return
}
sel := fmt.Sprintf(`pod=%q,namespace=%q,container="loki"`, u.Pod, u.Namespace)
if v, ok := promQueryValue(base, fmt.Sprintf(`max_over_time(sum(rate(container_cpu_usage_seconds_total{%s}[5m]))[24h:1m])`, sel)); ok {
milli := int64(v * 1000)
u.CPUPeakMilli = &milli
}
if v, ok := promQueryValue(base, fmt.Sprintf(`max_over_time(sum(container_memory_working_set_bytes{%s})[24h:1m])`, sel)); ok {
b := int64(v)
u.MemPeakBytes = &b
}
if v, ok := promQueryValue(base, fmt.Sprintf(`100 * sum(rate(container_cpu_cfs_throttled_periods_total{pod=%q,namespace=%q}[15m])) / sum(rate(container_cpu_cfs_periods_total{pod=%q,namespace=%q}[15m]))`, u.Pod, u.Namespace, u.Pod, u.Namespace)); ok {
u.ThrottlePercent = &v
}
}
// ---------- host-docker ----------
func (s *DeploymentService) dockerUsage(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(context.Background(), 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 = "Docker 部署无历史峰值;如需 24h 峰值请接入 cAdvisor/Prometheus"
}
usage.buildHints()
return usage, nil
}
// ---------- host-binary ----------
func (s *DeploymentService) processUsage(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(context.Background(), 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() {
cpuRef := u.CPUCurrentMilli
if u.CPUPeakMilli != nil && *u.CPUPeakMilli > cpuRef {
cpuRef = *u.CPUPeakMilli
}
memRef := u.MemCurrentBytes
if u.MemPeakBytes != nil && *u.MemPeakBytes > memRef {
memRef = *u.MemPeakBytes
}
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.ThrottlePercent != nil && *u.ThrottlePercent >= 10 {
notes = append(notes, fmt.Sprintf("CPU 限流 %.0f%%", *u.ThrottlePercent))
}
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, "、")
}
}
// ---------- Prometheus ----------
// promQueryValue 查 Prometheus 瞬时值;未配置或失败返回 ok=false(页面降级)。
func promQueryValue(base, expr string) (float64, bool) {
client := &http.Client{Timeout: 8 * time.Second}
u := strings.TrimSuffix(base, "/") + "/api/v1/query?query=" + url.QueryEscape(expr)
resp, err := client.Get(u)
if err != nil {
return 0, false
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return 0, false
}
var payload struct {
Data struct {
Result []struct {
Value [2]interface{} `json:"value"`
} `json:"result"`
} `json:"data"`
}
if json.NewDecoder(resp.Body).Decode(&payload) != nil || len(payload.Data.Result) == 0 {
return 0, false
}
if len(payload.Data.Result[0].Value) < 2 {
return 0, false
}
raw, ok := payload.Data.Result[0].Value[1].(string)
if !ok {
return 0, false
}
return parseFloatSafe(raw), true
}
// ---------- 单位解析 ----------
// 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))
}