From 7e3356e32bb3cbd9b48ac9507f6b5faddb95e29c Mon Sep 17 00:00:00 2001 From: hermes-agent Date: Wed, 28 Oct 2026 20:01:38 +0000 Subject: [PATCH] =?UTF-8?q?feat(collectors):=20=E4=B8=BB=E6=9C=BA=E9=87=87?= =?UTF-8?q?=E9=9B=86=E5=99=A8=E5=AE=B9=E5=99=A8=E4=BC=98=E5=85=88=E6=89=98?= =?UTF-8?q?=E7=AE=A1=20+=20=E9=87=87=E9=9B=86=E5=99=A8=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E6=9F=A5=E7=9C=8B=20+=20=E6=8E=A8=E9=80=81=E5=9C=B0=E5=9D=80?= =?UTF-8?q?=E7=BA=A0=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现象 1. 主机采集器只有 systemd 一条路，跨发行版差异大（这轮已踩过 param 被删、日志路径、 时间戳格式三处）；进程存活/开机自启也没有统一保证。 2. 「采集器日志」在主机模式下无法从平台查看，排「起没起来/在报什么错」必须 SSH。 3. 实测发现：Loki 部署把 listenIP 填成 0.0.0.0 时，部署记录 endpoint 是 http://0.0.0.0:3100，采集器照抄去推 -> connect: connection refused， 而容器/systemd 都是 running、平台 status=active —— 日志静默丢失。 根因 1. 托管方式写死在代码里（promtail 二进制 + systemd 单元），没有能力探测。 2. 没有对外暴露采集器自身日志的接口。 3. 渲染 promtail 配置时直接用 deployment.Endpoint，没有校验它是不是「可连接的推送地址」； 另外容器用 bridge 网络，127.0.0.1 在容器里指向容器自身，同机 Loki 也会推失败。 修法 - services/collector_host_runtime.go（新增） * detectHostCapabilities：一次 SSH 往返探测 架构/init/systemd/docker可用/开机自启/SELinux/journald * chooseHostRunMode：docker 优先 -> systemd 兜底 -> 两者都没有则明确报错（不裸跑进程） * installHostDocker：grafana/promtail:3.4.0 容器，--restart=always --memory=256m --cpus=1 --log-opt max-size=10m,max-file=3，挂 /etc/promtail(:ro) /var/lib/promtail /var/log(:ro) (+journald 相关)，SELinux Enforcing 时 --security-opt label=disable； 先 docker run --rm -check-syntax 校验配置，再替换正式配置 * waitContainerRunning：看 Status + RestartCount，起来就崩立即报错并带回 docker logs * hostPushEndpoint：0.0.0.0/:: 换真实主机 IP；容器模式下 127.0.0.1/localhost 也换； 换不了明确报错 * CollectorLogs：主机 docker logs / journalctl，K8s DaemonSet Pod 日志、Sidecar fluent-bit 容器日志 - collector_deployer.go：installHost 探测->选模式->渲染配置->分发；systemd 路径从 docker 迁回时 清理同名容器；卸载时容器与 systemd 两种方式都清（RunMode 对老记录是空的） - 拉不到镜像（内网无外网）时降级 systemd 二进制托管，其它错误原样抛出 - models/collector.go：新增 runMode 并纳入 Sanitized；services.Update 白名单同步 - 接口 GET /api/v1/collectors/:id/logs?tail=N；前端卡片加「运行方式」+「日志」弹窗 - build.sh：detect_go 的 INFO 走 stderr（原来被打进 GO_CMD，构建必失败） 验证（172.19.16.8 / pre-cex-ops-jenkins，collector id=2） - 安装后 runMode=docker，容器 Up，RestartPolicy=always，systemd promtail 自动停用 - 推送地址自动纠正为 http://172.19.16.8:3100，Loki 近 10 分钟查到 4919 条 job=system/job=app、host=pre-cex-ops-jenkins - docker 垫片模拟「docker 不可用」-> 自动回退 runMode=systemd、unit active+enabled、 日志接口 source=systemd；移除垫片重装 -> 切回 docker - 卸载：容器 0 个、unit 文件不存在、/etc/promtail 不存在、runMode 清空；重装恢复 running/0 重启 - K8s 采集器日志接口 source=kubernetes（顺带暴露 429 限流，见后续日志量治理） - go vet 干净；go test ./... 全绿（新增 chooseHostRunMode / hostPushEndpoint 等 6 组用例）； tsc -b 通过 - 镜像 lghost/lokistack:latest = :v0.1 = sha256:906a6d5369d7，Pod imageID 一致 --- backend/controllers/collector.go | 25 ++ backend/models/collector.go | 21 +- backend/routers/router.go | 1 + backend/services/collector.go | 32 +- backend/services/collector_deployer.go | 254 ++++++++--- backend/services/collector_host_runtime.go | 409 ++++++++++++++++++ .../services/collector_host_runtime_test.go | 95 ++++ backend/services/host_push_endpoint_test.go | 112 +++++ build.sh | 4 +- frontend/src/api/collector/index.ts | 13 + frontend/src/i18n/locales/en/collectors.ts | 18 + frontend/src/i18n/locales/zh/collectors.ts | 19 + .../collectors/components/CollectorCard.tsx | 12 +- .../components/CollectorLogsDialog.tsx | 141 ++++++ frontend/src/pages/collectors/index.tsx | 6 + frontend/src/types/index.ts | 2 + 16 files changed, 1095 insertions(+), 69 deletions(-) create mode 100644 backend/services/collector_host_runtime.go create mode 100644 backend/services/collector_host_runtime_test.go create mode 100644 backend/services/host_push_endpoint_test.go create mode 100644 frontend/src/pages/collectors/components/CollectorLogsDialog.tsx diff --git a/backend/controllers/collector.go b/backend/controllers/collector.go index 7113f2d..ce60047 100644 --- a/backend/controllers/collector.go +++ b/backend/controllers/collector.go @@ -1,6 +1,8 @@ package controllers import ( + "strconv" + "lokistack/dto" "lokistack/models" "lokistack/services" @@ -131,3 +133,26 @@ func (ctrl *CollectorController) Uninstall(c *gin.Context) { go ctrl.service.Uninstall(uint(id)) utils.Success(c, gin.H{"message": "已开始卸载采集器"}) } + +// Logs 返回采集器最近的运行日志（主机：docker logs / journalctl；K8s：Pod 日志）。 +// 主机模式下采集器日志原来只能登机器看 docker logs / journalctl，容器化后这是 +// 顺带拿到的能力，排「采集器起没起来、在报什么错」不用再 SSH。 +func (ctrl *CollectorController) Logs(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + tail := 200 + if v := c.Query("tail"); v != "" { + if n, convErr := strconv.Atoi(v); convErr == nil && n > 0 { + tail = n + } + } + out, err := ctrl.service.CollectorLogs(uint(id), tail) + if err != nil { + utils.Error(c, 400, err.Error()) + return + } + utils.Success(c, out) +} diff --git a/backend/models/collector.go b/backend/models/collector.go index a807bb3..ae4d85e 100644 --- a/backend/models/collector.go +++ b/backend/models/collector.go @@ -37,14 +37,18 @@ type Collector struct { // 存下来才能让「配置预览」和线上真实配置一致。 // SystemLogPaths 安装时探测到的系统日志绝对路径（Ubuntu/Debian 与 RHEL 系不同）， // 存下来让「预览配置」和实际下发保持一致。 - SystemLogPaths []string `json:"systemLogPaths,omitempty"` - ResolvedHostname string `json:"resolvedHostname,omitempty"` - SyslogFormat string `json:"syslogFormat,omitempty"` // iso8601 | syslog - Status string `json:"status"` // pending | active | error - StatusMessage string `json:"statusMessage"` - Config string `json:"config"` // 生成的配置快照 - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + SystemLogPaths []string `json:"systemLogPaths,omitempty"` + ResolvedHostname string `json:"resolvedHostname,omitempty"` + SyslogFormat string `json:"syslogFormat,omitempty"` // iso8601 | syslog + // RunMode 主机采集器的托管方式：docker | systemd。 + // 安装时由目标机能力探测决定（docker 优先、systemd 兜底），必须落库： + // 卸载/升级要知道该删容器还是删 systemd 单元，否则会留孤儿进程继续采日志。 + RunMode string `json:"runMode,omitempty"` + Status string `json:"status"` // pending | active | error + StatusMessage string `json:"statusMessage"` + Config string `json:"config"` // 生成的配置快照 + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } func (c *Collector) Sanitized() map[string]interface{} { @@ -89,6 +93,7 @@ func (c *Collector) Sanitized() map[string]interface{} { "systemLogPaths": c.SystemLogPaths, "resolvedHostname": c.ResolvedHostname, "syslogFormat": c.SyslogFormat, + "runMode": c.RunMode, "status": c.Status, "statusMessage": c.StatusMessage, "createdAt": c.CreatedAt, diff --git a/backend/routers/router.go b/backend/routers/router.go index 80e6c3b..0c7168a 100644 --- a/backend/routers/router.go +++ b/backend/routers/router.go @@ -197,6 +197,7 @@ func SetupRouter() *gin.Engine { authorized.POST("/collectors/:id/install", collectorCtrl.Install) authorized.POST("/collectors/:id/uninstall", collectorCtrl.Uninstall) authorized.POST("/collectors/:id/preview", collectorCtrl.Preview) + authorized.GET("/collectors/:id/logs", collectorCtrl.Logs) // Static files - use executable directory as base distPath := findDistPath() diff --git a/backend/services/collector.go b/backend/services/collector.go index e0da484..a1169c2 100644 --- a/backend/services/collector.go +++ b/backend/services/collector.go @@ -107,8 +107,17 @@ func (s *CollectorService) renderConfig(c *models.Collector) (string, error) { if hostname == "" && c.Host != nil { hostname = c.Host.HostIP } + // 预览要与线上一致：这里也走一遍推送地址纠正（0.0.0.0 / 容器内回环） + pushEndpoint := endpoint + hostIP := "" + if c.Host != nil { + hostIP = c.Host.HostIP + } + if v, perr := hostPushEndpoint(endpoint, hostIP, c.RunMode == HostRunModeDocker); perr == nil { + pushEndpoint = v + } return PromtailConfigForHost(HostPromtailOptions{ - Endpoint: endpoint, + Endpoint: pushEndpoint, Cluster: "host", LabelsJSON: c.Labels, Hostname: hostname, @@ -184,6 +193,19 @@ func (s *CollectorService) PreviewConfig(id uint, updates map[string]interface{} if c.ResolvedHostname == "" { warnings = append(warnings, "尚未成功安装过：host 标签将使用平台登记的主机名，实际安装时会替换为目标机真实 hostname") } + // 推送地址填写了通配/回环地址时提前提示：这类地址采集器推不上去， + // 表现是「容器/systemd 都 running，但 Loki 里一条日志都没有」。 + rawEndpoint := "" + if c.LokiDeployment != nil { + rawEndpoint = c.LokiDeployment.Endpoint + } + hostIP := "" + if c.Host != nil { + hostIP = c.Host.HostIP + } + if fixed, perr := hostPushEndpoint(rawEndpoint, hostIP, c.RunMode == HostRunModeDocker); perr == nil && fixed != rawEndpoint { + warnings = append(warnings, fmt.Sprintf("Loki 推送地址 %s 不可用，部署时会自动改为 %s（建议修正 Loki 部署的监听地址）", rawEndpoint, fixed)) + } } return map[string]interface{}{"config": cfg, "warnings": warnings, "hostname": c.ResolvedHostname, "syslogFormat": c.SyslogFormat}, nil } @@ -412,6 +434,10 @@ func (s *CollectorService) Update(id uint, updates map[string]interface{}) error if v, ok := updates["systemLogPaths"].([]string); ok && len(v) > 0 { db.Collectors[i].SystemLogPaths = v } + // runMode：安装时探测出的托管方式（docker | systemd），卸载后清空。 + if v, ok := updates["runMode"].(string); ok { + db.Collectors[i].RunMode = v + } if v, ok := updates["config"].(string); ok && v != "" { db.Collectors[i].Config = v } @@ -478,6 +504,8 @@ func (s *CollectorService) Install(id uint) { if len(collector.SystemLogPaths) > 0 { updates["systemLogPaths"] = collector.SystemLogPaths } + // 托管方式（docker | systemd）由安装流程探测后写回，前端据此展示运行方式 + updates["runMode"] = collector.RunMode s.Update(id, updates) } @@ -493,5 +521,5 @@ func (s *CollectorService) Uninstall(id uint) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() NewCollectorDeployer(collector, collector.LokiDeployment, nil).Uninstall(ctx) - s.Update(id, map[string]interface{}{"status": "pending", "statusMessage": "采集器已卸载"}) + s.Update(id, map[string]interface{}{"status": "pending", "statusMessage": "采集器已卸载", "runMode": ""}) } diff --git a/backend/services/collector_deployer.go b/backend/services/collector_deployer.go index 3350c59..32b4bb5 100644 --- a/backend/services/collector_deployer.go +++ b/backend/services/collector_deployer.go @@ -3,6 +3,7 @@ package services import ( "context" "encoding/json" + "errors" "fmt" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "strings" @@ -69,8 +70,19 @@ func (d *CollectorDeployer) Uninstall(ctx context.Context) error { return nil } -// ===== host-promtail：SSH 安装 Promtail 二进制 + systemd ===== +// ===== host-promtail：托管方式自动探测（docker 优先，systemd 兜底） ===== +// installHost 主机采集器安装入口。 +// +// 「进程活着」这件事不该由平台自己扛：平台的职责是「期望状态 + 配置下发 + 健康校验」， +// 存活/开机自启交给成熟的监督者。所以按目标机能力自动选托管方式： +// +// ① 有可用 Docker（docker info 成功）→ 容器（--restart=always，跨发行版抹平差异） +// ② 无 Docker 但有 systemd → 二进制 + systemd 单元 +// ③ 两者都没有 → 明确报错，绝不裸跑进程 +// +// 第 ③ 条是硬约束：`nohup promtail &` 起的进程主机重启即消失，平台不主动巡检就会 +// 静默断采；要补齐就得自己实现存活检测/PID 管理/开机自启，等于重造一个更差的 supervisor。 func (d *CollectorDeployer) installHost(ctx context.Context, endpoint string) error { host := d.collector.Host if host == nil { @@ -82,57 +94,74 @@ func (d *CollectorDeployer) installHost(ctx context.Context, endpoint string) er } defer client.Close() - // 检测架构 - archOut, _, _ := client.Execute(ctx, "uname -m") - arch := strings.TrimSpace(archOut) - switch arch { - case "x86_64": - arch = "amd64" - case "aarch64", "arm64": - arch = "arm64" - default: - return fmt.Errorf("不支持的主机架构: %s", arch) + caps := detectHostCapabilities(ctx, client) + d.log("info", fmt.Sprintf("目标机探测：架构 %s / init %s / docker %s / SELinux %s", + caps.Arch, caps.InitName, caps.DockerSummary(), caps.SELinux), "") + + if err := ensureHostPromtailDirs(ctx, client); err != nil { + return err } - // 确定 Promtail 版本（优先使用关联的 Loki 版本） - version := "3.0.0" - if d.loki != nil && d.loki.Version != "" { - version = d.loki.Version + // 先定托管方式，再渲染配置：容器模式下回环地址必须换掉（bridge 里 127.0.0.1 指容器自己） + mode, err := chooseHostRunMode(caps) + if err != nil { + return err } - // 创建必要目录 - d.log("info", "准备目录 /opt/promtail /etc/promtail /var/lib/promtail", "") - if _, _, err := client.Execute(ctx, "mkdir -p /opt/promtail /etc/promtail /var/lib/promtail"); err != nil { - return fmt.Errorf("创建 Promtail 目录失败: %w", err) + // 配置生成与上传（写 .new，两种托管方式共用；校验通过才替换正式配置） + conf, err := d.prepareHostConfig(ctx, client, host, endpoint, mode == HostRunModeDocker) + if err != nil { + return err } - - // 下载 Promtail - d.log("info", fmt.Sprintf("下载 Promtail %s（%s）", version, arch), "") - url := fmt.Sprintf("https://github.com/grafana/loki/releases/download/v%s/promtail-linux-%s.zip", version, arch) - dl := fmt.Sprintf("cd /tmp && curl -fsSL -o p.zip '%s' && unzip -o p.zip -d /opt/promtail/ && chmod +x /opt/promtail/promtail-linux-%s", url, arch) - if _, _, err := client.Execute(ctx, dl); err != nil { - return fmt.Errorf("下载 Promtail 失败: %w", err) + if mode == HostRunModeDocker { + err := d.installHostDocker(ctx, client, conf, caps) + if err == nil { + return nil + } + // 目标机拉不到镜像（内网/无外网）时，systemd 二进制托管是可行的退路； + // 其它错误（配置非法、容器崩溃）必须原样抛出，不能被回退掩盖。 + if errors.Is(err, errImagePullFailed) && caps.HasSystemd { + d.log("warn", fmt.Sprintf("容器方式不可用，回退为 systemd 二进制托管：%v", err), "") + return d.installHostSystemd(ctx, client, conf, caps) + } + return err } - client.Execute(ctx, "ln -sf /opt/promtail/promtail-linux-* /usr/local/bin/promtail 2>/dev/null || true") + d.log("warn", "目标机没有可用的 Docker，回退为 systemd 二进制托管（跨发行版差异更大，建议先装 Docker）", "") + return d.installHostSystemd(ctx, client, conf, caps) +} - // 生成并上传配置 +// prepareHostConfig 探测目标机身份并渲染/上传 promtail.yaml.new。 +// +// 三件事必须在安装期探测，不能写死： +// - 真实主机名（promtail 不展开 ${HOSTNAME}，写死会在 Loki 里留下一堆字面量标签） +// - 系统日志时间戳格式（TencentOS 是 ISO8601，Debian 是传统 syslog） +// - 系统日志路径（Ubuntu 只有 syslog/auth.log，没有 messages） +func (d *CollectorDeployer) prepareHostConfig(ctx context.Context, client *SSHClient, host *models.Host, endpoint string, dockerMode bool) (string, error) { logPath := d.collector.LogPath if logPath == "" { logPath = "/var/log/*.log" } - // 目标机真实身份与 syslog 格式只能在安装时探测（见 Bug3 / Bug4）： - // - host 标签写真实主机名，promtail 不做 ${HOSTNAME} 展开； - // - 系统日志按实际格式（ISO8601 / 传统 syslog）选正则。 + // 推送地址纠正：Loki 的 listenIP 填成 0.0.0.0 时 endpoint 是必然连不上的 + // http://0.0.0.0:3100，采集器会 running 但一条日志都推不上去。 + hostIP := "" + if host != nil { + hostIP = host.HostIP + } + pushEndpoint, err := hostPushEndpoint(endpoint, hostIP, dockerMode) + if err != nil { + return "", err + } + if pushEndpoint != endpoint { + d.log("warn", fmt.Sprintf("Loki 推送地址不可用（%s），已改为 %s —— 建议把 Loki 部署的监听地址改成实际 IP/域名", endpoint, pushEndpoint), "") + } hostname := detectHostname(ctx, client, host) syslogFormat := detectSyslogFormat(ctx, client) - d.log("info", fmt.Sprintf("目标机 %s，syslog 格式 %s", hostname, syslogFormat), "") - - // 系统日志路径同样必须探测：Ubuntu 只有 syslog/auth.log，没有 messages sysLogPaths := detectSystemLogPaths(ctx, client) - d.log("info", fmt.Sprintf("系统日志路径：%s", strings.Join(sysLogPaths, ",")), "") + d.log("info", fmt.Sprintf("目标机 %s，syslog 格式 %s，系统日志 %s", + hostname, syslogFormat, strings.Join(sysLogPaths, ",")), "") conf, err := PromtailConfigForHost(HostPromtailOptions{ - Endpoint: endpoint, + Endpoint: pushEndpoint, Cluster: "host", LabelsJSON: d.collector.Labels, Hostname: hostname, @@ -142,29 +171,127 @@ func (d *CollectorDeployer) installHost(ctx context.Context, endpoint string) er Jobs: d.collector.ScrapeJobs, }) if err != nil { - return fmt.Errorf("生成 Promtail 配置失败（现有配置未改动）: %w", err) + return "", fmt.Errorf("生成 Promtail 配置失败（现有配置未改动）: %w", err) } d.collector.ResolvedHostname = hostname d.collector.SyslogFormat = syslogFormat d.collector.SystemLogPaths = sysLogPaths - // 先写临时文件并做语法校验，通过后才替换正式配置 —— - // 配置坏了会把正在跑的采集干掉，不能直接覆盖。 - d.log("info", "上传 Promtail 配置并校验", "") - if _, _, err := client.Execute(ctx, fmt.Sprintf("cat > /etc/promtail/promtail.yaml.new << 'PROMTAIL_EOF'\n%s\nPROMTAIL_EOF", conf)); err != nil { - return fmt.Errorf("上传配置失败: %w", err) + // 先写 .new：配置坏了会把正在跑的采集干掉，不能直接覆盖正式配置 + d.log("info", "上传 Promtail 配置", "") + if err := uploadDockerFile(ctx, client, promtailConfigPath+".new", conf, "644"); err != nil { + return "", fmt.Errorf("上传配置失败: %w", err) } - if out, _, err := client.Execute(ctx, "/usr/local/bin/promtail -check-syntax -config.file=/etc/promtail/promtail.yaml.new 2>&1"); err != nil { - // 老版本 promtail 可能没有 -check-syntax：只有明确报配置错误才阻断 + return conf, nil +} + +// installHostDocker 以容器方式托管采集器（首选路径）。 +func (d *CollectorDeployer) installHostDocker(ctx context.Context, client *SSHClient, conf string, caps *HostCapabilities) error { + name := promtailContainerName(d.collector.ID) + image := ResolveImageFor("promtail", "grafana/promtail", DefaultPromtailVersion) + + d.log("info", "拉取采集器镜像 "+image, "") + if _, stderr, err := client.Execute(ctx, "docker pull "+image); err != nil { + return fmt.Errorf("%w（目标机需能访问镜像仓库，可先在平台侧配置镜像加速）: %v: %s", + errImagePullFailed, err, truncateForMessage(strings.TrimSpace(stderr), 300)) + } + + // 用容器自己校验配置：3.x 的 -check-syntax 能挡住路径/正则/缩进错误 + d.log("info", "校验 Promtail 配置", "") + checkCmd := fmt.Sprintf("docker run --rm -v %s:%s:ro %s -check-syntax -config.file=%s.new 2>&1", + promtailConfigDir, promtailConfigDir, image, promtailConfigPath) + out, _, err := client.Execute(ctx, checkCmd) + if err != nil { msg := strings.TrimSpace(out) + if strings.Contains(msg, "flag provided but not defined") { + d.log("info", "当前 Promtail 版本不支持 -check-syntax，跳过语法校验", "") + } else { + return fmt.Errorf("Promtail 配置校验未通过（现有采集未改动）：%s", truncateForMessage(msg, 600)) + } + } + if err := promoteHostConfig(ctx, client); err != nil { + return err + } + + // 从 systemd 迁到容器：不先停旧服务就会两边同时采同一批文件（日志重复一倍） + if out, _, err := client.Execute(ctx, "systemctl is-active "+promtailServiceName+" 2>/dev/null"); err == nil && strings.TrimSpace(out) == "active" { + client.Execute(ctx, "systemctl stop "+promtailServiceName+" 2>/dev/null; systemctl disable "+promtailServiceName+" 2>/dev/null") + d.log("info", "已停止旧的 systemd promtail 服务（改为容器托管，避免重复采集）", "") + } + + client.Execute(ctx, "docker rm -f "+name+" >/dev/null 2>&1 || true") + + mounts := []string{ + "-v " + promtailConfigDir + ":" + promtailConfigDir + ":ro", + "-v " + promtailPositionsDir + ":" + promtailPositionsDir, + "-v /var/log:/var/log:ro", + } + if caps.HasJournald { + // 采 systemd journal 时需要（Ubuntu 最小安装无 rsyslog 的场景） + mounts = append(mounts, + "-v /run/log/journal:/run/log/journal:ro", + "-v /etc/machine-id:/etc/machine-id:ro") + } + // SELinux Enforcing 时 bind mount 会被拒绝（容器里读不到 /var/log）； + // 采集器容器不承载多租户隔离，直接关标签检查，比给宿主机目录重新打标签安全。 + secOpt := "" + if strings.EqualFold(caps.SELinux, "Enforcing") { + secOpt = " --security-opt label=disable" + } + runCmd := fmt.Sprintf( + "docker run -d --name %s --restart=always --memory=256m --cpus=1 "+ + "--log-opt max-size=10m --log-opt max-file=3%s %s %s -config.file=%s", + name, secOpt, strings.Join(mounts, " "), image, promtailConfigPath) + d.log("info", "启动采集器容器 "+name, "") + if _, stderr, err := client.Execute(ctx, runCmd); err != nil { + return fmt.Errorf("启动采集器容器失败: %v: %s", err, truncateForMessage(strings.TrimSpace(stderr), 400)) + } + + // docker run 返回 0 也可能是「起来就崩」——必须查容器状态与重启次数 + if err := waitContainerRunning(ctx, client, name, 30*time.Second); err != nil { + logs, _, _ := client.Execute(ctx, fmt.Sprintf("docker logs --tail 30 %s 2>&1", name)) + return fmt.Errorf("采集器容器未正常运行：%v\n最近日志：\n%s", err, truncateForMessage(strings.TrimSpace(logs), 900)) + } + + d.collector.RunMode = HostRunModeDocker + d.collector.Config = conf + d.log("success", fmt.Sprintf("Promtail 已以容器方式运行（%s，--restart=always）", name), "") + return nil +} + +// installHostSystemd 无 Docker 环境下的兜底路径：二进制 + systemd 单元。 +func (d *CollectorDeployer) installHostSystemd(ctx context.Context, client *SSHClient, conf string, caps *HostCapabilities) error { + arch, err := normalizeArch(caps.Arch) + if err != nil { + return err + } + + // 确定 Promtail 版本（优先使用关联的 Loki 版本） + version := "3.0.0" + if d.loki != nil && d.loki.Version != "" { + version = d.loki.Version + } + d.log("info", fmt.Sprintf("下载 Promtail %s（%s）", version, arch), "") + url := fmt.Sprintf("https://github.com/grafana/loki/releases/download/v%s/promtail-linux-%s.zip", version, arch) + dl := fmt.Sprintf("cd /tmp && curl -fsSL -o p.zip '%s' && unzip -o p.zip -d /opt/promtail/ && chmod +x /opt/promtail/promtail-linux-%s", url, arch) + if _, _, err := client.Execute(ctx, dl); err != nil { + return fmt.Errorf("下载 Promtail 失败: %w", err) + } + client.Execute(ctx, "ln -sf /opt/promtail/promtail-linux-* /usr/local/bin/promtail 2>/dev/null || true") + + // 语法校验通过才替换正式配置 + d.log("info", "校验 Promtail 配置", "") + if out, _, err := client.Execute(ctx, "/usr/local/bin/promtail -check-syntax -config.file="+promtailConfigPath+".new 2>&1"); err != nil { + msg := strings.TrimSpace(out) + // 老版本 promtail 可能没有 -check-syntax：只有明确报配置错误才阻断 if strings.Contains(msg, "flag provided but not defined") { d.log("info", "当前 promtail 不支持 -check-syntax，跳过语法校验", "") } else { return fmt.Errorf("Promtail 配置校验未通过（现有配置未改动）：%s", truncateForMessage(msg, 600)) } } - if _, _, err := client.Execute(ctx, "mv -f /etc/promtail/promtail.yaml.new /etc/promtail/promtail.yaml"); err != nil { - return fmt.Errorf("替换 Promtail 配置失败: %w", err) + if err := promoteHostConfig(ctx, client); err != nil { + return err } // systemd 服务（生产级） @@ -177,7 +304,7 @@ After=network-online.target Type=simple User=root Group=root -ExecStart=/usr/local/bin/promtail -config.file=/etc/promtail/promtail.yaml +ExecStart=/usr/local/bin/promtail -config.file=` + promtailConfigPath + ` ExecReload=/bin/kill -HUP $MAINPID Restart=always RestartSec=10 @@ -189,18 +316,18 @@ StandardError=append:/var/log/promtail.log WantedBy=multi-user.target ` d.log("info", "安装 systemd 服务", "") - if _, _, err := client.Execute(ctx, fmt.Sprintf("cat > /etc/systemd/system/promtail.service << 'EOF'\n%s\nEOF", service)); err != nil { + if err := uploadDockerFile(ctx, client, "/etc/systemd/system/"+promtailServiceName+".service", service, "644"); err != nil { return fmt.Errorf("创建 systemd 服务失败: %w", err) } - if _, _, err := client.Execute(ctx, "systemctl daemon-reload && systemctl enable promtail && systemctl restart promtail"); err != nil { + if _, _, err := client.Execute(ctx, "systemctl daemon-reload && systemctl enable "+promtailServiceName+" && systemctl restart "+promtailServiceName); err != nil { return fmt.Errorf("启动 Promtail 失败: %w", err) } // 必须校验服务真的起来了： // 原来只判断 systemctl 命令有没有执行错误，结果 Promtail 因配置/参数错误 - // 崩溃重启 40+ 次，平台仍然显示"采集器已部署"，用户以为采集是好的。 - if err := waitServiceActive(ctx, client, "promtail", 20*time.Second); err != nil { - out, _, _ := client.Execute(ctx, "journalctl -u promtail -n 15 --no-pager 2>/dev/null | tail -15") + // 崩溃重启 40+ 次，平台仍然显示「采集器已部署」，用户以为采集是好的。 + if err := waitServiceActive(ctx, client, promtailServiceName, 20*time.Second); err != nil { + out, _, _ := client.Execute(ctx, "journalctl -u "+promtailServiceName+" -n 15 --no-pager 2>/dev/null | tail -15") tail := strings.TrimSpace(out) if tail == "" { out, _, _ = client.Execute(ctx, "tail -10 /var/log/promtail.log 2>/dev/null") @@ -209,9 +336,13 @@ WantedBy=multi-user.target return fmt.Errorf("Promtail 启动后未处于运行状态：%v\n最近日志：\n%s", err, truncateForMessage(tail, 900)) } + // 反向清理：同名容器若存在（从 docker 模式切回来），停掉避免重复采集 + client.Execute(ctx, "docker rm -f "+promtailContainerName(d.collector.ID)+" >/dev/null 2>&1 || true") + // 保存配置快照 + d.collector.RunMode = HostRunModeSystemd d.collector.Config = conf - d.log("success", "Promtail 部署完成", "") + d.log("success", "Promtail 已以 systemd 服务方式运行", "") return nil } @@ -291,10 +422,19 @@ func (d *CollectorDeployer) uninstallHost(ctx context.Context) error { return err } defer client.Close() - client.Execute(ctx, "systemctl stop promtail 2>/dev/null || true") - client.Execute(ctx, "systemctl disable promtail 2>/dev/null || true") - client.Execute(ctx, "rm -rf /etc/systemd/system/promtail.service /usr/local/bin/promtail /etc/promtail /opt/promtail /var/lib/promtail") - client.Execute(ctx, "systemctl daemon-reload") + + name := promtailContainerName(d.collector.ID) + // 两种托管方式都清一遍：RunMode 对老记录是空的，且用户可能手工装过另一种。 + // 顺序固定：先停容器/服务，再删文件 —— 否则进程还在跑就没了配置。 + if out, _, err := client.Execute(ctx, "docker rm -f "+name+" 2>&1"); err == nil { + d.log("info", "已删除采集器容器 "+strings.TrimSpace(out), "") + } + client.Execute(ctx, "systemctl stop "+promtailServiceName+" 2>/dev/null; systemctl disable "+promtailServiceName+" 2>/dev/null") + client.Execute(ctx, "rm -f /etc/systemd/system/"+promtailServiceName+".service") + client.Execute(ctx, "systemctl daemon-reload 2>/dev/null || true") + client.Execute(ctx, "rm -rf "+promtailConfigDir+" /opt/promtail "+promtailPositionsDir+" /usr/local/bin/promtail") + d.log("info", "已清理采集器配置、positions 与二进制", "") + d.collector.RunMode = "" return nil } diff --git a/backend/services/collector_host_runtime.go b/backend/services/collector_host_runtime.go new file mode 100644 index 0000000..7805b76 --- /dev/null +++ b/backend/services/collector_host_runtime.go @@ -0,0 +1,409 @@ +package services + +import ( + "context" + "errors" + "fmt" + "net" + "net/url" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + + "lokistack/models" +) + +// 主机采集器的两种托管方式。 +// +// 为什么要落库：安装/升级/卸载必须走**同一条**路径。不记录的话，卸载时不知道该 +// 删容器还是删 systemd 单元 → 留下孤儿进程继续采日志（重复数据 + 无法收敛）。 +const ( + HostRunModeDocker = "docker" + HostRunModeSystemd = "systemd" +) + +// DefaultPromtailVersion 采集器容器镜像版本（与 DaemonSet 模板保持一致）。 +const DefaultPromtailVersion = "3.4.0" + +const ( + promtailServiceName = "promtail" + promtailConfigDir = "/etc/promtail" + promtailConfigPath = promtailConfigDir + "/promtail.yaml" + promtailPositionsDir = "/var/lib/promtail" +) + +func promtailContainerName(id uint) string { return fmt.Sprintf("lokistack-promtail-%d", id) } + +// hostPushEndpoint 把「Loki 监听地址」纠正成「采集器真正能连的推送地址」。 +// +// 两种必须纠正的情况（本会话实测踩到的）： +// +// ① 用户在主机部署 Loki 时把 listenIP 填成 0.0.0.0 / :: +// → 部署记录里的 endpoint 变成 http://0.0.0.0:3100，采集器照抄过去推 +// → connect: connection refused，日志静默丢失（容器/serve 都是 running，最难查） +// ② 容器模式下的回环地址 +// → 容器用 bridge 网络，127.0.0.1 指的是容器自己，不是宿主机 +// （systemd 模式不受影响，所以只在 docker 模式下改） +// +// 纠正不了（没有主机 IP 可替换）时明确报错，不返回一个必然失败的地址。 +func hostPushEndpoint(endpoint, hostIP string, dockerMode bool) (string, error) { + raw := strings.TrimSpace(endpoint) + if raw == "" { + return "", fmt.Errorf("缺少 Loki 推送地址（endpoint）") + } + u, err := url.Parse(raw) + if err != nil || u.Host == "" { + return raw, nil // 非标准形态（如 k8s svc 短名）不做处理 + } + host := u.Hostname() + wildcard := host == "0.0.0.0" || host == "::" || host == "[::]" + loopback := host == "localhost" || host == "127.0.0.1" || host == "::1" + if !wildcard && !(dockerMode && loopback) { + return raw, nil + } + ip := strings.TrimSpace(hostIP) + if ip == "" || net.ParseIP(ip) == nil { + return "", fmt.Errorf("Loki 推送地址是 %s（%s），但取不到目标主机 IP 用于替换："+ + "请把 Loki 部署的监听地址改成实际可访问的 IP 或域名", raw, map[bool]string{true: "通配地址无法作为推送目标", false: "容器内 127.0.0.1 指向容器自身"}[wildcard]) + } + port := u.Port() + if port == "" { + port = "3100" + } + u.Host = net.JoinHostPort(ip, port) + return u.String(), nil +} + +// errImagePullFailed 拉不到采集器镜像（目标机访问不了镜像仓库）。 +// 单独成哨兵错误：这种情况可以退回 systemd 二进制托管，而配置错误、容器起不来 +// 这类问题不该回退（回退只会把真实错误藏起来）。 +var errImagePullFailed = errors.New("拉取采集器镜像失败") + +// HostCapabilities 是目标主机的托管能力探测结果（一次 SSH 往返拿全）。 +type HostCapabilities struct { + Arch string // x86_64 / aarch64 … + InitName string // systemd / init / … + HasSystemd bool // /run/systemd/system 存在（systemd 正在跑，不是只装了命令） + HasDocker bool // 有 docker 命令 + DockerUsable bool // docker info 成功（daemon 在跑 + 当前用户有权限） + DockerVersion string + DockerEnabled string // enabled / disabled / unknown + SELinux string // Enforcing / Permissive / Disabled / unknown + HasJournald bool // 存在持久化或运行时 journal 目录 +} + +// DockerSummary 供操作日志展示（别在日志里输出空字符串）。 +func (c *HostCapabilities) DockerSummary() string { + if !c.HasDocker { + return "未安装" + } + if !c.DockerUsable { + if c.DockerVersion != "" { + return "已安装但不可用（daemon 未运行或无权限）" + } + return "已安装但不可用" + } + desc := c.DockerVersion + if desc == "" { + desc = "可用" + } + return desc + "（开机自启 " + c.DockerEnabled + "）" +} + +// detectHostCapabilities 探测目标机能否用 docker / systemd 托管长驻进程。 +// +// 一次合并命令的原因：每次 SSH 往返都有 100~300ms 开销，装机流程本来就要跑十几步， +// 逐条探测会让「点部署后等很久」的体感更差。 +func detectHostCapabilities(ctx context.Context, client *SSHClient) *HostCapabilities { + caps := &HostCapabilities{InitName: "unknown", SELinux: "unknown", DockerEnabled: "unknown"} + script := `echo "ARCH=$(uname -m 2>/dev/null)" +echo "INIT=$(ps -p 1 -o comm= 2>/dev/null | tr -d ' ')" +[ -d /run/systemd/system ] && echo "SYSTEMD=1" || echo "SYSTEMD=0" +if command -v docker >/dev/null 2>&1; then + echo "DOCKER=1" + echo "DOCKER_VERSION=$(docker version --format '{{.Server.Version}}' 2>/dev/null)" + docker info >/dev/null 2>&1 && echo "DOCKER_USABLE=1" || echo "DOCKER_USABLE=0" + echo "DOCKER_ENABLED=$(systemctl is-enabled docker 2>/dev/null || echo unknown)" +else + echo "DOCKER=0" + echo "DOCKER_USABLE=0" +fi +echo "SELINUX=$(getenforce 2>/dev/null || echo unknown)" +if [ -d /run/log/journal ] || [ -d /var/log/journal ]; then echo "JOURNALD=1"; else echo "JOURNALD=0"; fi` + out, _, err := client.Execute(ctx, script) + if err != nil && strings.TrimSpace(out) == "" { + return caps + } + for _, line := range strings.Split(out, "\n") { + line = strings.TrimSpace(line) + key, value, ok := strings.Cut(line, "=") + if !ok { + continue + } + switch key { + case "ARCH": + caps.Arch = value + case "INIT": + if value != "" { + caps.InitName = value + } + case "SYSTEMD": + caps.HasSystemd = value == "1" + case "DOCKER": + caps.HasDocker = value == "1" + case "DOCKER_VERSION": + caps.DockerVersion = value + case "DOCKER_USABLE": + caps.DockerUsable = value == "1" + case "DOCKER_ENABLED": + if value != "" { + caps.DockerEnabled = value + } + case "SELINUX": + if value != "" { + caps.SELinux = value + } + case "JOURNALD": + caps.HasJournald = value == "1" + } + } + return caps +} + +// normalizeArch 把 uname -m 转成发行包/镜像使用的架构名。 +func normalizeArch(arch string) (string, error) { + switch strings.TrimSpace(arch) { + case "x86_64", "amd64": + return "amd64", nil + case "aarch64", "arm64": + return "arm64", nil + } + return "", fmt.Errorf("不支持的主机架构: %s", arch) +} + +// chooseHostRunMode 决定托管方式：docker 优先 → systemd 兜底 → 都没有则明确报错。 +// +// 抽成纯函数是为了能测：这条判断决定「卸载时要删容器还是删 unit」， +// 判错的后果是孤儿进程继续采日志（重复数据）或卸载残留，靠人工回归太贵。 +func chooseHostRunMode(caps *HostCapabilities) (string, error) { + if caps == nil { + return "", fmt.Errorf("缺少目标机能力探测结果") + } + if caps.DockerUsable { + return HostRunModeDocker, nil + } + if caps.HasSystemd { + return HostRunModeSystemd, nil + } + return "", fmt.Errorf("目标机既没有可用的 Docker（docker info 失败）也没有 systemd，无法可靠托管采集进程：" + + "请先在目标机安装并启动 Docker 后重试。平台不会以裸进程方式运行 promtail（主机重启后采集会静默中断）") +} + +func ensureHostPromtailDirs(ctx context.Context, client *SSHClient) error { + if _, _, err := client.Execute(ctx, "mkdir -p "+promtailConfigDir+" "+promtailPositionsDir); err != nil { + return fmt.Errorf("创建 Promtail 目录失败: %w", err) + } + return nil +} + +// promoteHostConfig 把校验通过的 .new 覆盖为正式配置。 +func promoteHostConfig(ctx context.Context, client *SSHClient) error { + if _, _, err := client.Execute(ctx, "mv -f "+promtailConfigPath+".new "+promtailConfigPath); err != nil { + return fmt.Errorf("替换 Promtail 配置失败: %w", err) + } + return nil +} + +// waitContainerRunning 轮询容器真实状态。 +// +// `docker run -d` 返回 0 **不代表容器在跑**（镜像参数错会立刻退出）。 +// 这里同时看 Status 与 RestartCount：RestartCount 持续增长 = 崩溃循环， +// 必须立即报错并带回 docker logs —— 等价于 systemd 侧 waitServiceActive 的防护。 +func waitContainerRunning(ctx context.Context, client *SSHClient, name string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + last := "" + for time.Now().Before(deadline) { + out, _, _ := client.Execute(ctx, fmt.Sprintf("docker inspect -f '{{.State.Status}}|{{.RestartCount}}' %s 2>/dev/null", name)) + last = strings.TrimSpace(out) + status, restarts, ok := strings.Cut(last, "|") + if ok { + status = strings.TrimSpace(status) + restarts = strings.TrimSpace(restarts) + if status == "running" && restarts == "0" { + return nil + } + if restarts != "" && restarts != "0" { + return fmt.Errorf("容器 %s 处于重启循环（Status=%s, RestartCount=%s）", name, status, restarts) + } + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(2 * time.Second): + } + } + if last == "" { + last = "容器不存在" + } + return fmt.Errorf("容器 %s %s 内未进入 running 且无重启（当前 %s）", name, timeout, last) +} + +// ===== 采集器日志（Web 上直接看，不用登机器） ===== + +// CollectorLogs 返回采集器最近的运行日志。 +// +// 主机模式按托管方式自动选入口：容器 → `docker logs`；systemd → journalctl / 文件。 +// K8s 模式读 DaemonSet/采集 Pod 的容器日志。 +func (s *CollectorService) CollectorLogs(id uint, tail int) (map[string]interface{}, error) { + c, err := s.Get(id) + if err != nil { + return nil, err + } + if tail <= 0 { + tail = 200 + } + if tail > 2000 { + tail = 2000 + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + switch c.Type { + case "host-promtail": + return hostCollectorLogs(ctx, c, tail) + case "k8s-daemonset", "k8s-fluentbit-sidecar": + return k8sCollectorLogs(ctx, c, tail) + } + return nil, fmt.Errorf("不支持的采集器类型: %s", c.Type) +} + +func hostCollectorLogs(ctx context.Context, c *models.Collector, tail int) (map[string]interface{}, error) { + if c.Host == nil { + return nil, fmt.Errorf("该采集器未关联主机，无法读取日志") + } + client, err := NewSSHClientWithAuth(HostSSHAuth(c.Host)) + if err != nil { + return nil, err + } + defer client.Close() + + name := promtailContainerName(c.ID) + // ① 容器托管（首选）：docker logs 自带轮转，不需要额外清理 + if _, _, err := client.Execute(ctx, "docker container inspect "+name+" >/dev/null 2>&1"); err == nil { + out, stderr, _ := client.Execute(ctx, fmt.Sprintf("docker logs --tail %d --timestamps %s 2>&1", tail, name)) + return map[string]interface{}{ + "source": "docker", + "sourceLabel": "Docker 容器 " + name, + "content": strings.TrimSpace(out + stderr), + }, nil + } + + // ② systemd 托管（无 docker 的兜底路径，或历史安装） + if _, _, err := client.Execute(ctx, "[ -f /etc/systemd/system/"+promtailServiceName+".service ]"); err == nil { + out, _, _ := client.Execute(ctx, fmt.Sprintf("journalctl -u %s -n %d --no-pager 2>/dev/null", promtailServiceName, tail)) + content := strings.TrimSpace(out) + if content == "" { + out, _, _ = client.Execute(ctx, fmt.Sprintf("tail -n %d /var/log/promtail.log 2>/dev/null", tail)) + content = strings.TrimSpace(out) + } + return map[string]interface{}{ + "source": "systemd", + "sourceLabel": "systemd 服务 " + promtailServiceName, + "content": content, + }, nil + } + + return nil, fmt.Errorf("目标机 %s 上既没有采集器容器 %s，也没有 %s systemd 服务（可能尚未部署或已被手工清理）", + c.Host.HostIP, name, promtailServiceName) +} + +func k8sCollectorLogs(ctx context.Context, c *models.Collector, tail int) (map[string]interface{}, error) { + if c.Cluster == nil { + return nil, fmt.Errorf("该采集器未关联集群，无法读取日志") + } + ns := c.Namespace + if ns == "" && c.LokiDeployment != nil { + ns = c.LokiDeployment.Namespace + } + if ns == "" { + ns = "log" + } + client, err := clusterClient(c.Cluster, 30*time.Second) + if err != nil { + return nil, err + } + + // Sidecar 模式的日志在业务 Pod 的 fluent-bit 容器里，不是独立的 DaemonSet + if c.Type == "k8s-fluentbit-sidecar" { + return sidecarCollectorLogs(ctx, client, ns, c.TargetDeployment, tail) + } + + pods, err := client.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: "app=promtail"}) + if err != nil { + return nil, fmt.Errorf("读取采集器 Pod 失败（检查集群凭证是否具备 pods list/log 权限）: %w", err) + } + if len(pods.Items) == 0 { + return nil, fmt.Errorf("命名空间 %s 中没有找到 promtail Pod（采集器可能未部署成功）", ns) + } + return podContainerLogs(ctx, client, ns, runningPodName(pods.Items), "promtail", tail) +} + +// runningPodName 优先挑 Running 的 Pod：DaemonSet 里常有刚重建的 Pending 实例， +// 读它的日志只会得到一行「container is waiting to start」。 +func runningPodName(pods []corev1.Pod) string { + for _, p := range pods { + if p.Status.Phase == corev1.PodRunning { + return p.Name + } + } + if len(pods) > 0 { + return pods[0].Name + } + return "" +} + +func podContainerLogs(ctx context.Context, client kubernetes.Interface, ns, pod, container string, tail int) (map[string]interface{}, error) { + if pod == "" { + return nil, fmt.Errorf("命名空间 %s 中没有可读取日志的采集 Pod", ns) + } + lines := int64(tail) + raw, err := client.CoreV1().Pods(ns).GetLogs(pod, &corev1.PodLogOptions{TailLines: &lines, Container: container}).DoRaw(ctx) + if err != nil { + return nil, fmt.Errorf("读取 Pod %s/%s 的 %s 容器日志失败: %w", ns, pod, container, err) + } + return map[string]interface{}{ + "source": "kubernetes", + "sourceLabel": fmt.Sprintf("Pod %s/%s（容器 %s）", ns, pod, container), + "content": strings.TrimSpace(string(raw)), + }, nil +} + +// sidecarCollectorLogs 找业务 Deployment 的 Pod，读它里面 fluent-bit 容器的日志。 +func sidecarCollectorLogs(ctx context.Context, client kubernetes.Interface, ns, deployment string, tail int) (map[string]interface{}, error) { + if deployment == "" { + return nil, fmt.Errorf("该 Sidecar 采集器缺少目标 Deployment，无法定位日志") + } + pods, err := client.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, fmt.Errorf("读取命名空间 %s 的 Pod 失败（检查集群凭证权限）: %w", ns, err) + } + matched := make([]corev1.Pod, 0, len(pods.Items)) + for _, p := range pods.Items { + if !strings.HasPrefix(p.Name, deployment) { + continue + } + for _, ct := range p.Spec.Containers { + if ct.Name == "fluent-bit" { + matched = append(matched, p) + break + } + } + } + if len(matched) == 0 { + return nil, fmt.Errorf("命名空间 %s 中 %s 的 Pod 里没有 fluent-bit 容器（Sidecar 可能未注入成功）", ns, deployment) + } + return podContainerLogs(ctx, client, ns, runningPodName(matched), "fluent-bit", tail) +} diff --git a/backend/services/collector_host_runtime_test.go b/backend/services/collector_host_runtime_test.go new file mode 100644 index 0000000..631290f --- /dev/null +++ b/backend/services/collector_host_runtime_test.go @@ -0,0 +1,95 @@ +package services + +import ( + "strings" + "testing" +) + +// 托管方式选择是「卸载时删容器还是删 unit」的唯一依据，四条分支都要钉死。 +func TestChooseHostRunMode(t *testing.T) { + cases := []struct { + name string + caps *HostCapabilities + want string + wantErr string + }{ + { + name: "docker 可用 → 容器（即使同时有 systemd 也优先容器）", + caps: &HostCapabilities{HasDocker: true, DockerUsable: true, HasSystemd: true}, + want: HostRunModeDocker, + }, + { + name: "装了 docker 但 daemon 没起来 → 回退 systemd", + caps: &HostCapabilities{HasDocker: true, DockerUsable: false, HasSystemd: true}, + want: HostRunModeSystemd, + }, + { + name: "没有 docker 也没有 systemd → 明确报错，不裸跑", + caps: &HostCapabilities{HasDocker: false, DockerUsable: false, HasSystemd: false}, + wantErr: "既没有可用的 Docker", + }, + { + name: "探测结果缺失 → 报错而不是猜一个", + caps: nil, + wantErr: "缺少目标机能力探测结果", + }, + } + for _, c := range cases { + got, err := chooseHostRunMode(c.caps) + if c.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), c.wantErr) { + t.Fatalf("%s: 期望报错含 %q，实际 got=%q err=%v", c.name, c.wantErr, got, err) + } + continue + } + if err != nil { + t.Fatalf("%s: 不应报错，实际 %v", c.name, err) + } + if got != c.want { + t.Fatalf("%s: 期望 %q，实际 %q", c.name, c.want, got) + } + } +} + +func TestNormalizeArch(t *testing.T) { + for in, want := range map[string]string{ + "x86_64": "amd64", "amd64": "amd64", "aarch64": "arm64", "arm64": "arm64", + } { + got, err := normalizeArch(in) + if err != nil || got != want { + t.Fatalf("%s → %q (err=%v)，期望 %q", in, got, err, want) + } + } + if _, err := normalizeArch("mips64"); err == nil { + t.Fatal("未知架构应报错") + } +} + +// 容器名必须带上采集器 ID：同一台机器可能装多个主机采集器， +// 用固定名字会互相覆盖（第二个部署把第一个容器删掉）。 +func TestPromtailContainerNameIsPerCollector(t *testing.T) { + a, b := promtailContainerName(2), promtailContainerName(3) + if a == b { + t.Fatalf("不同采集器不能共用容器名: %s", a) + } + if !strings.HasPrefix(a, "lokistack-promtail-") { + t.Fatalf("容器名不符合约定: %s", a) + } +} + +// 探测摘要要能一眼区分「没装 docker」和「装了但起不来」——这两种处置完全不同。 +func TestDockerSummaryDistinguishesStates(t *testing.T) { + missing := (&HostCapabilities{}).DockerSummary() + broken := (&HostCapabilities{HasDocker: true}).DockerSummary() + ok := (&HostCapabilities{HasDocker: true, DockerUsable: true, DockerVersion: "27.5.1", DockerEnabled: "enabled"}).DockerSummary() + + if !strings.Contains(missing, "未安装") { + t.Fatalf("未安装 docker 的摘要不对: %s", missing) + } + if !strings.Contains(broken, "不可用") || strings.Contains(broken, "未安装") { + t.Fatalf("装了但不可用的摘要不对: %s", broken) + } + if !strings.Contains(ok, "27.5.1") || !strings.Contains(ok, "enabled") { + t.Fatalf("可用状态的摘要不对: %s", ok) + } +} diff --git a/backend/services/host_push_endpoint_test.go b/backend/services/host_push_endpoint_test.go new file mode 100644 index 0000000..00e1adf --- /dev/null +++ b/backend/services/host_push_endpoint_test.go @@ -0,0 +1,112 @@ +package services + +import ( + "strings" + "testing" +) + +// 推送地址纠正：这是「采集器 running 但一条日志都进不去」的根因防护。 +func TestHostPushEndpoint(t *testing.T) { + const ip = "172.19.16.8" + cases := []struct { + name string + endpoint string + hostIP string + dockerMode bool + want string + wantErr string + }{ + { + name: "正常 IP 不动", + endpoint: "http://172.19.16.8:3100", + hostIP: ip, + want: "http://172.19.16.8:3100", + }, + { + name: "k8s svc 地址不动（命名空间短名照原样）", + endpoint: "http://loki.log.svc.cluster.local:3100", + hostIP: ip, + want: "http://loki.log.svc.cluster.local:3100", + }, + { + // 本会话实测踩到：listenIP 填 0.0.0.0 → endpoint 变成 0.0.0.0:3100 → 推送被拒 + name: "通配地址 0.0.0.0 换成真实主机 IP（两种模式都要换）", + endpoint: "http://0.0.0.0:3100", + hostIP: ip, + want: "http://172.19.16.8:3100", + }, + { + name: "容器模式下 127.0.0.1 必须换成主机 IP（bridge 里指向容器自己）", + endpoint: "http://127.0.0.1:3100", + hostIP: ip, + dockerMode: true, + want: "http://172.19.16.8:3100", + }, + { + name: "systemd 模式下 127.0.0.1 保持原样（同机部署是合法的）", + endpoint: "http://127.0.0.1:3100", + hostIP: ip, + want: "http://127.0.0.1:3100", + }, + { + name: "容器模式下 localhost 同样换掉", + endpoint: "http://localhost:3100", + hostIP: ip, + dockerMode: true, + want: "http://172.19.16.8:3100", + }, + { + name: "缺端口时补 3100", + endpoint: "http://0.0.0.0", + hostIP: ip, + want: "http://172.19.16.8:3100", + }, + { + name: "空 endpoint 直接报错", + endpoint: " ", + hostIP: ip, + wantErr: "缺少 Loki 推送地址", + }, + { + name: "拿不到主机 IP 时明确报错，而不是返回必然失败的地址", + endpoint: "http://0.0.0.0:3100", + hostIP: "", + wantErr: "取不到目标主机 IP", + }, + } + for _, c := range cases { + got, err := hostPushEndpoint(c.endpoint, c.hostIP, c.dockerMode) + if c.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), c.wantErr) { + t.Fatalf("%s: 期望报错含 %q，实际 got=%q err=%v", c.name, c.wantErr, got, err) + } + continue + } + if err != nil { + t.Fatalf("%s: 不应报错: %v", c.name, err) + } + if got != c.want { + t.Fatalf("%s: 期望 %q，实际 %q", c.name, c.want, got) + } + } +} + +// 渲染出来的 promtail 配置里不能再出现 0.0.0.0 这类推不上去的地址。 +func TestHostPromtailConfigUsesReachableEndpoint(t *testing.T) { + fixed, err := hostPushEndpoint("http://0.0.0.0:3100", "172.19.16.8", true) + if err != nil { + t.Fatalf("纠正失败: %v", err) + } + cfg, err := PromtailConfigForHost(HostPromtailOptions{ + Endpoint: fixed, Cluster: "host", Hostname: "h1", LogPath: "/var/log/a.log", + }) + if err != nil { + t.Fatalf("渲染失败: %v", err) + } + if strings.Contains(cfg, "0.0.0.0") { + t.Fatalf("配置里仍是通配地址:\n%s", cfg) + } + if !strings.Contains(cfg, "url: http://172.19.16.8:3100/loki/api/v1/push") { + t.Fatalf("推送地址不是纠正后的地址:\n%s", cfg) + } +} diff --git a/build.sh b/build.sh index ff9e935..baa1878 100755 --- a/build.sh +++ b/build.sh @@ -45,7 +45,9 @@ detect_node() { detect_go() { if command -v go &> /dev/null; then GO_VERSION=$(go version 2>/dev/null | awk '{print $3}') - echo "[INFO] Go detected: $GO_VERSION" + # INFO 必须走 stderr：detect_go 的 stdout 会被当作命令路径捕获（ + # 原来把 INFO 打到 stdout，GO_CMD 变成 "[INFO] Go detected: go1.24" → 构建直接失败） + echo "[INFO] Go detected: $GO_VERSION" >&2 echo "go" elif [ -f "$HOME/go/bin/go" ]; then echo "$HOME/go/bin/go" diff --git a/frontend/src/api/collector/index.ts b/frontend/src/api/collector/index.ts index ba67658..3dac531 100644 --- a/frontend/src/api/collector/index.ts +++ b/frontend/src/api/collector/index.ts @@ -11,6 +11,15 @@ export interface CollectorPreview { syslogFormat?: string; } +export interface CollectorLogs { + /** 日志来源：docker | systemd | kubernetes */ + source: string; + /** 展示用的来源描述（容器名 / systemd 单元 / Pod） */ + sourceLabel: string; + /** 日志正文（可能为空 = 采集器尚未产生日志） */ + content: string; +} + export const collectorApi = { list: (params: ListParams = {}, options?: RequestOptions) => http.get>(`/collectors${buildQuery(params)}`, options), @@ -42,4 +51,8 @@ export const collectorApi = { install: (id: number) => http.post<{ message: string }>(`/collectors/${id}/install`), uninstall: (id: number) => http.post<{ message: string }>(`/collectors/${id}/uninstall`), + + /** 读取采集器最近运行日志（主机：docker logs / journalctl；K8s：Pod 日志） */ + logs: (id: number, tail = 200) => + http.get(`/collectors/${id}/logs?tail=${tail}`), }; diff --git a/frontend/src/i18n/locales/en/collectors.ts b/frontend/src/i18n/locales/en/collectors.ts index f82f174..25dc966 100644 --- a/frontend/src/i18n/locales/en/collectors.ts +++ b/frontend/src/i18n/locales/en/collectors.ts @@ -25,14 +25,32 @@ const collectors = { namespace: 'Namespace', logPath: 'Log Path', target: 'Target', + runMode: 'Run Mode', unknown: 'Unknown', }, + runMode: { + docker: 'Docker container', + systemd: 'systemd service', + }, action: { install: 'Deploy', uninstall: 'Uninstall', + logs: 'Logs', editAria: 'Edit collection policy {{name}}', deleteAria: 'Delete collector {{name}}', }, + logs: { + title: 'Collector logs · {{name}}', + description: 'Recent collector logs (host: docker logs / journalctl; K8s: collector Pod logs)', + source: 'Source: {{source}}', + tailLines: '{{count}} lines', + refresh: 'Refresh', + empty: 'No log output yet (the collector may have just started, or nothing matched so far)', + copy: 'Copy', + copied: 'Copied', + close: 'Close', + failed: 'Failed to read collector logs', + }, toast: { loadOptionsFailed: 'Failed to load Loki deployments, clusters, or hosts', installFailed: 'Failed to deploy collector', diff --git a/frontend/src/i18n/locales/zh/collectors.ts b/frontend/src/i18n/locales/zh/collectors.ts index dd5d35f..266a9f4 100644 --- a/frontend/src/i18n/locales/zh/collectors.ts +++ b/frontend/src/i18n/locales/zh/collectors.ts @@ -26,14 +26,33 @@ const collectors = { namespace: '命名空间', logPath: '日志路径', target: '目标', + runMode: '运行方式', unknown: '未知', }, + // 主机采集器托管方式（安装时自动探测：docker 优先、systemd 兜底） + runMode: { + docker: 'Docker 容器', + systemd: 'systemd 服务', + }, action: { install: '部署', uninstall: '卸载', + logs: '日志', editAria: '编辑采集策略 {{name}}', deleteAria: '删除采集器 {{name}}', }, + logs: { + title: '采集器日志 · {{name}}', + description: '采集器最近运行日志（主机：docker logs / journalctl；K8s：采集 Pod 日志）', + source: '来源：{{source}}', + tailLines: '{{count}} 行', + refresh: '刷新', + empty: '暂无日志输出（采集器可能刚启动，或暂时没有匹配的日志行）', + copy: '复制', + copied: '已复制', + close: '关闭', + failed: '读取采集器日志失败', + }, toast: { loadOptionsFailed: '加载 Loki 部署/集群/主机数据失败', installFailed: '部署采集器失败', diff --git a/frontend/src/pages/collectors/components/CollectorCard.tsx b/frontend/src/pages/collectors/components/CollectorCard.tsx index 78fc7de..61d3532 100644 --- a/frontend/src/pages/collectors/components/CollectorCard.tsx +++ b/frontend/src/pages/collectors/components/CollectorCard.tsx @@ -5,6 +5,7 @@ import '@/i18n/locales/zh/collectors' import '@/i18n/locales/en/collectors' import { RadioTower, + ScrollText, Trash2, Upload, Download, @@ -36,16 +37,23 @@ interface CollectorCardProps { onUninstall: (id: number) => void onDelete: (c: Collector) => void onEdit: (c: Collector) => void + onLogs: (c: Collector) => void } /** 采集策略卡片 — 统一 ResourceCard 样式 */ export const CollectorCard = memo(function CollectorCard({ - collector, operating, onInstall, onUninstall, onDelete, onEdit, + collector, operating, onInstall, onUninstall, onDelete, onEdit, onLogs, }: CollectorCardProps) { const { t } = useTranslation('collectors') const meta = COLLECTOR_TYPE_META[collector.type] ?? FALLBACK_META const Icon = meta.icon const isBusy = operating === collector.id + // 运行方式只对主机采集器有意义（容器 vs systemd 单元）； + // K8s 类型的运行形态已由类型徽章（DaemonSet / Sidecar）表达，不重复占一行。 + const runModeLabel = + collector.type === 'host-promtail' && collector.runMode + ? t(`runMode.${collector.runMode}`) + : '' return ( } fields={[ { label: t('field.loki'), value: collector.lokiDeployment?.name || t('field.unknown') }, + ...(runModeLabel ? [{ label: t('field.runMode'), value: runModeLabel }] : []), ...(collector.cluster ? [{ label: t('field.cluster'), value: collector.cluster.name }] : []), ...(collector.host ? [{ label: t('field.host'), value: `${collector.host.name}(${collector.host.hostIP})` }] : []), ...(collector.namespace ? [{ label: t('field.namespace'), value: collector.namespace }] : []), @@ -67,6 +76,7 @@ export const CollectorCard = memo(function CollectorCard({ onEdit(collector)} /> onInstall(collector.id)} /> onUninstall(collector.id)} /> + onLogs(collector)} /> onDelete(collector)} /> } diff --git a/frontend/src/pages/collectors/components/CollectorLogsDialog.tsx b/frontend/src/pages/collectors/components/CollectorLogsDialog.tsx new file mode 100644 index 0000000..7971b91 --- /dev/null +++ b/frontend/src/pages/collectors/components/CollectorLogsDialog.tsx @@ -0,0 +1,141 @@ +import { memo, useCallback, useEffect, useState } from 'react' +import { useTranslation } from 'react-i18next' +// collectors 域词典(zh/en 自注册模块,副作用导入须先于 useTranslation 使用) +import '@/i18n/locales/zh/collectors' +import '@/i18n/locales/en/collectors' +import { RefreshCw } from 'lucide-react' +import { Button } from '@/components/ui/button' +import { Dialog, DialogContent, DialogDescription, DialogHeader, DialogTitle } from '@/components/ui/dialog' +import { collectorApi, type CollectorLogs } from '@/api' +import { LoadingSpinner } from '@/components/LoadingSpinner' +import { toast } from 'sonner' +import type { Collector } from '@/types' + +interface CollectorLogsDialogProps { + collector: Collector | null + onOpenChange: (open: boolean) => void +} + +/** 可选的日志行数（默认 200：够看启动报错，又不至于把弹窗卡住） */ +const TAIL_OPTIONS = [200, 500, 1000] + +/** + * 采集器运行日志弹窗 —— 主机采集器在 docker 模式下是 `docker logs`， + * systemd 模式下是 `journalctl -u promtail`；K8s 采集器是采集 Pod 的容器日志。 + * 排「采集器起没起来、在报什么错」不用再登机器。 + */ +export const CollectorLogsDialog = memo(function CollectorLogsDialog({ + collector, + onOpenChange, +}: CollectorLogsDialogProps) { + const { t } = useTranslation('collectors') + const [logs, setLogs] = useState(null) + const [loading, setLoading] = useState(false) + const [error, setError] = useState('') + const [tail, setTail] = useState(TAIL_OPTIONS[0]) + + const load = useCallback( + async (id: number, lines: number) => { + setLoading(true) + setError('') + try { + setLogs(await collectorApi.logs(id, lines)) + } catch (err) { + // 后端会指名原因（未部署 / 没有容器也没有 systemd 单元 / 缺 pods/log 权限），直接展示 + setError(err instanceof Error ? err.message : t('logs.failed')) + setLogs(null) + } finally { + setLoading(false) + } + }, + [t] + ) + + useEffect(() => { + if (!collector) { + setLogs(null) + setError('') + return + } + setTail(TAIL_OPTIONS[0]) + void load(collector.id, TAIL_OPTIONS[0]) + }, [collector, load]) + + const copying = Boolean(logs?.content) + + return ( + + + + {t('logs.title', { name: collector?.name ?? '' })} + + {logs?.sourceLabel ? t('logs.source', { source: logs.sourceLabel }) : t('logs.description')} + + + +
+ {TAIL_OPTIONS.map((n) => ( + + ))} + +
+ + {loading ? ( + + ) : error ? ( +
+ {error} +
+ ) : ( + 
+            {logs?.content || t('logs.empty')}
+
+ )} + +
+ + +
+ + + ) +}) diff --git a/frontend/src/pages/collectors/index.tsx b/frontend/src/pages/collectors/index.tsx index 30e73d6..cdcdd38 100644 --- a/frontend/src/pages/collectors/index.tsx +++ b/frontend/src/pages/collectors/index.tsx @@ -18,6 +18,7 @@ import type { Cluster, Collector, Host, LokiDeployment } from '@/types' import { CollectorCard } from './components/CollectorCard' import { CollectorFormDialog } from './components/CollectorFormDialog' import { CollectorEditDialog } from './components/CollectorEditDialog' +import { CollectorLogsDialog } from './components/CollectorLogsDialog' /** * 日志采集页 — 采集策略编排：列表走 useListQuery（取消+版本号）， @@ -38,6 +39,8 @@ const CollectorsPage = memo(function CollectorsPage() { const [formOpen, setFormOpen] = useState(false) const [editTarget, setEditTarget] = useState(null) const [deleteTarget, setDeleteTarget] = useState(null) + /** 日志弹窗目标（null = 关闭） */ + const [logsTarget, setLogsTarget] = useState(null) const [deleting, setDeleting] = useState(false) /** 部署/卸载指令进行中的采集器 ID（按钮 loading + 防重复点击） */ const [operating, setOperating] = useState(null) @@ -146,6 +149,7 @@ const CollectorsPage = memo(function CollectorsPage() { onUninstall={(id) => handleAction(id, 'uninstall')} onDelete={setDeleteTarget} onEdit={setEditTarget} + onLogs={setLogsTarget} /> ))} @@ -169,6 +173,8 @@ const CollectorsPage = memo(function CollectorsPage() { { if (!open) setEditTarget(null) }} onSaved={query.refresh} /> + { if (!open) setLogsTarget(null) }} /> + { diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 0634244..35dee91 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -225,6 +225,8 @@ export interface Collector { resolvedHostname?: string; /** 上次安装探测到的 syslog 时间戳格式：iso8601 | rfc3164 */ syslogFormat?: string; + /** 主机采集器的托管方式（安装时自动探测决定）：docker | systemd */ + runMode?: 'docker' | 'systemd' | ''; status: string; statusMessage: string; createdAt: string;