feat: 自动备份与保留策略、健康持续化与状态页、指标历史与趋势图,并修隧道判定误报与仪表盘布局(v1.2.11)

- 自动备份:每日一份(按最新备份是否在 24h 内判定,重启不重复)、保留份数可配、列表/下载/恢复/删除,恢复前先校验归档;顺手修 writeArchive 逐个检查 tar/gzip/文件 Close()(审查报告 §二)

- 健康持续化:只存状态变更(up/down/unknown)并推导可用率与不可用区间;公开状态页默认关闭、访问码存 SHA-256、按 IP 限流、只暴露域名与可用率

- 隧道判定修复:RouteHealthCandidate 新增 tunnel_known,frps 管理接口不可用时不再把「判不出来」当成「不通」(此前会误报「隧道连续不通」);汇总与徽标同样区分「无法判定」

- 指标历史:metrics_samples 纵向点表(本机 CPU/内存/磁盘 + 每规则上下行,30 天保留、查询按桶平均、不做降采样表);仪表盘趋势卡支持 1h/24h/7d

- 仪表盘布局:运行状态并入顶部统计区(运行时长补进「服务」卡);性能并入系统资源那一行,整行 4 列

- 测试:backup/metrics/frp/api/app 新增用例;gofmt、go build/vet/test、vue-tsc、vite build 全绿

- agent 无改动,AgentVersion 保持 0.12.2
This commit is contained in:
2026-09-25 13:05:58 +08:00
parent c8264bf280
commit 74d731d0ac
46 changed files with 2967 additions and 175 deletions
+96 -3
View File
@@ -7,14 +7,16 @@ import (
"path/filepath"
"github.com/fonu/fonu/internal/backup"
"github.com/fonu/fonu/internal/settings"
)
type BackupHandler struct {
svc *backup.Service
svc *backup.Service
settings *settings.Store
}
func NewBackupHandler(svc *backup.Service) *BackupHandler {
return &BackupHandler{svc: svc}
func NewBackupHandler(svc *backup.Service, settingsStore *settings.Store) *BackupHandler {
return &BackupHandler{svc: svc, settings: settingsStore}
}
func (h *BackupHandler) Export(w http.ResponseWriter, r *http.Request) {
@@ -72,3 +74,94 @@ func (h *BackupHandler) Restore(w http.ResponseWriter, r *http.Request) {
_ = os.Remove(tmpPath)
writeJSON(w, http.StatusOK, map[string]string{"message": "备份已恢复,建议重启 Fonu"})
}
// ListArchives 列出留在磁盘上的备份(自动备份 + 手动「立即备份」)。
// 与「立即下载」不同:这些文件长期留在 <DataDir>/backups 下。
func (h *BackupHandler) ListArchives(w http.ResponseWriter, r *http.Request) {
archives, err := h.svc.List()
if err != nil {
writeError(r, w, http.StatusInternalServerError, "读取备份列表失败", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"archives": archives,
"keep": backup.KeepCount(r.Context(), h.settings),
})
}
// CreateArchive 立即备份一份并保留在服务器上,顺带按保留份数剪枝
func (h *BackupHandler) CreateArchive(w http.ResponseWriter, r *http.Request) {
path, err := h.svc.Export(r.Context())
if err != nil {
writeError(r, w, http.StatusInternalServerError, "创建备份失败", err)
return
}
name := filepath.Base(path)
// 写失败的备份不能留在列表里冒充可用备份
if _, err := h.svc.Verify(r.Context(), path); err != nil {
_ = h.svc.Remove(name)
writeError(r, w, http.StatusInternalServerError, "备份校验失败:"+err.Error(), err)
return
}
removed, err := h.svc.Prune(backup.KeepCount(r.Context(), h.settings))
if err != nil {
writeError(r, w, http.StatusInternalServerError, "备份已创建,但清理旧备份失败:"+err.Error(), err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"name": name, "removed": removed})
}
// DownloadArchive 下载指定备份;用 ServeContent 支持断点续传
func (h *BackupHandler) DownloadArchive(w http.ResponseWriter, r *http.Request) {
path, err := h.svc.Path(r.PathValue("name"))
if err != nil {
writeError(r, w, http.StatusBadRequest, err.Error())
return
}
file, err := os.Open(path)
if err != nil {
writeError(r, w, http.StatusNotFound, "备份不存在")
return
}
defer file.Close()
info, err := file.Stat()
if err != nil {
writeError(r, w, http.StatusInternalServerError, "读取备份失败", err)
return
}
w.Header().Set("Content-Type", "application/gzip")
w.Header().Set("Content-Disposition", "attachment; filename="+filepath.Base(path))
http.ServeContent(w, r, filepath.Base(path), info.ModTime(), file)
}
// RestoreArchive 从指定备份恢复。恢复会覆盖数据库与证书,前端必须二次确认
func (h *BackupHandler) RestoreArchive(w http.ResponseWriter, r *http.Request) {
path, err := h.svc.Path(r.PathValue("name"))
if err != nil {
writeError(r, w, http.StatusBadRequest, err.Error())
return
}
// 恢复前先校验:半个包不能拿去覆盖现有数据
if _, err := h.svc.Verify(r.Context(), path); err != nil {
writeError(r, w, http.StatusBadRequest, "该备份不可用:"+err.Error(), err)
return
}
if err := h.svc.Restore(r.Context(), path); err != nil {
writeError(r, w, http.StatusBadRequest, err.Error(), err)
return
}
writeJSON(w, http.StatusOK, map[string]string{"message": "备份已恢复,建议重启 Fonu"})
}
// DeleteArchive 删除指定备份
func (h *BackupHandler) DeleteArchive(w http.ResponseWriter, r *http.Request) {
if err := h.svc.Remove(r.PathValue("name")); err != nil {
if err.Error() == "备份不存在" {
writeError(r, w, http.StatusNotFound, err.Error())
return
}
writeError(r, w, http.StatusBadRequest, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
+17 -15
View File
@@ -12,6 +12,7 @@ import (
"github.com/fonu/fonu/internal/ddns"
"github.com/fonu/fonu/internal/discovery"
"github.com/fonu/fonu/internal/frp"
"github.com/fonu/fonu/internal/metrics"
"github.com/fonu/fonu/internal/notify"
"github.com/fonu/fonu/internal/service"
"github.com/fonu/fonu/internal/settings"
@@ -19,19 +20,20 @@ import (
)
type Deps struct {
Config config.Config
Logger *slog.Logger
Auth *auth.Service
Proxy *service.ProxyService
DDNS *ddns.Service
ACME *acme.Service
Settings *settings.Store
ChinaCIDR *chinacidr.Service
Backup *backup.Service
Discovery *discovery.Service
FRPMulti *frp.Service
Traffic *traffic.Collector
Notify *notify.Service
StaticFS fs.FS
StartedAt string
Config config.Config
Logger *slog.Logger
Auth *auth.Service
Proxy *service.ProxyService
DDNS *ddns.Service
ACME *acme.Service
Settings *settings.Store
ChinaCIDR *chinacidr.Service
Backup *backup.Service
Metrics *metrics.Store
Discovery *discovery.Service
FRPMulti *frp.Service
Traffic *traffic.Collector
Notify *notify.Service
StaticFS fs.FS
StartedAt string
}
+18
View File
@@ -7,6 +7,7 @@ import (
"net/http"
"strconv"
"strings"
"time"
"github.com/fonu/fonu/internal/frp"
)
@@ -439,6 +440,23 @@ func (h *FRPMultiHandler) AgentRouteConf(w http.ResponseWriter, r *http.Request)
writeJSON(w, http.StatusOK, map[string]any{"proxy_id": proxyID, "domain": domain, "content": content})
}
// RouteUptime 返回每条反代规则在最近窗口内的可用率与不可用区间。
// 数据来自后台巡检落库的「状态变更」序列,不做现场探测。
func (h *FRPMultiHandler) RouteUptime(w http.ResponseWriter, r *http.Request) {
hours := 24
if raw := strings.TrimSpace(r.URL.Query().Get("hours")); raw != "" {
if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 && parsed <= 24*7 {
hours = parsed
}
}
stats, err := h.svc.RouteUptime(r.Context(), time.Duration(hours)*time.Hour)
if err != nil {
writeError(r, w, http.StatusInternalServerError, "读取可用率失败", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"hours": hours, "items": stats})
}
// AgentRouteVersions 返回规则首个域名在 VPS 上的当前 vhost 与历史版本,供差异对比与回滚。
func (h *FRPMultiHandler) AgentRouteVersions(w http.ResponseWriter, r *http.Request) {
serverID, _, domain, status, msg := h.agentRouteTarget(r, r.PathValue("id"), r.PathValue("proxyID"))
+47
View File
@@ -0,0 +1,47 @@
package api
import (
"net/http"
"strconv"
"time"
"github.com/fonu/fonu/internal/metrics"
)
type MetricsHandler struct {
store *metrics.Store
}
func NewMetricsHandler(store *metrics.Store) *MetricsHandler {
return &MetricsHandler{store: store}
}
// History 返回指标历史:窗口内的序列已按桶取平均,前端拿到就能直接画。
// hours 最多 30 天(=保留期),buckets 最多 metrics.MaxBuckets。
func (h *MetricsHandler) History(w http.ResponseWriter, r *http.Request) {
hours := 24
if raw := r.URL.Query().Get("hours"); raw != "" {
if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 && parsed <= 24*30 {
hours = parsed
}
}
buckets := 48
if raw := r.URL.Query().Get("buckets"); raw != "" {
if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 {
buckets = parsed
}
}
if buckets > metrics.MaxBuckets {
buckets = metrics.MaxBuckets
}
series, err := h.store.Query(r.Context(), time.Duration(hours)*time.Hour, buckets)
if err != nil {
writeError(r, w, http.StatusInternalServerError, "读取指标历史失败", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"hours": hours,
"buckets": buckets,
"series": series,
})
}
+41 -28
View File
@@ -13,25 +13,29 @@ import (
func NewRouter(deps Deps) http.Handler {
logsHandler := NewLogsHandler(deps.Config)
r := &Router{
authHandler: NewAuthHandler(deps.Auth, deps.Notify),
proxyHandler: NewProxyHandler(deps.Config, deps.Proxy, logsHandler, deps.Traffic),
nginxHandler: NewNginxHandler(deps.Proxy),
statusHandler: NewStatusHandler(deps.Config, deps.Proxy, deps.DDNS, deps.ACME, deps.StartedAt, deps.Config.NginxPIDFile),
ddnsHandler: NewDDNSHandler(deps.DDNS),
certHandler: NewCertHandler(deps.ACME),
settingsHandler: NewSettingsHandler(deps.Settings, deps.Proxy, deps.Notify),
chinaCIDRHandler: NewChinaCIDRHandler(deps.ChinaCIDR),
logsHandler: logsHandler,
backupHandler: NewBackupHandler(deps.Backup),
discoveryHandler: NewDiscoveryHandler(deps.Discovery),
frpmultiHandler: NewFRPMultiHandler(deps.FRPMulti),
authSvc: deps.Auth,
staticFS: deps.StaticFS,
authHandler: NewAuthHandler(deps.Auth, deps.Notify),
proxyHandler: NewProxyHandler(deps.Config, deps.Proxy, logsHandler, deps.Traffic),
nginxHandler: NewNginxHandler(deps.Proxy),
statusHandler: NewStatusHandler(deps.Config, deps.Proxy, deps.DDNS, deps.ACME, deps.StartedAt, deps.Config.NginxPIDFile),
ddnsHandler: NewDDNSHandler(deps.DDNS),
certHandler: NewCertHandler(deps.ACME),
settingsHandler: NewSettingsHandler(deps.Settings, deps.Proxy, deps.Notify),
chinaCIDRHandler: NewChinaCIDRHandler(deps.ChinaCIDR),
logsHandler: logsHandler,
backupHandler: NewBackupHandler(deps.Backup, deps.Settings),
statusPageHandler: NewStatusPageHandler(deps.Settings, deps.FRPMulti),
metricsHandler: NewMetricsHandler(deps.Metrics),
discoveryHandler: NewDiscoveryHandler(deps.Discovery),
frpmultiHandler: NewFRPMultiHandler(deps.FRPMulti),
authSvc: deps.Auth,
staticFS: deps.StaticFS,
}
mux := http.NewServeMux()
mux.HandleFunc("GET /api/version", Version(deps.Config))
// 唯一一个匿名只读接口:默认关闭、可设访问码、按 IP 限流
mux.HandleFunc("GET /api/status-page", r.statusPageHandler.Get)
mux.HandleFunc("GET /api/auth/status", r.authHandler.Status)
mux.HandleFunc("POST /api/auth/login", r.authHandler.Login)
mux.HandleFunc("POST /api/auth/logout", r.authHandler.Logout)
@@ -88,9 +92,15 @@ func NewRouter(deps Deps) http.Handler {
protect("GET /api/logs/access", r.logsHandler.Access)
protect("GET /api/logs/error", r.logsHandler.Error)
protect("GET /api/logs/system", r.logsHandler.System)
protect("GET /api/metrics/history", r.metricsHandler.History)
protect("GET /api/logs/stream", r.logsHandler.Stream)
protect("GET /api/backup/export", r.backupHandler.Export)
protect("POST /api/backup/restore", r.backupHandler.Restore)
protect("GET /api/backup/archives", r.backupHandler.ListArchives)
protect("POST /api/backup/archives", r.backupHandler.CreateArchive)
protect("GET /api/backup/archives/{name}", r.backupHandler.DownloadArchive)
protect("POST /api/backup/archives/{name}/restore", r.backupHandler.RestoreArchive)
protect("DELETE /api/backup/archives/{name}", r.backupHandler.DeleteArchive)
protect("GET /api/discovery/scan", r.discoveryHandler.Scan)
if deps.FRPMulti != nil {
frpHandler := r.frpmultiHandler
@@ -122,6 +132,7 @@ func NewRouter(deps Deps) http.Handler {
protect("GET /api/frpmulti/servers/{id}/agent/routes", frpHandler.AgentRoutes)
protect("GET /api/frpmulti/servers/{id}/agent/routes/health", frpHandler.AgentRouteHealth)
protect("GET /api/frpmulti/routes/summary", frpHandler.AgentRouteHealthSummary)
protect("GET /api/frpmulti/routes/uptime", frpHandler.RouteUptime)
protect("GET /api/frpmulti/servers/{id}/agent/routes/drift", frpHandler.AgentRouteDrift)
protect("POST /api/frpmulti/servers/{id}/agent/routes/redeploy", frpHandler.AgentRedeployRoutes)
protect("GET /api/frpmulti/servers/{id}/agent/metrics", frpHandler.AgentMetrics)
@@ -165,20 +176,22 @@ func NewRouter(deps Deps) http.Handler {
}
type Router struct {
authHandler *AuthHandler
proxyHandler *ProxyHandler
nginxHandler *NginxHandler
statusHandler *StatusHandler
ddnsHandler *DDNSHandler
certHandler *CertHandler
settingsHandler *SettingsHandler
chinaCIDRHandler *ChinaCIDRHandler
logsHandler *LogsHandler
backupHandler *BackupHandler
discoveryHandler *DiscoveryHandler
frpmultiHandler *FRPMultiHandler
authSvc *auth.Service
staticFS fs.FS
authHandler *AuthHandler
proxyHandler *ProxyHandler
nginxHandler *NginxHandler
statusHandler *StatusHandler
ddnsHandler *DDNSHandler
certHandler *CertHandler
settingsHandler *SettingsHandler
chinaCIDRHandler *ChinaCIDRHandler
logsHandler *LogsHandler
backupHandler *BackupHandler
statusPageHandler *StatusPageHandler
metricsHandler *MetricsHandler
discoveryHandler *DiscoveryHandler
frpmultiHandler *FRPMultiHandler
authSvc *auth.Service
staticFS fs.FS
}
func (r *Router) spaHandler() http.Handler {
+39
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"net/http"
"strconv"
"strings"
"github.com/fonu/fonu/internal/notify"
"github.com/fonu/fonu/internal/service"
@@ -29,6 +30,11 @@ func (h *SettingsHandler) Get(w http.ResponseWriter, r *http.Request) {
return
}
notify.SanitizeSettings(values)
// 状态页访问码只回掩码:既不下发明文也不下发哈希
if values[settings.KeyStatusPageCodeHash] != "" {
values[settings.KeyStatusPageCode] = MaskedStatusPageCode
}
delete(values, settings.KeyStatusPageCodeHash)
writeJSON(w, http.StatusOK, values)
}
@@ -80,6 +86,13 @@ func (h *SettingsHandler) Put(w http.ResponseWriter, r *http.Request) {
settings.KeyNotifyFrpsFailThreshold: true,
settings.KeyNotifyAgentFailThreshold: true,
settings.KeyNotifyDriftCooldownHours: true,
settings.KeyBackupAutoEnabled: true,
settings.KeyRouteHealthEnabled: true,
settings.KeyStatusPageEnabled: true,
settings.KeyMetricsHistoryEnabled: true,
settings.KeyStatusPageCode: true,
settings.KeyBackupKeepCount: true,
settings.KeyNotifyOnBackupFailure: true,
settings.KeyTrustedProxy: true,
settings.KeyGlobalIPBlacklist: true,
settings.KeyGlobalIPWhitelist: true,
@@ -98,6 +111,19 @@ func (h *SettingsHandler) Put(w http.ResponseWriter, r *http.Request) {
}
filtered[key] = value
}
// 状态页访问码走「掩码往返」:客户端提交明文,这里落 SHA-256;
// 空串 = 清除访问码,掩码值 = 保持不变。哈希字段不接受客户端直接写入。
if code, ok := filtered[settings.KeyStatusPageCode]; ok {
delete(filtered, settings.KeyStatusPageCode)
trimmed := strings.TrimSpace(code)
switch {
case trimmed == MaskedStatusPageCode:
case trimmed == "":
filtered[settings.KeyStatusPageCodeHash] = ""
default:
filtered[settings.KeyStatusPageCodeHash] = hashStatusPageCode(trimmed)
}
}
if h.notify != nil && h.notify.Store() != nil {
if in, ok := notify.ParseSaveInputFromMap(filtered); ok {
if err := h.notify.Store().Save(r.Context(), in); err != nil {
@@ -127,6 +153,11 @@ func (h *SettingsHandler) Put(w http.ResponseWriter, r *http.Request) {
return
}
notify.SanitizeSettings(values)
// 状态页访问码只回掩码:既不下发明文也不下发哈希
if values[settings.KeyStatusPageCodeHash] != "" {
values[settings.KeyStatusPageCode] = MaskedStatusPageCode
}
delete(values, settings.KeyStatusPageCodeHash)
writeJSON(w, http.StatusOK, values)
}
@@ -163,6 +194,7 @@ func isNotifySettingKey(key string) bool {
settings.KeyNotifyFrpsFailThreshold,
settings.KeyNotifyAgentFailThreshold,
settings.KeyNotifyDriftCooldownHours,
settings.KeyNotifyOnBackupFailure,
settings.KeyNotifyWebhookURL:
return true
default:
@@ -199,6 +231,13 @@ func (h *SettingsHandler) validateSetting(key, value string) error {
return fmt.Errorf("统计窗口需在 10–3600 秒之间")
}
}
case settings.KeyBackupKeepCount:
if value != "" {
n, err := strconv.Atoi(value)
if err != nil || n < 1 || n > 60 {
return fmt.Errorf("备份保留份数需在 1–60 之间")
}
}
case settings.KeyNotifyLoginFailureThreshold:
if value != "" {
n, err := strconv.Atoi(value)
+173
View File
@@ -0,0 +1,173 @@
package api
import (
"crypto/sha256"
"encoding/hex"
"net"
"net/http"
"strings"
"sync"
"time"
"github.com/fonu/fonu/internal/frp"
"github.com/fonu/fonu/internal/notify"
"github.com/fonu/fonu/internal/settings"
)
// statusPageWindow / statusPageLimit 匿名访问的限流窗口与配额:
// 状态页会被扫描器盯上,这里比登录接口更宽松但仍有限度。
const (
statusPageWindow = time.Minute
statusPageLimit = 60
)
// StatusPageHandler 提供**匿名只读**状态页数据,默认关闭。
//
// 刻意只暴露域名与可用率:不带内网地址、端口、证书路径、服务端地址——
// 状态页是要挂到公网上的东西,泄露拓扑就等于把内网结构摊开给别人看。
type StatusPageHandler struct {
settings *settings.Store
frp *frp.Service
limiter *statusPageLimiter
}
func NewStatusPageHandler(settingsStore *settings.Store, frpSvc *frp.Service) *StatusPageHandler {
return &StatusPageHandler{
settings: settingsStore,
frp: frpSvc,
limiter: newStatusPageLimiter(),
}
}
type statusPageRoute struct {
Domain string `json:"domain"`
State string `json:"state"`
Uptime float64 `json:"uptime_pct"`
DownMinutes int `json:"down_minutes"`
UnknownMinutes int `json:"unknown_minutes"`
}
type statusPagePayload struct {
Title string `json:"title"`
CodeRequired bool `json:"code_required"`
CheckedAt string `json:"checked_at"`
Up int `json:"up"`
Down int `json:"down"`
Unknown int `json:"unknown"`
Routes []statusPageRoute `json:"routes"`
}
// Get 是唯一的匿名入口:关闭时一律 404(与「不存在」不可区分),
// 设了访问码时校验 ?code=,校验失败也回 404。
func (h *StatusPageHandler) Get(w http.ResponseWriter, r *http.Request) {
if h == nil || h.settings == nil {
writeError(r, w, http.StatusNotFound, "状态页未开启")
return
}
if !h.limiter.allow(clientIPForLimit(r)) {
writeError(r, w, http.StatusTooManyRequests, "请求过于频繁")
return
}
enabled, _ := h.settings.GetBool(r.Context(), settings.KeyStatusPageEnabled)
if !enabled || h.frp == nil {
writeError(r, w, http.StatusNotFound, "状态页未开启")
return
}
codeHash, err := h.settings.Get(r.Context(), settings.KeyStatusPageCodeHash)
if err != nil {
codeHash = ""
}
codeRequired := strings.TrimSpace(codeHash) != ""
if codeRequired && hashStatusPageCode(r.URL.Query().Get("code")) != codeHash {
writeError(r, w, http.StatusNotFound, "状态页未开启")
return
}
stats, err := h.frp.RouteUptime(r.Context(), 24*time.Hour)
if err != nil {
writeError(r, w, http.StatusInternalServerError, "读取状态失败", err)
return
}
payload := statusPagePayload{
Title: "Fonu 服务状态",
CodeRequired: codeRequired,
CheckedAt: time.Now().UTC().Format(time.RFC3339),
Routes: make([]statusPageRoute, 0, len(stats)),
}
for _, stat := range stats {
switch stat.State {
case frp.RouteHealthUp:
payload.Up++
case frp.RouteHealthDown:
payload.Down++
default:
payload.Unknown++
}
payload.Routes = append(payload.Routes, statusPageRoute{
Domain: stat.Domain,
State: stat.State,
Uptime: stat.Uptime,
DownMinutes: stat.DownMinutes,
UnknownMinutes: stat.UnknownMinutes,
})
}
writeJSON(w, http.StatusOK, payload)
}
func hashStatusPageCode(code string) string {
sum := sha256.Sum256([]byte(strings.TrimSpace(code)))
return hex.EncodeToString(sum[:])
}
// clientIPForLimit 取限流用的客户端 IP
func clientIPForLimit(r *http.Request) string {
host, _, err := net.SplitHostPort(r.RemoteAddr)
if err != nil {
return r.RemoteAddr
}
return host
}
// statusPageLimiter 按 IP 的滑动窗口限流;只给匿名状态页用
type statusPageLimiter struct {
mu sync.Mutex
hits map[string][]time.Time
now func() time.Time
}
func newStatusPageLimiter() *statusPageLimiter {
return &statusPageLimiter{hits: map[string][]time.Time{}, now: time.Now}
}
func (l *statusPageLimiter) allow(ip string) bool {
now := l.now()
cutoff := now.Add(-statusPageWindow)
l.mu.Lock()
defer l.mu.Unlock()
kept := make([]time.Time, 0, len(l.hits[ip])+1)
for _, at := range l.hits[ip] {
if at.After(cutoff) {
kept = append(kept, at)
}
}
if len(kept) >= statusPageLimit {
l.hits[ip] = kept
return false
}
l.hits[ip] = append(kept, now)
// 顺手清理长期不活跃的 IP,避免 map 无限增长
if len(l.hits) > 1024 {
for key, times := range l.hits {
if len(times) == 0 {
delete(l.hits, key)
}
}
}
return true
}
// MaskedStatusPageCode 供设置接口回显用(与通知模块的掩码约定一致)
const MaskedStatusPageCode = notify.MaskedSecret
+43
View File
@@ -0,0 +1,43 @@
package api
import (
"testing"
"time"
)
func TestStatusPageLimiterBlocksAfterLimit(t *testing.T) {
limiter := newStatusPageLimiter()
now := time.Date(2026, 9, 25, 10, 0, 0, 0, time.UTC)
limiter.now = func() time.Time { return now }
for i := 0; i < statusPageLimit; i++ {
if !limiter.allow("1.2.3.4") {
t.Fatalf("第 %d 次请求应放行", i+1)
}
}
if limiter.allow("1.2.3.4") {
t.Fatal("超过配额应拒绝")
}
// 限流按 IP 计:别的访客不该被牵连
if !limiter.allow("5.6.7.8") {
t.Fatal("其它 IP 不应被牵连")
}
// 窗口滑过后恢复
now = now.Add(statusPageWindow + time.Second)
if !limiter.allow("1.2.3.4") {
t.Fatal("窗口过后应恢复放行")
}
}
func TestHashStatusPageCodeHidesPlaintext(t *testing.T) {
hash := hashStatusPageCode("my-secret")
if hash == "my-secret" || len(hash) != 64 {
t.Fatalf("访问码应以 SHA-256 存储,实际 %q", hash)
}
if hashStatusPageCode(" my-secret ") != hash {
t.Fatal("前后空白裁剪后应得到同一哈希")
}
if hashStatusPageCode("other") == hash {
t.Fatal("不同访问码不应得到同一哈希")
}
}
+12 -1
View File
@@ -23,6 +23,7 @@ import (
"github.com/fonu/fonu/internal/discovery"
"github.com/fonu/fonu/internal/frp"
"github.com/fonu/fonu/internal/logstore"
"github.com/fonu/fonu/internal/metrics"
"github.com/fonu/fonu/internal/monitor"
"github.com/fonu/fonu/internal/nginx"
"github.com/fonu/fonu/internal/notify"
@@ -93,6 +94,8 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
ddnsSvc := ddns.NewService(ddnsStore, settingsStore, secretBox, logger, notifySvc)
acmeSvc := acme.NewService(cfg, certStore, ddnsSvc, settingsStore, proxySvc, logger, notifySvc)
backupSvc := backup.New(cfg.DataDir)
// 指标历史:与巡检同生命周期,采样跟随巡检轮次
metricsStore := metrics.NewStore(conn)
discoverySvc := discovery.New()
// 新增:多服务端 FRP 模块(接管原网关模式)
frpMultiSvc := frp.NewService(cfg, frp.NewStore(conn), secretBox, logger, certStore)
@@ -114,6 +117,7 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
Notify: notifySvc,
ChinaCIDR: chinaCIDRSvc,
Backup: backupSvc,
Metrics: metricsStore,
Discovery: discoverySvc,
FRPMulti: frpMultiSvc,
Traffic: trafficCollector,
@@ -140,7 +144,9 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
chinaHours = 24
}
// 巡检间隔比其它任务短:「连续 N 次不通」需要足够采样次数才能判定
monitorChecker := monitor.New(settingsStore, notifySvc, logger, monitorSource{frp: frpMultiSvc, logger: logger})
monitorChecker := monitor.New(settingsStore, notifySvc, logger, monitorSource{frp: frpMultiSvc, logger: logger, traffic: trafficCollector}, healthRecorder{frp: frpMultiSvc}, metricsStore)
// 自动备份:启动时会先跑一轮,靠「最新备份是否在 24h 内」避免频繁重启重复备份
backupRunner := backup.NewAutoRunner(backupSvc, settingsStore, notifySvc, logger)
sched := scheduler.New(
scheduler.Job{
Name: "ddns",
@@ -162,6 +168,11 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
Interval: 2 * time.Minute,
Run: monitorChecker.Run,
},
scheduler.Job{
Name: "backup",
Interval: backup.AutoInterval,
Run: backupRunner.Run,
},
)
sched.Start(ctx)
trafficCollector.Start(ctx)
+63 -3
View File
@@ -7,13 +7,15 @@ import (
"github.com/fonu/fonu/internal/frp"
"github.com/fonu/fonu/internal/monitor"
"github.com/fonu/fonu/internal/traffic"
)
// monitorSource 把 frp 服务适配成 monitor.Source:monitor 只关心「状态快照」,
// 怎么从 frp / agent 取是这里的事,两边因此互不依赖。
type monitorSource struct {
frp *frp.Service
logger *slog.Logger
frp *frp.Service
logger *slog.Logger
traffic *traffic.Collector
}
func (s monitorSource) Routes(ctx context.Context) ([]monitor.RouteStatus, error) {
@@ -30,7 +32,15 @@ func (s monitorSource) Routes(ctx context.Context) ([]monitor.RouteStatus, error
continue
}
for _, item := range health {
routes = append(routes, monitor.RouteStatus{Domain: item.Domain, TunnelOK: item.TunnelOK, Detail: item.TunnelDetail})
routes = append(routes, monitor.RouteStatus{
ServerID: server.ID,
Domain: item.Domain,
TunnelOK: item.TunnelOK,
DNSOK: item.DNSOK,
ServiceOK: item.ServiceOK,
Known: item.TunnelKnown,
Detail: item.TunnelDetail,
})
}
}
return routes, nil
@@ -98,6 +108,19 @@ func (s monitorSource) Drift(ctx context.Context) ([]monitor.DriftStatus, error)
return out, nil
}
// Traffic 返回若干域名的当前上下行速率:流量采集器内部按 host:port 累计,这里按域名合并。
func (s monitorSource) Traffic(ctx context.Context, domains []string) (map[string]monitor.TrafficRate, error) {
if s.traffic == nil {
return nil, nil
}
snapshot := s.traffic.SnapshotByHost(domains)
rates := make(map[string]monitor.TrafficRate, len(snapshot))
for host, item := range snapshot {
rates[host] = monitor.TrafficRate{Upload: item.UploadRate, Download: item.DownloadRate}
}
return rates, nil
}
func (s monitorSource) enabledServers(ctx context.Context) ([]frp.Server, error) {
servers, err := s.frp.ListServers(ctx)
if err != nil {
@@ -128,3 +151,40 @@ func driftKindText(kind string) string {
return kind
}
}
// healthRecorder 把 monitor 的健康结论翻成 frp 侧样本并落库(
// 两侧的状态字面量由 healthStateValue 显式翻译,改动一侧会在这里编译期报错)。
type healthRecorder struct {
frp *frp.Service
}
func (r healthRecorder) RecordRouteHealth(ctx context.Context, samples []monitor.HealthSample) error {
converted := make([]frp.RouteHealthSample, 0, len(samples))
for _, sample := range samples {
converted = append(converted, frp.RouteHealthSample{
ServerID: sample.ServerID,
Domain: sample.Domain,
State: healthStateValue(sample.State),
Reason: sample.Reason,
Detail: sample.Detail,
})
}
_, err := r.frp.RecordRouteHealth(ctx, converted)
return err
}
func (r healthRecorder) PruneRouteHealth(ctx context.Context) error {
_, err := r.frp.PruneRouteHealth(ctx)
return err
}
func healthStateValue(state string) string {
switch state {
case monitor.HealthUp:
return frp.RouteHealthUp
case monitor.HealthDown:
return frp.RouteHealthDown
default:
return frp.RouteHealthUnknown
}
}
+24
View File
@@ -0,0 +1,24 @@
package app
import (
"testing"
"github.com/fonu/fonu/internal/frp"
"github.com/fonu/fonu/internal/monitor"
)
// monitor 与 frp 两侧的状态字面量必须一致,落库前由 healthStateValue 翻译一次;
// 这个测试把翻译钉住,避免以后改一侧的常量而另一侧悄悄写错值。
func TestHealthStateValue(t *testing.T) {
cases := map[string]string{
monitor.HealthUp: frp.RouteHealthUp,
monitor.HealthDown: frp.RouteHealthDown,
monitor.HealthUnknown: frp.RouteHealthUnknown,
"其他": frp.RouteHealthUnknown,
}
for input, want := range cases {
if got := healthStateValue(input); got != want {
t.Fatalf("%q 应翻译为 %q,实际 %q", input, want, got)
}
}
}
+157
View File
@@ -0,0 +1,157 @@
package backup
import (
"archive/tar"
"compress/gzip"
"context"
"fmt"
"io"
"os"
"path/filepath"
"regexp"
"sort"
"time"
"github.com/fonu/fonu/internal/settings"
)
// 保留份数的边界:0/负数按默认值处理,上限防止把磁盘塞满
const (
DefaultKeepCount = 7
MaxKeepCount = 60
)
// archiveNamePattern 备份命名格式(Export 用 CreateTemp 生成,后缀是随机数字)。
// List / Path 都按它过滤,避免把 backups 目录里任意文件当成备份读写或删除。
var archiveNamePattern = regexp.MustCompile(`^fonu-backup-\d{8}-\d{6}-\d+\.tar\.gz$`)
// Archive 留在磁盘上的一份备份
type Archive struct {
Name string `json:"name"`
Size int64 `json:"size"`
ModTime time.Time `json:"mod_time"`
}
func (s *Service) backupsDir() string {
return filepath.Join(s.dataDir, "backups")
}
// List 按新的在前列出备份;目录不存在返回空列表(还没备份过是正常状态)。
func (s *Service) List() ([]Archive, error) {
entries, err := os.ReadDir(s.backupsDir())
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, err
}
archives := make([]Archive, 0, len(entries))
for _, entry := range entries {
if entry.IsDir() || !archiveNamePattern.MatchString(entry.Name()) {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
archives = append(archives, Archive{Name: entry.Name(), Size: info.Size(), ModTime: info.ModTime().UTC()})
}
// 文件名前缀就是时间戳,按名倒序即按时间倒序
sort.Slice(archives, func(i, j int) bool { return archives[i].Name > archives[j].Name })
return archives, nil
}
// Prune 只保留最新的 keep 份,返回删除数量;keep <= 0 表示不剪枝。
func (s *Service) Prune(keep int) (int, error) {
if keep <= 0 {
return 0, nil
}
archives, err := s.List()
if err != nil {
return 0, err
}
removed := 0
for _, archive := range archives[min(keep, len(archives)):] {
if err := s.Remove(archive.Name); err != nil {
return removed, err
}
removed++
}
return removed, nil
}
// Path 把备份名解析成绝对路径;名字必须匹配备份命名格式,避免被用来读写目录外的文件。
func (s *Service) Path(name string) (string, error) {
if !archiveNamePattern.MatchString(name) {
return "", fmt.Errorf("备份名无效")
}
return filepath.Join(s.backupsDir(), name), nil
}
// Remove 删除指定备份
func (s *Service) Remove(name string) error {
path, err := s.Path(name)
if err != nil {
return err
}
if err := os.Remove(path); err != nil {
if os.IsNotExist(err) {
return fmt.Errorf("备份不存在")
}
return err
}
return nil
}
// Verify 回读归档并返回条目数:确认这份备份不是写了一半的残件,
// 同时要求包内必须有 fonu.db(否则多半不是 Fonu 的备份)。
func (s *Service) Verify(ctx context.Context, path string) (int, error) {
file, err := os.Open(path)
if err != nil {
return 0, err
}
defer file.Close()
gr, err := gzip.NewReader(file)
if err != nil {
return 0, fmt.Errorf("备份文件无法解压:%w", err)
}
defer gr.Close()
tr := tar.NewReader(gr)
count := 0
hasDatabase := false
for {
header, err := tr.Next()
if err == io.EOF {
break
}
if err != nil {
return count, fmt.Errorf("备份文件已损坏:%w", err)
}
if err := ctx.Err(); err != nil {
return count, err
}
count++
if header.Name == "fonu.db" {
hasDatabase = true
}
}
if !hasDatabase {
return count, fmt.Errorf("备份里没有 fonu.db,可能不是 Fonu 的备份")
}
return count, nil
}
// KeepCount 读取保留份数(未配置或越界时回落到默认值)。
// 自动备份任务与「立即备份」共用它,避免两处各写一套默认值。
func KeepCount(ctx context.Context, store *settings.Store) int {
keep, err := store.GetInt(ctx, settings.KeyBackupKeepCount)
if err != nil || keep <= 0 {
return DefaultKeepCount
}
if keep > MaxKeepCount {
return MaxKeepCount
}
return keep
}
+89
View File
@@ -0,0 +1,89 @@
package backup
import (
"context"
"fmt"
"log/slog"
"path/filepath"
"time"
"github.com/fonu/fonu/internal/notify"
"github.com/fonu/fonu/internal/settings"
)
// AutoInterval 自动备份间隔。scheduler 启动时会先跑一轮,靠「最新备份是否还在间隔内」
// 判断该不该备份,因此频繁重启(NAS 上很常见)不会重复备份。
const AutoInterval = 24 * time.Hour
// AutoRunner 定期导出一份归档、回读校验、再按保留份数剪枝。
type AutoRunner struct {
svc *Service
settings *settings.Store
notify *notify.Service
logger *slog.Logger
now func() time.Time
}
func NewAutoRunner(svc *Service, settingsStore *settings.Store, notifySvc *notify.Service, logger *slog.Logger) *AutoRunner {
if logger == nil {
logger = slog.Default()
}
return &AutoRunner{svc: svc, settings: settingsStore, notify: notifySvc, logger: logger, now: time.Now}
}
// Run 跑一轮。失败只通知、不返回错误:scheduler 的任务没有重试机制,通知才是可见性来源。
func (r *AutoRunner) Run(ctx context.Context) {
if r == nil || r.svc == nil || r.settings == nil {
return
}
enabled, err := r.settings.GetBool(ctx, settings.KeyBackupAutoEnabled)
if err != nil || !enabled {
return
}
archives, err := r.svc.List()
if err != nil {
r.logger.Warn("读取备份列表失败", "module", "BACKUP", "error", err.Error())
return
}
if len(archives) > 0 && !dueForBackup(archives[0].ModTime, r.now(), AutoInterval) {
return
}
path, err := r.svc.Export(ctx)
if err != nil {
r.fail(ctx, fmt.Sprintf("导出备份失败:%s", err.Error()))
return
}
entries, err := r.svc.Verify(ctx, path)
if err != nil {
// 校验不过的归档不算一份可用备份:删掉并如实告警,别让用户以为已经有备份
if removeErr := r.svc.Remove(filepath.Base(path)); removeErr != nil {
r.logger.Warn("删除校验失败的备份失败", "module", "BACKUP", "error", removeErr.Error())
}
r.fail(ctx, fmt.Sprintf("备份校验失败:%s", err.Error()))
return
}
removed, err := r.svc.Prune(KeepCount(ctx, r.settings))
if err != nil {
r.logger.Warn("剪枝旧备份失败", "module", "BACKUP", "error", err.Error())
}
r.logger.Info("自动备份完成", "module", "BACKUP",
"file", filepath.Base(path), "entries", entries, "removed", removed)
}
// dueForBackup 判断该不该做一次备份:没有备份,或最新一份已超过间隔。
func dueForBackup(newest, now time.Time, interval time.Duration) bool {
if newest.IsZero() {
return true
}
return now.Sub(newest) >= interval
}
func (r *AutoRunner) fail(ctx context.Context, message string) {
r.logger.Error("自动备份失败", "module", "BACKUP", "error", message)
if r.notify != nil {
r.notify.Alert(ctx, notify.EventBackupFailure, "自动备份失败", message)
}
}
+130
View File
@@ -0,0 +1,130 @@
package backup
import (
"context"
"os"
"path/filepath"
"testing"
"time"
)
func TestDueForBackup(t *testing.T) {
now := time.Date(2026, 9, 25, 10, 0, 0, 0, time.UTC)
// 还没有备份 → 立刻做一份
if !dueForBackup(time.Time{}, now, AutoInterval) {
t.Fatal("没有备份时应立即备份")
}
// 刚备份过 → 跳过:这正是重启不会重复备份的原因
if dueForBackup(now.Add(-time.Hour), now, AutoInterval) {
t.Fatal("间隔内的备份不应重复执行")
}
// 正好到间隔 → 该做了
if !dueForBackup(now.Add(-AutoInterval), now, AutoInterval) {
t.Fatal("到达间隔应再备份一次")
}
// 超过间隔 → 该做
if !dueForBackup(now.Add(-25*time.Hour), now, AutoInterval) {
t.Fatal("超过间隔应再备份一次")
}
}
func TestExportVerifyRoundTrip(t *testing.T) {
dir := t.TempDir()
if err := os.WriteFile(filepath.Join(dir, "fonu.db"), []byte("db"), 0o600); err != nil {
t.Fatal(err)
}
svc := New(dir)
path, err := svc.Export(context.Background())
if err != nil {
t.Fatalf("export: %v", err)
}
entries, err := svc.Verify(context.Background(), path)
if err != nil {
t.Fatalf("verify: %v", err)
}
if entries < 1 {
t.Fatalf("条目数应至少为 1,实际 %d", entries)
}
archives, err := svc.List()
if err != nil {
t.Fatalf("list: %v", err)
}
if len(archives) != 1 {
t.Fatalf("应列出 1 份备份,实际 %d", len(archives))
}
if archives[0].Size == 0 || !archiveNamePattern.MatchString(archives[0].Name) {
t.Fatalf("备份条目异常:%+v", archives[0])
}
}
func TestVerifyRejectsGarbage(t *testing.T) {
svc := New(t.TempDir())
path := filepath.Join(svc.backupsDir(), "fonu-backup-20260925-000000-1.tar.gz")
if err := os.MkdirAll(svc.backupsDir(), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, []byte("not a gzip stream"), 0o600); err != nil {
t.Fatal(err)
}
if _, err := svc.Verify(context.Background(), path); err == nil {
t.Fatal("非 gzip 内容应校验失败")
}
}
func TestListPruneAndPath(t *testing.T) {
svc := New(t.TempDir())
if err := os.MkdirAll(svc.backupsDir(), 0o755); err != nil {
t.Fatal(err)
}
for _, name := range []string{
"fonu-backup-20260923-100000-1.tar.gz",
"fonu-backup-20260924-100000-2.tar.gz",
"fonu-backup-20260925-100000-3.tar.gz",
} {
if err := os.WriteFile(filepath.Join(svc.backupsDir(), name), []byte("x"), 0o600); err != nil {
t.Fatal(err)
}
}
// 不符合命名格式的文件不参与:备份目录里可能混着手工放进来的东西
if err := os.WriteFile(filepath.Join(svc.backupsDir(), "notes.txt"), []byte("x"), 0o600); err != nil {
t.Fatal(err)
}
archives, err := svc.List()
if err != nil {
t.Fatalf("list: %v", err)
}
if len(archives) != 3 {
t.Fatalf("应只列出 3 份备份,实际 %d", len(archives))
}
if archives[0].Name != "fonu-backup-20260925-100000-3.tar.gz" {
t.Fatalf("应按时间倒序,实际首条 %s", archives[0].Name)
}
removed, err := svc.Prune(2)
if err != nil {
t.Fatalf("prune: %v", err)
}
if removed != 1 {
t.Fatalf("应删除 1 份,实际 %d", removed)
}
if archives, _ = svc.List(); len(archives) != 2 {
t.Fatalf("剪枝后应剩 2 份,实际 %d", len(archives))
}
// 名称校验:不能让备份名把读写带出备份目录
for _, name := range []string{"../fonu.db", "notes.txt", "", "fonu-backup-x.tar.gz"} {
if _, err := svc.Path(name); err == nil {
t.Fatalf("非法备份名应拒绝:%q", name)
}
}
if err := svc.Remove("fonu-backup-20260920-100000-9.tar.gz"); err == nil {
t.Fatal("删除不存在的备份应报错")
}
if _, err := svc.Prune(0); err != nil {
t.Fatalf("keep<=0 应直接返回:%v", err)
}
}
+43 -28
View File
@@ -26,45 +26,60 @@ func New(dataDir string) *Service {
return &Service{dataDir: dataDir}
}
// backupTimeFmt 归档文件名里的时间戳;List / Path 都按同一套格式校验
const backupTimeFmt = "20060102-150405"
// Export 导出一份归档并返回路径;文件留在 <DataDir>/backups 下,由调用方决定下载后删除还是长期保留。
func (s *Service) Export(ctx context.Context) (string, error) {
tmpDir := filepath.Join(s.dataDir, "backups")
if err := os.MkdirAll(tmpDir, 0o755); err != nil {
if err := os.MkdirAll(s.backupsDir(), 0o755); err != nil {
return "", err
}
name := fmt.Sprintf("fonu-backup-%s.tar.gz", time.Now().Format("20060102-His"))
outPath := filepath.Join(tmpDir, name)
file, err := os.Create(outPath)
// CreateTemp 保证同一秒内多次导出不会覆盖已有备份(文件名带随机后缀)
file, err := os.CreateTemp(s.backupsDir(), fmt.Sprintf("fonu-backup-%s-*.tar.gz", time.Now().Format(backupTimeFmt)))
if err != nil {
return "", err
}
defer file.Close()
gw := gzip.NewWriter(file)
tw := tar.NewWriter(gw)
entries := []string{
"fonu.db",
"certs",
"nginx",
}
for _, entry := range entries {
if err := addToArchive(ctx, tw, s.dataDir, entry); err != nil {
_ = tw.Close()
_ = gw.Close()
_ = os.Remove(outPath)
return "", err
}
}
if err := tw.Close(); err != nil {
return "", err
}
if err := gw.Close(); err != nil {
outPath := file.Name()
if err := s.writeArchive(ctx, file, outPath); err != nil {
return "", err
}
return outPath, nil
}
// writeArchive 写归档,逐层检查关闭错误。
//
// Close 的错误必须往上抛:tar/gzip 的尾部与文件系统的 flush 都发生在 Close,
// 忽略它就会出现「备份文件不完整、函数却返回成功」——这是最坏的一种假成功。
// 任一步失败都删掉半成品,不留一个看起来能用的备份。
func (s *Service) writeArchive(ctx context.Context, file *os.File, outPath string) error {
discard := func() {
_ = file.Close()
_ = os.Remove(outPath)
}
gw := gzip.NewWriter(file)
tw := tar.NewWriter(gw)
for _, entry := range []string{"fonu.db", "certs", "nginx"} {
if err := addToArchive(ctx, tw, s.dataDir, entry); err != nil {
discard()
return err
}
}
if err := tw.Close(); err != nil {
discard()
return err
}
if err := gw.Close(); err != nil {
discard()
return err
}
if err := file.Close(); err != nil {
_ = os.Remove(outPath)
return err
}
return nil
}
func (s *Service) Restore(ctx context.Context, archivePath string) error {
file, err := os.Open(archivePath)
if err != nil {
+233
View File
@@ -0,0 +1,233 @@
package frp
import (
"context"
"fmt"
"sort"
"strings"
"time"
)
// 健康状态取值。unknown 表示「判不出来」(frps 管理接口未配或不可用),
// 不能当成 down —— 这正是 tunnel_known 存在的理由。
const (
RouteHealthUp = "up"
RouteHealthDown = "down"
RouteHealthUnknown = "unknown"
// RouteHealthRetention 状态变更记录的保留期
RouteHealthRetention = 30 * 24 * time.Hour
)
// RouteHealthSample 是一轮巡检得到的一条规则健康结论
type RouteHealthSample struct {
ServerID int64
Domain string
State string
Reason string
Detail string
}
// RouteHealthEvent 一条状态变更记录
type RouteHealthEvent struct {
ServerID int64 `json:"server_id"`
Domain string `json:"domain"`
State string `json:"state"`
Reason string `json:"reason,omitempty"`
Detail string `json:"detail,omitempty"`
At string `json:"at"`
}
// RouteHealthIncident 一段不可用区间;To 为空表示到窗口结束仍在持续
type RouteHealthIncident struct {
From string `json:"from"`
To string `json:"to,omitempty"`
Reason string `json:"reason,omitempty"`
Detail string `json:"detail,omitempty"`
Minutes int `json:"minutes"`
}
// RouteUptimeStat 一条规则在窗口内的可用率与不可用区间
type RouteUptimeStat struct {
ServerID int64 `json:"server_id"`
Domain string `json:"domain"`
State string `json:"state"`
Uptime float64 `json:"uptime_pct"`
DownMinutes int `json:"down_minutes"`
UnknownMinutes int `json:"unknown_minutes"`
CoveredMinutes int `json:"covered_minutes"`
Incidents []RouteHealthIncident `json:"incidents"`
}
// RecordRouteHealth 只写「状态变更」:与上一条同状态就不写,避免表被每 2 分钟的采样撑大。
// 返回写入条数。
func (s *Service) RecordRouteHealth(ctx context.Context, samples []RouteHealthSample) (int, error) {
if len(samples) == 0 {
return 0, nil
}
current, err := s.store.latestRouteHealthStates(ctx)
if err != nil {
return 0, err
}
now := time.Now().UTC().Format(time.RFC3339)
written := 0
for _, sample := range samples {
domain := strings.ToLower(strings.TrimSpace(sample.Domain))
if domain == "" || sample.State == "" {
continue
}
key := routeHealthKey(sample.ServerID, domain)
if state, ok := current[key]; ok && state == sample.State {
continue
}
if _, err := s.store.db.ExecContext(ctx,
`INSERT INTO route_health_events(server_id, domain, state, reason, detail, at) VALUES (?, ?, ?, ?, ?, ?)`,
sample.ServerID, domain, sample.State, sample.Reason, sample.Detail, now); err != nil {
return written, err
}
written++
}
return written, nil
}
// RouteUptime 返回窗口内每条规则的可用率与不可用区间
func (s *Service) RouteUptime(ctx context.Context, window time.Duration) ([]RouteUptimeStat, error) {
if window <= 0 {
window = 24 * time.Hour
}
now := time.Now().UTC()
since := now.Add(-window)
events, err := s.store.routeHealthEventsSince(ctx, since)
if err != nil {
return nil, err
}
return ComputeRouteUptime(events, since, now), nil
}
// PruneRouteHealth 清理超过保留期的状态变更
func (s *Service) PruneRouteHealth(ctx context.Context) (int64, error) {
return s.store.pruneRouteHealthEvents(ctx, time.Now().UTC().Add(-RouteHealthRetention))
}
func routeHealthKey(serverID int64, domain string) string {
return fmt.Sprintf("%d|%s", serverID, domain)
}
// ComputeRouteUptime 由状态变更序列推导 [from, to] 窗口内的可用率。
//
// 覆盖时长只算「有记录可依」的部分:窗口开始前有记录时,其状态延续到窗口开始;
// 窗口内才第一次出现的规则,从它第一条记录开始算。窗口内完全没有记录的规则不出现在结果里。
// 可用率按 up/(up+down) 计,unknown 不计入分母(判不出来不能算不可用),但单独给出时长。
func ComputeRouteUptime(events []RouteHealthEvent, from, to time.Time) []RouteUptimeStat {
type point struct {
at time.Time
serverID int64
domain string
state string
reason, detail string
}
byRule := map[string][]point{}
order := make([]string, 0, len(events))
for _, event := range events {
at, err := time.Parse(time.RFC3339, event.At)
if err != nil {
continue
}
key := routeHealthKey(event.ServerID, event.Domain)
if _, ok := byRule[key]; !ok {
order = append(order, key)
}
byRule[key] = append(byRule[key], point{
at: at, serverID: event.ServerID, domain: event.Domain,
state: event.State, reason: event.Reason, detail: event.Detail,
})
}
sort.Strings(order)
stats := make([]RouteUptimeStat, 0, len(order))
for _, key := range order {
points := byRule[key]
sort.Slice(points, func(i, j int) bool { return points[i].at.Before(points[j].at) })
// 窗口起点的生效状态:取 from 之前最后一条;没有则从第一条记录开始覆盖
start := from
index := 0
state := ""
reason, detail := "", ""
for index < len(points) && !points[index].at.After(from) {
state, reason, detail = points[index].state, points[index].reason, points[index].detail
index++
}
if state == "" {
if index >= len(points) {
continue
}
start = points[index].at
state, reason, detail = points[index].state, points[index].reason, points[index].detail
index++
}
stat := RouteUptimeStat{
ServerID: points[0].serverID,
Domain: points[0].domain,
State: state,
Incidents: make([]RouteHealthIncident, 0, 2),
}
upMinutes := 0
// stat.State 要是「窗口末尾的状态」,不是窗口开始时的状态
lastState := state
for {
end := to
ongoing := true
if index < len(points) {
end = points[index].at
ongoing = false
}
minutes := int(end.Sub(start).Minutes())
if minutes < 0 {
minutes = 0
}
switch state {
case RouteHealthUp:
upMinutes += minutes
case RouteHealthDown:
stat.DownMinutes += minutes
incident := RouteHealthIncident{
From: start.UTC().Format(time.RFC3339),
Reason: reason,
Detail: detail,
Minutes: minutes,
}
if !ongoing {
incident.To = end.UTC().Format(time.RFC3339)
}
stat.Incidents = append(stat.Incidents, incident)
default:
stat.UnknownMinutes += minutes
}
stat.CoveredMinutes += minutes
lastState = state
if index >= len(points) {
break
}
start = points[index].at
state, reason, detail = points[index].state, points[index].reason, points[index].detail
index++
}
stat.State = lastState
observed := upMinutes + stat.DownMinutes
if observed > 0 {
stat.Uptime = float64(upMinutes) / float64(observed) * 100
} else {
// 窗口内既没有 up 也没有 down(只有 unknown)时不编造可用率
stat.Uptime = 100
if stat.UnknownMinutes > 0 {
stat.Uptime = 0
}
}
stats = append(stats, stat)
}
return stats
}
+85
View File
@@ -0,0 +1,85 @@
package frp
import (
"context"
"database/sql"
"time"
)
// latestRouteHealthStates 取每条规则最新一条记录的状态,用于「同状态不重复写」判断
func (s *Store) latestRouteHealthStates(ctx context.Context) (map[string]string, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT e.server_id, e.domain, e.state
FROM route_health_events e
JOIN (
SELECT server_id, domain, MAX(id) AS id
FROM route_health_events
GROUP BY server_id, domain
) m ON m.id = e.id`)
if err != nil {
return nil, err
}
defer rows.Close()
states := map[string]string{}
for rows.Next() {
var (
serverID int64
domain string
state string
)
if err := rows.Scan(&serverID, &domain, &state); err != nil {
return nil, err
}
states[routeHealthKey(serverID, domain)] = state
}
return states, rows.Err()
}
// routeHealthEventsSince 取窗口内的变更,外加每条规则在窗口开始前的最后一条
// ——否则「窗口开始时是什么状态」就无从得知了。
func (s *Store) routeHealthEventsSince(ctx context.Context, since time.Time) ([]RouteHealthEvent, error) {
stamp := since.UTC().Format(time.RFC3339)
rows, err := s.db.QueryContext(ctx, `
SELECT server_id, domain, state, COALESCE(reason, ''), COALESCE(detail, ''), at
FROM route_health_events WHERE at >= ?
UNION ALL
SELECT e.server_id, e.domain, e.state, COALESCE(e.reason, ''), COALESCE(e.detail, ''), e.at
FROM route_health_events e
JOIN (
SELECT server_id, domain, MAX(id) AS id
FROM route_health_events WHERE at < ?
GROUP BY server_id, domain
) m ON m.id = e.id`, stamp, stamp)
if err != nil {
return nil, err
}
defer rows.Close()
events := make([]RouteHealthEvent, 0, 32)
for rows.Next() {
var event RouteHealthEvent
if err := rows.Scan(&event.ServerID, &event.Domain, &event.State, &event.Reason, &event.Detail, &event.At); err != nil {
return nil, err
}
events = append(events, event)
}
return events, rows.Err()
}
// pruneRouteHealthEvents 清理 before 之前的记录,返回删除条数
func (s *Store) pruneRouteHealthEvents(ctx context.Context, before time.Time) (int64, error) {
res, err := s.db.ExecContext(ctx,
`DELETE FROM route_health_events WHERE at < ?`, before.UTC().Format(time.RFC3339))
if err != nil {
return 0, err
}
affected, err := res.RowsAffected()
if err != nil {
if err == sql.ErrNoRows {
return 0, nil
}
return 0, err
}
return affected, nil
}
+8 -5
View File
@@ -4,16 +4,19 @@ import "testing"
func TestSummarizeRouteHealth(t *testing.T) {
items := []RouteHealthCandidate{
{DNSOK: true, CertOK: true, TunnelOK: true, ServiceOK: true},
{DNSOK: true, CertOK: false, TunnelOK: true, ServiceOK: false},
{DNSOK: false, CertOK: true, TunnelOK: false, ServiceOK: true},
{DNSOK: true, CertOK: true, TunnelOK: true, TunnelKnown: true, ServiceOK: true},
{DNSOK: true, CertOK: false, TunnelOK: true, TunnelKnown: true, ServiceOK: false},
{DNSOK: false, CertOK: true, TunnelOK: false, TunnelKnown: true, ServiceOK: true},
// frps 管理接口不可用:隧道状态判不出来,既不算通也不是不通
{DNSOK: true, CertOK: true, ServiceOK: true},
}
summary := summarizeRouteHealth(2, []string{"vps-b"}, items)
if summary.Servers != 2 || summary.Routes != 3 || summary.Healthy != 1 {
if summary.Servers != 2 || summary.Routes != 4 || summary.Healthy != 1 {
t.Fatalf("unexpected summary counts: %+v", summary)
}
if summary.DNSFailed != 1 || summary.CertMissing != 1 || summary.TunnelDown != 1 || summary.ServiceDown != 1 {
if summary.DNSFailed != 1 || summary.CertMissing != 1 || summary.TunnelDown != 1 ||
summary.TunnelUnknown != 1 || summary.ServiceDown != 1 {
t.Fatalf("unexpected segment counts: %+v", summary)
}
if len(summary.FailedServers) != 1 || summary.FailedServers[0] != "vps-b" {
+108
View File
@@ -0,0 +1,108 @@
package frp
import (
"testing"
"time"
)
func findUptimeStat(t *testing.T, stats []RouteUptimeStat, domain string) RouteUptimeStat {
t.Helper()
for _, stat := range stats {
if stat.Domain == domain {
return stat
}
}
t.Fatalf("结果里没有 %s:%+v", domain, stats)
return RouteUptimeStat{}
}
func TestComputeRouteUptimeIncidentAndUnknown(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
to := from.Add(24 * time.Hour)
at := func(offset time.Duration) string { return from.Add(offset).Format(time.RFC3339) }
events := []RouteHealthEvent{
// 窗口开始前就是 up:状态延续到窗口开始
{ServerID: 1, Domain: "a.com", State: RouteHealthUp, At: at(-2 * time.Hour)},
{ServerID: 1, Domain: "a.com", State: RouteHealthDown, At: at(time.Hour), Reason: "tunnel"},
{ServerID: 1, Domain: "a.com", State: RouteHealthUp, At: at(2 * time.Hour)},
// 窗口内才第一次出现
{ServerID: 1, Domain: "b.com", State: RouteHealthUp, At: at(6 * time.Hour)},
// 判不出来
{ServerID: 2, Domain: "c.com", State: RouteHealthUnknown, At: at(-time.Hour)},
}
stats := ComputeRouteUptime(events, from, to)
if len(stats) != 3 {
t.Fatalf("应有 3 条规则的统计,实际 %d:%+v", len(stats), stats)
}
a := findUptimeStat(t, stats, "a.com")
if a.CoveredMinutes != 24*60 {
t.Fatalf("窗口开始前的状态应延续,覆盖 24h,实际 %d 分钟", a.CoveredMinutes)
}
if a.DownMinutes != 60 {
t.Fatalf("不可用应为 60 分钟,实际 %d", a.DownMinutes)
}
if a.Uptime < 95 || a.Uptime > 96 {
t.Fatalf("可用率应约 95.8%%,实际 %.2f", a.Uptime)
}
if len(a.Incidents) != 1 || a.Incidents[0].Minutes != 60 || a.Incidents[0].Reason != "tunnel" {
t.Fatalf("不可用区间不对:%+v", a.Incidents)
}
if a.Incidents[0].To == "" {
t.Fatal("已恢复的区间应有结束时间")
}
if a.State != RouteHealthUp {
t.Fatalf("窗口末尾状态应为 up,实际 %s", a.State)
}
b := findUptimeStat(t, stats, "b.com")
if b.CoveredMinutes != 18*60 {
t.Fatalf("窗口内才出现的规则应从第一条记录起算 18h,实际 %d", b.CoveredMinutes)
}
if b.Uptime != 100 || b.DownMinutes != 0 {
t.Fatalf("没有任何 down 时可用率应为 100:%+v", b)
}
c := findUptimeStat(t, stats, "c.com")
if c.UnknownMinutes != 24*60 || c.DownMinutes != 0 {
t.Fatalf("全程 unknown 不该算作不可用:%+v", c)
}
if c.Uptime != 0 {
t.Fatalf("全程判不出来时不能编造可用率(不给 100),实际 %.2f", c.Uptime)
}
}
func TestComputeRouteUptimeOngoingIncident(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
to := from.Add(24 * time.Hour)
events := []RouteHealthEvent{
{ServerID: 1, Domain: "a.com", State: RouteHealthUp, At: from.Add(0).Format(time.RFC3339)},
{ServerID: 1, Domain: "a.com", State: RouteHealthDown, At: from.Add(20 * time.Hour).Format(time.RFC3339), Reason: "service"},
}
stats := ComputeRouteUptime(events, from, to)
stat := findUptimeStat(t, stats, "a.com")
if stat.DownMinutes != 4*60 {
t.Fatalf("到窗口结束仍在持续应计 240 分钟,实际 %d", stat.DownMinutes)
}
if len(stat.Incidents) != 1 {
t.Fatalf("应有 1 段区间:%+v", stat.Incidents)
}
if stat.Incidents[0].To != "" {
t.Fatalf("仍在持续的区间不应有结束时间,实际 %q", stat.Incidents[0].To)
}
if stat.State != RouteHealthDown {
t.Fatalf("窗口末尾状态应为 down,实际 %s", stat.State)
}
}
func TestComputeRouteUptimeSkipsRulesWithoutEvents(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
// 窗口内没有任何记录 → 不出现在结果里,而不是给一个 100%
if stats := ComputeRouteUptime(nil, from, from.Add(time.Hour)); len(stats) != 0 {
t.Fatalf("没有记录时不该有统计:%+v", stats)
}
}
+26 -16
View File
@@ -870,12 +870,14 @@ func (s *Service) InstallAgentNginx(ctx context.Context, serverID int64) (map[st
// 健康判定与规则页一致:四段(DNS/证书/隧道/内网服务)全通过才算健康;
// 一个域名可同时命中多段失败,因此各段计数之和可能大于 Routes-Healthy。
type RouteHealthSummary struct {
Servers int `json:"servers"`
Routes int `json:"routes"`
Healthy int `json:"healthy"`
DNSFailed int `json:"dns_failed"`
CertMissing int `json:"cert_missing"`
TunnelDown int `json:"tunnel_down"`
Servers int `json:"servers"`
Routes int `json:"routes"`
Healthy int `json:"healthy"`
DNSFailed int `json:"dns_failed"`
CertMissing int `json:"cert_missing"`
TunnelDown int `json:"tunnel_down"`
// TunnelUnknown 是「frps 管理接口不可用、隧道状态判不出来」的条数
TunnelUnknown int `json:"tunnel_unknown"`
ServiceDown int `json:"service_down"`
FailedServers []string `json:"failed_servers,omitempty"`
}
@@ -911,7 +913,11 @@ func summarizeRouteHealth(servers int, failedServers []string, items []RouteHeal
if !item.CertOK {
summary.CertMissing++
}
if !item.TunnelOK {
switch {
case !item.TunnelKnown:
// frps 管理接口不可用时隧道状态判不出来,不能计入「不通」
summary.TunnelUnknown++
case !item.TunnelOK:
summary.TunnelDown++
}
if !item.ServiceOK {
@@ -926,13 +932,16 @@ func summarizeRouteHealth(servers int, failedServers []string, items []RouteHeal
// RouteHealthCandidate 公网反代规则的四段健康状态(DNS/证书/隧道/内网服务)。
type RouteHealthCandidate struct {
ProxyID int64 `json:"proxy_id"`
Domain string `json:"domain"`
DNSOK bool `json:"dns_ok"`
DNSDetail string `json:"dns_detail,omitempty"`
CertOK bool `json:"cert_ok"`
CertDetail string `json:"cert_detail,omitempty"`
TunnelOK bool `json:"tunnel_ok"`
ProxyID int64 `json:"proxy_id"`
Domain string `json:"domain"`
DNSOK bool `json:"dns_ok"`
DNSDetail string `json:"dns_detail,omitempty"`
CertOK bool `json:"cert_ok"`
CertDetail string `json:"cert_detail,omitempty"`
TunnelOK bool `json:"tunnel_ok"`
// TunnelKnown 表示「隧道状态是判得出来的」:frps 管理接口没配或连不上时 TunnelOK 恒为 false,
// 但那是「无法判定」而不是「不通」——告警与汇总都必须按它区分,否则会误报。
TunnelKnown bool `json:"tunnel_known"`
TunnelName string `json:"tunnel_name,omitempty"`
TunnelDetail string `json:"tunnel_detail,omitempty"`
ServiceOK bool `json:"service_ok"`
@@ -1067,6 +1076,8 @@ func (s *Service) RouteHealth(ctx context.Context, serverID int64) ([]RouteHealt
// frp 会自动给代理名加 "<user>." 前缀(如 admin.ssssss),两种形态都接受。
item.TunnelName = proxy.Name
if dashboardReady {
// dashboard 可用:这一次的隧道状态是「判得出来」的,不管结论是通还是不通
item.TunnelKnown = true
// 优先按域名索引匹配(V2 API 返回 spec.http.customDomains,代理名含 user 前缀不可靠),
// 再回退代理名(裸名或 user.前缀名)
switch {
@@ -1088,8 +1099,7 @@ func (s *Service) RouteHealth(ctx context.Context, serverID int64) ([]RouteHealt
}
if item.TunnelOK {
item.TunnelDetail = "frps 已注册: " + item.TunnelName
} else {
// 诊断透出:dashboard 实际返回的代理名列表(供前端悬停排查)
} else { // 诊断透出:dashboard 实际返回的代理名列表(供前端悬停排查)
names := make([]string, 0, len(tunnelNames))
for name := range tunnelNames {
if !strings.HasPrefix(name, "domain:") {
+253
View File
@@ -0,0 +1,253 @@
// Package metrics 保存指标历史(纵向 (scope, metric, at) 点表)并按窗口查询。
//
// 采样跟随巡检轮次(2 分钟),行量极小,所以不建降采样表:查询时按桶取平均即可,
// 这样只有一套数据、没有「原始数据与聚合数据不一致」这类问题。
package metrics
import (
"context"
"database/sql"
"sort"
"time"
)
const (
// ScopeHost 本机资源;server:<id> / route:<serverID>:<domain> 留给后续序列
ScopeHost = "host"
// 指标名:与前端展示一一对应,改动即改动契约
MetricCPU = "cpu_percent"
MetricMemUsed = "mem_used_bytes"
MetricMemTotal = "mem_total_bytes"
MetricDiskUsed = "disk_used_bytes"
MetricDiskTotal = "disk_total_bytes"
// 流量序列按规则区分:scope 形如 route:<serverID>:<domain>
MetricTrafficIn = "traffic_in_rate"
MetricTrafficOut = "traffic_out_rate"
// MaxBuckets 单次查询最多返回的桶数(前端最多画这么多点)
MaxBuckets = 720
// Retention 保留期
Retention = 30 * 24 * time.Hour
)
// HostMetrics 是本机序列的固定集合(顺序即前端展示顺序)
var HostMetrics = []string{MetricCPU, MetricMemUsed, MetricMemTotal, MetricDiskUsed, MetricDiskTotal}
// Sample 一条待写入的样本
type Sample struct {
Scope string
Metric string
Value float64
}
// Point 查询结果里的一个点(bucket 为平均值;At 是该桶的起始时刻)
type Point struct {
At string `json:"at"`
Value float64 `json:"value"`
}
// Series 一条序列
type Series struct {
Scope string `json:"scope"`
Metric string `json:"metric"`
Points []Point `json:"points"`
}
type Store struct {
db *sql.DB
}
func NewStore(db *sql.DB) *Store {
return &Store{db: db}
}
// Record 批量写入一批样本(同一事务,避免每点一个来回)
func (s *Store) Record(ctx context.Context, samples []Sample) error {
if len(samples) == 0 {
return nil
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
now := time.Now().UTC().Format(time.RFC3339)
stmt, err := tx.PrepareContext(ctx,
`INSERT INTO metrics_samples(scope, metric, value, at) VALUES (?, ?, ?, ?)`)
if err != nil {
_ = tx.Rollback()
return err
}
for _, sample := range samples {
if sample.Scope == "" || sample.Metric == "" {
continue
}
if _, err := stmt.ExecContext(ctx, sample.Scope, sample.Metric, sample.Value, now); err != nil {
_ = stmt.Close()
_ = tx.Rollback()
return err
}
}
if err := stmt.Close(); err != nil {
_ = tx.Rollback()
return err
}
return tx.Commit()
}
// Prune 删除超过保留期的样本
func (s *Store) Prune(ctx context.Context) (int64, error) {
res, err := s.db.ExecContext(ctx,
`DELETE FROM metrics_samples WHERE at < ?`, time.Now().UTC().Add(-Retention).Format(time.RFC3339))
if err != nil {
return 0, err
}
affected, err := res.RowsAffected()
if err != nil {
return 0, nil
}
return affected, nil
}
// Query 取窗口内的序列并按 buckets 个等宽桶聚合(每桶取平均)。
// buckets <= 0 时按原始点返回;都会按 MaxBuckets 截断。
func (s *Store) Query(ctx context.Context, window time.Duration, buckets int) ([]Series, error) {
if window <= 0 {
window = 24 * time.Hour
}
if buckets <= 0 {
buckets = 48
}
if buckets > MaxBuckets {
buckets = MaxBuckets
}
to := time.Now().UTC()
from := to.Add(-window)
rows, err := s.db.QueryContext(ctx, `
SELECT scope, metric, at, value FROM metrics_samples
WHERE at >= ? ORDER BY scope, metric, at`, from.Format(time.RFC3339))
if err != nil {
return nil, err
}
defer rows.Close()
type raw struct {
scope, metric string
at time.Time
value float64
}
grouped := map[string][]raw{}
order := make([]string, 0, 8)
for rows.Next() {
var (
scope, metric, at string
value float64
)
if err := rows.Scan(&scope, &metric, &at, &value); err != nil {
return nil, err
}
parsed, err := time.Parse(time.RFC3339, at)
if err != nil {
continue
}
key := scope + "\x00" + metric
if _, ok := grouped[key]; !ok {
order = append(order, key)
}
grouped[key] = append(grouped[key], raw{scope: scope, metric: metric, at: parsed, value: value})
}
if err := rows.Err(); err != nil {
return nil, err
}
series := make([]Series, 0, len(order))
for _, key := range order {
points := grouped[key]
at := make([]time.Time, 0, len(points))
values := make([]float64, 0, len(points))
for _, item := range points {
at = append(at, item.at)
values = append(values, item.value)
}
if buckets <= 0 {
buckets = 48
}
series = append(series, Series{
Scope: points[0].scope,
Metric: points[0].metric,
Points: Bucketize(at, values, from, to, buckets),
})
}
// 固定顺序:本机序列按 HostMetrics 的顺序排,便于前端稳定取用
sort.SliceStable(series, func(i, j int) bool {
return metricOrder(series[i].Metric) < metricOrder(series[j].Metric)
})
return series, nil
}
// Bucketize 把 (时刻, 值) 序列按 [from, to] 等宽分成 buckets 个桶并取平均。
// 空桶不产出点(前端画线时自动跨过,不会出现「凭空补 0」的假谷底)。
func Bucketize(at []time.Time, values []float64, from, to time.Time, buckets int) []Point {
if buckets <= 0 || to.Before(from) || len(at) == 0 {
return nil
}
span := to.Sub(from)
if span <= 0 {
return nil
}
width := span / time.Duration(buckets)
if width <= 0 {
return nil
}
sums := make([]float64, buckets)
counts := make([]int, buckets)
for i, moment := range at {
if i >= len(values) {
break
}
if moment.Before(from) || moment.After(to) {
continue
}
index := int(moment.Sub(from) / width)
if index >= buckets {
index = buckets - 1
}
if index < 0 {
continue
}
sums[index] += values[i]
counts[index]++
}
points := make([]Point, 0, buckets)
for i := 0; i < buckets; i++ {
if counts[i] == 0 {
continue
}
points = append(points, Point{
At: from.Add(time.Duration(i) * width).UTC().Format(time.RFC3339),
Value: sums[i] / float64(counts[i]),
})
}
return points
}
func metricOrder(metric string) int {
for index, name := range HostMetrics {
if name == metric {
return index
}
}
return len(HostMetrics) + 1
}
// Percent 由「已用 / 总量」算百分比:总量为 0 时返回 0(而不是 NaN)
func Percent(used, total float64) float64 {
if total <= 0 {
return 0
}
return used / total * 100
}
+82
View File
@@ -0,0 +1,82 @@
package metrics
import (
"testing"
"time"
)
func TestBucketizeAveragesAndSkipsEmptyBuckets(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
to := from.Add(time.Hour)
// 12 个 5 分钟桶:第 0 桶两个点(取平均)、第 1 桶一个、第 11 桶一个
at := []time.Time{
from.Add(time.Minute),
from.Add(2 * time.Minute),
from.Add(6 * time.Minute),
from.Add(59 * time.Minute),
}
values := []float64{10, 20, 30, 40}
points := Bucketize(at, values, from, to, 12)
if len(points) != 3 {
t.Fatalf("空桶不该产出点(否则图上会出现凭空补 0 的谷底),实际 %d 个:%+v", len(points), points)
}
if points[0].Value != 15 {
t.Fatalf("第 0 桶应为 (10+20)/2=15,实际 %v", points[0].Value)
}
if points[0].At != from.Format(time.RFC3339) {
t.Fatalf("桶时间戳应为桶起点,实际 %q", points[0].At)
}
if points[1].Value != 30 || points[2].Value != 40 {
t.Fatalf("后续桶取值不对:%+v", points)
}
}
func TestBucketizeDropsOutOfWindowPoints(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
to := from.Add(time.Hour)
at := []time.Time{from.Add(-time.Minute), from.Add(30 * time.Minute), to.Add(time.Minute)}
values := []float64{1, 2, 3}
points := Bucketize(at, values, from, to, 12)
if len(points) != 1 || points[0].Value != 2 {
t.Fatalf("窗口外的点应被丢掉,实际 %+v", points)
}
if points := Bucketize(at[:1], values[:1], from, to, 12); len(points) != 0 {
t.Fatalf("全在窗口外时应返回空,实际 %+v", points)
}
}
func TestBucketizeRejectsDegenerateInput(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
values := []float64{1}
if points := Bucketize([]time.Time{from}, values, from, from, 12); points != nil {
t.Fatalf("零宽窗口应返回 nil,实际 %+v", points)
}
if points := Bucketize([]time.Time{from}, values, from, from.Add(time.Hour), 0); points != nil {
t.Fatalf("桶数为 0 应返回 nil,实际 %+v", points)
}
if points := Bucketize(nil, nil, from, from.Add(time.Hour), 12); points != nil {
t.Fatalf("没有样本应返回 nil,实际 %+v", points)
}
}
func TestBucketizeKeepsPointsShorterThanValues(t *testing.T) {
from := time.Date(2026, 9, 25, 0, 0, 0, 0, time.UTC)
points := Bucketize([]time.Time{from.Add(time.Minute)}, nil, from, from.Add(time.Hour), 12)
if len(points) != 0 {
t.Fatalf("值与时刻数量不一致时应忽略多出的时刻,实际 %+v", points)
}
}
func TestPercent(t *testing.T) {
if got := Percent(0, 0); got != 0 {
t.Fatalf("总量为 0 应返回 0,实际 %v", got)
}
if got := Percent(50, 200); got != 25 {
t.Fatalf("期望 25,实际 %v", got)
}
}
+244 -21
View File
@@ -14,18 +14,49 @@ import (
"sync"
"time"
"github.com/fonu/fonu/internal/metrics"
"github.com/fonu/fonu/internal/notify"
"github.com/fonu/fonu/internal/settings"
"github.com/fonu/fonu/internal/sysinfo"
)
// RouteStatus 是巡检关心的隧道状态。
// RouteStatus 是巡检关心的隧道与健康状态。
type RouteStatus struct {
ServerID int64
Domain string
TunnelOK bool
// Known 表示这次的隧道状态是判得出来的(frps 管理接口可用)。
// 判不出来时既不能计入连续失败,也不能当成恢复。
Known bool
TunnelOK bool
DNSOK bool
ServiceOK bool
Detail string
// 健康结论取值,与 frp 侧的 RouteHealthUp/Down/Unknown 保持一致,
// 由 app 侧适配时显式翻译一次。
}
const (
HealthUp = "up"
HealthDown = "down"
HealthUnknown = "unknown"
)
// HealthSample 是一条规则本轮的健康结论,交给落库层(monitor 不关心它写哪张表)。
type HealthSample struct {
ServerID int64
Domain string
State string
Reason string
Detail string
}
// HealthRecorder 持久化健康结论,由 frp 服务实现。
type HealthRecorder interface {
RecordRouteHealth(ctx context.Context, samples []HealthSample) error
PruneRouteHealth(ctx context.Context) error
}
// ServerStatus 是巡检关心的公网服务端状态:只有 DesiredState 表示「用户要它运行」时才判定掉线。
type ServerStatus struct {
ID int64
@@ -52,12 +83,25 @@ type DriftStatus struct {
Items []string
}
// Source 提供巡检需要的外部状态;由 app 侧把 frp 服务适配过来,monitor 因此不依赖 frp 包。
// TrafficRate 某个域名的当前速率(字节/秒)
type TrafficRate struct {
Upload float64
Download float64
}
// Source 提供巡检需要的外部状态;由 app 侧把 frp 服务与流量采集器适配过来。
type Source interface {
Routes(ctx context.Context) ([]RouteStatus, error)
Servers(ctx context.Context) ([]ServerStatus, error)
Agents(ctx context.Context) ([]AgentStatus, error)
Drift(ctx context.Context) ([]DriftStatus, error)
Traffic(ctx context.Context, domains []string) (map[string]TrafficRate, error)
}
// MetricsRecorder 落库指标历史;*metrics.Store 直接实现它。
type MetricsRecorder interface {
Record(ctx context.Context, samples []metrics.Sample) error
Prune(ctx context.Context) (int64, error)
}
// Checker 每轮都重新读设置项:改阈值或开关不需要重启。
@@ -66,16 +110,20 @@ type Checker struct {
notify *notify.Service
logger *slog.Logger
source Source
health HealthRecorder
metrics MetricsRecorder
mu sync.Mutex
cpuPrev sysinfo.CPUSample
resourceAlerts map[string]time.Time
failures map[string]int
driftAlerted map[int64]bool
driftNotified map[int64]time.Time
mu sync.Mutex
cpuPrev sysinfo.CPUSample
resourceAlerts map[string]time.Time
failures map[string]int
driftAlerted map[int64]bool
driftNotified map[int64]time.Time
lastHealthPrune time.Time
lastMetricsPrune time.Time
}
func New(settingsStore *settings.Store, notifySvc *notify.Service, logger *slog.Logger, source Source) *Checker {
func New(settingsStore *settings.Store, notifySvc *notify.Service, logger *slog.Logger, source Source, health HealthRecorder, metricsRecorder MetricsRecorder) *Checker {
if logger == nil {
logger = slog.Default()
}
@@ -84,6 +132,8 @@ func New(settingsStore *settings.Store, notifySvc *notify.Service, logger *slog.
notify: notifySvc,
logger: logger,
source: source,
health: health,
metrics: metricsRecorder,
resourceAlerts: map[string]time.Time{},
failures: map[string]int{},
driftAlerted: map[int64]bool{},
@@ -96,27 +146,86 @@ func (c *Checker) Run(ctx context.Context) {
if c == nil || c.settings == nil {
return
}
c.checkResources(ctx)
c.runHostChecks(ctx)
c.checkRoutes(ctx)
c.checkServers(ctx)
c.checkAgents(ctx)
c.checkDrift(ctx)
}
func (c *Checker) checkResources(ctx context.Context) {
// 开关关闭时连采集都省掉(关掉的功能不该有后台开销)
if enabled, _ := c.settings.GetBool(ctx, settings.KeyNotifyOnResourceThreshold); !enabled {
// runHostChecks 一次采样同时喂给「资源阈值告警」与「指标历史」。
// CPU 占用率靠两次采样的差值得出,采两次会互相干扰,所以采样点只留这一个。
func (c *Checker) runHostChecks(ctx context.Context) {
alertEnabled, _ := c.settings.GetBool(ctx, settings.KeyNotifyOnResourceThreshold)
// 指标历史默认开启:没配置过时按开处理(存了 "0"/"false" 才关)
recordEnabled := c.metricsHistoryEnabled(ctx)
// 两件事都关着就连采样都省掉(关掉的功能不该有后台开销)
if !alertEnabled && !recordEnabled {
return
}
cpuLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyCPUThreshold)
memoryLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyMemoryThreshold)
diskLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyDiskThreshold)
c.mu.Lock()
stats, sample := sysinfo.Read(c.cpuPrev)
c.cpuPrev = sample
c.mu.Unlock()
if alertEnabled {
c.checkResourceLimits(ctx, stats)
}
if recordEnabled {
c.recordHostMetrics(ctx, stats)
}
}
// metricsHistoryEnabled 读指标历史开关:未配置按开启处理(只有显式存了 0/false 才关)
func (c *Checker) metricsHistoryEnabled(ctx context.Context) bool {
if c.metrics == nil {
return false
}
raw, err := c.settings.Get(ctx, settings.KeyMetricsHistoryEnabled)
if err != nil {
return true
}
return raw != "0" && !strings.EqualFold(raw, "false")
}
// recordHostMetrics 记本机序列;采样间隔就是巡检轮次(2 分钟),行量很小
func (c *Checker) recordHostMetrics(ctx context.Context, stats sysinfo.Stats) {
samples := []metrics.Sample{
{Scope: metrics.ScopeHost, Metric: metrics.MetricCPU, Value: stats.CPUPercent},
{Scope: metrics.ScopeHost, Metric: metrics.MetricMemUsed, Value: float64(stats.MemUsedBytes)},
{Scope: metrics.ScopeHost, Metric: metrics.MetricMemTotal, Value: float64(stats.MemTotalBytes)},
{Scope: metrics.ScopeHost, Metric: metrics.MetricDiskUsed, Value: float64(stats.DiskUsedBytes)},
{Scope: metrics.ScopeHost, Metric: metrics.MetricDiskTotal, Value: float64(stats.DiskTotalBytes)},
}
if err := c.metrics.Record(ctx, samples); err != nil {
c.logger.Warn("记录指标历史失败", "module", "MONITOR", "error", err.Error())
}
c.pruneMetricsIfDue(ctx)
}
// pruneMetricsIfDue 每天清理一次超期样本
func (c *Checker) pruneMetricsIfDue(ctx context.Context) {
c.mu.Lock()
due := c.lastMetricsPrune.IsZero() || time.Since(c.lastMetricsPrune) >= 24*time.Hour
if due {
c.lastMetricsPrune = time.Now()
}
c.mu.Unlock()
if !due {
return
}
if _, err := c.metrics.Prune(ctx); err != nil {
c.logger.Warn("清理指标历史失败", "module", "MONITOR", "error", err.Error())
}
}
// checkResourceLimits 只做阈值判定:采样已由 runHostChecks 统一完成
func (c *Checker) checkResourceLimits(ctx context.Context, stats sysinfo.Stats) {
cpuLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyCPUThreshold)
memoryLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyMemoryThreshold)
diskLimit, _ := c.settings.GetInt(ctx, settings.KeyNotifyDiskThreshold)
limits := []struct {
metric string
label string
@@ -164,11 +273,14 @@ func (c *Checker) checkRoutes(ctx context.Context) {
if c.source == nil {
return
}
if enabled, _ := c.settings.GetBool(ctx, settings.KeyNotifyOnTunnelDown); !enabled {
return
recordHealth, _ := c.settings.GetBool(ctx, settings.KeyRouteHealthEnabled)
if c.health == nil {
recordHealth = false
}
limit, _ := c.settings.GetInt(ctx, settings.KeyNotifyTunnelFailThreshold)
if limit <= 0 {
alertEnabled, _ := c.settings.GetBool(ctx, settings.KeyNotifyOnTunnelDown)
recordMetrics := c.metricsHistoryEnabled(ctx)
// 三件事都关着就别去探测:RouteHealth 是这一轮里最贵的调用
if !recordHealth && !alertEnabled && !recordMetrics {
return
}
@@ -177,10 +289,48 @@ func (c *Checker) checkRoutes(ctx context.Context) {
c.logger.Warn("巡检隧道状态失败", "module", "MONITOR", "error", err.Error())
return
}
// 落库先于告警:判不出来的那一档也要记(unknown 与 down 必须区分)
if recordHealth {
samples := make([]HealthSample, 0, len(routes))
for _, route := range routes {
if strings.TrimSpace(route.Domain) == "" {
continue
}
state, reason := healthStateOf(route)
samples = append(samples, HealthSample{
ServerID: route.ServerID,
Domain: route.Domain,
State: state,
Reason: reason,
Detail: route.Detail,
})
}
if err := c.health.RecordRouteHealth(ctx, samples); err != nil {
c.logger.Warn("记录规则健康状态失败", "module", "MONITOR", "error", err.Error())
}
c.pruneHealthIfDue(ctx)
}
// 流量序列用同一轮的规则列表落库,不再单独探测一轮
if recordMetrics {
c.recordRouteTraffic(ctx, routes)
}
if !alertEnabled {
return
}
limit, _ := c.settings.GetInt(ctx, settings.KeyNotifyTunnelFailThreshold)
if limit <= 0 {
return
}
for _, route := range routes {
if strings.TrimSpace(route.Domain) == "" {
continue
}
if !route.Known {
continue
}
_, event := c.track("route:"+route.Domain, route.TunnelOK, limit)
switch event {
case failureEventDown:
@@ -193,6 +343,79 @@ func (c *Checker) checkRoutes(ctx context.Context) {
}
}
// healthStateOf 把一段状态翻成与 frp 侧一致的结论 + 失败环节
func healthStateOf(route RouteStatus) (string, string) {
if !route.Known {
return HealthUnknown, "dashboard"
}
switch {
case !route.DNSOK:
return HealthDown, "dns"
case !route.TunnelOK:
return HealthDown, "tunnel"
case !route.ServiceOK:
return HealthDown, "service"
default:
return HealthUp, ""
}
}
// recordRouteTraffic 记每条规则的上下行速率(scope 按服务端与域名区分)
func (c *Checker) recordRouteTraffic(ctx context.Context, routes []RouteStatus) {
if c.source == nil || c.metrics == nil {
return
}
domains := make([]string, 0, len(routes))
for _, route := range routes {
if domain := strings.TrimSpace(route.Domain); domain != "" {
domains = append(domains, domain)
}
}
if len(domains) == 0 {
return
}
rates, err := c.source.Traffic(ctx, domains)
if err != nil {
c.logger.Warn("读取流量快照失败", "module", "MONITOR", "error", err.Error())
return
}
samples := make([]metrics.Sample, 0, len(routes)*2)
for _, route := range routes {
domain := strings.ToLower(strings.TrimSpace(route.Domain))
rate, ok := rates[domain]
if !ok {
continue
}
scope := fmt.Sprintf("route:%d:%s", route.ServerID, domain)
samples = append(samples,
metrics.Sample{Scope: scope, Metric: metrics.MetricTrafficIn, Value: rate.Upload},
metrics.Sample{Scope: scope, Metric: metrics.MetricTrafficOut, Value: rate.Download},
)
}
if len(samples) == 0 {
return
}
if err := c.metrics.Record(ctx, samples); err != nil {
c.logger.Warn("记录流量历史失败", "module", "MONITOR", "error", err.Error())
}
}
// pruneHealthIfDue 每天清理一次超期记录:不在每轮都删,省一次写事务
func (c *Checker) pruneHealthIfDue(ctx context.Context) {
c.mu.Lock()
due := c.lastHealthPrune.IsZero() || time.Since(c.lastHealthPrune) >= 24*time.Hour
if due {
c.lastHealthPrune = time.Now()
}
c.mu.Unlock()
if !due {
return
}
if err := c.health.PruneRouteHealth(ctx); err != nil {
c.logger.Warn("清理健康记录失败", "module", "MONITOR", "error", err.Error())
}
}
// checkServers 公网服务端掉线:进程没在跑,或隧道客户端连不上 frps。
// 只对「用户要它运行」的服务端判定,手动停掉的不报。
func (c *Checker) checkServers(ctx context.Context) {
+3
View File
@@ -14,6 +14,7 @@ const (
EventFrpsServerDown = "frps_server_down"
EventAgentUnreachable = "agent_unreachable"
EventRouteDrift = "route_drift"
EventBackupFailure = "backup_failure"
MaskedSecret = "********"
@@ -125,6 +126,7 @@ type Config struct {
FrpsFailThreshold int `json:"frps_fail_threshold"`
AgentFailThreshold int `json:"agent_fail_threshold"`
DriftCooldownHours int `json:"drift_cooldown_hours"`
OnBackupFailure bool `json:"on_backup_failure"`
}
type RuntimeConfig struct {
@@ -166,6 +168,7 @@ type SaveInput struct {
FrpsFailThreshold int
AgentFailThreshold int
DriftCooldownHours int
OnBackupFailure bool
}
type TestInput struct {
+2
View File
@@ -96,6 +96,8 @@ func eventMetaFor(event string) eventMeta {
return eventMeta{"📡", "隧道"}
case EventRouteDrift:
return eventMeta{"🧭", "配置"}
case EventBackupFailure:
return eventMeta{"💾", "系统"}
case "test":
return eventMeta{"🔔", "系统"}
default:
+2
View File
@@ -252,6 +252,8 @@ func (s *Service) enabledFor(cfg RuntimeConfig, event string) bool {
return cfg.OnAgentUnreachable
case EventRouteDrift:
return cfg.OnRouteDrift
case EventBackupFailure:
return cfg.OnBackupFailure
default:
return true
}
+7
View File
@@ -99,6 +99,8 @@ func (s *Store) load(ctx context.Context, decrypt bool) (Config, runtimeSecrets,
cfg.OnFrpsDown = flag(s.settings, ctx, settings.KeyNotifyOnFrpsDown, false)
cfg.OnAgentUnreachable = flag(s.settings, ctx, settings.KeyNotifyOnAgentUnreachable, false)
cfg.OnRouteDrift = flag(s.settings, ctx, settings.KeyNotifyOnRouteDrift, false)
// 备份失败属于「你以为有备份其实没有」,默认开启
cfg.OnBackupFailure = flag(s.settings, ctx, settings.KeyNotifyOnBackupFailure, true)
} else {
legacyDDNS := flag(s.settings, ctx, settings.KeyNotifyOnDDNSError, true)
legacyCert := flag(s.settings, ctx, settings.KeyNotifyOnCertError, true)
@@ -202,6 +204,7 @@ func hasNotifyEventSettings(ctx context.Context, store *settings.Store) bool {
settings.KeyNotifyOnFrpsDown,
settings.KeyNotifyOnAgentUnreachable,
settings.KeyNotifyOnRouteDrift,
settings.KeyNotifyOnBackupFailure,
}
for _, key := range keys {
raw, err := store.Get(ctx, key)
@@ -271,6 +274,9 @@ func (s *Store) Save(ctx context.Context, in SaveInput) error {
if err := s.settings.SetBool(ctx, settings.KeyNotifyOnRouteDrift, in.OnRouteDrift); err != nil {
return err
}
if err := s.settings.SetBool(ctx, settings.KeyNotifyOnBackupFailure, in.OnBackupFailure); err != nil {
return err
}
if err := s.settings.SetInt(ctx, settings.KeyNotifyFrpsFailThreshold, clampThreshold(in.FrpsFailThreshold, DefaultFrpsFailThreshold)); err != nil {
return err
}
@@ -523,6 +529,7 @@ func ParseSaveInputFromMap(values map[string]string) (SaveInput, bool) {
in.OnFrpsDown = values[settings.KeyNotifyOnFrpsDown] == "1" || strings.EqualFold(values[settings.KeyNotifyOnFrpsDown], "true")
in.OnAgentUnreachable = values[settings.KeyNotifyOnAgentUnreachable] == "1" || strings.EqualFold(values[settings.KeyNotifyOnAgentUnreachable], "true")
in.OnRouteDrift = values[settings.KeyNotifyOnRouteDrift] == "1" || strings.EqualFold(values[settings.KeyNotifyOnRouteDrift], "true")
in.OnBackupFailure = values[settings.KeyNotifyOnBackupFailure] == "1" || strings.EqualFold(values[settings.KeyNotifyOnBackupFailure], "true")
in.FrpsFailThreshold = intSettingFromMap(values, settings.KeyNotifyFrpsFailThreshold, DefaultFrpsFailThreshold)
in.AgentFailThreshold = intSettingFromMap(values, settings.KeyNotifyAgentFailThreshold, DefaultAgentFailThreshold)
in.DriftCooldownHours = intSettingFromMap(values, settings.KeyNotifyDriftCooldownHours, DefaultDriftCooldownHours)
+29 -19
View File
@@ -52,25 +52,35 @@ const (
KeyNotifyFrpsFailThreshold = "notify_frps_fail_threshold"
KeyNotifyAgentFailThreshold = "notify_agent_fail_threshold"
KeyNotifyDriftCooldownHours = "notify_drift_cooldown_hours"
KeyTrustedProxy = "trusted_proxy_json"
KeyGlobalIPBlacklist = "global_ip_blacklist"
KeyGlobalIPWhitelist = "global_ip_whitelist"
KeyChinaCIDRUpdateHours = "china_cidr_update_interval_hours"
KeyChinaCIDRSourceV4 = "china_cidr_source_url_v4"
KeyChinaCIDRSourceV6 = "china_cidr_source_url_v6"
KeyChinaCIDRUpdatedAt = "china_cidr_updated_at"
KeyChinaCIDRCountV4 = "china_cidr_count_v4"
KeyChinaCIDRCountV6 = "china_cidr_count_v6"
KeyChinaCIDRLastError = "china_cidr_last_error"
KeyFRPEnabled = "frp_enabled"
KeyFRPServerAddr = "frp_server_addr"
KeyFRPServerPort = "frp_server_port"
KeyFRPAuthToken = "frp_auth_token"
KeyFRPTLSEnabled = "frp_tls_enabled"
KeyFRPCustomDomains = "frp_custom_domains"
KeyFRPLastError = "frp_last_error"
KeyFRPStartedAt = "frp_started_at"
KeyNginxGlobalMode = "nginx_global_mode"
KeyBackupAutoEnabled = "backup_auto_enabled"
KeyBackupKeepCount = "backup_keep_count"
KeyRouteHealthEnabled = "route_health_enabled"
KeyStatusPageEnabled = "status_page_enabled"
KeyMetricsHistoryEnabled = "metrics_history_enabled"
// KeyStatusPageCodeHash 存访问码的 SHA-256;明文只在保存请求里出现一次
KeyStatusPageCodeHash = "status_page_code_hash"
// KeyStatusPageCode 只是「保存请求」里用的字段名,不会被直接写库
KeyStatusPageCode = "status_page_code"
KeyNotifyOnBackupFailure = "notify_on_backup_failure"
KeyTrustedProxy = "trusted_proxy_json"
KeyGlobalIPBlacklist = "global_ip_blacklist"
KeyGlobalIPWhitelist = "global_ip_whitelist"
KeyChinaCIDRUpdateHours = "china_cidr_update_interval_hours"
KeyChinaCIDRSourceV4 = "china_cidr_source_url_v4"
KeyChinaCIDRSourceV6 = "china_cidr_source_url_v6"
KeyChinaCIDRUpdatedAt = "china_cidr_updated_at"
KeyChinaCIDRCountV4 = "china_cidr_count_v4"
KeyChinaCIDRCountV6 = "china_cidr_count_v6"
KeyChinaCIDRLastError = "china_cidr_last_error"
KeyFRPEnabled = "frp_enabled"
KeyFRPServerAddr = "frp_server_addr"
KeyFRPServerPort = "frp_server_port"
KeyFRPAuthToken = "frp_auth_token"
KeyFRPTLSEnabled = "frp_tls_enabled"
KeyFRPCustomDomains = "frp_custom_domains"
KeyFRPLastError = "frp_last_error"
KeyFRPStartedAt = "frp_started_at"
KeyNginxGlobalMode = "nginx_global_mode"
)
var Defaults = map[string]string{
+55 -2
View File
@@ -228,6 +228,15 @@ func (c *Collector) pruneHost(st *hostState, now time.Time) {
}
func (c *Collector) aggregateEndpoints(endpoints []proxy.Endpoint, now time.Time) (uploadTotal, downloadTotal int64, uploadRate, downloadRate float64, connections int) {
keys := make([]string, 0, len(endpoints))
for _, ep := range endpoints {
keys = append(keys, EndpointKey(ep.Hostname, ep.Port))
}
return c.aggregateHostKeys(keys, now)
}
// aggregateHostKeys 汇总若干 host:port 条目的总量、速率与活跃连接数
func (c *Collector) aggregateHostKeys(keys []string, now time.Time) (uploadTotal, downloadTotal int64, uploadRate, downloadRate float64, connections int) {
c.mu.RLock()
defer c.mu.RUnlock()
@@ -235,8 +244,7 @@ func (c *Collector) aggregateEndpoints(endpoints []proxy.Endpoint, now time.Time
var oldest time.Time
clientSet := map[string]time.Time{}
for _, ep := range endpoints {
key := EndpointKey(ep.Hostname, ep.Port)
for _, key := range keys {
st, ok := c.hosts[key]
if !ok {
continue
@@ -297,6 +305,51 @@ func (c *Collector) SnapshotForRules(rules []proxy.Rule) []RuleTraffic {
return out
}
// HostTraffic 单个域名的当前速率与总量(指标历史按域名落点)
type HostTraffic struct {
Hostname string `json:"hostname"`
UploadTotal int64 `json:"upload_total"`
DownloadTotal int64 `json:"download_total"`
UploadRate float64 `json:"upload_rate"`
DownloadRate float64 `json:"download_rate"`
Connections int `json:"connections"`
}
// SnapshotByHost 按域名汇总当前速率:内部键是 host:port,
// 同一域名的多个端口会合并,与 aggregateEndpoints 的算法保持一致。
func (c *Collector) SnapshotByHost(hostnames []string) map[string]HostTraffic {
now := time.Now()
out := map[string]HostTraffic{}
for _, host := range hostnames {
host = strings.ToLower(strings.TrimSpace(host))
if host == "" {
continue
}
prefix := host + ":"
c.mu.RLock()
keys := make([]string, 0, 2)
for key := range c.hosts {
if strings.HasPrefix(key, prefix) {
keys = append(keys, key)
}
}
c.mu.RUnlock()
if len(keys) == 0 {
continue
}
up, down, upRate, downRate, conns := c.aggregateHostKeys(keys, now)
out[host] = HostTraffic{
Hostname: host,
UploadTotal: up,
DownloadTotal: down,
UploadRate: upRate,
DownloadRate: downRate,
Connections: conns,
}
}
return out
}
func (c *Collector) ClientsForRule(rule proxy.Rule) []ClientConn {
now := time.Now()
activeCutoff := now.Add(-activeWindow)