记录节点/集群的流量统计以备将来使用

internal/db/models/stats/node_cluster_traffic_daily_stat_dao.go

@@ -0,0 +1,50 @@
+package stats
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+)
+
+type NodeClusterTrafficDailyStatDAO dbs.DAO
+
+func NewNodeClusterTrafficDailyStatDAO() *NodeClusterTrafficDailyStatDAO {
+	return dbs.NewDAO(&NodeClusterTrafficDailyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeNodeClusterTrafficDailyStats",
+			Model:  new(NodeClusterTrafficDailyStat),
+			PkName: "id",
+		},
+	}).(*NodeClusterTrafficDailyStatDAO)
+}
+
+var SharedNodeClusterTrafficDailyStatDAO *NodeClusterTrafficDailyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedNodeClusterTrafficDailyStatDAO = NewNodeClusterTrafficDailyStatDAO()
+	})
+}
+
+// 增加流量
+func (this *NodeClusterTrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, clusterId int64, 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{
+			"clusterId": clusterId,
+			"day":       day,
+			"bytes":     bytes,
+		}, maps.Map{
+			"bytes": dbs.SQL("bytes+:bytes"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}

internal/db/models/stats/node_cluster_traffic_daily_stat_dao_test.go

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

internal/db/models/stats/node_cluster_traffic_daily_stat_model.go

@@ -0,0 +1,20 @@
+package stats
+
+// 总的流量统计（按天）
+type NodeClusterTrafficDailyStat struct {
+	Id        uint64 `field:"id"`        // ID
+	ClusterId uint32 `field:"clusterId"` // 集群ID
+	Day       string `field:"day"`       // YYYYMMDD
+	Bytes     uint64 `field:"bytes"`     // 流量字节
+}
+
+type NodeClusterTrafficDailyStatOperator struct {
+	Id        interface{} // ID
+	ClusterId interface{} // 集群ID
+	Day       interface{} // YYYYMMDD
+	Bytes     interface{} // 流量字节
+}
+
+func NewNodeClusterTrafficDailyStatOperator() *NodeClusterTrafficDailyStatOperator {
+	return &NodeClusterTrafficDailyStatOperator{}
+}

internal/db/models/stats/node_cluster_traffic_daily_stat_model_ext.go

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

internal/db/models/stats/node_traffic_daily_stat_dao.go

@@ -0,0 +1,50 @@
+package stats
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/maps"
+)
+
+type NodeTrafficDailyStatDAO dbs.DAO
+
+func NewNodeTrafficDailyStatDAO() *NodeTrafficDailyStatDAO {
+	return dbs.NewDAO(&NodeTrafficDailyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeNodeTrafficDailyStats",
+			Model:  new(NodeTrafficDailyStat),
+			PkName: "id",
+		},
+	}).(*NodeTrafficDailyStatDAO)
+}
+
+var SharedNodeTrafficDailyStatDAO *NodeTrafficDailyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedNodeTrafficDailyStatDAO = NewNodeTrafficDailyStatDAO()
+	})
+}
+
+// 增加流量
+func (this *NodeTrafficDailyStatDAO) IncreaseDailyBytes(tx *dbs.Tx, nodeId int64, 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{
+			"nodeId": nodeId,
+			"day":    day,
+			"bytes":  bytes,
+		}, maps.Map{
+			"bytes": dbs.SQL("bytes+:bytes"),
+		})
+	if err != nil {
+		return err
+	}
+	return nil
+}

internal/db/models/stats/node_traffic_daily_stat_dao_test.go

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

internal/db/models/stats/node_traffic_daily_stat_model.go

@@ -0,0 +1,20 @@
+package stats
+
+// 总的流量统计（按天）
+type NodeTrafficDailyStat struct {
+	Id     uint64 `field:"id"`     // ID
+	NodeId uint32 `field:"nodeId"` // 集群ID
+	Day    string `field:"day"`    // YYYYMMDD
+	Bytes  uint64 `field:"bytes"`  // 流量字节
+}
+
+type NodeTrafficDailyStatOperator struct {
+	Id     interface{} // ID
+	NodeId interface{} // 集群ID
+	Day    interface{} // YYYYMMDD
+	Bytes  interface{} // 流量字节
+}
+
+func NewNodeTrafficDailyStatOperator() *NodeTrafficDailyStatOperator {
+	return &NodeTrafficDailyStatOperator{}
+}

internal/db/models/stats/node_traffic_daily_stat_model_ext.go

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

internal/db/models/stats/traffic_daily_stat_dao.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"

internal/db/models/stats/traffic_daily_stat_dao_test.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 import (
 	_ "github.com/go-sql-driver/mysql"
@@ -12,7 +12,7 @@ func TestTrafficDailyStatDAO_IncreaseDayBytes(t *testing.T) {
 	dbs.NotifyReady()
 
 	now := time.Now()
-	err := SharedTrafficDailyStatDAO.IncreaseDayBytes(nil, timeutil.Format("Ymd"), 1)
+	err := SharedTrafficDailyStatDAO.IncreaseDailyBytes(nil, timeutil.Format("Ymd"), 1)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/db/models/stats/traffic_daily_stat_model.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 // 总的流量统计
 type TrafficDailyStat struct {

internal/db/models/stats/traffic_daily_stat_model_ext.go

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

internal/db/models/stats/traffic_hourly_stat_dao.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"

internal/db/models/stats/traffic_hourly_stat_dao_test.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 import (
 	_ "github.com/go-sql-driver/mysql"
@@ -12,7 +12,7 @@ func TestTrafficHourlyStatDAO_IncreaseDayBytes(t *testing.T) {
 	dbs.NotifyReady()
 
 	now := time.Now()
-	err := SharedTrafficHourlyStatDAO.IncreaseDayBytes(nil, timeutil.Format("YmdH"), 1)
+	err := SharedTrafficHourlyStatDAO.IncreaseHourlyBytes(nil, timeutil.Format("YmdH"), 1)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/db/models/stats/traffic_hourly_stat_model.go

@@ -1,4 +1,4 @@
-package models
+package stats
 
 // 总的流量统计（按小时）
 type TrafficHourlyStat struct {

internal/db/models/stats/traffic_hourly_stat_model_ext.go

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

internal/db/models/traffic_daily_stat_model_ext.go

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

internal/db/models/traffic_hourly_stat_model_ext.go

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

internal/rpc/services/service_admin.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
 	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
@@ -515,7 +516,7 @@ func (this *AdminService) ComposeAdminDashboard(ctx context.Context, req *pb.Com
 
 	// 按日流量统计
 	dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
-	dailyTrafficStats, err := models.SharedTrafficDailyStatDAO.FindDailyStats(tx, dayFrom, timeutil.Format("Ymd"))
+	dailyTrafficStats, err := stats.SharedTrafficDailyStatDAO.FindDailyStats(tx, dayFrom, timeutil.Format("Ymd"))
 	if err != nil {
 		return nil, err
 	}
@@ -529,7 +530,7 @@ func (this *AdminService) ComposeAdminDashboard(ctx context.Context, req *pb.Com
 	// 小时流量统计
 	hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
 	hourTo := timeutil.Format("YmdH")
-	hourlyTrafficStats, err := models.SharedTrafficHourlyStatDAO.FindHourlyStats(tx, hourFrom, hourTo)
+	hourlyTrafficStats, err := stats.SharedTrafficHourlyStatDAO.FindHourlyStats(tx, hourFrom, hourTo)
 	if err != nil {
 		return nil, err
 	}

internal/rpc/services/service_server_daily_stat.go

@@ -3,6 +3,7 @@ package services
 import (
 	"context"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	timeutil "github.com/iwind/TeaGo/utils/time"
 )
@@ -14,7 +15,7 @@ type ServerDailyStatService struct {
 
 // 上传统计
 func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context, req *pb.UploadServerDailyStatsRequest) (*pb.RPCSuccess, error) {
-	_, err := this.ValidateNode(ctx)
+	nodeId, err := this.ValidateNode(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -30,19 +31,37 @@ func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context,
 	// TODO 将来改成每小时入库一次
 	for _, stat := range req.Stats {
 		// 总体流量（按天）
-		err = models.SharedTrafficDailyStatDAO.IncreaseDailyBytes(tx, timeutil.FormatTime("Ymd", stat.CreatedAt), stat.Bytes)
+		err = stats.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)
+		err = stats.SharedTrafficHourlyStatDAO.IncreaseHourlyBytes(tx, timeutil.FormatTime("YmdH", stat.CreatedAt), stat.Bytes)
+		if err != nil {
+			return nil, err
+		}
+
+		// 节点流量
+		if nodeId > 0 {
+			err = stats.SharedNodeTrafficDailyStatDAO.IncreaseDailyBytes(tx, nodeId, timeutil.FormatTime("Ymd", stat.CreatedAt), stat.Bytes)
+			if err != nil {
+				return nil, err
+			}
+
+			// 集群流量
+			clusterId, err := models.SharedNodeDAO.FindNodeClusterId(tx, nodeId)
+			if err != nil {
+				return nil, err
+			}
+			if clusterId > 0 {
+				err = stats.SharedNodeClusterTrafficDailyStatDAO.IncreaseDailyBytes(tx, clusterId, timeutil.FormatTime("Ymd", stat.CreatedAt), stat.Bytes)
 				if err != nil {
 					return nil, err
 				}
 			}
-
+		}
-	// TODO 集群流量/节点流量
+	}
 
 	return this.Success()
 }