diff --git a/backend/.backend.pid b/backend/.backend.pid index 04506c8..722d656 100644 --- a/backend/.backend.pid +++ b/backend/.backend.pid @@ -1 +1 @@ -46832 +93376 diff --git a/backend/controllers/collector.go b/backend/controllers/collector.go new file mode 100644 index 0000000..b55cddd --- /dev/null +++ b/backend/controllers/collector.go @@ -0,0 +1,122 @@ +package controllers + +import ( + "lokideploy/models" + "lokideploy/services" + "lokideploy/utils" + + "github.com/gin-gonic/gin" +) + +type CollectorController struct { + service *services.CollectorService +} + +func NewCollectorController() *CollectorController { + return &CollectorController{service: services.NewCollectorService()} +} + +type CreateCollectorRequest struct { + Name string `json:"name" binding:"required"` + Type string `json:"type" binding:"required"` + LokiDeploymentID uint `json:"lokiDeploymentID" binding:"required"` + HostID *uint `json:"hostID"` + ClusterID *uint `json:"clusterID"` + TargetDeployment string `json:"targetDeployment"` + Namespace string `json:"namespace"` + LogPath string `json:"logPath"` + Labels string `json:"labels"` +} + +func (ctrl *CollectorController) Create(c *gin.Context) { + var req CreateCollectorRequest + if !utils.BindAndValidate(c, &req) { + return + } + collector, err := ctrl.service.Create(&services.CreateCollectorRequest{ + Name: req.Name, Type: req.Type, LokiDeploymentID: req.LokiDeploymentID, + HostID: req.HostID, ClusterID: req.ClusterID, TargetDeployment: req.TargetDeployment, + Namespace: req.Namespace, LogPath: req.LogPath, Labels: req.Labels, + }) + if err != nil { + utils.Error(c, 400, err.Error()) + return + } + utils.Success(c, models.SanitizedJSON(collector)) +} + +func (ctrl *CollectorController) Get(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + collector, err := ctrl.service.Get(uint(id)) + if err != nil { + utils.NotFound(c, "采集器不存在") + return + } + utils.Success(c, models.SanitizedJSON(collector)) +} + +func (ctrl *CollectorController) List(c *gin.Context) { + page, pageSize := utils.Pagination(c) + collectors, total, err := ctrl.service.List(page, pageSize) + if err != nil { + utils.Error(c, 500, err.Error()) + return + } + utils.Success(c, utils.PageData{ + List: models.SanitizedJSON(collectors), Total: total, Page: page, PageSize: pageSize, + }) +} + +func (ctrl *CollectorController) Update(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + var updates map[string]interface{} + if !utils.BindAndValidate(c, &updates) { + return + } + if err := ctrl.service.Update(uint(id), updates); err != nil { + utils.Error(c, 400, err.Error()) + return + } + utils.Success(c, gin.H{"message": "更新成功"}) +} + +func (ctrl *CollectorController) Delete(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + if err := ctrl.service.Delete(uint(id)); err != nil { + utils.Error(c, 400, err.Error()) + return + } + utils.Success(c, gin.H{"message": "删除成功"}) +} + +func (ctrl *CollectorController) Install(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + go ctrl.service.Install(uint(id)) + utils.Success(c, gin.H{"message": "已开始部署采集器"}) +} + +func (ctrl *CollectorController) Uninstall(c *gin.Context) { + id, err := utils.IntParam(c, "id") + if err != nil { + utils.BadRequest(c, "ID 无效") + return + } + go ctrl.service.Uninstall(uint(id)) + utils.Success(c, gin.H{"message": "已开始卸载采集器"}) +} diff --git a/backend/models/collector.go b/backend/models/collector.go new file mode 100644 index 0000000..a183157 --- /dev/null +++ b/backend/models/collector.go @@ -0,0 +1,66 @@ +package models + +import "time" + +// Collector 日志采集器（部署客户端），关联一个已部署的 Loki 作为推送目标 +type Collector struct { + ID uint `json:"id"` + Name string `json:"name"` + Type string `json:"type"` // host-promtail | k8s-daemonset | k8s-fluentbit-sidecar + LokiDeploymentID uint `json:"lokiDeploymentID"` + LokiDeployment *LokiDeployment `json:"lokiDeployment,omitempty"` + RefType string `json:"refType"` // host | cluster + HostID *uint `json:"hostID,omitempty"` + Host *Host `json:"host,omitempty"` + ClusterID *uint `json:"clusterID,omitempty"` + Cluster *Cluster `json:"cluster,omitempty"` + TargetDeployment string `json:"targetDeployment"` // sidecar 专用：目标 deployment 名 + Namespace string `json:"namespace"` // k8s 命名空间（默认 logging） + LogPath string `json:"logPath"` // 采集日志路径 + Labels string `json:"labels"` // 额外标签（JSON） + 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{} { + var loki map[string]interface{} + if c.LokiDeployment != nil { + loki = map[string]interface{}{ + "id": c.LokiDeployment.ID, + "name": c.LokiDeployment.Name, + "endpoint": c.LokiDeployment.Endpoint, + "status": c.LokiDeployment.Status, + } + } + var host map[string]interface{} + if c.Host != nil { + host = c.Host.Sanitized() + } + var cluster map[string]interface{} + if c.Cluster != nil { + cluster = c.Cluster.Sanitized() + } + return map[string]interface{}{ + "id": c.ID, + "name": c.Name, + "type": c.Type, + "lokiDeploymentID": c.LokiDeploymentID, + "lokiDeployment": loki, + "refType": c.RefType, + "hostID": c.HostID, + "host": host, + "clusterID": c.ClusterID, + "cluster": cluster, + "targetDeployment": c.TargetDeployment, + "namespace": c.Namespace, + "logPath": c.LogPath, + "labels": c.Labels, + "status": c.Status, + "statusMessage": c.StatusMessage, + "createdAt": c.CreatedAt, + "updatedAt": c.UpdatedAt, + } +} diff --git a/backend/models/db.go b/backend/models/db.go index 2c1feed..d0f8394 100644 --- a/backend/models/db.go +++ b/backend/models/db.go @@ -15,6 +15,7 @@ type JsonDB struct { Hosts []Host `json:"hosts"` Storages []StorageConfig `json:"storages"` Deployments []LokiDeployment `json:"deployments"` + Collectors []Collector `json:"collectors"` Logs []DeploymentLog `json:"logs"` Settings []SystemSetting `json:"settings"` sequences map[string]uint @@ -37,6 +38,7 @@ func InitDB() error { Hosts: []Host{}, Storages: []StorageConfig{}, Deployments: []LokiDeployment{}, + Collectors: []Collector{}, Logs: []DeploymentLog{}, Settings: []SystemSetting{}, sequences: map[string]uint{ @@ -45,6 +47,7 @@ func InitDB() error { "hosts": 0, "storages": 0, "deployments": 0, + "collectors": 0, "logs": 0, }, } @@ -57,6 +60,7 @@ func InitDB() error { for _, h := range d.Hosts { if h.ID > d.sequences["hosts"] { d.sequences["hosts"] = h.ID } } for _, s := range d.Storages { if s.ID > d.sequences["storages"] { d.sequences["storages"] = s.ID } } for _, de := range d.Deployments { if de.ID > d.sequences["deployments"] { d.sequences["deployments"] = de.ID } } + for _, col := range d.Collectors { if col.ID > d.sequences["collectors"] { d.sequences["collectors"] = col.ID } } for _, l := range d.Logs { if l.ID > d.sequences["logs"] { d.sequences["logs"] = l.ID } } } } diff --git a/backend/models/sanitizer.go b/backend/models/sanitizer.go index 6a6a6d9..095725f 100644 --- a/backend/models/sanitizer.go +++ b/backend/models/sanitizer.go @@ -37,6 +37,13 @@ func SanitizedJSON(v any) any { return nil } return val.Sanitized() + case Collector: + return val.Sanitized() + case *Collector: + if val == nil { + return nil + } + return val.Sanitized() // Slices case []User: result := make([]map[string]interface{}, len(val)) @@ -62,6 +69,12 @@ func SanitizedJSON(v any) any { result[i] = item.Sanitized() } return result + case []Collector: + result := make([]map[string]interface{}, len(val)) + for i, item := range val { + result[i] = item.Sanitized() + } + return result default: return v } diff --git a/backend/routers/router.go b/backend/routers/router.go index 275d4ae..f55b89d 100644 --- a/backend/routers/router.go +++ b/backend/routers/router.go @@ -28,6 +28,7 @@ func SetupRouter() *gin.Engine { hostCtrl := controllers.NewHostController() storageCtrl := controllers.NewStorageController() deploymentCtrl := controllers.NewDeploymentController() + collectorCtrl := controllers.NewCollectorController() // API v1 api := r.Group("/api/v1") @@ -83,6 +84,15 @@ func SetupRouter() *gin.Engine { authorized.GET("/deployments/:id/logs", deploymentCtrl.GetLogs) authorized.GET("/deployments/stats", deploymentCtrl.GetStats) + // Collectors（日志采集器 / 部署客户端） + authorized.GET("/collectors", collectorCtrl.List) + authorized.GET("/collectors/:id", collectorCtrl.Get) + authorized.POST("/collectors", collectorCtrl.Create) + authorized.PUT("/collectors/:id", collectorCtrl.Update) + authorized.DELETE("/collectors/:id", collectorCtrl.Delete) + authorized.POST("/collectors/:id/install", collectorCtrl.Install) + authorized.POST("/collectors/:id/uninstall", collectorCtrl.Uninstall) + // Static files - use executable directory as base distPath := findDistPath() diff --git a/backend/services/collector.go b/backend/services/collector.go new file mode 100644 index 0000000..12c139b --- /dev/null +++ b/backend/services/collector.go @@ -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": "采集器已卸载"}) +} diff --git a/backend/services/collector_deployer.go b/backend/services/collector_deployer.go new file mode 100644 index 0000000..b1aebb1 --- /dev/null +++ b/backend/services/collector_deployer.go @@ -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 +} diff --git a/backend/services/collector_templates.go b/backend/services/collector_templates.go new file mode 100644 index 0000000..c97ab1f --- /dev/null +++ b/backend/services/collector_templates.go @@ -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" +} diff --git a/frontend/.frontend.pid b/frontend/.frontend.pid index c87407e..ff58705 100644 --- a/frontend/.frontend.pid +++ b/frontend/.frontend.pid @@ -1 +1 @@ -46900 +93417 diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 8aa626c..d81a812 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -6,6 +6,7 @@ import Clusters from '@/pages/Clusters' import Hosts from '@/pages/Hosts' import Storage from '@/pages/Storage' import Deployments from '@/pages/Deployments' +import Collectors from '@/pages/Collectors' import Settings from '@/pages/Settings' function RequireAuth({ children }: { children: React.ReactNode }) { @@ -60,6 +61,14 @@ export default function App() { } /> + + + + } + /> = { + 'host-promtail': { label: '主机 Promtail', icon: Monitor, color: 'bg-purple-100 text-purple-700' }, + 'k8s-daemonset': { label: 'K8s DaemonSet', icon: Cloud, color: 'bg-blue-100 text-blue-700' }, + 'k8s-fluentbit-sidecar': { label: 'K8s Sidecar', icon: Server, color: 'bg-cyan-100 text-cyan-700' }, +} + +const getStatusColor = (status: string) => { + switch (status) { + case 'active': + return 'bg-emerald-100 text-emerald-700' + case 'error': + return 'bg-red-100 text-red-700' + case 'pending': + return 'bg-amber-100 text-amber-700' + default: + return 'bg-slate-100 text-slate-600' + } +} + +export default function Collectors() { + const [collectors, setCollectors] = useState([]) + const [deployments, setDeployments] = useState([]) + const [clusters, setClusters] = useState([]) + const [hosts, setHosts] = useState([]) + const [loading, setLoading] = useState(true) + const [open, setOpen] = useState(false) + const [busy, setBusy] = useState(null) + + // Form + const [name, setName] = useState('') + const [type, setType] = useState<'host-promtail' | 'k8s-daemonset' | 'k8s-fluentbit-sidecar'>('host-promtail') + const [lokiID, setLokiID] = useState('') + const [hostID, setHostID] = useState('') + const [clusterID, setClusterID] = useState('') + const [targetDeployment, setTargetDeployment] = useState('') + const [namespace, setNamespace] = useState('') + const [logPath, setLogPath] = useState('') + const [labels, setLabels] = useState('') + const [submitting, setSubmitting] = useState(false) + + useEffect(() => { + loadAll() + }, []) + + const loadAll = async () => { + try { + const [cRes, dRes, clRes, hRes] = await Promise.all([ + collectorApi.list(1, 100), + deploymentApi.list(1, 100), + clusterApi.list(1, 100), + hostApi.list(1, 100), + ]) + setCollectors(cRes.list) + setDeployments(dRes.list) + setClusters(clRes.list) + setHosts(hRes.list) + } catch (err) { + console.error('Failed to load collectors:', err) + } finally { + setLoading(false) + } + } + + const resetForm = () => { + setName('') + setType('host-promtail') + setLokiID('') + setHostID('') + setClusterID('') + setTargetDeployment('') + setNamespace('') + setLogPath('') + setLabels('') + } + + const handleSubmit = async (e: React.FormEvent) => { + e.preventDefault() + if (!lokiID) { + alert('请选择关联的 Loki 部署') + return + } + setSubmitting(true) + try { + await collectorApi.create({ + name, + type, + lokiDeploymentID: Number(lokiID), + hostID: type === 'host-promtail' && hostID ? Number(hostID) : undefined, + clusterID: type !== 'host-promtail' && clusterID ? Number(clusterID) : undefined, + targetDeployment: type === 'k8s-fluentbit-sidecar' ? targetDeployment : undefined, + namespace: type !== 'host-promtail' ? namespace : undefined, + logPath, + labels, + }) + setOpen(false) + resetForm() + loadAll() + } catch (err: unknown) { + const message = err instanceof Error ? err.message : '创建采集器失败' + alert(message) + } finally { + setSubmitting(false) + } + } + + const handleDelete = async (id: number) => { + if (!confirm('确定删除此采集器？（将自动卸载）')) return + try { + await collectorApi.remove(id) + loadAll() + } catch (err) { + console.error('Failed to delete collector:', err) + } + } + + const handleAction = async (id: number, action: 'install' | 'uninstall') => { + setBusy(id) + try { + await collectorApi[action](id) + setTimeout(loadAll, 3000) + } catch (err) { + console.error('Collector action failed:', err) + } finally { + setTimeout(() => setBusy(null), 3000) + } + } + + return ( +
+ {/* Header */} +
+
+

部署客户端

+

部署日志采集器，将日志推送到关联的 Loki

+
+ + + + + + + 新建日志采集器 + +
+
+
+ + setName(e.target.value)} placeholder="app-logs-collector" required /> +
+
+ + +
+
+ +
+ + +
+ + {type === 'host-promtail' && ( +
+ + +
+ )} + + {type !== 'host-promtail' && ( + <> +
+ + +
+
+
+ + setNamespace(e.target.value)} placeholder={type === 'k8s-daemonset' ? 'logging' : 'default'} /> +
+ {type === 'k8s-fluentbit-sidecar' && ( +
+ + setTargetDeployment(e.target.value)} placeholder="my-app" /> +
+ )} +
+ + )} + +
+ + setLogPath(e.target.value)} placeholder={type === 'host-promtail' ? '/var/log/*.log' : '/var/log/app/*.log'} /> +
+ + +
+ + +
+ + {/* List */} + {loading ? ( +
+ +
+ ) : collectors.length === 0 ? ( + + + +

暂无采集器

+

新建一个日志采集器，关联已部署的 Loki

+ + + ) : ( +
+ {collectors.map((c) => { + const meta = typeMeta[c.type] || { label: c.type, icon: RadioTower, color: 'bg-slate-100 text-slate-600' } + const Icon = meta.icon + return ( + + +
+
+ +
+

{c.name}

+

关联 Loki：{c.lokiDeployment?.name || '未知'}

+
+
+ + {statusLabel(c.status)} + +
+ +
+
+ 类型 + {meta.label} +
+ {c.host && ( +
+ 主机 + {c.host.name}（{c.host.hostIP}） +
+ )} + {c.cluster && ( +
+ 集群 + {c.cluster.name} +
+ )} + {c.targetDeployment && ( +
+ 目标 Deployment + {c.targetDeployment} +
+ )} + {c.namespace && ( +
+ 命名空间 + {c.namespace} +
+ )} + {c.logPath && ( +
+ 日志路径 + {c.logPath} +
+ )} + {c.statusMessage && ( +
{c.statusMessage}
+ )} +
+ +
+ + + +
+ + + ) + })} +
+ )} +
+ ) +} diff --git a/frontend/src/pages/Deployments.tsx b/frontend/src/pages/Deployments.tsx index e09a71c..9b857d2 100644 --- a/frontend/src/pages/Deployments.tsx +++ b/frontend/src/pages/Deployments.tsx @@ -12,6 +12,7 @@ import { Clock, PauseCircle, Activity, + ExternalLink, } from 'lucide-react' import { Button } from '@/components/ui/button' import { Input } from '@/components/ui/input' @@ -240,7 +241,7 @@ export default function Deployments() { {/* Header */}
-

部署管理

+

部署Loki

部署与管理 Grafana Loki 日志聚合服务

@@ -506,6 +507,18 @@ export default function Deployments() { 日志 + {d.endpoint && ( + + )}