实现集群看板

internal/db/models/stats/node_cluster_traffic_daily_stat_dao.go

@@ -2,14 +2,33 @@ package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type NodeClusterTrafficDailyStatDAO dbs.DAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedNodeClusterTrafficDailyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("NodeClusterTrafficDailyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func NewNodeClusterTrafficDailyStatDAO() *NodeClusterTrafficDailyStatDAO {
 	return dbs.NewDAO(&NodeClusterTrafficDailyStatDAO{
 		DAOObject: dbs.DAOObject{
@@ -29,22 +48,69 @@ func init() {
 	})
 }
 
-// 增加流量
-func (this *NodeClusterTrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, clusterId int64, day string, bytes int64) error {
+// IncreaseDailyStat 增加统计数据
+func (this *NodeClusterTrafficDailyStatDAO) IncreaseDailyStat(tx *dbs.Tx, clusterId int64, day string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
 	if len(day) != 8 {
 		return errors.New("invalid day '" + day + "'")
 	}
 	err := this.Query(tx).
 		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
 		InsertOrUpdateQuickly(maps.Map{
-			"clusterId": clusterId,
-			"day":       day,
-			"bytes":     bytes,
+			"clusterId":           clusterId,
+			"day":                 day,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
 		}, maps.Map{
-			"bytes": dbs.SQL("bytes+:bytes"),
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
 		})
 	if err != nil {
 		return err
 	}
 	return nil
 }
+
+// FindDailyStats 获取日期之间统计
+func (this *NodeClusterTrafficDailyStatDAO) FindDailyStats(tx *dbs.Tx, clusterId int64, dayFrom string, dayTo string) (result []*NodeClusterTrafficDailyStat, err error) {
+	ones, err := this.Query(tx).
+		Attr("clusterId", clusterId).
+		Between("day", dayFrom, dayTo).
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	dayMap := map[string]*NodeClusterTrafficDailyStat{} // day => Stat
+	for _, one := range ones {
+		stat := one.(*NodeClusterTrafficDailyStat)
+		dayMap[stat.Day] = stat
+	}
+	days, err := utils.RangeDays(dayFrom, dayTo)
+	if err != nil {
+		return nil, err
+	}
+	for _, day := range days {
+		stat, ok := dayMap[day]
+		if ok {
+			result = append(result, stat)
+		} else {
+			result = append(result, &NodeClusterTrafficDailyStat{Day: day})
+		}
+	}
+	return result, nil
+}
+
+// Clean 清理历史数据
+func (this *NodeClusterTrafficDailyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("day", day).
+		Delete()
+	return err
+}

internal/db/models/stats/node_cluster_traffic_daily_stat_model.go

@@ -1,18 +1,24 @@
 package stats
 
-// 总的流量统计（按天）
+// NodeClusterTrafficDailyStat 总的流量统计（按天）
 type NodeClusterTrafficDailyStat struct {
-	Id        uint64 `field:"id"`        // ID
-	ClusterId uint32 `field:"clusterId"` // 集群ID
-	Day       string `field:"day"`       // YYYYMMDD
-	Bytes     uint64 `field:"bytes"`     // 流量字节
+	Id                  uint64 `field:"id"`                  // ID
+	ClusterId           uint32 `field:"clusterId"`           // 集群ID
+	Day                 string `field:"day"`                 // YYYYMMDD
+	Bytes               uint64 `field:"bytes"`               // 流量字节
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存的请求数
 }
 
 type NodeClusterTrafficDailyStatOperator struct {
-	Id        interface{} // ID
-	ClusterId interface{} // 集群ID
-	Day       interface{} // YYYYMMDD
-	Bytes     interface{} // 流量字节
+	Id                  interface{} // ID
+	ClusterId           interface{} // 集群ID
+	Day                 interface{} // YYYYMMDD
+	Bytes               interface{} // 流量字节
+	CachedBytes         interface{} // 缓存流量
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存的请求数
 }
 
 func NewNodeClusterTrafficDailyStatOperator() *NodeClusterTrafficDailyStatOperator {

internal/db/models/stats/node_traffic_daily_stat_dao.go

@@ -2,14 +2,32 @@ package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type NodeTrafficDailyStatDAO dbs.DAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedNodeTrafficDailyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("NodeTrafficDailyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func NewNodeTrafficDailyStatDAO() *NodeTrafficDailyStatDAO {
 	return dbs.NewDAO(&NodeTrafficDailyStatDAO{
 		DAOObject: dbs.DAOObject{
@@ -29,22 +47,42 @@ func init() {
 	})
 }
 
-// 增加流量
-func (this *NodeTrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, nodeId int64, day string, bytes int64) error {
+// IncreaseDailyStat 增加统计数据
+func (this *NodeTrafficDailyStatDAO) IncreaseDailyStat(tx *dbs.Tx, clusterId int64, role string, nodeId int64, day string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
 	if len(day) != 8 {
 		return errors.New("invalid day '" + day + "'")
 	}
 	err := this.Query(tx).
 		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
 		InsertOrUpdateQuickly(maps.Map{
-			"nodeId": nodeId,
-			"day":    day,
-			"bytes":  bytes,
+			"clusterId":           clusterId,
+			"role":                role,
+			"nodeId":              nodeId,
+			"day":                 day,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
 		}, maps.Map{
-			"bytes": dbs.SQL("bytes+:bytes"),
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
 		})
 	if err != nil {
 		return err
 	}
 	return nil
 }
+
+// Clean 清理历史数据
+func (this *NodeTrafficDailyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("day", day).
+		Delete()
+	return err
+}

internal/db/models/stats/node_traffic_daily_stat_model.go

@@ -1,18 +1,28 @@
 package stats
 
-// 总的流量统计（按天）
+// NodeTrafficDailyStat 总的流量统计（按天）
 type NodeTrafficDailyStat struct {
-	Id     uint64 `field:"id"`     // ID
-	NodeId uint32 `field:"nodeId"` // 集群ID
-	Day    string `field:"day"`    // YYYYMMDD
-	Bytes  uint64 `field:"bytes"`  // 流量字节
+	Id                  uint64 `field:"id"`                  // ID
+	Role                string `field:"role"`                // 节点角色
+	ClusterId           uint32 `field:"clusterId"`           // 集群ID
+	NodeId              uint32 `field:"nodeId"`              // 集群ID
+	Day                 string `field:"day"`                 // YYYYMMDD
+	Bytes               uint64 `field:"bytes"`               // 流量字节
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存的请求数
 }
 
 type NodeTrafficDailyStatOperator struct {
-	Id     interface{} // ID
-	NodeId interface{} // 集群ID
-	Day    interface{} // YYYYMMDD
-	Bytes  interface{} // 流量字节
+	Id                  interface{} // ID
+	Role                interface{} // 节点角色
+	ClusterId           interface{} // 集群ID
+	NodeId              interface{} // 集群ID
+	Day                 interface{} // YYYYMMDD
+	Bytes               interface{} // 流量字节
+	CachedBytes         interface{} // 缓存流量
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存的请求数
 }
 
 func NewNodeTrafficDailyStatOperator() *NodeTrafficDailyStatOperator {

internal/db/models/stats/node_traffic_hourly_stat_dao.go

@@ -0,0 +1,134 @@
+package stats
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
+)
+
+type NodeTrafficHourlyStatDAO dbs.DAO
+
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedNodeTrafficHourlyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("NodeTrafficHourlyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
+func NewNodeTrafficHourlyStatDAO() *NodeTrafficHourlyStatDAO {
+	return dbs.NewDAO(&NodeTrafficHourlyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeNodeTrafficHourlyStats",
+			Model:  new(NodeTrafficHourlyStat),
+			PkName: "id",
+		},
+	}).(*NodeTrafficHourlyStatDAO)
+}
+
+var SharedNodeTrafficHourlyStatDAO *NodeTrafficHourlyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedNodeTrafficHourlyStatDAO = NewNodeTrafficHourlyStatDAO()
+	})
+}
+
+// IncreaseHourlyStat 增加统计数据
+func (this *NodeTrafficHourlyStatDAO) IncreaseHourlyStat(tx *dbs.Tx, clusterId int64, role string, nodeId int64, hour string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
+	if len(hour) != 10 {
+		return errors.New("invalid hour '" + hour + "'")
+	}
+	err := this.Query(tx).
+		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
+		InsertOrUpdateQuickly(maps.Map{
+			"clusterId":           clusterId,
+			"role":                role,
+			"nodeId":              nodeId,
+			"hour":                hour,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
+		}, maps.Map{
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// FindHourlyStatsWithClusterId 获取小时之间统计
+func (this *NodeTrafficHourlyStatDAO) FindHourlyStatsWithClusterId(tx *dbs.Tx, clusterId int64, hourFrom string, hourTo string) (result []*NodeTrafficHourlyStat, err error) {
+	ones, err := this.Query(tx).
+		Attr("clusterId", clusterId).
+		Between("hour", hourFrom, hourTo).
+		Result("hour, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
+		Group("hour").
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	hourMap := map[string]*NodeTrafficHourlyStat{} // hour => Stat
+	for _, one := range ones {
+		stat := one.(*NodeTrafficHourlyStat)
+		hourMap[stat.Hour] = stat
+	}
+	hours, err := utils.RangeHours(hourFrom, hourTo)
+	if err != nil {
+		return nil, err
+	}
+	for _, hour := range hours {
+		stat, ok := hourMap[hour]
+		if ok {
+			result = append(result, stat)
+		} else {
+			result = append(result, &NodeTrafficHourlyStat{Hour: hour})
+		}
+	}
+	return result, nil
+}
+
+// FindTopNodeStatsWithClusterId 取得一定时间内的节点排行数据
+func (this *NodeTrafficHourlyStatDAO) FindTopNodeStatsWithClusterId(tx *dbs.Tx, clusterId int64, hourFrom string, hourTo string) (result []*NodeTrafficHourlyStat, err error) {
+	// TODO 节点如果已经被删除，则忽略
+	_, err = this.Query(tx).
+		Attr("clusterId", clusterId).
+		Between("hour", hourFrom, hourTo).
+		Result("nodeId, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
+		Group("nodeId").
+		Desc("countRequests").
+		Slice(&result).
+		FindAll()
+	return
+}
+
+// Clean 清理历史数据
+func (this *NodeTrafficHourlyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var hour = timeutil.Format("Ymd00", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("hour", hour).
+		Delete()
+	return err
+}

internal/db/models/stats/node_traffic_hourly_stat_dao_test.go

@@ -0,0 +1,6 @@
+package stats
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	_ "github.com/iwind/TeaGo/bootstrap"
+)

internal/db/models/stats/node_traffic_hourly_stat_model.go

@@ -0,0 +1,30 @@
+package stats
+
+// NodeTrafficHourlyStat 总的流量统计（按天）
+type NodeTrafficHourlyStat struct {
+	Id                  uint64 `field:"id"`                  // ID
+	Role                string `field:"role"`                // 节点角色
+	ClusterId           uint32 `field:"clusterId"`           // 集群ID
+	NodeId              uint32 `field:"nodeId"`              // 集群ID
+	Hour                string `field:"hour"`                // YYYYMMDDHH
+	Bytes               uint64 `field:"bytes"`               // 流量字节
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存的请求数
+}
+
+type NodeTrafficHourlyStatOperator struct {
+	Id                  interface{} // ID
+	Role                interface{} // 节点角色
+	ClusterId           interface{} // 集群ID
+	NodeId              interface{} // 集群ID
+	Hour                interface{} // YYYYMMDDHH
+	Bytes               interface{} // 流量字节
+	CachedBytes         interface{} // 缓存流量
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存的请求数
+}
+
+func NewNodeTrafficHourlyStatOperator() *NodeTrafficHourlyStatOperator {
+	return &NodeTrafficHourlyStatOperator{}
+}

internal/db/models/stats/node_traffic_hourly_stat_model_ext.go

@@ -0,0 +1 @@
+package stats

internal/db/models/stats/server_domain_hourly_stat_dao.go

@@ -0,0 +1,103 @@
+package stats
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
+)
+
+type ServerDomainHourlyStatDAO dbs.DAO
+
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedServerDomainHourlyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("ServerDomainHourlyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
+func NewServerDomainHourlyStatDAO() *ServerDomainHourlyStatDAO {
+	return dbs.NewDAO(&ServerDomainHourlyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeServerDomainHourlyStats",
+			Model:  new(ServerDomainHourlyStat),
+			PkName: "id",
+		},
+	}).(*ServerDomainHourlyStatDAO)
+}
+
+var SharedServerDomainHourlyStatDAO *ServerDomainHourlyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedServerDomainHourlyStatDAO = NewServerDomainHourlyStatDAO()
+	})
+}
+
+// IncreaseHourlyStat 增加统计数据
+func (this *ServerDomainHourlyStatDAO) IncreaseHourlyStat(tx *dbs.Tx, clusterId int64, nodeId int64, serverId int64, domain string, hour string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
+	if len(hour) != 10 {
+		return errors.New("invalid hour '" + hour + "'")
+	}
+	err := this.Query(tx).
+		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
+		InsertOrUpdateQuickly(maps.Map{
+			"clusterId":           clusterId,
+			"nodeId":              nodeId,
+			"serverId":            serverId,
+			"hour":                hour,
+			"domain":              domain,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
+		}, maps.Map{
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// FindTopDomainStatsWithClusterId 取得一定时间内的节点排行数据
+func (this *ServerDomainHourlyStatDAO) FindTopDomainStatsWithClusterId(tx *dbs.Tx, clusterId int64, hourFrom string, hourTo string) (result []*ServerDomainHourlyStat, err error) {
+	// TODO 节点如果已经被删除，则忽略
+	_, err = this.Query(tx).
+		Attr("clusterId", clusterId).
+		Between("hour", hourFrom, hourTo).
+		Result("domain, MIN(serverId) AS serverId, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
+		Group("domain").
+		Desc("countRequests").
+		Slice(&result).
+		FindAll()
+	return
+}
+
+// Clean 清理历史数据
+func (this *ServerDomainHourlyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var hour = timeutil.Format("Ymd00", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("hour", hour).
+		Delete()
+	return err
+}

internal/db/models/stats/server_domain_hourly_stat_dao_test.go

@@ -0,0 +1,6 @@
+package stats
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	_ "github.com/iwind/TeaGo/bootstrap"
+)

internal/db/models/stats/server_domain_hourly_stat_model.go

@@ -0,0 +1,32 @@
+package stats
+
+// ServerDomainHourlyStat 服务域名统计
+type ServerDomainHourlyStat struct {
+	Id                  uint64 `field:"id"`                  // ID
+	ClusterId           uint32 `field:"clusterId"`           // 集群ID
+	NodeId              uint32 `field:"nodeId"`              // 节点ID
+	ServerId            uint32 `field:"serverId"`            // 服务ID
+	Domain              string `field:"domain"`              // 域名
+	Hour                string `field:"hour"`                // YYYYMMDDHH
+	Bytes               uint64 `field:"bytes"`               // 流量
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存请求
+}
+
+type ServerDomainHourlyStatOperator struct {
+	Id                  interface{} // ID
+	ClusterId           interface{} // 集群ID
+	NodeId              interface{} // 节点ID
+	ServerId            interface{} // 服务ID
+	Domain              interface{} // 域名
+	Hour                interface{} // YYYYMMDDHH
+	Bytes               interface{} // 流量
+	CachedBytes         interface{} // 缓存流量
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存请求
+}
+
+func NewServerDomainHourlyStatOperator() *ServerDomainHourlyStatOperator {
+	return &ServerDomainHourlyStatOperator{}
+}

internal/db/models/stats/server_domain_hourly_stat_model_ext.go

@@ -0,0 +1 @@
+package stats

internal/db/models/stats/traffic_daily_stat_dao.go

@@ -2,15 +2,33 @@ package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type TrafficDailyStatDAO dbs.DAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedTrafficDailyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("TrafficDailyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func NewTrafficDailyStatDAO() *TrafficDailyStatDAO {
 	return dbs.NewDAO(&TrafficDailyStatDAO{
 		DAOObject: dbs.DAOObject{
@@ -30,18 +48,27 @@ func init() {
 	})
 }
 
-// 增加流量
-func (this *TrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, day string, bytes int64) error {
+// IncreaseDailyStat 增加统计数据
+func (this *TrafficDailyStatDAO) IncreaseDailyStat(tx *dbs.Tx, day string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
 	if len(day) != 8 {
 		return errors.New("invalid day '" + day + "'")
 	}
 	err := this.Query(tx).
 		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
 		InsertOrUpdateQuickly(maps.Map{
-			"day":   day,
-			"bytes": bytes,
+			"day":                 day,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
 		}, maps.Map{
-			"bytes": dbs.SQL("bytes+:bytes"),
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
 		})
 	if err != nil {
 		return err
@@ -49,11 +76,14 @@ func (this *TrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, day string, byte
 	return nil
 }
 
-// 获取日期之间统计
+// FindDailyStats 获取日期之间统计
 func (this *TrafficDailyStatDAO) FindDailyStats(tx *dbs.Tx, dayFrom string, dayTo string) (result []*TrafficDailyStat, err error) {
 	ones, err := this.Query(tx).
 		Between("day", dayFrom, dayTo).
 		FindAll()
+	if err != nil {
+		return nil, err
+	}
 	dayMap := map[string]*TrafficDailyStat{} // day => Stat
 	for _, one := range ones {
 		stat := one.(*TrafficDailyStat)
@@ -73,3 +103,12 @@ func (this *TrafficDailyStatDAO) FindDailyStats(tx *dbs.Tx, dayFrom string, dayT
 	}
 	return result, nil
 }
+
+// Clean 清理历史数据
+func (this *TrafficDailyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("day", day).
+		Delete()
+	return err
+}

internal/db/models/stats/traffic_daily_stat_model.go

@@ -1,16 +1,22 @@
 package stats
 
-// 总的流量统计
+// TrafficDailyStat 总的流量统计（按天）
 type TrafficDailyStat struct {
-	Id    uint64 `field:"id"`    // ID
-	Day   string `field:"day"`   // YYYYMMDD
-	Bytes uint64 `field:"bytes"` // 流量字节
+	Id                  uint64 `field:"id"`                  // ID
+	Day                 string `field:"day"`                 // YYYYMMDD
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	Bytes               uint64 `field:"bytes"`               // 流量字节
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存请求数
 }
 
 type TrafficDailyStatOperator struct {
-	Id    interface{} // ID
-	Day   interface{} // YYYYMMDD
-	Bytes interface{} // 流量字节
+	Id                  interface{} // ID
+	Day                 interface{} // YYYYMMDD
+	CachedBytes         interface{} // 缓存流量
+	Bytes               interface{} // 流量字节
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存请求数
 }
 
 func NewTrafficDailyStatOperator() *TrafficDailyStatOperator {

internal/db/models/stats/traffic_hourly_stat_dao.go

@@ -2,15 +2,33 @@ package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/maps"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type TrafficHourlyStatDAO dbs.DAO
 
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(24 * time.Hour)
+		go func() {
+			for range ticker.C {
+				err := SharedTrafficHourlyStatDAO.Clean(nil, 60) // 只保留60天
+				if err != nil {
+					remotelogs.Error("TrafficHourlyStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		}()
+	})
+}
+
 func NewTrafficHourlyStatDAO() *TrafficHourlyStatDAO {
 	return dbs.NewDAO(&TrafficHourlyStatDAO{
 		DAOObject: dbs.DAOObject{
@@ -30,18 +48,27 @@ func init() {
 	})
 }
 
-// 增加流量
-func (this *TrafficHourlyStatDAO) IncreaseHourlyBytes(tx *dbs.Tx, hour string, bytes int64) error {
+// IncreaseHourlyStat 增加流量
+func (this *TrafficHourlyStatDAO) IncreaseHourlyStat(tx *dbs.Tx, hour string, bytes int64, cachedBytes int64, countRequests int64, countCachedRequests int64) error {
 	if len(hour) != 10 {
 		return errors.New("invalid hour '" + hour + "'")
 	}
 	err := this.Query(tx).
 		Param("bytes", bytes).
+		Param("cachedBytes", cachedBytes).
+		Param("countRequests", countRequests).
+		Param("countCachedRequests", countCachedRequests).
 		InsertOrUpdateQuickly(maps.Map{
-			"hour":  hour,
-			"bytes": bytes,
+			"hour":                hour,
+			"bytes":               bytes,
+			"cachedBytes":         cachedBytes,
+			"countRequests":       countRequests,
+			"countCachedRequests": countCachedRequests,
 		}, maps.Map{
-			"bytes": dbs.SQL("bytes+:bytes"),
+			"bytes":               dbs.SQL("bytes+:bytes"),
+			"cachedBytes":         dbs.SQL("cachedBytes+:cachedBytes"),
+			"countRequests":       dbs.SQL("countRequests+:countRequests"),
+			"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
 		})
 	if err != nil {
 		return err
@@ -49,11 +76,14 @@ func (this *TrafficHourlyStatDAO) IncreaseHourlyBytes(tx *dbs.Tx, hour string, b
 	return nil
 }
 
-// 获取日期之间统计
+// FindHourlyStats 获取小时之间统计
 func (this *TrafficHourlyStatDAO) FindHourlyStats(tx *dbs.Tx, hourFrom string, hourTo string) (result []*TrafficHourlyStat, err error) {
 	ones, err := this.Query(tx).
 		Between("hour", hourFrom, hourTo).
 		FindAll()
+	if err != nil {
+		return nil, err
+	}
 	hourMap := map[string]*TrafficHourlyStat{} // hour => Stat
 	for _, one := range ones {
 		stat := one.(*TrafficHourlyStat)
@@ -73,3 +103,12 @@ func (this *TrafficHourlyStatDAO) FindHourlyStats(tx *dbs.Tx, hourFrom string, h
 	}
 	return result, nil
 }
+
+// Clean 清理历史数据
+func (this *TrafficHourlyStatDAO) Clean(tx *dbs.Tx, days int) error {
+	var hour = timeutil.Format("Ymd00", time.Now().AddDate(0, 0, -days))
+	_, err := this.Query(tx).
+		Lt("hour", hour).
+		Delete()
+	return err
+}

internal/db/models/stats/traffic_hourly_stat_model.go

@@ -1,16 +1,22 @@
 package stats
 
-// 总的流量统计（按小时）
+// TrafficHourlyStat 总的流量统计（按小时）
 type TrafficHourlyStat struct {
-	Id    uint64 `field:"id"`    // ID
-	Hour  string `field:"hour"`  // YYYYMMDDHH
-	Bytes uint64 `field:"bytes"` // 流量字节
+	Id                  uint64 `field:"id"`                  // ID
+	Hour                string `field:"hour"`                // YYYYMMDDHH
+	Bytes               uint64 `field:"bytes"`               // 流量字节
+	CachedBytes         uint64 `field:"cachedBytes"`         // 缓存流量
+	CountRequests       uint64 `field:"countRequests"`       // 请求数
+	CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存请求数
 }
 
 type TrafficHourlyStatOperator struct {
-	Id    interface{} // ID
-	Hour  interface{} // YYYYMMDDHH
-	Bytes interface{} // 流量字节
+	Id                  interface{} // ID
+	Hour                interface{} // YYYYMMDDHH
+	Bytes               interface{} // 流量字节
+	CachedBytes         interface{} // 缓存流量
+	CountRequests       interface{} // 请求数
+	CountCachedRequests interface{} // 缓存请求数
 }
 
 func NewTrafficHourlyStatOperator() *TrafficHourlyStatOperator {