mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-29 05:56:38 +08:00
fix: 修复应用日志异常膨胀
This commit is contained in:
@@ -22,6 +22,10 @@ sidebar: false
|
||||
|
||||
### 修复
|
||||
|
||||
- 修复应用日志异常膨胀:Agent 配置同步加锁避免并发重复上报,成功且版本/checksum 未变时跳过重复 apply 日志;Server 入库前对相同成功记录去重;Flared 配置未变更时不再上报 apply 日志。
|
||||
|
||||
- 修复应用日志页「清空」无效果:原按钮仅重置筛选;新增「清空日志」入口并对接 `/api/v1/d/apply-logs/cleanup`,支持确认后删除全部记录。
|
||||
|
||||
- 配置版本列表按 `created_at` 倒序展示,最新发布版本固定显示在列表顶部。
|
||||
|
||||
- 修复 WAF 规则组保存/绑定网站时报 `of_waf_rule_group_bindings_pkey` 冲突:PostgreSQL 在迁移导入显式 ID 后同步绑定表序列,并在写入前自动校正序列。
|
||||
|
||||
@@ -3,11 +3,22 @@
|
||||
import {useCallback, useEffect, useMemo, useState} from "react"
|
||||
import Link from "next/link"
|
||||
import {useSearchParams} from "next/navigation"
|
||||
import {ClipboardList, Eye, RefreshCw, Search} from "lucide-react"
|
||||
import {ClipboardList, Eye, RefreshCw, Search, Trash2} from "lucide-react"
|
||||
import {toast} from "sonner"
|
||||
|
||||
import {EmptyStateWithBorder} from "@/components/layout/empty"
|
||||
import {ErrorInline} from "@/components/layout/error"
|
||||
import {LoadingStateWithBorder} from "@/components/layout/loading"
|
||||
import {
|
||||
AlertDialog,
|
||||
AlertDialogAction,
|
||||
AlertDialogCancel,
|
||||
AlertDialogContent,
|
||||
AlertDialogDescription,
|
||||
AlertDialogFooter,
|
||||
AlertDialogHeader,
|
||||
AlertDialogTitle,
|
||||
} from "@/components/ui/alert-dialog"
|
||||
import {Badge} from "@/components/ui/badge"
|
||||
import {Button} from "@/components/ui/button"
|
||||
import {Input} from "@/components/ui/input"
|
||||
@@ -78,6 +89,8 @@ export function ApplyLogsPageClient() {
|
||||
|
||||
const [selectedLog, setSelectedLog] = useState<ApplyLogItem | null>(null)
|
||||
const [detailOpen, setDetailOpen] = useState(false)
|
||||
const [cleanupOpen, setCleanupOpen] = useState(false)
|
||||
const [cleaning, setCleaning] = useState(false)
|
||||
|
||||
const summary = useMemo(() => {
|
||||
const nodeIds = new Set(rows.map((item) => item.node_id))
|
||||
@@ -126,12 +139,31 @@ export function ApplyLogsPageClient() {
|
||||
setNodeFilter(nodeFilterInput.trim())
|
||||
}
|
||||
|
||||
const handleReset = () => {
|
||||
const handleResetFilters = () => {
|
||||
setNodeFilterInput("")
|
||||
setNodeFilter("")
|
||||
setPageNo(1)
|
||||
}
|
||||
|
||||
const handleCleanupLogs = async () => {
|
||||
setCleaning(true)
|
||||
try {
|
||||
const result = await ApplyLogService.cleanup({ delete_all: true })
|
||||
toast.success("应用日志已清空", {
|
||||
description: `共删除 ${result.deleted_count} 条记录`,
|
||||
})
|
||||
setCleanupOpen(false)
|
||||
setPageNo(1)
|
||||
await fetchLogs()
|
||||
} catch (err) {
|
||||
toast.error("清空应用日志失败", {
|
||||
description: err instanceof Error ? err.message : "未知错误",
|
||||
})
|
||||
} finally {
|
||||
setCleaning(false)
|
||||
}
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="py-6 px-1 space-y-6">
|
||||
<div className="flex flex-col gap-4 sm:flex-row sm:items-center sm:justify-between">
|
||||
@@ -150,6 +182,15 @@ export function ApplyLogsPageClient() {
|
||||
<RefreshCw className={`size-3.5 mr-1 ${loading ? "animate-spin" : ""}`} />
|
||||
刷新
|
||||
</Button>
|
||||
<Button
|
||||
variant="outline"
|
||||
size="sm"
|
||||
onClick={() => setCleanupOpen(true)}
|
||||
disabled={loading || cleaning || total === 0}
|
||||
>
|
||||
<Trash2 className="size-3.5 mr-1" />
|
||||
清空日志
|
||||
</Button>
|
||||
<Button variant="outline" size="sm" asChild>
|
||||
<Link href="/nodes">返回节点</Link>
|
||||
</Button>
|
||||
@@ -213,8 +254,8 @@ export function ApplyLogsPageClient() {
|
||||
<Button size="sm" onClick={handleSearch}>
|
||||
筛选
|
||||
</Button>
|
||||
<Button variant="outline" size="sm" onClick={handleReset}>
|
||||
清空
|
||||
<Button variant="outline" size="sm" onClick={handleResetFilters}>
|
||||
重置筛选
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
@@ -319,6 +360,30 @@ export function ApplyLogsPageClient() {
|
||||
open={detailOpen}
|
||||
onOpenChange={setDetailOpen}
|
||||
/>
|
||||
|
||||
<AlertDialog open={cleanupOpen} onOpenChange={setCleanupOpen}>
|
||||
<AlertDialogContent>
|
||||
<AlertDialogHeader>
|
||||
<AlertDialogTitle>确认清空应用日志?</AlertDialogTitle>
|
||||
<AlertDialogDescription>
|
||||
该操作将删除全部 {total} 条应用日志记录,且不可撤销。清空后节点仍会照常上报新的应用结果。
|
||||
</AlertDialogDescription>
|
||||
</AlertDialogHeader>
|
||||
<AlertDialogFooter>
|
||||
<AlertDialogCancel disabled={cleaning}>取消</AlertDialogCancel>
|
||||
<AlertDialogAction
|
||||
className="bg-destructive text-destructive-foreground hover:bg-destructive/90"
|
||||
disabled={cleaning}
|
||||
onClick={(event) => {
|
||||
event.preventDefault()
|
||||
void handleCleanupLogs()
|
||||
}}
|
||||
>
|
||||
{cleaning ? "清空中..." : "确认清空"}
|
||||
</AlertDialogAction>
|
||||
</AlertDialogFooter>
|
||||
</AlertDialogContent>
|
||||
</AlertDialog>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"log/slog"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
openrestyrender "github.com/Rain-kl/Wavelet/pkg/render/openresty"
|
||||
|
||||
@@ -49,6 +50,7 @@ type Service struct {
|
||||
nginxManager NginxManager
|
||||
stateStore *state.Store
|
||||
pagesDir string
|
||||
syncMu sync.Mutex
|
||||
}
|
||||
|
||||
// SetPagesDir sets the local directory used for pages deployment packages.
|
||||
@@ -76,6 +78,9 @@ func (s *Service) SyncOnStartup(ctx context.Context, target *protocol.ActiveConf
|
||||
}
|
||||
|
||||
func (s *Service) sync(ctx context.Context, startup bool, target *protocol.ActiveConfigMeta) error {
|
||||
s.syncMu.Lock()
|
||||
defer s.syncMu.Unlock()
|
||||
|
||||
mode := syncMode(startup)
|
||||
snapshot, currentChecksum, err := s.loadSyncState()
|
||||
if err != nil {
|
||||
@@ -94,6 +99,9 @@ func (s *Service) sync(ctx context.Context, startup bool, target *protocol.Activ
|
||||
|
||||
// ForceSyncOnce clears any blocked target state then unconditionally fetches and applies the active config.
|
||||
func (s *Service) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
|
||||
s.syncMu.Lock()
|
||||
defer s.syncMu.Unlock()
|
||||
|
||||
snapshot, err := s.stateStore.Load()
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -161,12 +169,24 @@ func (s *Service) applyRenderedConfig(ctx context.Context, mode string, snapshot
|
||||
mainConfigChecksum := checksumString(rendered.mainConfig)
|
||||
routeConfigChecksum := checksumString(rendered.routeConfig)
|
||||
slog.Info("applying new openresty config", "mode", mode, "from_version", snapshot.CurrentVersion, "to_version", config.Version, "old_checksum", currentChecksum, "new_checksum", config.Checksum)
|
||||
alreadySynced := snapshotMatchesTarget(snapshot, config.Version, config.Checksum)
|
||||
outcome, message := normalizeApplyOutcome(s.nginxManager.Apply(ctx, rendered.mainConfig, rendered.routeConfig, rendered.supportFiles))
|
||||
applyResult := updateSnapshotFromApplyOutcome(mode, snapshot, config, outcome, message)
|
||||
|
||||
if err := s.stateStore.Save(snapshot); err != nil {
|
||||
return err
|
||||
}
|
||||
if !shouldReportApplyLog(alreadySynced, applyResult.reportResult) {
|
||||
slog.Debug("skipping duplicate apply log report", "version", config.Version, "checksum", config.Checksum, "result", applyResult.reportResult)
|
||||
if applyResult.reportResult == ApplyResultFailed {
|
||||
return outcomeError(config.Version, applyResult.message)
|
||||
}
|
||||
if err := s.syncReferencedWAFIPGroups(ctx, rendered.supportFiles); err != nil {
|
||||
slog.Error("sync referenced waf ip groups failed", "version", config.Version, "error", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err := s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{
|
||||
NodeID: snapshot.NodeID,
|
||||
Version: config.Version,
|
||||
|
||||
@@ -479,6 +479,44 @@ func TestSyncOnceReportsNoopWhenVersionChangesButChecksumMatches(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncOnStartupSkipsDuplicateSuccessReportWhenStateAlreadySynced(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
config: protocol.ActiveConfigResponse{
|
||||
Version: "20260309-003",
|
||||
Checksum: "checksum-3",
|
||||
SourceConfigJSON: testSourceConfigJSON("auto", 80),
|
||||
CreatedAt: time.Now().Format(time.RFC3339),
|
||||
},
|
||||
}
|
||||
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
|
||||
nodeID, err := stateStore.EnsureNodeID()
|
||||
if err != nil {
|
||||
t.Fatalf("EnsureNodeID failed: %v", err)
|
||||
}
|
||||
if err = stateStore.Save(&state.Snapshot{
|
||||
NodeID: nodeID,
|
||||
CurrentVersion: "20260309-003",
|
||||
CurrentChecksum: "checksum-3",
|
||||
}); err != nil {
|
||||
t.Fatalf("failed to seed state: %v", err)
|
||||
}
|
||||
|
||||
manager := &fakeManager{currentChecksum: "checksum-3"}
|
||||
service := New(client, manager, stateStore)
|
||||
if err = service.SyncOnStartup(context.Background(), &protocol.ActiveConfigMeta{
|
||||
Version: "20260309-003",
|
||||
Checksum: "checksum-3",
|
||||
}); err != nil {
|
||||
t.Fatalf("SyncOnStartup failed: %v", err)
|
||||
}
|
||||
if len(client.reports) != 0 {
|
||||
t.Fatalf("expected startup sync to skip duplicate success report, got %+v", client.reports)
|
||||
}
|
||||
if len(manager.applyMainContents) != 1 {
|
||||
t.Fatal("expected startup sync to still refresh local config once")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncOnceDoesNotRepeatNoopReportWhenStateAlreadyMatches(t *testing.T) {
|
||||
client := &fakeClient{}
|
||||
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
|
||||
|
||||
@@ -203,3 +203,18 @@ func updateSnapshotFromApplyOutcome(mode string, snapshot *state.Snapshot, confi
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func snapshotMatchesTarget(snapshot *state.Snapshot, version string, checksum string) bool {
|
||||
if snapshot == nil {
|
||||
return false
|
||||
}
|
||||
return strings.TrimSpace(snapshot.CurrentVersion) == strings.TrimSpace(version) &&
|
||||
strings.TrimSpace(snapshot.CurrentChecksum) == strings.TrimSpace(checksum)
|
||||
}
|
||||
|
||||
func shouldReportApplyLog(alreadySynced bool, result string) bool {
|
||||
if result != ApplyResultSuccess {
|
||||
return true
|
||||
}
|
||||
return !alreadySynced
|
||||
}
|
||||
|
||||
@@ -95,12 +95,13 @@ func (m *Manager) GetCurrentConfigChecksum() string {
|
||||
}
|
||||
|
||||
// UpdateConfig reconciles running frpc processes with the latest tunnel configuration.
|
||||
func (m *Manager) UpdateConfig(ctx context.Context, newConfig *service.FlaredTunnelConfigResponse) error {
|
||||
// The returned bool indicates whether the active config version or checksum changed.
|
||||
func (m *Manager) UpdateConfig(ctx context.Context, newConfig *service.FlaredTunnelConfigResponse) (bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if newConfig == nil {
|
||||
return nil
|
||||
return false, nil
|
||||
}
|
||||
|
||||
versionChanged := newConfig.Version != m.currentVersion || newConfig.Checksum != m.currentChecksum
|
||||
@@ -111,7 +112,7 @@ func (m *Manager) UpdateConfig(ctx context.Context, newConfig *service.FlaredTun
|
||||
}
|
||||
|
||||
if err := os.MkdirAll(m.cfg.DataDir, dataDirPerm); err != nil {
|
||||
return fmt.Errorf("create data dir failed: %w", err)
|
||||
return false, fmt.Errorf("create data dir failed: %w", err)
|
||||
}
|
||||
|
||||
activeRelays := make(map[string]struct{})
|
||||
@@ -155,9 +156,9 @@ func (m *Manager) UpdateConfig(ctx context.Context, newConfig *service.FlaredTun
|
||||
if versionChanged {
|
||||
m.currentVersion = newConfig.Version
|
||||
m.currentChecksum = newConfig.Checksum
|
||||
return m.saveState()
|
||||
return true, m.saveState()
|
||||
}
|
||||
return nil
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func (m *Manager) restartProcess(ctx context.Context, relayID string, configPath string) {
|
||||
|
||||
@@ -119,7 +119,7 @@ func TestStartProcessSuccess(t *testing.T) {
|
||||
Proxies: nil,
|
||||
}
|
||||
|
||||
err := m.UpdateConfig(context.Background(), newConfig)
|
||||
_, err := m.UpdateConfig(context.Background(), newConfig)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to UpdateConfig: %v", err)
|
||||
}
|
||||
@@ -160,7 +160,7 @@ func TestStartProcessFailureAndBackoff(t *testing.T) {
|
||||
Proxies: nil,
|
||||
}
|
||||
|
||||
_ = m.UpdateConfig(context.Background(), newConfig)
|
||||
_, _ = m.UpdateConfig(context.Background(), newConfig)
|
||||
|
||||
assertStatusEventually(t, m, "relay-1", "error", 4*time.Second)
|
||||
|
||||
@@ -208,7 +208,7 @@ func TestUnexpectedExit0CPUProtection(t *testing.T) {
|
||||
Proxies: nil,
|
||||
}
|
||||
|
||||
_ = m.UpdateConfig(context.Background(), newConfig)
|
||||
_, _ = m.UpdateConfig(context.Background(), newConfig)
|
||||
|
||||
assertStatusEventually(t, m, "relay-1", "stopped", 4*time.Second)
|
||||
|
||||
@@ -249,7 +249,7 @@ func TestBackoffReset(t *testing.T) {
|
||||
Proxies: nil,
|
||||
}
|
||||
|
||||
_ = m.UpdateConfig(context.Background(), newConfig)
|
||||
_, _ = m.UpdateConfig(context.Background(), newConfig)
|
||||
|
||||
// Wait to crash
|
||||
assertStatusEventually(t, m, "relay-1", "error", 4*time.Second)
|
||||
@@ -323,7 +323,7 @@ func TestUpdateConfigKillsOrphanProcessBeforeRestart(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
if err := m.UpdateConfig(context.Background(), newConfig); err != nil {
|
||||
if _, err := m.UpdateConfig(context.Background(), newConfig); err != nil {
|
||||
t.Fatalf("failed to UpdateConfig: %v", err)
|
||||
}
|
||||
|
||||
@@ -361,7 +361,7 @@ func TestStopCancelsRunningProcesses(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
if err := m.UpdateConfig(context.Background(), newConfig); err != nil {
|
||||
if _, err := m.UpdateConfig(context.Background(), newConfig); err != nil {
|
||||
t.Fatalf("failed to UpdateConfig: %v", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -68,19 +68,24 @@ func (s *Service) doSync(ctx context.Context) {
|
||||
|
||||
// 不在 sync 层做版本早退,由 frpcManager.UpdateConfig 负责判断。
|
||||
// 原因:重启后进程全部消失,即使版本/checksum 未变,仍需重新拉起 frpc 进程。
|
||||
err = s.frpcManager.UpdateConfig(ctx, configResp)
|
||||
|
||||
result := "success"
|
||||
message := "apply success"
|
||||
configChanged, err := s.frpcManager.UpdateConfig(ctx, configResp)
|
||||
if err != nil {
|
||||
result = "failed"
|
||||
message = err.Error()
|
||||
slog.Error("failed to apply tunnel config", "error", err)
|
||||
} else {
|
||||
slog.Info("tunnel config applied successfully", "version", configResp.Version)
|
||||
s.reportApplyLog(ctx, configResp, "failed", err.Error())
|
||||
return
|
||||
}
|
||||
if configChanged {
|
||||
slog.Info("tunnel config applied successfully", "version", configResp.Version)
|
||||
s.reportApplyLog(ctx, configResp, "success", "apply success")
|
||||
return
|
||||
}
|
||||
slog.Debug("tunnel config unchanged, skipping apply log report", "version", configResp.Version)
|
||||
}
|
||||
|
||||
// Report apply log
|
||||
func (s *Service) reportApplyLog(ctx context.Context, configResp *service.FlaredTunnelConfigResponse, result string, message string) {
|
||||
if configResp == nil {
|
||||
return
|
||||
}
|
||||
logPayload := service.ApplyLogPayload{
|
||||
Version: configResp.Version,
|
||||
Result: result,
|
||||
|
||||
@@ -181,6 +181,17 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
return nil, errors.New(errInvalidApplyResult)
|
||||
}
|
||||
|
||||
latest, err := model.GetLatestOpenFlareApplyLogByNodeID(ctx, payload.NodeID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if model.IsRepeatSuccessApplyLog(latest, payload.Version, payload.Checksum, payload.Result) {
|
||||
if err := updateNodeFromApplyLog(ctx, payload, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return latest, nil
|
||||
}
|
||||
|
||||
log := &model.OpenFlareApplyLog{
|
||||
NodeID: payload.NodeID,
|
||||
Version: payload.Version,
|
||||
@@ -198,23 +209,11 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
return nil, errors.New("database not initialized")
|
||||
}
|
||||
|
||||
err := conn.Transaction(func(tx *gorm.DB) error {
|
||||
record := &model.OpenFlareNode{}
|
||||
if err := tx.Where("node_id = ?", payload.NodeID).First(record).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
record.Status = nodeStatusOnline
|
||||
record.LastSeenAt = &now
|
||||
if payload.Result == applyResultOK {
|
||||
record.CurrentVersion = payload.Version
|
||||
record.LastError = ""
|
||||
} else {
|
||||
record.LastError = payload.Message
|
||||
}
|
||||
err = conn.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(log).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Model(record).Select("status", "last_seen_at", "current_version", "last_error").Updates(record).Error
|
||||
return updateNodeFromApplyLogTx(tx, payload, now)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -222,6 +221,32 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
return log, nil
|
||||
}
|
||||
|
||||
func updateNodeFromApplyLog(ctx context.Context, payload ApplyLogPayload, now time.Time) error {
|
||||
conn := db.DB(ctx)
|
||||
if conn == nil {
|
||||
return errors.New("database not initialized")
|
||||
}
|
||||
return conn.Transaction(func(tx *gorm.DB) error {
|
||||
return updateNodeFromApplyLogTx(tx, payload, now)
|
||||
})
|
||||
}
|
||||
|
||||
func updateNodeFromApplyLogTx(tx *gorm.DB, payload ApplyLogPayload, now time.Time) error {
|
||||
record := &model.OpenFlareNode{}
|
||||
if err := tx.Where("node_id = ?", payload.NodeID).First(record).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
record.Status = nodeStatusOnline
|
||||
record.LastSeenAt = &now
|
||||
if payload.Result == applyResultOK {
|
||||
record.CurrentVersion = payload.Version
|
||||
record.LastError = ""
|
||||
} else {
|
||||
record.LastError = payload.Message
|
||||
}
|
||||
return tx.Model(record).Select("status", "last_seen_at", "current_version", "last_error").Updates(record).Error
|
||||
}
|
||||
|
||||
// ValidateDiscoveryToken delegates to the node package discovery token helper.
|
||||
func ValidateDiscoveryToken(ctx context.Context, token string) error {
|
||||
return node.ValidateDiscoveryToken(ctx, token)
|
||||
|
||||
@@ -170,6 +170,17 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
return nil, errors.New("result 仅支持 success、warning 或 failed")
|
||||
}
|
||||
|
||||
latest, err := model.GetLatestOpenFlareApplyLogByNodeID(ctx, payload.NodeID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if model.IsRepeatSuccessApplyLog(latest, payload.Version, payload.Checksum, payload.Result) {
|
||||
if err := updateFlaredNodeFromApplyLog(ctx, payload, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return latest, nil
|
||||
}
|
||||
|
||||
log := &model.OpenFlareApplyLog{
|
||||
NodeID: payload.NodeID,
|
||||
Version: payload.Version,
|
||||
@@ -182,24 +193,11 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
CreatedAt: now,
|
||||
}
|
||||
|
||||
err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var node model.OpenFlareNode
|
||||
if err := tx.Where("node_id = ?", payload.NodeID).First(&node).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
node.Status = nodeStatusOnline
|
||||
lastSeen := now
|
||||
node.LastSeenAt = &lastSeen
|
||||
if payload.Result == applyResultOK {
|
||||
node.CurrentVersion = payload.Version
|
||||
node.LastError = ""
|
||||
} else {
|
||||
node.LastError = payload.Message
|
||||
}
|
||||
err = db.DB(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(log).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Model(&node).Select("status", "last_seen_at", "current_version", "last_error").Updates(&node).Error
|
||||
return updateFlaredNodeFromApplyLogTx(tx, payload, now)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -207,6 +205,33 @@ func ReportApplyLog(ctx context.Context, payload ApplyLogPayload) (*model.OpenFl
|
||||
return log, nil
|
||||
}
|
||||
|
||||
func updateFlaredNodeFromApplyLog(ctx context.Context, payload ApplyLogPayload, now time.Time) error {
|
||||
conn := db.DB(ctx)
|
||||
if conn == nil {
|
||||
return errors.New("database not initialized")
|
||||
}
|
||||
return conn.Transaction(func(tx *gorm.DB) error {
|
||||
return updateFlaredNodeFromApplyLogTx(tx, payload, now)
|
||||
})
|
||||
}
|
||||
|
||||
func updateFlaredNodeFromApplyLogTx(tx *gorm.DB, payload ApplyLogPayload, now time.Time) error {
|
||||
var node model.OpenFlareNode
|
||||
if err := tx.Where("node_id = ?", payload.NodeID).First(&node).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
node.Status = nodeStatusOnline
|
||||
lastSeen := now
|
||||
node.LastSeenAt = &lastSeen
|
||||
if payload.Result == applyResultOK {
|
||||
node.CurrentVersion = payload.Version
|
||||
node.LastError = ""
|
||||
} else {
|
||||
node.LastError = payload.Message
|
||||
}
|
||||
return tx.Model(&node).Select("status", "last_seen_at", "current_version", "last_error").Updates(&node).Error
|
||||
}
|
||||
|
||||
func getActiveConfigVersion(ctx context.Context) (*configVersionRow, error) {
|
||||
conn := db.DB(ctx)
|
||||
if conn == nil {
|
||||
|
||||
@@ -6,6 +6,7 @@ package model
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
@@ -83,6 +84,39 @@ func CountOpenFlareApplyLogs(ctx context.Context, nodeID string) (int64, error)
|
||||
return total, nil
|
||||
}
|
||||
|
||||
// GetLatestOpenFlareApplyLogByNodeID returns the most recent apply log for a node.
|
||||
func GetLatestOpenFlareApplyLogByNodeID(ctx context.Context, nodeID string) (*OpenFlareApplyLog, error) {
|
||||
nodeID = strings.TrimSpace(nodeID)
|
||||
if nodeID == "" {
|
||||
return nil, errors.New("node_id is required")
|
||||
}
|
||||
|
||||
conn := db.DB(ctx)
|
||||
if conn == nil {
|
||||
return nil, errors.New(errDatabaseNotInitialized)
|
||||
}
|
||||
|
||||
var log OpenFlareApplyLog
|
||||
err := conn.Where("node_id = ?", nodeID).Order("id desc").First(&log).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &log, nil
|
||||
}
|
||||
|
||||
// IsRepeatSuccessApplyLog reports whether the payload repeats an already-recorded success entry.
|
||||
func IsRepeatSuccessApplyLog(latest *OpenFlareApplyLog, version, checksum, result string) bool {
|
||||
if latest == nil || result != "success" {
|
||||
return false
|
||||
}
|
||||
return latest.Result == "success" &&
|
||||
strings.TrimSpace(latest.Version) == strings.TrimSpace(version) &&
|
||||
strings.TrimSpace(latest.Checksum) == strings.TrimSpace(checksum)
|
||||
}
|
||||
|
||||
// GetLatestOpenFlareApplyLogsByNodeIDs returns the latest apply log per node id.
|
||||
func GetLatestOpenFlareApplyLogsByNodeIDs(ctx context.Context, nodeIDs []string) (map[string]*OpenFlareApplyLog, error) {
|
||||
result := make(map[string]*OpenFlareApplyLog)
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/glebarez/sqlite"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func setupApplyLogModelTestDB(t *testing.T) func() {
|
||||
t.Helper()
|
||||
|
||||
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
|
||||
DisableForeignKeyConstraintWhenMigrating: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, sqliteDB.AutoMigrate(&OpenFlareApplyLog{}))
|
||||
|
||||
db.SetDB(sqliteDB)
|
||||
return func() {
|
||||
db.SetDB(nil)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsRepeatSuccessApplyLog(t *testing.T) {
|
||||
latest := &OpenFlareApplyLog{
|
||||
Version: "20260615-001",
|
||||
Checksum: "checksum-a",
|
||||
Result: "success",
|
||||
}
|
||||
|
||||
assert.True(t, IsRepeatSuccessApplyLog(latest, "20260615-001", "checksum-a", "success"))
|
||||
assert.False(t, IsRepeatSuccessApplyLog(latest, "20260615-002", "checksum-a", "success"))
|
||||
assert.False(t, IsRepeatSuccessApplyLog(latest, "20260615-001", "checksum-b", "success"))
|
||||
assert.False(t, IsRepeatSuccessApplyLog(latest, "20260615-001", "checksum-a", "failed"))
|
||||
assert.False(t, IsRepeatSuccessApplyLog(nil, "20260615-001", "checksum-a", "success"))
|
||||
}
|
||||
|
||||
func TestGetLatestOpenFlareApplyLogByNodeID(t *testing.T) {
|
||||
cleanup := setupApplyLogModelTestDB(t)
|
||||
defer cleanup()
|
||||
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
require.NoError(t, db.DB(ctx).Create(&OpenFlareApplyLog{
|
||||
NodeID: "node-1",
|
||||
Version: "v1",
|
||||
Result: "success",
|
||||
Checksum: "checksum-1",
|
||||
CreatedAt: now.Add(-time.Hour),
|
||||
}).Error)
|
||||
require.NoError(t, db.DB(ctx).Create(&OpenFlareApplyLog{
|
||||
NodeID: "node-1",
|
||||
Version: "v2",
|
||||
Result: "success",
|
||||
Checksum: "checksum-2",
|
||||
CreatedAt: now,
|
||||
}).Error)
|
||||
|
||||
latest, err := GetLatestOpenFlareApplyLogByNodeID(ctx, "node-1")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, latest)
|
||||
assert.Equal(t, "v2", latest.Version)
|
||||
|
||||
missing, err := GetLatestOpenFlareApplyLogByNodeID(ctx, "node-missing")
|
||||
require.NoError(t, err)
|
||||
assert.Nil(t, missing)
|
||||
}
|
||||
Reference in New Issue
Block a user