chore: sync local updates
This commit is contained in:
295
mengyaping-backend/storage/probe.go
Normal file
295
mengyaping-backend/storage/probe.go
Normal file
@@ -0,0 +1,295 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"mengyaping-backend/config"
|
||||
"mengyaping-backend/models"
|
||||
)
|
||||
|
||||
func latestToRecord(p *MonitorProbeLatest) models.MonitorRecord {
|
||||
return models.MonitorRecord{
|
||||
WebsiteID: p.WebsiteID,
|
||||
URLID: p.URLID,
|
||||
URL: p.URL,
|
||||
StatusCode: p.StatusCode,
|
||||
Latency: p.LatencyMs,
|
||||
IsUp: p.IsUp,
|
||||
Error: p.ErrorText,
|
||||
CheckedAt: p.CheckedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// AddRecord 写入最新探测 + 按小时/按天汇总（不插逐条历史）
|
||||
func (s *Storage) AddRecord(record models.MonitorRecord) error {
|
||||
hourAt := record.CheckedAt.Truncate(time.Hour)
|
||||
statDate := time.Date(record.CheckedAt.Year(), record.CheckedAt.Month(), record.CheckedAt.Day(),
|
||||
0, 0, 0, 0, record.CheckedAt.Location())
|
||||
upDelta := 0
|
||||
if record.IsUp {
|
||||
upDelta = 1
|
||||
}
|
||||
|
||||
return s.db.Transaction(func(tx *gorm.DB) error {
|
||||
lp := MonitorProbeLatest{
|
||||
WebsiteID: record.WebsiteID,
|
||||
URLID: record.URLID,
|
||||
URL: record.URL,
|
||||
StatusCode: record.StatusCode,
|
||||
LatencyMs: record.Latency,
|
||||
IsUp: record.IsUp,
|
||||
ErrorText: record.Error,
|
||||
CheckedAt: record.CheckedAt,
|
||||
}
|
||||
if err := tx.Save(&lp).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Exec(`
|
||||
INSERT INTO monitor_probe_hour (website_id, url_id, hour_at, probe_count, up_count, latency_sum)
|
||||
VALUES (?, ?, ?, 1, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
probe_count = probe_count + 1,
|
||||
up_count = up_count + VALUES(up_count),
|
||||
latency_sum = latency_sum + VALUES(latency_sum)
|
||||
`, record.WebsiteID, record.URLID, hourAt, upDelta, record.Latency).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Exec(`
|
||||
INSERT INTO monitor_probe_day (website_id, url_id, stat_date, probe_count, up_count, latency_sum)
|
||||
VALUES (?, ?, ?, 1, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
probe_count = probe_count + 1,
|
||||
up_count = up_count + VALUES(up_count),
|
||||
latency_sum = latency_sum + VALUES(latency_sum)
|
||||
`, record.WebsiteID, record.URLID, statDate.Format("2006-01-02"), upDelta, record.Latency).Error
|
||||
})
|
||||
}
|
||||
|
||||
// GetLatestRecord 最新一条（无记录时返回 nil；用 Find 而非 First，避免 GORM 将「未找到」打成 record not found 日志）
|
||||
func (s *Storage) GetLatestRecord(websiteID, urlID string) *models.MonitorRecord {
|
||||
var p MonitorProbeLatest
|
||||
if err := s.db.Where("website_id = ? AND url_id = ?", websiteID, urlID).Limit(1).Find(&p).Error; err != nil {
|
||||
return nil
|
||||
}
|
||||
if p.WebsiteID == "" {
|
||||
return nil
|
||||
}
|
||||
m := latestToRecord(&p)
|
||||
return &m
|
||||
}
|
||||
|
||||
// GetRecords 兼容旧接口：按小时聚合展开为「代表点」（每探测窗口 1 条），仅用于仍依赖 slice 的逻辑；新逻辑请用汇总 API
|
||||
func (s *Storage) GetRecords(websiteID, urlID string, since time.Time) []models.MonitorRecord {
|
||||
rows := s.getProbeHourRows(websiteID, urlID, since)
|
||||
var out []models.MonitorRecord
|
||||
for _, h := range rows {
|
||||
if h.ProbeCount <= 0 {
|
||||
continue
|
||||
}
|
||||
avgLat := h.LatencySum / int64(h.ProbeCount)
|
||||
isUp := h.UpCount*2 >= h.ProbeCount
|
||||
out = append(out, models.MonitorRecord{
|
||||
WebsiteID: websiteID,
|
||||
URLID: urlID,
|
||||
StatusCode: 0,
|
||||
Latency: avgLat,
|
||||
IsUp: isUp,
|
||||
CheckedAt: h.HourAt,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *Storage) getProbeHourRows(websiteID, urlID string, since time.Time) []MonitorProbeHour {
|
||||
var rows []MonitorProbeHour
|
||||
s.db.Where("website_id = ? AND url_id = ? AND hour_at >= ?", websiteID, urlID, since).
|
||||
Order("hour_at ASC").Find(&rows)
|
||||
return rows
|
||||
}
|
||||
|
||||
// GetProbeHourRowsForURL 某 URL 自 since 起的小时汇总行（升序）
|
||||
func (s *Storage) GetProbeHourRowsForURL(websiteID, urlID string, since time.Time) []MonitorProbeHour {
|
||||
return s.getProbeHourRows(websiteID, urlID, since)
|
||||
}
|
||||
|
||||
// GetProbeLatestKeyMap 全表 latest（每 URL 一行），key = website_id + "\x00" + url_id
|
||||
func (s *Storage) GetProbeLatestKeyMap() map[string]models.MonitorRecord {
|
||||
var rows []MonitorProbeLatest
|
||||
if err := s.db.Find(&rows).Error; err != nil {
|
||||
log.Printf("GetProbeLatestKeyMap: %v", err)
|
||||
return map[string]models.MonitorRecord{}
|
||||
}
|
||||
m := make(map[string]models.MonitorRecord, len(rows))
|
||||
for i := range rows {
|
||||
k := rows[i].WebsiteID + "\x00" + rows[i].URLID
|
||||
m[k] = latestToRecord(&rows[i])
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// GroupProbeHoursSince 近窗口内全部小时汇总，按站点、URL 分组（列表接口一次查出）
|
||||
func (s *Storage) GroupProbeHoursSince(since time.Time) map[string]map[string][]MonitorProbeHour {
|
||||
var rows []MonitorProbeHour
|
||||
if err := s.db.Where("hour_at >= ?", since).Order("website_id, url_id, hour_at").Find(&rows).Error; err != nil {
|
||||
log.Printf("GroupProbeHoursSince: %v", err)
|
||||
return map[string]map[string][]MonitorProbeHour{}
|
||||
}
|
||||
out := make(map[string]map[string][]MonitorProbeHour)
|
||||
for i := range rows {
|
||||
h := rows[i]
|
||||
if out[h.WebsiteID] == nil {
|
||||
out[h.WebsiteID] = make(map[string][]MonitorProbeHour)
|
||||
}
|
||||
out[h.WebsiteID][h.URLID] = append(out[h.WebsiteID][h.URLID], h)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// LoadAllWebsiteDayRollups 自 since 日起各站每日汇总 + 90d 整体可用率（单条 SQL 聚合）
|
||||
func (s *Storage) LoadAllWebsiteDayRollups(since time.Time) (map[string][]models.DailyStats, map[string]float64) {
|
||||
sinceDay := time.Date(since.Year(), since.Month(), since.Day(), 0, 0, 0, 0, since.Location())
|
||||
sinceStr := sinceDay.Format("2006-01-02")
|
||||
type aggRow struct {
|
||||
Wid string `gorm:"column:wid"`
|
||||
D time.Time `gorm:"column:d"`
|
||||
Pc int64 `gorm:"column:pc"`
|
||||
Uc int64 `gorm:"column:uc"`
|
||||
Ls int64 `gorm:"column:ls"`
|
||||
}
|
||||
var agg []aggRow
|
||||
if err := s.db.Raw(`
|
||||
SELECT website_id AS wid, stat_date AS d, SUM(probe_count) AS pc, SUM(up_count) AS uc, SUM(latency_sum) AS ls
|
||||
FROM monitor_probe_day
|
||||
WHERE stat_date >= ?
|
||||
GROUP BY website_id, stat_date
|
||||
ORDER BY website_id, stat_date
|
||||
`, sinceStr).Scan(&agg).Error; err != nil {
|
||||
log.Printf("LoadAllWebsiteDayRollups: %v", err)
|
||||
return map[string][]models.DailyStats{}, map[string]float64{}
|
||||
}
|
||||
|
||||
daily := make(map[string][]models.DailyStats)
|
||||
totP := make(map[string]int64)
|
||||
totU := make(map[string]int64)
|
||||
for _, r := range agg {
|
||||
st := models.DailyStats{
|
||||
Date: r.D,
|
||||
TotalCount: int(r.Pc),
|
||||
UpCount: int(r.Uc),
|
||||
}
|
||||
if r.Pc > 0 {
|
||||
st.AvgLatency = r.Ls / r.Pc
|
||||
st.Uptime = float64(r.Uc) / float64(r.Pc) * 100
|
||||
}
|
||||
daily[r.Wid] = append(daily[r.Wid], st)
|
||||
totP[r.Wid] += r.Pc
|
||||
totU[r.Wid] += r.Uc
|
||||
}
|
||||
uptime := make(map[string]float64)
|
||||
for wid, p := range totP {
|
||||
if p > 0 {
|
||||
uptime[wid] = float64(totU[wid]) / float64(p) * 100
|
||||
}
|
||||
}
|
||||
return daily, uptime
|
||||
}
|
||||
|
||||
// GetProbeHourTotals 某 URL 在时间窗口内汇总（来自小时表）
|
||||
func (s *Storage) GetProbeHourTotals(websiteID, urlID string, since time.Time) (probeTotal int64, upTotal int64, latencySum int64) {
|
||||
type row struct {
|
||||
P int64
|
||||
U int64
|
||||
L int64
|
||||
}
|
||||
var r row
|
||||
s.db.Model(&MonitorProbeHour{}).
|
||||
Select("COALESCE(SUM(probe_count),0), COALESCE(SUM(up_count),0), COALESCE(SUM(latency_sum),0)").
|
||||
Where("website_id = ? AND url_id = ? AND hour_at >= ?", websiteID, urlID, since).
|
||||
Scan(&r)
|
||||
return r.P, r.U, r.L
|
||||
}
|
||||
|
||||
// GetHourlyStatsForURL 7 天折线用（已聚合）
|
||||
func (s *Storage) GetHourlyStatsForURL(websiteID, urlID string, since time.Time) []models.HourlyStats {
|
||||
rows := s.getProbeHourRows(websiteID, urlID, since)
|
||||
out := make([]models.HourlyStats, 0, len(rows))
|
||||
for _, h := range rows {
|
||||
st := models.HourlyStats{
|
||||
Hour: h.HourAt,
|
||||
TotalCount: h.ProbeCount,
|
||||
UpCount: h.UpCount,
|
||||
}
|
||||
if h.ProbeCount > 0 {
|
||||
st.AvgLatency = h.LatencySum / int64(h.ProbeCount)
|
||||
st.Uptime = float64(h.UpCount) / float64(h.ProbeCount) * 100
|
||||
}
|
||||
out = append(out, st)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// GetWebsiteDailyAggregates 站点维度按日合并（多 URL 汇总到同一天）
|
||||
func (s *Storage) GetWebsiteDailyAggregates(websiteID string, since time.Time) []models.DailyStats {
|
||||
type aggRow struct {
|
||||
D time.Time `gorm:"column:d"`
|
||||
Pc int64 `gorm:"column:pc"`
|
||||
Uc int64 `gorm:"column:uc"`
|
||||
Ls int64 `gorm:"column:ls"`
|
||||
}
|
||||
var rows []aggRow
|
||||
sinceDay := time.Date(since.Year(), since.Month(), since.Day(), 0, 0, 0, 0, since.Location())
|
||||
s.db.Raw(`
|
||||
SELECT stat_date AS d, SUM(probe_count) AS pc, SUM(up_count) AS uc, SUM(latency_sum) AS ls
|
||||
FROM monitor_probe_day
|
||||
WHERE website_id = ? AND stat_date >= ?
|
||||
GROUP BY stat_date
|
||||
ORDER BY stat_date ASC
|
||||
`, websiteID, sinceDay.Format("2006-01-02")).Scan(&rows)
|
||||
|
||||
out := make([]models.DailyStats, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
st := models.DailyStats{
|
||||
Date: r.D,
|
||||
TotalCount: int(r.Pc),
|
||||
UpCount: int(r.Uc),
|
||||
}
|
||||
if r.Pc > 0 {
|
||||
st.AvgLatency = r.Ls / r.Pc
|
||||
st.Uptime = float64(r.Uc) / float64(r.Pc) * 100
|
||||
}
|
||||
out = append(out, st)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// GetWebsiteDayProbeTotals 整站在日期范围内探测次数汇总（用于 90 天可用率）
|
||||
func (s *Storage) GetWebsiteDayProbeTotals(websiteID string, since time.Time) (probeTotal int64, upTotal int64) {
|
||||
sinceDay := time.Date(since.Year(), since.Month(), since.Day(), 0, 0, 0, 0, since.Location())
|
||||
type row struct {
|
||||
P int64
|
||||
U int64
|
||||
}
|
||||
var r row
|
||||
s.db.Model(&MonitorProbeDay{}).
|
||||
Select("COALESCE(SUM(probe_count),0), COALESCE(SUM(up_count),0)").
|
||||
Where("website_id = ? AND stat_date >= ?", websiteID, sinceDay.Format("2006-01-02")).
|
||||
Scan(&r)
|
||||
return r.P, r.U
|
||||
}
|
||||
|
||||
func (s *Storage) purgeProbeRollups() {
|
||||
cfg := config.GetConfig()
|
||||
now := time.Now()
|
||||
dayCutoff := now.AddDate(0, 0, -cfg.Monitor.HistoryDays).Truncate(24 * time.Hour)
|
||||
if err := s.db.Where("stat_date < ?", dayCutoff.Format("2006-01-02")).Delete(&MonitorProbeDay{}).Error; err != nil {
|
||||
log.Printf("清理 monitor_probe_day: %v", err)
|
||||
}
|
||||
// 小时图保留约 10 天（覆盖 7 天曲线 + 余量）
|
||||
hourCutoff := now.Add(-10 * 24 * time.Hour).Truncate(time.Hour)
|
||||
if err := s.db.Where("hour_at < ?", hourCutoff).Delete(&MonitorProbeHour{}).Error; err != nil {
|
||||
log.Printf("清理 monitor_probe_hour: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user