feat: 新增「部署客户端」日志采集器模块

三种采集器(host-promtail / k8s-daemonset / k8s-fluentbit-sidecar),每种关联已部署 Loki:
- host-promtail:SSH 安装 Promtail(复用 deployBinary)
- k8s-daemonset:helm 安装 Promtail DaemonSet(复用 deployHelm)
- k8s-fluentbit-sidecar:kubectl patch 注入 Fluent Bit sidecar

含 Collector 数据模型、三个部署器、配置模板、CRUD API、前端页面。
附带 UI 优化:ErrorBoundary、部署卡片浏览器访问按钮、菜单名称调整。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Logan.LiandClaude committed 2026-06-29 19:26:41 +08:00
1 parent 2a93de600d
commit 673b2c60c8
17 files changed
+1442 -4

No files matched your search

+268
View File
@@ -0,0 +1,268 @@
package services
import (
"context"
"fmt"
"time"
"lokideploy/models"
"lokideploy/utils"
)
type CollectorService struct{}
func NewCollectorService() *CollectorService {
return &CollectorService{}
}
type CreateCollectorRequest struct {
Name string
Type string
LokiDeploymentID uint
HostID *uint
ClusterID *uint
TargetDeployment string
Namespace string
LogPath string
Labels string
}
// fillAssociations 填充关联的 Loki / Host / Cluster(用于配置生成与展示)
func (s *CollectorService) fillAssociations(c *models.Collector) {
db := models.GetDB()
db.RLock()
defer db.RUnlock()
for i := range db.Deployments {
if db.Deployments[i].ID == c.LokiDeploymentID {
d := db.Deployments[i]
c.LokiDeployment = &d
break
}
}
if c.HostID != nil {
for i := range db.Hosts {
if db.Hosts[i].ID == *c.HostID {
h := db.Hosts[i]
c.Host = &h
break
}
}
}
if c.ClusterID != nil {
for i := range db.Clusters {
if db.Clusters[i].ID == *c.ClusterID {
cl := db.Clusters[i]
c.Cluster = &cl
break
}
}
}
}
// generateConfigSnapshot 按类型生成配置快照
func (s *CollectorService) generateConfigSnapshot(c *models.Collector) string {
endpoint := ""
if c.LokiDeployment != nil {
endpoint = c.LokiDeployment.Endpoint
}
switch c.Type {
case "host-promtail":
return PromtailConfig(endpoint, c.LogPath)
case "k8s-daemonset":
return PromtailHelmValues(endpoint)
case "k8s-fluentbit-sidecar":
host, port := parseLokiURL(endpoint)
app := c.TargetDeployment
ns := c.Namespace
if ns == "" {
ns = "default"
}
return FluentbitConfig(host, port, c.LogPath, app, ns)
}
return ""
}
func (s *CollectorService) Create(req *CreateCollectorRequest) (*models.Collector, error) {
switch req.Type {
case "host-promtail", "k8s-daemonset", "k8s-fluentbit-sidecar":
default:
return nil, fmt.Errorf("不支持的采集器类型: %s", req.Type)
}
if req.LokiDeploymentID == 0 {
return nil, fmt.Errorf("请选择关联的 Loki 部署")
}
if req.Type == "host-promtail" && (req.HostID == nil || *req.HostID == 0) {
return nil, fmt.Errorf("主机采集器需要选择目标主机")
}
if (req.Type == "k8s-daemonset" || req.Type == "k8s-fluentbit-sidecar") && (req.ClusterID == nil || *req.ClusterID == 0) {
return nil, fmt.Errorf("K8s 采集器需要选择目标集群")
}
if req.Type == "k8s-fluentbit-sidecar" && req.TargetDeployment == "" {
return nil, fmt.Errorf("Sidecar 采集器需要填写目标 Deployment 名称")
}
refType := "cluster"
if req.Type == "host-promtail" {
refType = "host"
}
collector := models.Collector{
ID: models.GetDB().NextID("collectors"),
Name: req.Name,
Type: req.Type,
LokiDeploymentID: req.LokiDeploymentID,
RefType: refType,
HostID: req.HostID,
ClusterID: req.ClusterID,
TargetDeployment: req.TargetDeployment,
Namespace: req.Namespace,
LogPath: req.LogPath,
Labels: req.Labels,
Status: "pending",
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
s.fillAssociations(&collector)
if collector.LokiDeployment == nil {
return nil, fmt.Errorf("关联的 Loki 部署不存在")
}
collector.Config = s.generateConfigSnapshot(&collector)
db := models.GetDB()
db.Lock()
db.Collectors = append(db.Collectors, collector)
db.Unlock()
go db.Save()
go s.Install(collector.ID)
return &collector, nil
}
func (s *CollectorService) Get(id uint) (*models.Collector, error) {
db := models.GetDB()
db.RLock()
var found *models.Collector
for i := range db.Collectors {
if db.Collectors[i].ID == id {
c := db.Collectors[i]
found = &c
break
}
}
db.RUnlock()
if found == nil {
return nil, utils.ErrNotFound
}
s.fillAssociations(found)
return found, nil
}
func (s *CollectorService) List(page, pageSize int) ([]models.Collector, int64, error) {
db := models.GetDB()
db.RLock()
total := int64(len(db.Collectors))
start := (page - 1) * pageSize
if start < 0 || start >= len(db.Collectors) {
db.RUnlock()
return []models.Collector{}, total, nil
}
end := start + pageSize
if end > len(db.Collectors) {
end = len(db.Collectors)
}
result := make([]models.Collector, end-start)
copy(result, db.Collectors[start:end])
db.RUnlock()
for i := range result {
s.fillAssociations(&result[i])
}
return result, total, nil
}
func (s *CollectorService) Update(id uint, updates map[string]interface{}) error {
db := models.GetDB()
db.Lock()
defer db.Unlock()
for i := range db.Collectors {
if db.Collectors[i].ID == id {
if v, ok := updates["name"].(string); ok && v != "" {
db.Collectors[i].Name = v
}
if v, ok := updates["status"].(string); ok && v != "" {
db.Collectors[i].Status = v
}
if v, ok := updates["statusMessage"].(string); ok {
db.Collectors[i].StatusMessage = v
}
if v, ok := updates["logPath"].(string); ok {
db.Collectors[i].LogPath = v
}
if v, ok := updates["labels"].(string); ok {
db.Collectors[i].Labels = v
}
db.Collectors[i].UpdatedAt = time.Now()
go db.Save()
return nil
}
}
return utils.ErrNotFound
}
func (s *CollectorService) Delete(id uint) error {
collector, err := s.Get(id)
if err != nil {
return utils.ErrNotFound
}
db := models.GetDB()
db.Lock()
for i, c := range db.Collectors {
if c.ID == id {
db.Collectors = append(db.Collectors[:i], db.Collectors[i+1:]...)
break
}
}
db.Unlock()
go db.Save()
// 异步卸载(用拷贝的 collector,避免删除后取不到)
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
NewCollectorDeployer(collector, collector.LokiDeployment, nil).Uninstall(ctx)
}()
return nil
}
// Install 异步部署采集器
func (s *CollectorService) Install(id uint) {
collector, err := s.Get(id)
if err != nil {
return
}
if collector.LokiDeployment == nil {
s.Update(id, map[string]interface{}{"status": "error", "statusMessage": "关联的 Loki 不存在"})
return
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
if err := NewCollectorDeployer(collector, collector.LokiDeployment, nil).Install(ctx); err != nil {
s.Update(id, map[string]interface{}{"status": "error", "statusMessage": err.Error()})
return
}
s.Update(id, map[string]interface{}{"status": "active", "statusMessage": "采集器已部署"})
}
// Uninstall 异步卸载采集器
func (s *CollectorService) Uninstall(id uint) {
collector, err := s.Get(id)
if err != nil {
return
}
if collector.LokiDeployment == nil {
return
}
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": "采集器已卸载"})
}
+331
View File
@@ -0,0 +1,331 @@
package services
import (
"context"
"fmt"
"os"
"os/exec"
"strings"
"lokideploy/models"
)
// CollectorDeployer 采集器部署器,按 type 分发到三种安装/卸载流程
type CollectorDeployer struct {
collector *models.Collector
loki *models.LokiDeployment
logFunc func(level, message, detail string)
}
func NewCollectorDeployer(c *models.Collector, l *models.LokiDeployment, logFunc func(level, message, detail string)) *CollectorDeployer {
return &CollectorDeployer{collector: c, loki: l, logFunc: logFunc}
}
func (d *CollectorDeployer) log(level, message, detail string) {
if d.logFunc != nil {
d.logFunc(level, message, detail)
}
}
// kubeconfigEnv 构造 kubectl/helm 所需的 KUBECONFIG 环境变量,返回 cleanup 删除临时文件
func kubeconfigEnv(cluster *models.Cluster) (env []string, cleanup func(), err error) {
kubeconfig, err := resolveKubeconfig(cluster)
if err != nil {
return nil, nil, err
}
kubeFile, err := os.CreateTemp("", "kube-*.yaml")
if err != nil {
return nil, nil, err
}
kubeFile.WriteString(kubeconfig)
kubeFile.Close()
env = append(os.Environ(), fmt.Sprintf("KUBECONFIG=%s", kubeFile.Name()))
cleanup = func() { os.Remove(kubeFile.Name()) }
return env, cleanup, nil
}
// Install 按采集器类型分发安装
func (d *CollectorDeployer) Install(ctx context.Context) error {
endpoint := d.loki.Endpoint
if endpoint == "" {
return fmt.Errorf("关联的 Loki 尚未部署,缺少推送地址(endpoint),请先成功部署该 Loki")
}
d.log("info", fmt.Sprintf("开始部署采集器 %s(类型 %s)", d.collector.Name, d.collector.Type), "")
var err error
switch d.collector.Type {
case "host-promtail":
err = d.installHost(ctx, endpoint)
case "k8s-daemonset":
err = d.installDaemonset(ctx, endpoint)
case "k8s-fluentbit-sidecar":
err = d.installSidecar(ctx, endpoint)
default:
return fmt.Errorf("不支持的采集器类型: %s", d.collector.Type)
}
if err != nil {
d.log("error", fmt.Sprintf("采集器部署失败: %v", err), "")
return err
}
d.log("success", "采集器部署完成", "")
return nil
}
// Uninstall 按采集器类型分发卸载
func (d *CollectorDeployer) Uninstall(ctx context.Context) error {
switch d.collector.Type {
case "host-promtail":
return d.uninstallHost(ctx)
case "k8s-daemonset":
return d.uninstallDaemonset(ctx)
case "k8s-fluentbit-sidecar":
return d.uninstallSidecar(ctx)
}
return nil
}
// ===== host-promtail:SSH 安装 Promtail 二进制 + systemd =====
func (d *CollectorDeployer) installHost(ctx context.Context, endpoint string) error {
host := d.collector.Host
if host == nil {
return fmt.Errorf("主机采集器缺少目标主机")
}
password, err := DecryptCred(host.Password)
if err != nil {
return err
}
client, err := NewSSHClient(host.HostIP, host.Port, host.User, password)
if err != nil {
return err
}
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)
}
// 下载 Promtail
d.log("info", fmt.Sprintf("下载 Promtail(%s)", arch), "")
url := fmt.Sprintf("https://github.com/grafana/loki/releases/download/v3.0.0/promtail-linux-%s.zip", arch)
dl := fmt.Sprintf("mkdir -p /opt/promtail /etc/promtail && 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 失败: %v", err)
}
client.Execute(ctx, "ln -sf /opt/promtail/promtail-linux-* /usr/local/bin/promtail 2>/dev/null || true")
// 配置
logPath := d.collector.LogPath
if logPath == "" {
logPath = "/var/log/*.log"
}
conf := PromtailConfig(endpoint, logPath)
client.Execute(ctx, fmt.Sprintf("cat > /etc/promtail/promtail.yaml << 'EOF'\n%s\nEOF", conf))
// systemd 服务
service := `[Unit]
Description=Promtail Log Collector
After=network.target
[Service]
Type=simple
ExecStart=/usr/local/bin/promtail -config.file=/etc/promtail/promtail.yaml
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
`
client.Execute(ctx, fmt.Sprintf("cat > /etc/systemd/system/promtail.service << 'EOF'\n%s\nEOF", service))
if _, _, err := client.Execute(ctx, "systemctl daemon-reload && systemctl enable promtail && systemctl restart promtail"); err != nil {
return fmt.Errorf("启动 Promtail 失败: %v", err)
}
return nil
}
func (d *CollectorDeployer) uninstallHost(ctx context.Context) error {
host := d.collector.Host
if host == nil {
return nil
}
password, _ := DecryptCred(host.Password)
client, err := NewSSHClient(host.HostIP, host.Port, host.User, password)
if err != nil {
return err
}
defer client.Close()
client.Execute(ctx, "systemctl stop promtail 2>/dev/null || true; systemctl disable promtail 2>/dev/null || true; rm -f /etc/systemd/system/promtail.service /usr/local/bin/promtail; systemctl daemon-reload")
return nil
}
// ===== k8s-daemonset:helm 安装 Promtail DaemonSet =====
func (d *CollectorDeployer) installDaemonset(ctx context.Context, endpoint string) error {
cluster := d.collector.Cluster
if cluster == nil {
return fmt.Errorf("DaemonSet 采集器缺少目标集群")
}
env, cleanup, err := kubeconfigEnv(cluster)
if err != nil {
return err
}
defer cleanup()
ns := d.collector.Namespace
if ns == "" {
ns = "logging"
}
values := PromtailHelmValues(endpoint)
valuesFile, err := os.CreateTemp("", "promtail-values-*.yaml")
if err != nil {
return err
}
defer os.Remove(valuesFile.Name())
valuesFile.WriteString(values)
valuesFile.Close()
d.log("info", fmt.Sprintf("helm install promtail(命名空间 %s)", ns), "")
// 先添加 grafana repo(幂等)
repo := exec.CommandContext(ctx, "helm", "repo", "add", "grafana", "https://grafana.github.io/helm-charts")
repo.Env = env
repo.CombinedOutput()
exec.CommandContext(ctx, "helm", "repo", "update").CombinedOutput()
cmd := exec.CommandContext(ctx, "helm", "upgrade", "--install", "promtail", "grafana/promtail",
"-f", valuesFile.Name(), "--namespace", ns, "--create-namespace", "--wait", "--timeout", "5m")
cmd.Env = env
out, err := cmd.CombinedOutput()
if err != nil {
return fmt.Errorf("helm 安装 Promtail 失败: %v\n%s", err, string(out))
}
return nil
}
func (d *CollectorDeployer) uninstallDaemonset(ctx context.Context) error {
cluster := d.collector.Cluster
if cluster == nil {
return nil
}
env, cleanup, err := kubeconfigEnv(cluster)
if err != nil {
return err
}
defer cleanup()
ns := d.collector.Namespace
if ns == "" {
ns = "logging"
}
cmd := exec.CommandContext(ctx, "helm", "uninstall", "promtail", "--namespace", ns)
cmd.Env = env
cmd.CombinedOutput()
return nil
}
// ===== k8s-fluentbit-sidecar:kubectl patch 注入 Fluent Bit sidecar =====
func (d *CollectorDeployer) installSidecar(ctx context.Context, endpoint string) error {
cluster := d.collector.Cluster
if cluster == nil {
return fmt.Errorf("Sidecar 采集器缺少目标集群")
}
target := d.collector.TargetDeployment
if target == "" {
return fmt.Errorf("Sidecar 采集器缺少目标 Deployment")
}
env, cleanup, err := kubeconfigEnv(cluster)
if err != nil {
return err
}
defer cleanup()
ns := d.collector.Namespace
if ns == "" {
ns = "default"
}
logPath := d.collector.LogPath
if logPath == "" {
logPath = "/var/log/app/*.log"
}
app := target
// 1. 检查目标 deployment 是否存在
d.log("info", fmt.Sprintf("检查目标 Deployment %s/%s", ns, target), "")
check := exec.CommandContext(ctx, "kubectl", "get", "deployment", target, "-n", ns)
check.Env = env
if out, err := check.CombinedOutput(); err != nil {
return fmt.Errorf("目标 Deployment 不存在: %s/%s\n%s", ns, target, string(out))
}
// 2. 幂等检查:是否已有 fluent-bit 容器
getNames := exec.CommandContext(ctx, "kubectl", "get", "deployment", target, "-n", ns, "-o", "jsonpath={.spec.template.spec.containers[*].name}")
getNames.Env = env
namesOut, _ := getNames.CombinedOutput()
if strings.Contains(string(namesOut), "fluent-bit") {
d.log("warn", "目标 Deployment 已含 fluent-bit 容器,跳过注入", "")
return nil
}
// 3. 部署 ConfigMap
lokiHost, lokiPort := parseLokiURL(endpoint)
conf := FluentbitConfig(lokiHost, lokiPort, logPath, app, ns)
cm := FluentbitConfigMapYAML(conf, ns)
cmFile, _ := os.CreateTemp("", "fluent-bit-cm-*.yaml")
defer os.Remove(cmFile.Name())
cmFile.WriteString(cm)
cmFile.Close()
d.log("info", "部署 Fluent Bit ConfigMap", "")
applyCm := exec.CommandContext(ctx, "kubectl", "apply", "-f", cmFile.Name())
applyCm.Env = env
if out, err := applyCm.CombinedOutput(); err != nil {
return fmt.Errorf("ConfigMap 部署失败: %v\n%s", err, string(out))
}
// 4. patch 注入 sidecar
d.log("info", fmt.Sprintf("为 %s 注入 Fluent Bit Sidecar", target), "")
patchFile, _ := os.CreateTemp("", "sidecar-patch-*.yaml")
defer os.Remove(patchFile.Name())
patchFile.WriteString(SidecarPatchYAML)
patchFile.Close()
patch := exec.CommandContext(ctx, "kubectl", "patch", "deployment", target, "-n", ns, "--type=strategic", "--patch-file", patchFile.Name())
patch.Env = env
if out, err := patch.CombinedOutput(); err != nil {
return fmt.Errorf("Sidecar 注入失败: %v\n%s", err, string(out))
}
// 5. 等待 rollout
d.log("info", "等待 Pod 重建", "")
rollout := exec.CommandContext(ctx, "kubectl", "rollout", "status", "deployment/"+target, "-n", ns, "--timeout=5m")
rollout.Env = env
rollout.CombinedOutput()
return nil
}
func (d *CollectorDeployer) uninstallSidecar(ctx context.Context) error {
cluster := d.collector.Cluster
if cluster == nil || d.collector.TargetDeployment == "" {
return nil
}
env, cleanup, err := kubeconfigEnv(cluster)
if err != nil {
return err
}
defer cleanup()
ns := d.collector.Namespace
if ns == "" {
ns = "default"
}
patchFile, _ := os.CreateTemp("", "sidecar-rm-*.yaml")
defer os.Remove(patchFile.Name())
patchFile.WriteString(SidecarRemovePatchYAML)
patchFile.Close()
patch := exec.CommandContext(ctx, "kubectl", "patch", "deployment", d.collector.TargetDeployment, "-n", ns, "--type=strategic", "--patch-file", patchFile.Name())
patch.Env = env
patch.CombinedOutput()
return nil
}
+183
View File
@@ -0,0 +1,183 @@
package services
import (
"fmt"
"strings"
)
// PromtailConfig 主机 Promtail 配置(推送至 lokiEndpoint,采集 logPath)
func PromtailConfig(lokiEndpoint, logPath string) string {
return fmt.Sprintf(`server:
http_listen_port: 9080
grpc_listen_port: 0
log_level: info
positions:
filename: /tmp/positions.yaml
clients:
- url: %s/loki/api/v1/push
scrape_configs:
- job_name: app
static_configs:
- targets: [localhost]
labels:
job: app
host: ${HOSTNAME}
__path__: %s
`, lokiEndpoint, logPath)
}
// PromtailHelmValues DaemonSet 模式 Promtail Helm values
func PromtailHelmValues(lokiEndpoint string) string {
return fmt.Sprintf(`config:
clients:
- url: %s/loki/api/v1/push
daemonset:
enabled: true
serviceAccount:
create: true
resources:
limits:
cpu: 200m
memory: 128Mi
requests:
cpu: 100m
memory: 64Mi
extraVolumes:
- name: varlog
hostPath:
path: /var/log
- name: varlibdockercontainers
hostPath:
path: /var/lib/docker/containers
extraVolumeMounts:
- name: varlog
mountPath: /var/log
readOnly: true
- name: varlibdockercontainers
mountPath: /var/lib/docker/containers
readOnly: true
`, lokiEndpoint)
}
// FluentbitConfig Sidecar 模式 Fluent Bit 配置(采集 logPath,推送至 lokiHost:lokiPort)
func FluentbitConfig(lokiHost, lokiPort, logPath, app, namespace string) string {
return fmt.Sprintf(`[SERVICE]
Flush 5
Log_Level info
HTTP_Server On
HTTP_Listen 0.0.0.0
HTTP_Port 2020
[INPUT]
Name tail
Path %s
Tag app.%s
Refresh_Interval 5
Mem_Buf_Limit 5MB
Skip_Long_Lines On
Read_from_Head True
DB /tmp/flb.db
[FILTER]
Name modify
Match app.*
Add namespace %s
Add app %s
[OUTPUT]
Name loki
Match *
Host %s
Port %s
Labels job=fluent-bit, app=%s, namespace=%s
`, logPath, app, namespace, app, lokiHost, lokiPort, app, namespace)
}
// FluentbitConfigMapYAML 包装 fluent-bit.conf 为 ConfigMap YAML(部署到 namespace)
func FluentbitConfigMapYAML(conf, namespace string) string {
return fmt.Sprintf(`apiVersion: v1
kind: ConfigMap
metadata:
name: fluent-bit-config
namespace: %s
data:
fluent-bit.conf: |
%s
`, namespace, indentBlock(conf, " "))
}
// SidecarPatchYAML strategic merge patch:为 Deployment 注入 Fluent Bit sidecar 容器
const SidecarPatchYAML = `spec:
template:
spec:
volumes:
- name: app-logs
emptyDir: {}
- name: fluent-bit-config
configMap:
name: fluent-bit-config
containers:
- name: fluent-bit
image: cr.fluentbit.io/fluent/fluent-bit:3.1.1
imagePullPolicy: IfNotPresent
volumeMounts:
- name: app-logs
mountPath: /var/log/app
readOnly: true
- name: fluent-bit-config
mountPath: /fluent-bit/etc/
resources:
limits:
cpu: 200m
memory: 256Mi
requests:
cpu: 100m
memory: 128Mi
`
// SidecarRemovePatchYAML strategic merge patch:从 Deployment 移除 Fluent Bit sidecar(卸载)
const SidecarRemovePatchYAML = `spec:
template:
spec:
containers:
- name: fluent-bit
$patch: delete
volumes:
- name: app-logs
$patch: delete
- name: fluent-bit-config
$patch: delete
`
// indentBlock 给每行加前缀(ConfigMap 嵌入配置需缩进)
func indentBlock(s, pad string) string {
lines := strings.Split(s, "\n")
for i, l := range lines {
if l != "" {
lines[i] = pad + l
}
}
return strings.Join(lines, "\n")
}
// parseLokiURL 从 http://host:port 提取 host 与 port(Fluent Bit OUTPUT 需要)
func parseLokiURL(endpoint string) (host, port string) {
u := strings.TrimSpace(endpoint)
u = strings.TrimPrefix(u, "http://")
u = strings.TrimPrefix(u, "https://")
if i := strings.Index(u, "/"); i >= 0 {
u = u[:i]
}
if i := strings.LastIndex(u, ":"); i >= 0 {
return u[:i], u[i+1:]
}
return u, "3100"
}