fix: корректная агрегация top peers по диапазонам и починка hourly bucket
This commit is contained in:
+18
-4
@@ -96,10 +96,20 @@ func DashboardTrafficSummary() (vo.DashboardTrafficVo, error) {
|
|||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func DashboardTopPeers(fromMs int64, toMs int64, limit int) ([]vo.DashboardTopPeerVo, error) {
|
func DashboardTopPeers(fromMs int64, toMs int64, limit int, source string) ([]vo.DashboardTopPeerVo, error) {
|
||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 10
|
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)
|
rows := make([]vo.DashboardTopPeerVo, 0)
|
||||||
if tx := sqliteDB.Raw(`SELECT
|
if tx := sqliteDB.Raw(`SELECT
|
||||||
p.id AS peer_id,
|
p.id AS peer_id,
|
||||||
@@ -108,9 +118,9 @@ func DashboardTopPeers(fromMs int64, toMs int64, limit int) ([]vo.DashboardTopPe
|
|||||||
COALESCE(SUM(ts.rx_bytes),0) AS download,
|
COALESCE(SUM(ts.rx_bytes),0) AS download,
|
||||||
COALESCE(SUM(ts.tx_bytes),0) AS upload,
|
COALESCE(SUM(ts.tx_bytes),0) AS upload,
|
||||||
COALESCE(SUM(ts.rx_bytes + ts.tx_bytes),0) AS total
|
COALESCE(SUM(ts.rx_bytes + ts.tx_bytes),0) AS total
|
||||||
FROM traffic_sample ts
|
FROM `+sourceTable+` ts
|
||||||
JOIN peer p ON p.id = ts.peer_id
|
JOIN peer p ON p.id = ts.peer_id
|
||||||
WHERE ts.sampled_at BETWEEN ? AND ?
|
WHERE ts.`+timeColumn+` BETWEEN ? AND ?
|
||||||
GROUP BY p.id, p.name, p.remark
|
GROUP BY p.id, p.name, p.remark
|
||||||
ORDER BY total DESC
|
ORDER BY total DESC
|
||||||
LIMIT ?`, fromMs, toMs, limit).Scan(&rows); tx.Error != nil {
|
LIMIT ?`, fromMs, toMs, limit).Scan(&rows); tx.Error != nil {
|
||||||
@@ -141,6 +151,10 @@ func DashboardTrafficTimeseries(fromMs int64, toMs int64, bucketMs int64, source
|
|||||||
}
|
}
|
||||||
|
|
||||||
alignedFrom := fromMs - (fromMs % bucketMs)
|
alignedFrom := fromMs - (fromMs % bucketMs)
|
||||||
|
queryFrom := fromMs
|
||||||
|
if sourceTable == "traffic_aggregate_hourly" {
|
||||||
|
queryFrom = alignedFrom
|
||||||
|
}
|
||||||
if tx := sqliteDB.Raw(`SELECT
|
if tx := sqliteDB.Raw(`SELECT
|
||||||
(? + CAST((`+fromColumn+` - ?) / ? AS INTEGER) * ?) AS ts,
|
(? + CAST((`+fromColumn+` - ?) / ? AS INTEGER) * ?) AS ts,
|
||||||
COALESCE(SUM(rx_bytes),0) AS download,
|
COALESCE(SUM(rx_bytes),0) AS download,
|
||||||
@@ -148,7 +162,7 @@ func DashboardTrafficTimeseries(fromMs int64, toMs int64, bucketMs int64, source
|
|||||||
FROM `+sourceTable+`
|
FROM `+sourceTable+`
|
||||||
WHERE `+fromColumn+` BETWEEN ? AND ?
|
WHERE `+fromColumn+` BETWEEN ? AND ?
|
||||||
GROUP BY ts
|
GROUP BY ts
|
||||||
ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, fromMs, toMs).Scan(&rows); tx.Error != nil {
|
ORDER BY ts ASC`, alignedFrom, alignedFrom, bucketMs, bucketMs, queryFrom, toMs).Scan(&rows); tx.Error != nil {
|
||||||
logrus.Errorf("%v", tx.Error)
|
logrus.Errorf("%v", tx.Error)
|
||||||
return rows, errors.New(constant.SysError)
|
return rows, errors.New(constant.SysError)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
package entity
|
package entity
|
||||||
|
|
||||||
type TrafficAggregateDaily struct {
|
type TrafficAggregateDaily struct {
|
||||||
PeerId *int64 `gorm:"column:peer_id;default:0;primaryKey" json:"peerId"`
|
PeerId *int64 `gorm:"column:peer_id;default:0;primaryKey" json:"peerId"`
|
||||||
DayStart *int64 `gorm:"column:day_start;default:0;primaryKey" json:"dayStart"`
|
DayStart *int64 `gorm:"column:day_start;default:0;primaryKey" json:"dayStart"`
|
||||||
RxBytes *int64 `gorm:"column:rx_bytes;default:0" json:"rxBytes"`
|
RxBytes *int64 `gorm:"column:rx_bytes;default:0" json:"rxBytes"`
|
||||||
TxBytes *int64 `gorm:"column:tx_bytes;default:0" json:"txBytes"`
|
TxBytes *int64 `gorm:"column:tx_bytes;default:0" json:"txBytes"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func (TrafficAggregateDaily) TableName() string {
|
func (TrafficAggregateDaily) TableName() string {
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
package entity
|
package entity
|
||||||
|
|
||||||
type TrafficAggregateHourly struct {
|
type TrafficAggregateHourly struct {
|
||||||
PeerId *int64 `gorm:"column:peer_id;default:0;primaryKey" json:"peerId"`
|
PeerId *int64 `gorm:"column:peer_id;default:0;primaryKey" json:"peerId"`
|
||||||
HourStart *int64 `gorm:"column:hour_start;default:0;primaryKey" json:"hourStart"`
|
HourStart *int64 `gorm:"column:hour_start;default:0;primaryKey" json:"hourStart"`
|
||||||
RxBytes *int64 `gorm:"column:rx_bytes;default:0" json:"rxBytes"`
|
RxBytes *int64 `gorm:"column:rx_bytes;default:0" json:"rxBytes"`
|
||||||
TxBytes *int64 `gorm:"column:tx_bytes;default:0" json:"txBytes"`
|
TxBytes *int64 `gorm:"column:tx_bytes;default:0" json:"txBytes"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func (TrafficAggregateHourly) TableName() string {
|
func (TrafficAggregateHourly) TableName() string {
|
||||||
|
|||||||
@@ -90,11 +90,25 @@ func DashboardTimeseries(rangeKey string) (vo.DashboardTimeseriesVo, error) {
|
|||||||
|
|
||||||
func DashboardTopPeers(rangeKey string, limit int) ([]vo.DashboardTopPeerVo, error) {
|
func DashboardTopPeers(rangeKey string, limit int) ([]vo.DashboardTopPeerVo, error) {
|
||||||
nowMs := time.Now().UnixMilli()
|
nowMs := time.Now().UnixMilli()
|
||||||
|
normalizedRange := strings.TrimSpace(strings.ToLower(rangeKey))
|
||||||
fromMs := nowMs - int64(24*time.Hour/time.Millisecond)
|
fromMs := nowMs - int64(24*time.Hour/time.Millisecond)
|
||||||
if strings.TrimSpace(rangeKey) == "7d" {
|
source := "sample"
|
||||||
|
|
||||||
|
switch normalizedRange {
|
||||||
|
case "1h":
|
||||||
|
fromMs = nowMs - int64(time.Hour/time.Millisecond)
|
||||||
|
source = "sample"
|
||||||
|
case "7d":
|
||||||
fromMs = nowMs - int64(7*24*time.Hour/time.Millisecond)
|
fromMs = nowMs - int64(7*24*time.Hour/time.Millisecond)
|
||||||
|
source = "hourly"
|
||||||
|
case "24h", "":
|
||||||
|
fromMs = nowMs - int64(24*time.Hour/time.Millisecond)
|
||||||
|
source = "sample"
|
||||||
|
default:
|
||||||
|
fromMs = nowMs - int64(24*time.Hour/time.Millisecond)
|
||||||
|
source = "sample"
|
||||||
}
|
}
|
||||||
return dao.DashboardTopPeers(fromMs, nowMs, limit)
|
return dao.DashboardTopPeers(fromMs, nowMs, limit, source)
|
||||||
}
|
}
|
||||||
|
|
||||||
func DashboardSecurity() ([]vo.SecurityRiskVo, error) {
|
func DashboardSecurity() ([]vo.SecurityRiskVo, error) {
|
||||||
|
|||||||
Reference in New Issue
Block a user