diff --git a/backend/controllers/cluster.go b/backend/controllers/cluster.go index 7291fbe..dfa7cce 100644 --- a/backend/controllers/cluster.go +++ b/backend/controllers/cluster.go @@ -148,7 +148,7 @@ func (ctrl *ClusterController) Validate(c *gin.Context) { utils.BadRequest(c, "ID 无效") return } - go ctrl.service.Validate(uint(id)) + go ctrl.service.Validate(context.Background(), uint(id)) utils.Success(c, gin.H{"message": "已开始验证连接"}) } diff --git a/backend/controllers/deployment.go b/backend/controllers/deployment.go index 84aa7b0..a2a3dcb 100644 --- a/backend/controllers/deployment.go +++ b/backend/controllers/deployment.go @@ -217,7 +217,7 @@ func (ctrl *DeploymentController) GetUsage(c *gin.Context) { utils.BadRequest(c, "ID 无效") return } - usage, err := ctrl.service.GetUsage(uint(id)) + usage, err := ctrl.service.GetUsage(c.Request.Context(), uint(id)) if err != nil { utils.Error(c, 400, err.Error()) return diff --git a/backend/controllers/host.go b/backend/controllers/host.go index 4bb614f..7739d33 100644 --- a/backend/controllers/host.go +++ b/backend/controllers/host.go @@ -1,6 +1,7 @@ package controllers import ( + "context" "lokistack/dto" "lokistack/models" "lokistack/services" @@ -94,7 +95,7 @@ func (ctrl *HostController) Validate(c *gin.Context) { utils.BadRequest(c, "ID 无效") return } - go ctrl.service.Validate(uint(id)) + go ctrl.service.Validate(context.Background(), uint(id)) utils.Success(c, gin.H{"message": "validation started"}) } diff --git a/backend/repository/db.go b/backend/repository/db.go index 0ef44b7..4d61b55 100644 --- a/backend/repository/db.go +++ b/backend/repository/db.go @@ -264,11 +264,6 @@ func (d *JsonDB) fixUnreachableLogSourceURLs() { // LogSource : 查询页数据源下拉选不动(两个 option 同 value) // AlertRule : 规则列表出现两条同 id,编辑/删除操作命中错的那条 // NotifyChannel: 通知渠道覆盖错对象 -func (d *JsonDB) dedupeCollectionIDs() { - d.dedupeLogSourceIDs() - d.dedupeAlertRuleIDs() - d.dedupeNotifyChannelIDs() -} // dedupeAlertRuleIDs 重新编号重复的告警规则 ID。 // 同名规则(seed 重复执行导致)只保留第一条,其余删除。 @@ -539,6 +534,108 @@ func (d *JsonDB) migrateSchema() bool { return true } +// Reload 从磁盘重新加载内存态(备份恢复后调用): +// 成功解析才整体替换集合并重建 ID 序列;失败时保持原内存态并返回错误。 +// 必须在持有写锁的情况下调用。 +func (d *JsonDB) Reload() error { + data, err := os.ReadFile(d.dataFile) + if err != nil { + return err + } + var fresh JsonDB + if err := json.Unmarshal(data, &fresh); err != nil { + return err + } + d.Users = fresh.Users + d.Clusters = fresh.Clusters + d.Hosts = fresh.Hosts + d.Storages = fresh.Storages + d.Deployments = fresh.Deployments + d.Collectors = fresh.Collectors + d.Logs = fresh.Logs + d.Settings = fresh.Settings + d.LogSources = fresh.LogSources + d.AlertRules = fresh.AlertRules + d.AlertDeliveries = fresh.AlertDeliveries + d.NotifyChannels = fresh.NotifyChannels + d.LogPolicyGroups = fresh.LogPolicyGroups + d.AuditLogs = fresh.AuditLogs + d.DailyVolumeSnapshots = fresh.DailyVolumeSnapshots + d.SchemaVersion = fresh.SchemaVersion + d.sequences = map[string]uint{} + for _, u := range d.Users { + if u.ID > d.sequences["users"] { + d.sequences["users"] = u.ID + } + } + for _, cl := range d.Clusters { + if cl.ID > d.sequences["clusters"] { + d.sequences["clusters"] = cl.ID + } + } + 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 + } + } + for _, g := range d.LogPolicyGroups { + if g.ID > d.sequences["logPolicyGroups"] { + d.sequences["logPolicyGroups"] = g.ID + } + } + for _, s := range d.LogSources { + if s.ID > d.sequences["logSources"] { + d.sequences["logSources"] = s.ID + } + } + for _, ch := range d.NotifyChannels { + if ch.ID > d.sequences["notifyChannels"] { + d.sequences["notifyChannels"] = ch.ID + } + } + for _, r := range d.AlertRules { + if r.ID > d.sequences["alertRules"] { + d.sequences["alertRules"] = r.ID + } + } + for _, de := range d.AlertDeliveries { + if de.ID > d.sequences["alertDeliveries"] { + d.sequences["alertDeliveries"] = de.ID + } + } + for _, a := range d.AuditLogs { + if a.ID > d.sequences["auditLogs"] { + d.sequences["auditLogs"] = a.ID + } + } + for _, v := range d.DailyVolumeSnapshots { + if v.ID > d.sequences["dailyVolumeSnapshots"] { + d.sequences["dailyVolumeSnapshots"] = v.ID + } + } + return nil +} + // migrateLegacyDataFile 首次从历史命名(opsxloki.json / lokideploy.json)加载后, // 立即升级到 lokistack.json:数据原子写入新文件后删除旧文件。 // 不迁移的话回退链会让旧文件名(旧品牌残留)一直被沿用下去。 diff --git a/backend/services/backup.go b/backend/services/backup.go index f429e2b..06d5192 100644 --- a/backend/services/backup.go +++ b/backend/services/backup.go @@ -281,7 +281,12 @@ func encryptionKeyFingerprint() string { // ApplyStagedBackup 把暂存文件应用为正式 DB(调用方负责重启进程)。 func ApplyStagedBackup() (string, error) { - staged := filepath.Join(dataDirPath(), ".restore-pending", "opsxloki.json") + restoreDir := filepath.Join(dataDirPath(), ".restore-pending") + staged := filepath.Join(restoreDir, "lokistack.json") + if _, err := os.Stat(staged); err != nil { + // 兼容旧版暂存文件名 + staged = filepath.Join(restoreDir, "opsxloki.json") + } b, err := os.ReadFile(staged) if err != nil { return "", fmt.Errorf("没有待应用的恢复文件: %w", err) @@ -305,6 +310,16 @@ func ApplyStagedBackup() (string, error) { return "", err } _ = os.Remove(staged) + // 关键:立即把内存态切换为恢复后的数据。否则「替换文件 → 重启」之间, + // 任何一次防抖落盘都会用旧内存态把恢复文件覆盖回去,恢复静默失效。 + if db := repository.GetDB(); db != nil { + db.Lock() + if err := db.Reload(); err != nil { + db.Unlock() + return dbPath, fmt.Errorf("恢复文件已生效但内存重载失败(%v),请立即重启进程", err) + } + db.Unlock() + } return dbPath, nil } diff --git a/backend/services/cluster.go b/backend/services/cluster.go index 555a67c..e6b9009 100644 --- a/backend/services/cluster.go +++ b/backend/services/cluster.go @@ -107,7 +107,7 @@ func (s *ClusterService) Create(name, description, configType, apiServer, kubeco db.Unlock() db.RequestSave() - go s.Validate(cluster.ID) + go s.Validate(context.Background(), cluster.ID) return &cluster, nil } @@ -222,12 +222,12 @@ func (s *ClusterService) Delete(id uint) error { return utils.ErrNotFound } -func (s *ClusterService) Validate(id uint) { +func (s *ClusterService) Validate(ctx context.Context, id uint) { cluster, err := s.Get(id) if err != nil { return } - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() info, err := GetClusterInfo(ctx, cluster) diff --git a/backend/services/collector.go b/backend/services/collector.go index 7caafef..75a043a 100644 --- a/backend/services/collector.go +++ b/backend/services/collector.go @@ -4,7 +4,9 @@ import ( "context" "encoding/json" "fmt" + "log" "strings" + "sync" "time" "lokistack/models" @@ -12,10 +14,36 @@ import ( "lokistack/utils" ) -type CollectorService struct{} +type CollectorService struct { + // opsMu 保护 opsInflight:同一采集器的安装/卸载互斥去重。 + // 采集器是 fire-and-forget(接口立即返回),没有去重时连点「部署」 + // 会对同一主机并发 docker rm -f + docker run,容器名冲突/配置互踩。 + opsMu sync.Mutex + opsInflight map[uint]bool +} func NewCollectorService() *CollectorService { - return &CollectorService{} + return &CollectorService{opsInflight: map[uint]bool{}} +} + +// beginOp 声明开始一次安装/卸载;已在进行返回 false(直接忽略重复触发) +func (s *CollectorService) beginOp(id uint) bool { + s.opsMu.Lock() + defer s.opsMu.Unlock() + if s.opsInflight == nil { + s.opsInflight = map[uint]bool{} + } + if s.opsInflight[id] { + return false + } + s.opsInflight[id] = true + return true +} + +func (s *CollectorService) endOp(id uint) { + s.opsMu.Lock() + delete(s.opsInflight, id) + s.opsMu.Unlock() } type CreateCollectorRequest struct { @@ -475,17 +503,23 @@ func (s *CollectorService) Delete(id uint) error { } db.Unlock() db.RequestSave() - // 异步卸载(用拷贝的 collector,避免删除后取不到) + // 异步卸载(用拷贝的 collector,避免删除后取不到;失败留日志,远端不留孤儿容器也无提示) go func() { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() - NewCollectorDeployer(collector, collector.LokiDeployment, nil).Uninstall(ctx) + if err := NewCollectorDeployer(collector, collector.LokiDeployment, nil).Uninstall(ctx); err != nil { + log.Printf("[collector] 删除后卸载 %q(目标 %s)失败: %v", collector.Name, collector.Name, err) + } }() return nil } // Install 异步部署采集器 func (s *CollectorService) Install(id uint) { + if !s.beginOp(id) { + return // 同一采集器的安装/卸载正在进行,忽略重复触发 + } + defer s.endOp(id) collector, err := s.Get(id) if err != nil { return @@ -522,6 +556,10 @@ func (s *CollectorService) Install(id uint) { // Uninstall 异步卸载采集器 func (s *CollectorService) Uninstall(id uint) { + if !s.beginOp(id) { + return + } + defer s.endOp(id) collector, err := s.Get(id) if err != nil { return diff --git a/backend/services/deployment_usage.go b/backend/services/deployment_usage.go index 0b1309b..92ad693 100644 --- a/backend/services/deployment_usage.go +++ b/backend/services/deployment_usage.go @@ -69,18 +69,18 @@ type podMetricsResponse struct { } // GetUsage 采集某个 Loki 部署的资源用量(按部署模式分派)。 -func (s *DeploymentService) GetUsage(id uint) (*ResourceUsage, error) { +func (s *DeploymentService) GetUsage(ctx context.Context, id uint) (*ResourceUsage, error) { d, err := s.Get(id) if err != nil { return nil, err } switch d.Mode { case "k8s-yaml": - return s.k8sUsage(d) + return s.k8sUsage(ctx, d) case "host-docker": - return s.dockerUsage(d) + return s.dockerUsage(ctx, d) case "host-binary": - return s.processUsage(d) + return s.processUsage(ctx, d) default: return nil, fmt.Errorf("不支持的部署模式:%s", d.Mode) } @@ -88,7 +88,7 @@ func (s *DeploymentService) GetUsage(id uint) (*ResourceUsage, error) { // ---------- Kubernetes ---------- -func (s *DeploymentService) k8sUsage(d *models.LokiDeployment) (*ResourceUsage, error) { +func (s *DeploymentService) k8sUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Cluster == nil { return nil, fmt.Errorf("部署未关联 Kubernetes 集群") } @@ -96,7 +96,7 @@ func (s *DeploymentService) k8sUsage(d *models.LokiDeployment) (*ResourceUsage, if err != nil { return nil, err } - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() ns := d.Namespace @@ -192,7 +192,7 @@ func (s *DeploymentService) applyPrometheus(u *ResourceUsage) { // ---------- host-docker ---------- -func (s *DeploymentService) dockerUsage(d *models.LokiDeployment) (*ResourceUsage, error) { +func (s *DeploymentService) dockerUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Host == nil { return nil, fmt.Errorf("部署未关联主机") } @@ -202,7 +202,7 @@ func (s *DeploymentService) dockerUsage(d *models.LokiDeployment) (*ResourceUsag } defer client.Close() - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + ctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() name := dockerContainerName(d.ID) @@ -254,7 +254,7 @@ func (s *DeploymentService) dockerUsage(d *models.LokiDeployment) (*ResourceUsag // ---------- host-binary ---------- -func (s *DeploymentService) processUsage(d *models.LokiDeployment) (*ResourceUsage, error) { +func (s *DeploymentService) processUsage(ctx context.Context, d *models.LokiDeployment) (*ResourceUsage, error) { if d.Host == nil { return nil, fmt.Errorf("部署未关联主机") } @@ -264,7 +264,7 @@ func (s *DeploymentService) processUsage(d *models.LokiDeployment) (*ResourceUsa } defer client.Close() - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() usage := &ResourceUsage{Mode: d.Mode, Source: "binary", Pod: "loki", Node: d.Host.HostIP} diff --git a/backend/services/host.go b/backend/services/host.go index 256bc14..d5b74ed 100644 --- a/backend/services/host.go +++ b/backend/services/host.go @@ -88,7 +88,7 @@ func (s *HostService) Create(name, hostIP string, port int, user, authType, pass db.Unlock() db.RequestSave() - go s.Validate(host.ID) + go s.Validate(context.Background(), host.ID) return &host, nil } @@ -194,12 +194,12 @@ func (s *HostService) Delete(id uint) error { return utils.ErrNotFound } -func (s *HostService) Validate(id uint) { +func (s *HostService) Validate(ctx context.Context, id uint) { host, err := s.Get(id) if err != nil { return } - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() client, err := NewSSHClientWithAuth(HostSSHAuth(host))