实现总体流量按天统计、按小时统计，提供管理平台Dashboard API

internal/db/models/http_access_log_dao_test.go

@@ -41,7 +41,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0)
+	accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -68,7 +68,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page(t *testing.T) {
 	times := 0 // 防止循环次数太多
 	for {
 		before := time.Now()
-		accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0)
+		accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0)
 		cost := time.Since(before).Seconds()
 		if err != nil {
 			t.Fatal(err)
@@ -99,7 +99,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Reverse(t *testing.T) {
 	}
 
 	before := time.Now()
-	accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0)
+	accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0, false, 0)
 	cost := time.Since(before).Seconds()
 	if err != nil {
 		t.Fatal(err)
@@ -124,7 +124,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page_NotExists(t *testing.T) {
 	times := 0 // 防止循环次数太多
 	for {
 		before := time.Now()
-		accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd", time.Now().AddDate(0, 0, 1)), 0, false, false, 0, 0, 0)
+		accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd", time.Now().AddDate(0, 0, 1)), 0, false, false, 0, 0, 0, false, 0)
 		cost := time.Since(before).Seconds()
 		if err != nil {
 			t.Fatal(err)

internal/db/models/node_dao.go

@@ -163,6 +163,8 @@ func (this *NodeDAO) UpdateNode(tx *dbs.Tx, nodeId int64, name string, clusterId
 func (this *NodeDAO) CountAllEnabledNodes(tx *dbs.Tx) (int64, error) {
 	return this.Query(tx).
 		State(NodeStateEnabled).
+		Where("clusterId IN (SELECT id FROM "+SharedNodeClusterDAO.Table+" WHERE state=:clusterState)").
+		Param("clusterState", NodeClusterStateEnabled).
 		Count()
 }
 

internal/db/models/server_dao.go

@@ -592,6 +592,13 @@ func (this *ServerDAO) UpdateServerReverseProxy(tx *dbs.Tx, serverId int64, conf
 	return this.NotifyUpdate(tx, serverId)
 }
 
+// 计算所有可用服务数量
+func (this *ServerDAO) CountAllEnabledServers(tx *dbs.Tx) (int64, error) {
+	return this.Query(tx).
+		State(ServerStateEnabled).
+		Count()
+}
+
 // 计算所有可用服务数量
 func (this *ServerDAO) CountAllEnabledServersMatch(tx *dbs.Tx, groupId int64, keyword string, userId int64, clusterId int64, auditingFlag configutils.BoolState, protocolFamily string) (int64, error) {
 	query := this.Query(tx).

internal/db/models/traffic_daily_stat_dao.go

@@ -0,0 +1,75 @@
+package models
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"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"
+)
+
+type TrafficDailyStatDAO dbs.DAO
+
+func NewTrafficDailyStatDAO() *TrafficDailyStatDAO {
+	return dbs.NewDAO(&TrafficDailyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeTrafficDailyStats",
+			Model:  new(TrafficDailyStat),
+			PkName: "id",
+		},
+	}).(*TrafficDailyStatDAO)
+}
+
+var SharedTrafficDailyStatDAO *TrafficDailyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedTrafficDailyStatDAO = NewTrafficDailyStatDAO()
+	})
+}
+
+// 增加流量
+func (this *TrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, day string, bytes int64) error {
+	if len(day) != 8 {
+		return errors.New("invalid day '" + day + "'")
+	}
+	err := this.Query(tx).
+		Param("bytes", bytes).
+		InsertOrUpdateQuickly(maps.Map{
+			"day":   day,
+			"bytes": bytes,
+		}, maps.Map{
+			"bytes": dbs.SQL("bytes+:bytes"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// 获取日期之间统计
+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()
+	dayMap := map[string]*TrafficDailyStat{} // day => Stat
+	for _, one := range ones {
+		stat := one.(*TrafficDailyStat)
+		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, &TrafficDailyStat{Day: day})
+		}
+	}
+	return result, nil
+}

internal/db/models/traffic_daily_stat_dao_test.go

@@ -0,0 +1,20 @@
+package models
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/dbs"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"testing"
+	"time"
+)
+
+func TestTrafficDailyStatDAO_IncreaseDayBytes(t *testing.T) {
+	dbs.NotifyReady()
+
+	now := time.Now()
+	err := SharedTrafficDailyStatDAO.IncreaseDayBytes(nil, timeutil.Format("Ymd"), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok", time.Since(now).Seconds()*1000, "ms")
+}

internal/db/models/traffic_daily_stat_model.go

@@ -0,0 +1,18 @@
+package models
+
+// 总的流量统计
+type TrafficDailyStat struct {
+	Id    uint64 `field:"id"`    // ID
+	Day   string `field:"day"`   // YYYYMMDD
+	Bytes uint64 `field:"bytes"` // 流量字节
+}
+
+type TrafficDailyStatOperator struct {
+	Id    interface{} // ID
+	Day   interface{} // YYYYMMDD
+	Bytes interface{} // 流量字节
+}
+
+func NewTrafficDailyStatOperator() *TrafficDailyStatOperator {
+	return &TrafficDailyStatOperator{}
+}

internal/db/models/traffic_daily_stat_model_ext.go

@@ -0,0 +1 @@
+package models

internal/db/models/traffic_hourly_stat_dao.go

@@ -0,0 +1,75 @@
+package models
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"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"
+)
+
+type TrafficHourlyStatDAO dbs.DAO
+
+func NewTrafficHourlyStatDAO() *TrafficHourlyStatDAO {
+	return dbs.NewDAO(&TrafficHourlyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeTrafficHourlyStats",
+			Model:  new(TrafficHourlyStat),
+			PkName: "id",
+		},
+	}).(*TrafficHourlyStatDAO)
+}
+
+var SharedTrafficHourlyStatDAO *TrafficHourlyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedTrafficHourlyStatDAO = NewTrafficHourlyStatDAO()
+	})
+}
+
+// 增加流量
+func (this *TrafficHourlyStatDAO) IncreaseHourlyBytes(tx *dbs.Tx, hour string, bytes int64) error {
+	if len(hour) != 10 {
+		return errors.New("invalid hour '" + hour + "'")
+	}
+	err := this.Query(tx).
+		Param("bytes", bytes).
+		InsertOrUpdateQuickly(maps.Map{
+			"hour":  hour,
+			"bytes": bytes,
+		}, maps.Map{
+			"bytes": dbs.SQL("bytes+:bytes"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// 获取日期之间统计
+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()
+	hourMap := map[string]*TrafficHourlyStat{} // hour => Stat
+	for _, one := range ones {
+		stat := one.(*TrafficHourlyStat)
+		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, &TrafficHourlyStat{Hour: hour})
+		}
+	}
+	return result, nil
+}

internal/db/models/traffic_hourly_stat_dao_test.go

@@ -0,0 +1,20 @@
+package models
+
+import (
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/dbs"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"testing"
+	"time"
+)
+
+func TestTrafficHourlyStatDAO_IncreaseDayBytes(t *testing.T) {
+	dbs.NotifyReady()
+
+	now := time.Now()
+	err := SharedTrafficHourlyStatDAO.IncreaseDayBytes(nil, timeutil.Format("YmdH"), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok", time.Since(now).Seconds()*1000, "ms")
+}

internal/db/models/traffic_hourly_stat_model.go

@@ -0,0 +1,18 @@
+package models
+
+// 总的流量统计（按小时）
+type TrafficHourlyStat struct {
+	Id    uint64 `field:"id"`    // ID
+	Hour  string `field:"hour"`  // YYYYMMDDHH
+	Bytes uint64 `field:"bytes"` // 流量字节
+}
+
+type TrafficHourlyStatOperator struct {
+	Id    interface{} // ID
+	Hour  interface{} // YYYYMMDDHH
+	Bytes interface{} // 流量字节
+}
+
+func NewTrafficHourlyStatOperator() *TrafficHourlyStatOperator {
+	return &TrafficHourlyStatOperator{}
+}

internal/db/models/traffic_hourly_stat_model_ext.go

@@ -0,0 +1 @@
+package models

internal/rpc/services/service_admin.go

@@ -9,6 +9,8 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"time"
 )
 
 type AdminService struct {
@@ -450,3 +452,93 @@ func (this *AdminService) CheckAdminOTPWithUsername(ctx context.Context, req *pb
 	}
 	return &pb.CheckAdminOTPWithUsernameResponse{RequireOTP: otpIsOn}, nil
 }
+
+// 取得管理员Dashboard数据
+func (this *AdminService) ComposeAdminDashboard(ctx context.Context, req *pb.ComposeAdminDashboardRequest) (*pb.ComposeAdminDashboardResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	resp := &pb.ComposeAdminDashboardResponse{}
+
+	var tx = this.NullTx()
+
+	// 集群数
+	countClusters, err := models.SharedNodeClusterDAO.CountAllEnabledClusters(tx, "")
+	if err != nil {
+		return nil, err
+	}
+	resp.CountNodeClusters = countClusters
+
+	// 节点数
+	countNodes, err := models.SharedNodeDAO.CountAllEnabledNodes(tx)
+	if err != nil {
+		return nil, err
+	}
+	resp.CountNodes = countNodes
+
+	// 服务数
+	countServers, err := models.SharedServerDAO.CountAllEnabledServers(tx)
+	if err != nil {
+		return nil, err
+	}
+	resp.CountServers = countServers
+
+	// 用户数
+	countUsers, err := models.SharedUserDAO.CountAllEnabledUsers(tx, "")
+	if err != nil {
+		return nil, err
+	}
+	resp.CountUsers = countUsers
+
+	// API节点数
+	countAPINodes, err := models.SharedAPINodeDAO.CountAllEnabledAPINodes(tx)
+	if err != nil {
+		return nil, err
+	}
+	resp.CountAPINodes = countAPINodes
+
+	// 数据库节点数
+	countDBNodes, err := models.SharedDBNodeDAO.CountAllEnabledNodes(tx)
+	if err != nil {
+		return nil, err
+	}
+	resp.CountDBNodes = countDBNodes
+
+	// 用户节点数
+	countUserNodes, err := models.SharedUserNodeDAO.CountAllEnabledUserNodes(tx)
+	if err != nil {
+		return nil, err
+	}
+	resp.CountUserNodes = countUserNodes
+
+	// 按日流量统计
+	dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
+	dailyTrafficStats, err := models.SharedTrafficDailyStatDAO.FindDailyStats(tx, dayFrom, timeutil.Format("Ymd"))
+	if err != nil {
+		return nil, err
+	}
+	for _, stat := range dailyTrafficStats {
+		resp.DailyTrafficStats = append(resp.DailyTrafficStats, &pb.ComposeAdminDashboardResponse_DailyTrafficStat{
+			Day:   stat.Day,
+			Bytes: int64(stat.Bytes),
+		})
+	}
+
+	// 小时流量统计
+	hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
+	hourTo := timeutil.Format("YmdH")
+	hourlyTrafficStats, err := models.SharedTrafficHourlyStatDAO.FindHourlyStats(tx, hourFrom, hourTo)
+	if err != nil {
+		return nil, err
+	}
+	for _, stat := range hourlyTrafficStats {
+		resp.HourlyTrafficStats = append(resp.HourlyTrafficStats, &pb.ComposeAdminDashboardResponse_HourlyTrafficStat{
+			Hour:  stat.Hour,
+			Bytes: int64(stat.Bytes),
+		})
+	}
+
+	return resp, nil
+}

internal/rpc/services/service_server_daily_stat.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	timeutil "github.com/iwind/TeaGo/utils/time"
 )
 
 // 服务统计相关服务
@@ -24,5 +25,24 @@ func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context,
 	if err != nil {
 		return nil, err
 	}
+
+	// 写入其他统计表
+	// TODO 将来改成每小时入库一次
+	for _, stat := range req.Stats {
+		// 总体流量（按天）
+		err = models.SharedTrafficDailyStatDAO.IncreaseDailyBytes(tx, timeutil.FormatTime("Ymd", stat.CreatedAt), stat.Bytes)
+		if err != nil {
+			return nil, err
+		}
+
+		// 总体统计（按小时）
+		err = models.SharedTrafficHourlyStatDAO.IncreaseHourlyBytes(tx, timeutil.FormatTime("YmdH", stat.CreatedAt), stat.Bytes)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// TODO 集群流量/节点流量
+
 	return this.Success()
 }

internal/utils/time.go

@@ -0,0 +1,106 @@
+package utils
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/iwind/TeaGo/types"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"regexp"
+	"time"
+)
+
+// 计算日期之间的所有日期，格式为YYYYMMDD
+func RangeDays(dayFrom string, dayTo string) ([]string, error) {
+	ok, err := regexp.MatchString(`^\d{8}$`, dayFrom)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, errors.New("invalid 'dayFrom'")
+	}
+
+	ok, err = regexp.MatchString(`^\d{8}$`, dayTo)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, errors.New("invalid 'dayTo'")
+	}
+
+	if dayFrom > dayTo {
+		dayFrom, dayTo = dayTo, dayFrom
+	}
+
+	// 不能超过N天
+	maxDays := 100 - 1 // -1 是去掉默认加入的dayFrom
+	result := []string{dayFrom}
+
+	year := types.Int(dayFrom[:4])
+	month := types.Int(dayFrom[4:6])
+	day := types.Int(dayFrom[6:])
+	t := time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.Local)
+	for {
+		t = t.AddDate(0, 0, 1)
+		newDay := timeutil.Format("Ymd", t)
+		if newDay <= dayTo {
+			result = append(result, newDay)
+		} else {
+			break
+		}
+
+		maxDays--
+		if maxDays <= 0 {
+			break
+		}
+	}
+
+	return result, nil
+}
+
+// 计算小时之间的所有小时，格式为YYYYMMDDHH
+func RangeHours(hourFrom string, hourTo string) ([]string, error) {
+	ok, err := regexp.MatchString(`^\d{10}$`, hourFrom)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, errors.New("invalid 'hourFrom'")
+	}
+
+	ok, err = regexp.MatchString(`^\d{10}$`, hourTo)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, errors.New("invalid 'hourTo'")
+	}
+
+	if hourFrom > hourTo {
+		hourFrom, hourTo = hourTo, hourFrom
+	}
+
+	// 不能超过N天
+	maxHours := 100 - 1 // -1 是去掉默认加入的dayFrom
+	result := []string{hourFrom}
+
+	year := types.Int(hourFrom[:4])
+	month := types.Int(hourFrom[4:6])
+	day := types.Int(hourFrom[6:8])
+	hour := types.Int(hourFrom[8:])
+	t := time.Date(year, time.Month(month), day, hour, 0, 0, 0, time.Local)
+	for {
+		t = t.Add(1 * time.Hour)
+		newHour := timeutil.Format("YmdH", t)
+		if newHour <= hourTo {
+			result = append(result, newHour)
+		} else {
+			break
+		}
+
+		maxHours--
+		if maxHours <= 0 {
+			break
+		}
+	}
+
+	return result, nil
+}

internal/utils/time_test.go

@@ -0,0 +1,30 @@
+package utils
+
+import "testing"
+
+func TestRangeDays(t *testing.T) {
+	days, err := RangeDays("20210101", "20210115")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log(days)
+}
+
+
+func TestRangeHours(t *testing.T) {
+	{
+		hours, err := RangeHours("2021010100", "2021010123")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log(hours)
+	}
+
+	{
+		hours, err := RangeHours("2021010105", "2021010112")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log(hours)
+	}
+}