package dao import ( "github.com/sirupsen/logrus" "hy2xs-admin/model/entity" "hy2xs-admin/model/vo" "strings" "time" ) func SaveMetricSample(sample entity.MetricSample) error { if tx := sqliteDB.Save(&sample); tx.Error != nil { logrus.Errorf("%v", tx.Error) return ErrStorage } return nil } func LastMetricSample() (entity.MetricSample, error) { var sample entity.MetricSample if tx := sqliteDB.Model(&entity.MetricSample{}).Order("sampled_at desc").Limit(1).Find(&sample); tx.Error != nil { logrus.Errorf("%v", tx.Error) return sample, ErrStorage } return sample, nil } func CleanupMetricSample(olderThanMs int64) error { if !tableExists("metric_sample") { return nil } if tx := sqliteDB.Exec("DELETE FROM metric_sample WHERE sampled_at < ?", olderThanMs); tx.Error != nil { logrus.Errorf("%v", tx.Error) return ErrStorage } return nil } func DashboardPeerSummary(nowMs int64) (vo.DashboardPeerVo, error) { result := vo.DashboardPeerVo{} type row struct { Total int64 Enabled int64 Disabled int64 Expired int64 } var r row if tx := sqliteDB.Raw(`SELECT COUNT(1) AS total, COALESCE(SUM(CASE WHEN disabled = 0 THEN 1 ELSE 0 END),0) AS enabled, COALESCE(SUM(CASE WHEN disabled = 1 THEN 1 ELSE 0 END),0) AS disabled, COALESCE(SUM(CASE WHEN expires_at > 0 AND expires_at < ? THEN 1 ELSE 0 END),0) AS expired FROM peer`, nowMs).Scan(&r); tx.Error != nil { logrus.Errorf("%v", tx.Error) return result, ErrStorage } result.Total = r.Total result.Enabled = r.Enabled result.Disabled = r.Disabled result.Expired = r.Expired return result, nil } func DashboardTrafficSummary() (vo.DashboardTrafficVo, error) { result := vo.DashboardTrafficVo{} type row struct { Download int64 Upload int64 } var r row if tx := sqliteDB.Raw(`SELECT COALESCE(SUM(download_bytes),0) AS download, COALESCE(SUM(upload_bytes),0) AS upload FROM peer`).Scan(&r); tx.Error != nil { logrus.Errorf("%v", tx.Error) return result, ErrStorage } result.DownloadBytes = r.Download result.UploadBytes = r.Upload result.TotalBytes = r.Download + r.Upload result.SinceResetDownloadBytes = r.Download result.SinceResetUploadBytes = r.Upload nowTime := time.Now() dayStart := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location()).UnixMilli() var today row if tx := sqliteDB.Raw(`SELECT COALESCE(SUM(rx_bytes),0) AS download, COALESCE(SUM(tx_bytes),0) AS upload FROM traffic_sample WHERE sampled_at >= ?`, dayStart).Scan(&today); tx.Error == nil { result.TodayDownloadBytes = today.Download result.TodayUploadBytes = today.Upload } return result, nil } func DashboardTopPeers(fromMs int64, toMs int64, limit int, source string) ([]vo.DashboardTopPeerVo, error) { if limit <= 0 { limit = 10 } sourceTable := "traffic_sample" timeColumn := "sampled_at" switch strings.TrimSpace(strings.ToLower(source)) { case "hourly": sourceTable = "traffic_aggregate_hourly" timeColumn = "hour_start" default: sourceTable = "traffic_sample" timeColumn = "sampled_at" } rows := make([]vo.DashboardTopPeerVo, 0) if tx := sqliteDB.Raw(`SELECT p.id AS peer_id, p.name AS name, p.remark AS remark, COALESCE(SUM(ts.rx_bytes),0) AS download, COALESCE(SUM(ts.tx_bytes),0) AS upload, COALESCE(SUM(ts.rx_bytes + ts.tx_bytes),0) AS total FROM `+sourceTable+` ts JOIN peer p ON p.id = ts.peer_id WHERE ts.`+timeColumn+` BETWEEN ? AND ? GROUP BY p.id, p.name, p.remark ORDER BY total DESC LIMIT ?`, fromMs, toMs, limit).Scan(&rows); tx.Error != nil { logrus.Errorf("%v", tx.Error) return rows, ErrStorage } return rows, nil } func DashboardTrafficTimeseries(fromMs int64, toMs int64, bucketMs int64, source string) ([]vo.DashboardSeriesPointVo, error) { rows := make([]vo.DashboardSeriesPointVo, 0) if toMs < fromMs { return rows, nil } if bucketMs <= 0 { bucketMs = int64(time.Minute / time.Millisecond) } sourceTable := "traffic_sample" fromColumn := "sampled_at" switch strings.TrimSpace(strings.ToLower(source)) { case "hourly": sourceTable = "traffic_aggregate_hourly" fromColumn = "hour_start" default: sourceTable = "traffic_sample" fromColumn = "sampled_at" } alignedFrom := fromMs - (fromMs % bucketMs) queryFrom := fromMs if sourceTable == "traffic_aggregate_hourly" { queryFrom = alignedFrom } if tx := sqliteDB.Raw(`SELECT (? + CAST((`+fromColumn+` - ?) / ? AS INTEGER) * ?) AS ts, COALESCE(SUM(rx_bytes),0) AS download, COALESCE(SUM(tx_bytes),0) AS upload FROM `+sourceTable+` WHERE `+fromColumn+` BETWEEN ? AND ? GROUP BY ts ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, queryFrom, toMs).Scan(&rows); tx.Error != nil { logrus.Errorf("%v", tx.Error) return rows, ErrStorage } rowMap := make(map[int64]vo.DashboardSeriesPointVo, len(rows)) for _, item := range rows { rowMap[item.Ts] = item } filled := make([]vo.DashboardSeriesPointVo, 0, int((toMs-alignedFrom)/bucketMs)+1) for ts := alignedFrom; ts <= toMs; ts += bucketMs { if item, ok := rowMap[ts]; ok { filled = append(filled, item) continue } filled = append(filled, vo.DashboardSeriesPointVo{Ts: ts, Download: 0, Upload: 0}) } return filled, nil } func DashboardSystemTimeseries(fromMs int64, toMs int64, bucketMs int64) ([]vo.DashboardSeriesPointVo, error) { rows := make([]vo.DashboardSeriesPointVo, 0) if !tableExists("metric_sample") { return rows, nil } if toMs < fromMs { return rows, nil } if bucketMs <= 0 { bucketMs = int64(time.Minute / time.Millisecond) } alignedFrom := fromMs - (fromMs % bucketMs) if tx := sqliteDB.Raw(`SELECT (? + CAST((sampled_at - ?) / ? AS INTEGER) * ?) AS ts, AVG(cpu_percent) AS cpu, AVG(mem_percent) AS mem FROM metric_sample WHERE sampled_at BETWEEN ? AND ? GROUP BY ts ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, fromMs, toMs).Scan(&rows); tx.Error != nil { logrus.Errorf("%v", tx.Error) return rows, ErrStorage } return rows, nil }