Add monitoring retention controls (#481)

## Summary
- Add configurable monitoring retention days for node metrics, tunnel
metrics, service monitor results, and tunnel quality history.
- Add an admin-only storage summary endpoint and show database usage on
the config page.
- Keep tunnel quality cleanup active even when real-time quality probing
is disabled.

## Test Plan
- go test ./... (go-backend)
- pnpm run build (vite-frontend)
This commit is contained in:
sagit
2026-04-28 13:27:44 +08:00
committed by GitHub
15 changed files with 430 additions and 13 deletions
@@ -22,6 +22,7 @@ import (
"go-backend/internal/http/response"
"go-backend/internal/license"
"go-backend/internal/metrics"
"go-backend/internal/monitoring"
"go-backend/internal/security"
"go-backend/internal/store/repo"
"go-backend/internal/ws"
@@ -152,6 +153,7 @@ func (h *Handler) Register(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/config/list", h.getConfigs)
mux.HandleFunc("/api/v1/config/update", h.updateConfigs)
mux.HandleFunc("/api/v1/config/update-single", h.updateSingleConfig)
mux.HandleFunc("/api/v1/system/storage", h.storageSummary)
mux.HandleFunc("/api/v1/license/activate", h.licenseActivate)
mux.HandleFunc("/api/v1/backup/export", h.backupExport)
mux.HandleFunc("/api/v1/backup/import", h.backupImport)
@@ -1036,6 +1038,8 @@ func normalizeAndValidateConfigValue(key, value string) (string, error) {
default:
return "", fmt.Errorf("隧道质量检测开关配置值无效")
}
case monitoring.ConfigMonitorRetentionDays:
return monitoring.NormalizeMonitoringRetentionDays(value)
default:
return value, nil
}
@@ -0,0 +1,25 @@
package handler
import (
"net/http"
"go-backend/internal/http/response"
)
func (h *Handler) storageSummary(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet && r.Method != http.MethodPost {
response.WriteJSON(w, response.ErrDefault("请求失败"))
return
}
if h == nil || h.repo == nil {
response.WriteJSON(w, response.Err(-2, "repository not initialized"))
return
}
summary, err := h.repo.DatabaseStorageSummary()
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
response.WriteJSON(w, response.OK(summary))
}
@@ -8,6 +8,7 @@ import (
"sync/atomic"
"time"
"go-backend/internal/monitoring"
"go-backend/internal/store/model"
)
@@ -15,7 +16,6 @@ const (
tunnelQualityProbeInterval = 1 * time.Second
tunnelQualityProbeTimeout = 8 * time.Second
tunnelQualityPingTimeoutMs = 5000
tunnelQualityRetention = 24 * time.Hour // keep 24h of history
tunnelQualityPruneInterval = 10 * time.Minute
tunnelQualityReportInterval = 30 * time.Second // DB save interval
)
@@ -128,12 +128,19 @@ func (p *tunnelQualityProber) isEnabled() bool {
return p.handler.isTunnelQualityMonitoringEnabled()
}
func (p *tunnelQualityProber) retentionDays() int {
if p == nil || p.handler == nil || p.handler.repo == nil {
return monitoring.DefaultMonitorRetentionDays
}
cfg, err := p.handler.repo.GetConfigsByNames([]string{monitoring.ConfigMonitorRetentionDays})
if err != nil {
return monitoring.DefaultMonitorRetentionDays
}
return monitoring.MonitoringRetentionDaysFromConfigMap(cfg)
}
// maybePrune deletes old quality rows periodically (mirrors PruneServiceMonitorResults).
func (p *tunnelQualityProber) maybePrune() {
if !p.isEnabled() {
return
}
now := time.Now().UnixMilli()
if p.lastPrune > 0 && now-p.lastPrune < int64(tunnelQualityPruneInterval/time.Millisecond) {
return
@@ -145,7 +152,7 @@ func (p *tunnelQualityProber) maybePrune() {
return
}
cutoff := now - int64(tunnelQualityRetention/time.Millisecond)
cutoff := now - int64(time.Duration(p.retentionDays())*24*time.Hour/time.Millisecond)
if err := h.repo.PruneTunnelQualityResults(cutoff); err != nil {
log.Printf("tunnel_quality_prober: prune err=%v", err)
}
@@ -287,7 +294,7 @@ func (p *tunnelQualityProber) probeTunnel(tunnelID int64) {
hops = append(hops, hop)
break
}
fromNode, _ := h.getNodeRecord(source.NodeID)
targetIP, targetPort, resolveErr := resolveChainProbeTarget(fromNode, targetNode, target.Port, ipPreference, target.ConnectIP)
if resolveErr != nil {
@@ -105,6 +105,10 @@ func requiresAdmin(path string) bool {
return true
}
if strings.HasPrefix(path, "/api/v1/system/") {
return true
}
if strings.HasPrefix(path, "/api/v1/group/") {
return true
}
+17 -3
View File
@@ -6,6 +6,7 @@ import (
"sync"
"time"
"go-backend/internal/monitoring"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
)
@@ -31,7 +32,6 @@ type IngestionService struct {
nodeBuffer []*model.NodeMetric
nodeBufferMu sync.Mutex
flushInterval time.Duration
retentionDays int
}
func NewIngestionService(repo *repo.Repository) *IngestionService {
@@ -39,7 +39,6 @@ func NewIngestionService(repo *repo.Repository) *IngestionService {
repo: repo,
nodeBuffer: make([]*model.NodeMetric, 0, 500),
flushInterval: 30 * time.Second,
retentionDays: 7,
}
}
@@ -111,7 +110,22 @@ func (s *IngestionService) flushNodeMetrics() {
}
func (s *IngestionService) pruneMetrics() {
cutoff := time.Now().Add(-time.Duration(s.retentionDays) * 24 * time.Hour).UnixMilli()
s.pruneMetricsAt(time.Now())
}
func (s *IngestionService) retentionDaysFromConfig() int {
if s == nil || s.repo == nil {
return monitoring.DefaultMonitorRetentionDays
}
cfg, err := s.repo.GetConfigsByNames([]string{monitoring.ConfigMonitorRetentionDays})
if err != nil {
return monitoring.DefaultMonitorRetentionDays
}
return monitoring.MonitoringRetentionDaysFromConfigMap(cfg)
}
func (s *IngestionService) pruneMetricsAt(now time.Time) {
cutoff := now.Add(-time.Duration(s.retentionDaysFromConfig()) * 24 * time.Hour).UnixMilli()
if s.repo == nil {
return
}
+34 -1
View File
@@ -5,6 +5,7 @@ import (
"testing"
"time"
"go-backend/internal/store/model"
"go-backend/internal/store/repo"
)
@@ -215,7 +216,6 @@ func TestPruneMetrics(t *testing.T) {
defer r.Close()
svc := NewIngestionService(r)
svc.retentionDays = 1
info := SystemInfo{CPUUsage: 50.0, MemoryUsage: 60.0, DiskUsage: 30.0}
@@ -233,6 +233,39 @@ func TestPruneMetrics(t *testing.T) {
}
}
func TestPruneMetricsUsesConfiguredRetentionDays(t *testing.T) {
r, err := repo.Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
if err := r.UpsertConfig("monitor_retention_days", "2", now); err != nil {
t.Fatalf("upsert retention config: %v", err)
}
oldMetric := &model.NodeMetric{NodeID: 1, Timestamp: now - int64(3*24*time.Hour/time.Millisecond), CPUUsage: 10}
newMetric := &model.NodeMetric{NodeID: 1, Timestamp: now - int64(1*24*time.Hour/time.Millisecond), CPUUsage: 20}
if err := r.InsertNodeMetric(oldMetric); err != nil {
t.Fatalf("insert old metric: %v", err)
}
if err := r.InsertNodeMetric(newMetric); err != nil {
t.Fatalf("insert new metric: %v", err)
}
svc := NewIngestionService(r)
svc.pruneMetricsAt(time.UnixMilli(now))
metrics, err := r.GetNodeMetrics(1, now-int64(4*24*time.Hour/time.Millisecond), now+1000)
if err != nil {
t.Fatalf("get node metrics: %v", err)
}
if len(metrics) != 1 || metrics[0].CPUUsage != 20 {
t.Fatalf("expected only newer metric to remain, got %#v", metrics)
}
}
func TestMultipleNodes(t *testing.T) {
r, err := repo.Open(":memory:")
if err != nil {
@@ -0,0 +1,48 @@
package monitoring
import (
"fmt"
"strconv"
"strings"
)
const (
ConfigMonitorRetentionDays = "monitor_retention_days"
DefaultMonitorRetentionDays = 7
MinMonitorRetentionDays = 1
MaxMonitorRetentionDays = 3650
)
func MonitoringRetentionDaysFromConfigMap(cfg map[string]string) int {
if cfg == nil {
return DefaultMonitorRetentionDays
}
days, err := parseMonitoringRetentionDays(cfg[ConfigMonitorRetentionDays])
if err != nil {
return DefaultMonitorRetentionDays
}
return days
}
func NormalizeMonitoringRetentionDays(value string) (string, error) {
days, err := parseMonitoringRetentionDays(value)
if err != nil {
return "", err
}
return strconv.Itoa(days), nil
}
func parseMonitoringRetentionDays(value string) (int, error) {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
return 0, fmt.Errorf("监控数据保留天数不能为空")
}
days, err := strconv.Atoi(trimmed)
if err != nil {
return 0, fmt.Errorf("监控数据保留天数必须是整数")
}
if days < MinMonitorRetentionDays || days > MaxMonitorRetentionDays {
return 0, fmt.Errorf("监控数据保留天数必须在 %d 到 %d 之间", MinMonitorRetentionDays, MaxMonitorRetentionDays)
}
return days, nil
}
@@ -0,0 +1,40 @@
package monitoring
import "testing"
func TestMonitoringRetentionDaysFromConfigMap(t *testing.T) {
tests := []struct {
name string
cfg map[string]string
want int
}{
{"missing uses default", nil, 7},
{"valid custom", map[string]string{ConfigMonitorRetentionDays: "3"}, 3},
{"trimmed custom", map[string]string{ConfigMonitorRetentionDays: " 30 "}, 30},
{"invalid uses default", map[string]string{ConfigMonitorRetentionDays: "abc"}, 7},
{"too small uses default", map[string]string{ConfigMonitorRetentionDays: "0"}, 7},
{"too large uses default", map[string]string{ConfigMonitorRetentionDays: "3651"}, 7},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := MonitoringRetentionDaysFromConfigMap(tc.cfg); got != tc.want {
t.Fatalf("expected %d, got %d", tc.want, got)
}
})
}
}
func TestNormalizeMonitoringRetentionDays(t *testing.T) {
for _, value := range []string{"1", "7", "3650", " 30 "} {
if got, err := NormalizeMonitoringRetentionDays(value); err != nil || got == "" {
t.Fatalf("expected %q valid, got value=%q err=%v", value, got, err)
}
}
for _, value := range []string{"", "0", "-1", "3651", "abc", "1.5"} {
if got, err := NormalizeMonitoringRetentionDays(value); err == nil {
t.Fatalf("expected %q invalid, got value=%q", value, got)
}
}
}
+3 -2
View File
@@ -65,7 +65,8 @@ type TunnelQuality = model.TunnelQuality
// ─── Repository ──────────────────────────────────────────────────────
type Repository struct {
db *gorm.DB
db *gorm.DB
dbPath string
}
type FlowUploadCounterDelta struct {
@@ -198,7 +199,7 @@ func Open(path string) (*Repository, error) {
return nil, err
}
return &Repository{db: db}, nil
return &Repository{db: db, dbPath: path}, nil
}
func OpenPostgres(dsn string) (*Repository, error) {
@@ -0,0 +1,72 @@
package repo
import (
"errors"
"fmt"
"os"
)
type DatabaseStorageSummary struct {
DBType string `json:"dbType"`
DatabaseSizeBytes int64 `json:"databaseSizeBytes"`
DatabaseSizeText string `json:"databaseSizeText"`
}
func (r *Repository) DatabaseStorageSummary() (DatabaseStorageSummary, error) {
if r == nil || r.db == nil {
return DatabaseStorageSummary{}, errors.New("repository not initialized")
}
switch r.db.Dialector.Name() {
case "sqlite":
size, err := sqliteDatabaseFileSize(r.dbPath)
if err != nil {
return DatabaseStorageSummary{}, err
}
return DatabaseStorageSummary{DBType: "sqlite", DatabaseSizeBytes: size, DatabaseSizeText: formatDatabaseSize(size)}, nil
case "postgres":
var size int64
if err := r.db.Raw("SELECT pg_database_size(current_database())").Scan(&size).Error; err != nil {
return DatabaseStorageSummary{}, err
}
return DatabaseStorageSummary{DBType: "postgres", DatabaseSizeBytes: size, DatabaseSizeText: formatDatabaseSize(size)}, nil
default:
return DatabaseStorageSummary{}, fmt.Errorf("unsupported database dialect %q", r.db.Dialector.Name())
}
}
func sqliteDatabaseFileSize(path string) (int64, error) {
if path == "" || path == ":memory:" {
return 0, nil
}
var total int64
for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
info, err := os.Stat(candidate)
if err != nil {
if os.IsNotExist(err) {
continue
}
return 0, err
}
if !info.IsDir() {
total += info.Size()
}
}
return total, nil
}
func formatDatabaseSize(bytes int64) string {
if bytes < 1024 {
return fmt.Sprintf("%d B", bytes)
}
units := []string{"KB", "MB", "GB", "TB"}
value := float64(bytes) / 1024
for _, unit := range units {
if value < 1024 || unit == "TB" {
return fmt.Sprintf("%.1f %s", value, unit)
}
value /= 1024
}
return fmt.Sprintf("%d B", bytes)
}
@@ -0,0 +1,52 @@
package repo
import (
"path/filepath"
"testing"
"go-backend/internal/store/model"
)
func TestDatabaseStorageSummarySQLiteIncludesSize(t *testing.T) {
dbPath := filepath.Join(t.TempDir(), "storage.db")
r, err := Open(dbPath)
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
if err := r.InsertNodeMetric(&model.NodeMetric{NodeID: 1, Timestamp: 123, CPUUsage: 1}); err != nil {
t.Fatalf("insert metric: %v", err)
}
summary, err := r.DatabaseStorageSummary()
if err != nil {
t.Fatalf("storage summary: %v", err)
}
if summary.DBType != "sqlite" {
t.Fatalf("expected sqlite db type, got %q", summary.DBType)
}
if summary.DatabaseSizeBytes <= 0 {
t.Fatalf("expected database size > 0, got %d", summary.DatabaseSizeBytes)
}
if summary.DatabaseSizeText == "" {
t.Fatalf("expected formatted size")
}
}
func TestFormatDatabaseSize(t *testing.T) {
tests := []struct {
bytes int64
want string
}{
{bytes: 0, want: "0 B"},
{bytes: 512, want: "512 B"},
{bytes: 1024, want: "1.0 KB"},
{bytes: 1024 * 1024, want: "1.0 MB"},
}
for _, tc := range tests {
if got := formatDatabaseSize(tc.bytes); got != tc.want {
t.Fatalf("formatDatabaseSize(%d) = %q, want %q", tc.bytes, got, tc.want)
}
}
}
@@ -0,0 +1,64 @@
package contract_test
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"go-backend/internal/auth"
"go-backend/internal/http/response"
)
func TestStorageSummaryRequiresAdminAndReturnsSize(t *testing.T) {
secret := "storage-contract-secret"
router, _ := setupContractRouter(t, secret)
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
if err != nil {
t.Fatalf("generate admin token: %v", err)
}
userToken, err := auth.GenerateToken(2, "normal_user", 1, secret)
if err != nil {
t.Fatalf("generate user token: %v", err)
}
userReq := httptest.NewRequest(http.MethodGet, "/api/v1/system/storage", nil)
userReq.Header.Set("Authorization", userToken)
userRes := httptest.NewRecorder()
router.ServeHTTP(userRes, userReq)
var denied response.R
if err := json.NewDecoder(userRes.Body).Decode(&denied); err != nil {
t.Fatalf("decode denied response: %v", err)
}
if denied.Code != 403 {
t.Fatalf("expected 403 for non-admin, got %d", denied.Code)
}
adminReq := httptest.NewRequest(http.MethodGet, "/api/v1/system/storage", nil)
adminReq.Header.Set("Authorization", adminToken)
adminRes := httptest.NewRecorder()
router.ServeHTTP(adminRes, adminReq)
var out response.R
if err := json.NewDecoder(adminRes.Body).Decode(&out); err != nil {
t.Fatalf("decode admin response: %v", err)
}
if out.Code != 0 {
t.Fatalf("expected code 0, got %d: %s", out.Code, out.Msg)
}
data, ok := out.Data.(map[string]interface{})
if !ok {
t.Fatalf("expected object data, got %T", out.Data)
}
if data["dbType"] == "" {
t.Fatalf("expected dbType")
}
if _, ok := data["databaseSizeBytes"].(float64); !ok {
t.Fatalf("expected numeric databaseSizeBytes, got %T", data["databaseSizeBytes"])
}
if data["databaseSizeText"] == "" {
t.Fatalf("expected databaseSizeText")
}
}
+4
View File
@@ -41,6 +41,7 @@ import type {
MonitorPermissionApiItem,
MonitorAccessApiData,
TunnelQualityApiItem,
StorageSummaryApiData,
} from "./types";
import axios from "axios";
@@ -254,6 +255,9 @@ export const updateConfigs = (configMap: Record<string, string>) =>
export const updateConfig = (name: string, value: string) =>
Network.post("/config/update-single", { name, value });
export const getStorageSummary = () =>
Network.get<StorageSummaryApiData>("/system/storage");
export const activateLicense = (licenseKey: string) =>
Network.post("/license/activate", { license_key: licenseKey });
+6
View File
@@ -474,6 +474,12 @@ export interface ServiceMonitorLimitsApiData {
maxTimeoutSec: number;
}
export interface StorageSummaryApiData {
dbType: string;
databaseSizeBytes: number;
databaseSizeText: string;
}
export interface MonitorNodeApiItem {
id: number;
inx: number;
+43
View File
@@ -26,6 +26,7 @@ import {
importBackup,
getAnnouncement,
updateAnnouncement,
getStorageSummary,
type AnnouncementData,
} from "@/api";
import { BackIcon, SettingsIcon } from "@/components/icons";
@@ -146,6 +147,14 @@ const CONFIG_ITEMS: ConfigItem[] = [
"关闭后,前端停止自动刷新,后端停止实时隧道质量探测(全局配置)",
type: "switch",
},
{
key: "monitor_retention_days",
label: "监控数据保留天数",
placeholder: "7",
description:
"统一清理节点指标、隧道流量、服务监控结果和隧道质量历史;默认 7 天。",
type: "input",
},
{
key: "captcha_enabled",
label: "启用验证码",
@@ -220,6 +229,7 @@ const getInitialConfigs = (): Record<string, string> => {
"cloudflare_secret_key",
"forward_compact_mode",
"monitor_tunnel_quality_enabled",
"monitor_retention_days",
"ip",
"panel_domain",
"app_logo",
@@ -288,6 +298,7 @@ export default function ConfigPage() {
const [brandUploading, setBrandUploading] = useState<
Partial<Record<BrandPreviewKey, boolean>>
>({});
const [storageSummary, setStorageSummary] = useState("加载中...");
const canGoBack =
typeof window !== "undefined" &&
@@ -347,10 +358,26 @@ export default function ConfigPage() {
}
};
const loadStorageSummary = async () => {
try {
const response = await getStorageSummary();
if (response.code === 0 && response.data?.databaseSizeText) {
setStorageSummary(response.data.databaseSizeText);
return;
}
setStorageSummary("获取失败");
} catch {
setStorageSummary("获取失败");
}
};
useEffect(() => {
const timer = setTimeout(() => {
loadConfigs(initialConfigs);
loadAnnouncement();
loadStorageSummary();
}, 100);
return () => clearTimeout(timer);
@@ -1324,6 +1351,22 @@ export default function ConfigPage() {
</Select>
</div>
<Divider className="my-2" />
<div className="space-y-3">
<div className="flex flex-col gap-1">
<p className="text-sm font-medium text-gray-700 dark:text-gray-300">
数据库占用
</p>
<p className="text-xs text-gray-500 dark:text-gray-400">
当前后端数据库文件或实例占用空间,仅用于容量参考。
</p>
</div>
<div className="rounded-lg border border-divider bg-default-50/60 dark:bg-default-100/10 px-4 py-3 text-sm font-semibold text-default-800 dark:text-default-200">
{storageSummary}
</div>
</div>
<div className="flex justify-end pt-6 border-t border-divider/50 mt-4">
<Button
color="primary"