[统计]记录流量（以5分钟作为间隔）

internal/db/models/node_dao.go

@@ -469,13 +469,14 @@ func (this *NodeDAO) ComposeNodeConfig(nodeId int64) (*nodeconfigs.NodeConfig, e
 	}
 
 	config := &nodeconfigs.NodeConfig{
-		Id:      int64(node.Id),
-		NodeId:  node.UniqueId,
-		IsOn:    node.IsOn == 1,
-		Servers: nil,
-		Version: int64(node.Version),
-		Name:    node.Name,
-		MaxCPU:  types.Int32(node.MaxCPU),
+		Id:       int64(node.Id),
+		NodeId:   node.UniqueId,
+		IsOn:     node.IsOn == 1,
+		Servers:  nil,
+		Version:  int64(node.Version),
+		Name:     node.Name,
+		MaxCPU:   types.Int32(node.MaxCPU),
+		RegionId: int64(node.RegionId),
 	}
 
 	// 获取所有的服务

internal/db/models/server_daily_stat_dao.go

@@ -0,0 +1,57 @@
+package models
+
+import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	_ "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"
+)
+
+type ServerDailyStatDAO dbs.DAO
+
+func NewServerDailyStatDAO() *ServerDailyStatDAO {
+	return dbs.NewDAO(&ServerDailyStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeServerDailyStats",
+			Model:  new(ServerDailyStat),
+			PkName: "id",
+		},
+	}).(*ServerDailyStatDAO)
+}
+
+var SharedServerDailyStatDAO *ServerDailyStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedServerDailyStatDAO = NewServerDailyStatDAO()
+	})
+}
+
+// 提交数据
+func (this *ServerDailyStatDAO) SaveStats(stats []*pb.ServerDailyStat) error {
+	for _, stat := range stats {
+		day := timeutil.FormatTime("Ymd", stat.CreatedAt)
+		timeFrom := timeutil.FormatTime("His", stat.CreatedAt)
+		timeTo := timeutil.FormatTime("His", stat.CreatedAt+5*60) // 5分钟
+
+		_, _, err := this.Query().
+			Param("bytes", stat.Bytes).
+			InsertOrUpdate(maps.Map{
+				"serverId": stat.ServerId,
+				"regionId": stat.RegionId,
+				"bytes":    dbs.SQL("bytes+:bytes"),
+				"day":      day,
+				"timeFrom": timeFrom,
+				"timeTo":   timeTo,
+			}, maps.Map{
+				"bytes": dbs.SQL("bytes+:bytes"),
+			})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/db/models/server_daily_stat_dao_test.go

@@ -0,0 +1,39 @@
+package models
+
+import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	_ "github.com/go-sql-driver/mysql"
+	"testing"
+)
+
+func TestServerDailyStatDAO_SaveStats(t *testing.T) {
+	stats := []*pb.ServerDailyStat{
+		{
+			ServerId:  1,
+			RegionId:  2,
+			Bytes:     1,
+			CreatedAt: 1607671488,
+		},
+	}
+	err := NewServerDailyStatDAO().SaveStats(stats)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}
+
+func TestServerDailyStatDAO_SaveStats2(t *testing.T) {
+	stats := []*pb.ServerDailyStat{
+		{
+			ServerId:  1,
+			RegionId:  3,
+			Bytes:     1,
+			CreatedAt: 1607671488,
+		},
+	}
+	err := NewServerDailyStatDAO().SaveStats(stats)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}

internal/db/models/server_daily_stat_model.go

@@ -1,7 +1,7 @@
 package models
 
 // 计费流量统计
-type ServerStat struct {
+type ServerDailyStat struct {
 	Id       uint64 `field:"id"`       // ID
 	ServerId uint32 `field:"serverId"` // 服务ID
 	RegionId uint32 `field:"regionId"` // 区域ID
@@ -11,7 +11,7 @@ type ServerStat struct {
 	TimeTo   string `field:"timeTo"`   // 结束时间
 }
 
-type ServerStatOperator struct {
+type ServerDailyStatOperator struct {
 	Id       interface{} // ID
 	ServerId interface{} // 服务ID
 	RegionId interface{} // 区域ID
@@ -21,6 +21,6 @@ type ServerStatOperator struct {
 	TimeTo   interface{} // 结束时间
 }
 
-func NewServerStatOperator() *ServerStatOperator {
-	return &ServerStatOperator{}
+func NewServerDailyStatOperator() *ServerDailyStatOperator {
+	return &ServerDailyStatOperator{}
 }

internal/db/models/server_stat_dao.go

@@ -1,28 +0,0 @@
-package models
-
-import (
-	_ "github.com/go-sql-driver/mysql"
-	"github.com/iwind/TeaGo/Tea"
-	"github.com/iwind/TeaGo/dbs"
-)
-
-type ServerStatDAO dbs.DAO
-
-func NewServerStatDAO() *ServerStatDAO {
-	return dbs.NewDAO(&ServerStatDAO{
-		DAOObject: dbs.DAOObject{
-			DB:     Tea.Env,
-			Table:  "edgeServerStats",
-			Model:  new(ServerStat),
-			PkName: "id",
-		},
-	}).(*ServerStatDAO)
-}
-
-var SharedServerStatDAO *ServerStatDAO
-
-func init() {
-	dbs.OnReady(func() {
-		SharedServerStatDAO = NewServerStatDAO()
-	})
-}

internal/db/models/server_stat_dao_test.go

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

internal/nodes/api_node.go

@@ -203,6 +203,7 @@ func (this *APINode) listenRPC(listener net.Listener, tlsConfig *tls.Config) err
 	pb.RegisterACMETaskServiceServer(rpcServer, &services.ACMETaskService{})
 	pb.RegisterACMEAuthenticationServiceServer(rpcServer, &services.ACMEAuthenticationService{})
 	pb.RegisterUserServiceServer(rpcServer, &services.UserService{})
+	pb.RegisterServerDailyStatServiceServer(rpcServer, &services.ServerDailyStatService{})
 	err := rpcServer.Serve(listener)
 	if err != nil {
 		return errors.New("[API_NODE]start rpc failed: " + err.Error())

internal/rpc/services/service_server_daily_stat.go

@@ -0,0 +1,26 @@
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// 服务统计相关服务
+type ServerDailyStatService struct {
+	BaseService
+}
+
+// 上传统计
+func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context, req *pb.UploadServerDailyStatsRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateNode(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	err = models.SharedServerDailyStatDAO.SaveStats(req.Stats)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}