增加服务带宽统计

internal/db/models/metric_stat_dao.go

@@ -691,7 +691,6 @@ func (this *MetricStatDAO) Clean(tx *dbs.Tx) error {
 						Table(table).
 						Attr("itemId", item.Id).
 						Lte("createdDay", expiresDay).
-						UseIndex("createdDay").
 						Limit(10_000). // 一次性不要删除太多，防止阻塞其他操作
 						Delete()
 					return err

internal/db/models/metric_sum_stat_dao.go

@@ -237,7 +237,6 @@ func (this *MetricSumStatDAO) Clean(tx *dbs.Tx) error {
 						Attr("itemId", item.Id).
 						Where("(createdDay IS NULL OR createdDay<:day)").
 						Param("day", expiresDay).
-						UseIndex("createdDay").
 						Limit(10_000). // 一次性不要删除太多，防止阻塞其他操作
 						Delete()
 					return err

internal/db/models/stats/server_bandwidth_stat_dao.go

@@ -0,0 +1,128 @@
+package stats
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/goman"
+	"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"
+	"github.com/iwind/TeaGo/rands"
+	"github.com/iwind/TeaGo/types"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"sync"
+	"time"
+)
+
+type ServerBandwidthStatDAO dbs.DAO
+
+const (
+	ServerBandwidthStatTablePartials = 20 // 分表数量
+)
+
+func init() {
+	dbs.OnReadyDone(func() {
+		// 清理数据任务
+		var ticker = time.NewTicker(time.Duration(rands.Int(24, 48)) * time.Hour)
+		goman.New(func() {
+			for range ticker.C {
+				err := SharedServerBandwidthStatDAO.Clean(nil)
+				if err != nil {
+					remotelogs.Error("SharedServerBandwidthStatDAO", "clean expired data failed: "+err.Error())
+				}
+			}
+		})
+	})
+}
+
+func NewServerBandwidthStatDAO() *ServerBandwidthStatDAO {
+	return dbs.NewDAO(&ServerBandwidthStatDAO{
+		DAOObject: dbs.DAOObject{
+			DB:     Tea.Env,
+			Table:  "edgeServerBandwidthStats",
+			Model:  new(ServerBandwidthStat),
+			PkName: "id",
+		},
+	}).(*ServerBandwidthStatDAO)
+}
+
+var SharedServerBandwidthStatDAO *ServerBandwidthStatDAO
+
+func init() {
+	dbs.OnReady(func() {
+		SharedServerBandwidthStatDAO = NewServerBandwidthStatDAO()
+	})
+}
+
+// UpdateServerBandwidth 写入数据
+func (this *ServerBandwidthStatDAO) UpdateServerBandwidth(tx *dbs.Tx, serverId int64, day string, timeAt string, bytes int64) error {
+	if serverId <= 0 {
+		return errors.New("invalid server id '" + types.String(serverId) + "'")
+	}
+
+	return this.Query(tx).
+		Table(this.partialTable(serverId)).
+		Param("bytes", bytes).
+		InsertOrUpdateQuickly(maps.Map{
+			"serverId": serverId,
+			"day":      day,
+			"timeAt":   timeAt,
+			"bytes":    bytes,
+		}, maps.Map{
+			"bytes": dbs.SQL("bytes+:bytes"),
+		})
+}
+
+// FindServerStats 查找某个时间段的带宽统计
+// 参数：
+//   - day YYYYMMDD
+//   - timeAt HHII
+func (this *ServerBandwidthStatDAO) FindServerStats(tx *dbs.Tx, serverId int64, day string, timeFrom string, timeTo string) (result []*ServerBandwidthStat, err error) {
+	_, err = this.Query(tx).
+		Table(this.partialTable(serverId)).
+		Attr("serverId", serverId).
+		Attr("day", day).
+		Between("timeAt", timeFrom, timeTo).
+		Slice(&result).
+		FindAll()
+	return
+}
+
+// Clean 清理过期数据
+func (this *ServerBandwidthStatDAO) Clean(tx *dbs.Tx) error {
+	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -62)) // 保留大约2个月的数据
+	return this.runBatch(func(table string, locker *sync.Mutex) error {
+		_, err := this.Query(tx).
+			Table(table).
+			Lt("day", day).
+			Delete()
+		return err
+	})
+}
+
+// 批量执行
+func (this *ServerBandwidthStatDAO) runBatch(f func(table string, locker *sync.Mutex) error) error {
+	var locker = &sync.Mutex{}
+	var wg = sync.WaitGroup{}
+	wg.Add(ServerBandwidthStatTablePartials)
+	var resultErr error
+	for i := 0; i < ServerBandwidthStatTablePartials; i++ {
+		var table = this.partialTable(int64(i))
+		go func(table string) {
+			defer wg.Done()
+
+			err := f(table, locker)
+			if err != nil {
+				resultErr = err
+			}
+		}(table)
+	}
+	wg.Wait()
+	return resultErr
+}
+
+// 获取分区表
+func (this *ServerBandwidthStatDAO) partialTable(serverId int64) string {
+	return this.Table + "_" + types.String(serverId%int64(ServerBandwidthStatTablePartials))
+}

internal/db/models/stats/server_bandwidth_stat_dao_test.go

@@ -0,0 +1,48 @@
+package stats_test
+
+import (
+	"fmt"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
+	_ "github.com/go-sql-driver/mysql"
+	_ "github.com/iwind/TeaGo/bootstrap"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/rands"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"testing"
+	"time"
+)
+
+func TestServerBandwidthStatDAO_UpdateServerBandwidth(t *testing.T) {
+	var dao = stats.NewServerBandwidthStatDAO()
+	var tx *dbs.Tx
+	err := dao.UpdateServerBandwidth(tx, 1, timeutil.Format("Ymd"), timeutil.Format("Hi"), 1024)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}
+
+func TestSeverBandwidthStatDAO_InsertManyStats(t *testing.T) {
+	var dao = stats.NewServerBandwidthStatDAO()
+	var tx *dbs.Tx
+	for i := 0; i < 1_000_000; i++ {
+		var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -rands.Int(0, 200)))
+		var minute = fmt.Sprintf("%02d%02d", rands.Int(0, 23), rands.Int(0, 59))
+		err := dao.UpdateServerBandwidth(tx, 1, day, minute, 1024)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	t.Log("ok")
+}
+
+func TestServerBandwidthStatDAO_Clean(t *testing.T) {
+	var dao = stats.NewServerBandwidthStatDAO()
+	var tx *dbs.Tx
+	var before = time.Now()
+	err := dao.Clean(tx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok", time.Since(before).Seconds()*1000, "ms")
+}

internal/db/models/stats/server_bandwidth_stat_model.go

@@ -0,0 +1,22 @@
+package stats
+
+// ServerBandwidthStat 服务峰值带宽统计
+type ServerBandwidthStat struct {
+	Id       uint64 `field:"id"`       // ID
+	ServerId uint64 `field:"serverId"` // 服务ID
+	Day      string `field:"day"`      // 日期YYYYMMDD
+	TimeAt   string `field:"timeAt"`   // 时间点HHMM
+	Bytes    uint64 `field:"bytes"`    // 带宽字节
+}
+
+type ServerBandwidthStatOperator struct {
+	Id       interface{} // ID
+	ServerId interface{} // 服务ID
+	Day      interface{} // 日期YYYYMMDD
+	TimeAt   interface{} // 时间点HHMM
+	Bytes    interface{} // 带宽字节
+}
+
+func NewServerBandwidthStatOperator() *ServerBandwidthStatOperator {
+	return &ServerBandwidthStatOperator{}
+}

internal/db/models/stats/server_bandwidth_stat_model_ext.go

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

internal/nodes/api_node_services.go

@@ -14,579 +14,585 @@ import (
 // 注册服务
 func (this *APINode) registerServices(server *grpc.Server) {
 	{
-		instance := this.serviceInstance(&services.APITokenService{}).(*services.APITokenService)
+		var instance = this.serviceInstance(&services.APITokenService{}).(*services.APITokenService)
 		pb.RegisterAPITokenServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.AdminService{}).(*services.AdminService)
+		var instance = this.serviceInstance(&services.AdminService{}).(*services.AdminService)
 		pb.RegisterAdminServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeGrantService{}).(*services.NodeGrantService)
+		var instance = this.serviceInstance(&services.NodeGrantService{}).(*services.NodeGrantService)
 		pb.RegisterNodeGrantServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerService{}).(*services.ServerService)
+		var instance = this.serviceInstance(&services.ServerService{}).(*services.ServerService)
 		pb.RegisterServerServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeService{}).(*services.NodeService)
+		var instance = this.serviceInstance(&services.NodeService{}).(*services.NodeService)
 		pb.RegisterNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeClusterService{}).(*services.NodeClusterService)
+		var instance = this.serviceInstance(&services.NodeClusterService{}).(*services.NodeClusterService)
 		pb.RegisterNodeClusterServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeIPAddressService{}).(*services.NodeIPAddressService)
+		var instance = this.serviceInstance(&services.NodeIPAddressService{}).(*services.NodeIPAddressService)
 		pb.RegisterNodeIPAddressServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeIPAddressLogService{}).(*services.NodeIPAddressLogService)
+		var instance = this.serviceInstance(&services.NodeIPAddressLogService{}).(*services.NodeIPAddressLogService)
 		pb.RegisterNodeIPAddressLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeIPAddressThresholdService{}).(*services.NodeIPAddressThresholdService)
+		var instance = this.serviceInstance(&services.NodeIPAddressThresholdService{}).(*services.NodeIPAddressThresholdService)
 		pb.RegisterNodeIPAddressThresholdServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.APINodeService{}).(*services.APINodeService)
+		var instance = this.serviceInstance(&services.APINodeService{}).(*services.APINodeService)
 		pb.RegisterAPINodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.APIMethodStatService{}).(*services.APIMethodStatService)
+		var instance = this.serviceInstance(&services.APIMethodStatService{}).(*services.APIMethodStatService)
 		pb.RegisterAPIMethodStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.OriginService{}).(*services.OriginService)
+		var instance = this.serviceInstance(&services.OriginService{}).(*services.OriginService)
 		pb.RegisterOriginServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPWebService{}).(*services.HTTPWebService)
+		var instance = this.serviceInstance(&services.HTTPWebService{}).(*services.HTTPWebService)
 		pb.RegisterHTTPWebServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ReverseProxyService{}).(*services.ReverseProxyService)
+		var instance = this.serviceInstance(&services.ReverseProxyService{}).(*services.ReverseProxyService)
 		pb.RegisterReverseProxyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPGzipService{}).(*services.HTTPGzipService)
+		var instance = this.serviceInstance(&services.HTTPGzipService{}).(*services.HTTPGzipService)
 		pb.RegisterHTTPGzipServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPHeaderPolicyService{}).(*services.HTTPHeaderPolicyService)
+		var instance = this.serviceInstance(&services.HTTPHeaderPolicyService{}).(*services.HTTPHeaderPolicyService)
 		pb.RegisterHTTPHeaderPolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPHeaderService{}).(*services.HTTPHeaderService)
+		var instance = this.serviceInstance(&services.HTTPHeaderService{}).(*services.HTTPHeaderService)
 		pb.RegisterHTTPHeaderServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPPageService{}).(*services.HTTPPageService)
+		var instance = this.serviceInstance(&services.HTTPPageService{}).(*services.HTTPPageService)
 		pb.RegisterHTTPPageServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPAccessLogPolicyService{}).(*services.HTTPAccessLogPolicyService)
+		var instance = this.serviceInstance(&services.HTTPAccessLogPolicyService{}).(*services.HTTPAccessLogPolicyService)
 		pb.RegisterHTTPAccessLogPolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPCachePolicyService{}).(*services.HTTPCachePolicyService)
+		var instance = this.serviceInstance(&services.HTTPCachePolicyService{}).(*services.HTTPCachePolicyService)
 		pb.RegisterHTTPCachePolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPFirewallPolicyService{}).(*services.HTTPFirewallPolicyService)
+		var instance = this.serviceInstance(&services.HTTPFirewallPolicyService{}).(*services.HTTPFirewallPolicyService)
 		pb.RegisterHTTPFirewallPolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.FirewallService{}).(*services.FirewallService)
+		var instance = this.serviceInstance(&services.FirewallService{}).(*services.FirewallService)
 		pb.RegisterFirewallServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPLocationService{}).(*services.HTTPLocationService)
+		var instance = this.serviceInstance(&services.HTTPLocationService{}).(*services.HTTPLocationService)
 		pb.RegisterHTTPLocationServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPWebsocketService{}).(*services.HTTPWebsocketService)
+		var instance = this.serviceInstance(&services.HTTPWebsocketService{}).(*services.HTTPWebsocketService)
 		pb.RegisterHTTPWebsocketServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPRewriteRuleService{}).(*services.HTTPRewriteRuleService)
+		var instance = this.serviceInstance(&services.HTTPRewriteRuleService{}).(*services.HTTPRewriteRuleService)
 		pb.RegisterHTTPRewriteRuleServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.SSLCertService{}).(*services.SSLCertService)
+		var instance = this.serviceInstance(&services.SSLCertService{}).(*services.SSLCertService)
 		pb.RegisterSSLCertServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.SSLPolicyService{}).(*services.SSLPolicyService)
+		var instance = this.serviceInstance(&services.SSLPolicyService{}).(*services.SSLPolicyService)
 		pb.RegisterSSLPolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.SysSettingService{}).(*services.SysSettingService)
+		var instance = this.serviceInstance(&services.SysSettingService{}).(*services.SysSettingService)
 		pb.RegisterSysSettingServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPFirewallRuleGroupService{}).(*services.HTTPFirewallRuleGroupService)
+		var instance = this.serviceInstance(&services.HTTPFirewallRuleGroupService{}).(*services.HTTPFirewallRuleGroupService)
 		pb.RegisterHTTPFirewallRuleGroupServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPFirewallRuleSetService{}).(*services.HTTPFirewallRuleSetService)
+		var instance = this.serviceInstance(&services.HTTPFirewallRuleSetService{}).(*services.HTTPFirewallRuleSetService)
 		pb.RegisterHTTPFirewallRuleSetServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DBNodeService{}).(*services.DBNodeService)
+		var instance = this.serviceInstance(&services.DBNodeService{}).(*services.DBNodeService)
 		pb.RegisterDBNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeLogService{}).(*services.NodeLogService)
+		var instance = this.serviceInstance(&services.NodeLogService{}).(*services.NodeLogService)
 		pb.RegisterNodeLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeLoginService{}).(*services.NodeLoginService)
+		var instance = this.serviceInstance(&services.NodeLoginService{}).(*services.NodeLoginService)
 		pb.RegisterNodeLoginServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPAccessLogService{}).(*services.HTTPAccessLogService)
+		var instance = this.serviceInstance(&services.HTTPAccessLogService{}).(*services.HTTPAccessLogService)
 		pb.RegisterHTTPAccessLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageService{}).(*services.MessageService)
+		var instance = this.serviceInstance(&services.MessageService{}).(*services.MessageService)
 		pb.RegisterMessageServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageRecipientService{}).(*services.MessageRecipientService)
+		var instance = this.serviceInstance(&services.MessageRecipientService{}).(*services.MessageRecipientService)
 		pb.RegisterMessageRecipientServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageReceiverService{}).(*services.MessageReceiverService)
+		var instance = this.serviceInstance(&services.MessageReceiverService{}).(*services.MessageReceiverService)
 		pb.RegisterMessageReceiverServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageMediaService{}).(*services.MessageMediaService)
+		var instance = this.serviceInstance(&services.MessageMediaService{}).(*services.MessageMediaService)
 		pb.RegisterMessageMediaServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageRecipientGroupService{}).(*services.MessageRecipientGroupService)
+		var instance = this.serviceInstance(&services.MessageRecipientGroupService{}).(*services.MessageRecipientGroupService)
 		pb.RegisterMessageRecipientGroupServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageMediaInstanceService{}).(*services.MessageMediaInstanceService)
+		var instance = this.serviceInstance(&services.MessageMediaInstanceService{}).(*services.MessageMediaInstanceService)
 		pb.RegisterMessageMediaInstanceServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageTaskService{}).(*services.MessageTaskService)
+		var instance = this.serviceInstance(&services.MessageTaskService{}).(*services.MessageTaskService)
 		pb.RegisterMessageTaskServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MessageTaskLogService{}).(*services.MessageTaskLogService)
+		var instance = this.serviceInstance(&services.MessageTaskLogService{}).(*services.MessageTaskLogService)
 		pb.RegisterMessageTaskLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeGroupService{}).(*services.NodeGroupService)
+		var instance = this.serviceInstance(&services.NodeGroupService{}).(*services.NodeGroupService)
 		pb.RegisterNodeGroupServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeRegionService{}).(*services.NodeRegionService)
+		var instance = this.serviceInstance(&services.NodeRegionService{}).(*services.NodeRegionService)
 		pb.RegisterNodeRegionServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodePriceItemService{}).(*services.NodePriceItemService)
+		var instance = this.serviceInstance(&services.NodePriceItemService{}).(*services.NodePriceItemService)
 		pb.RegisterNodePriceItemServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerGroupService{}).(*services.ServerGroupService)
+		var instance = this.serviceInstance(&services.ServerGroupService{}).(*services.ServerGroupService)
 		pb.RegisterServerGroupServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.IPLibraryService{}).(*services.IPLibraryService)
+		var instance = this.serviceInstance(&services.IPLibraryService{}).(*services.IPLibraryService)
 		pb.RegisterIPLibraryServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.FileChunkService{}).(*services.FileChunkService)
+		var instance = this.serviceInstance(&services.FileChunkService{}).(*services.FileChunkService)
 		pb.RegisterFileChunkServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.FileService{}).(*services.FileService)
+		var instance = this.serviceInstance(&services.FileService{}).(*services.FileService)
 		pb.RegisterFileServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.RegionCountryService{}).(*services.RegionCountryService)
+		var instance = this.serviceInstance(&services.RegionCountryService{}).(*services.RegionCountryService)
 		pb.RegisterRegionCountryServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.RegionProvinceService{}).(*services.RegionProvinceService)
+		var instance = this.serviceInstance(&services.RegionProvinceService{}).(*services.RegionProvinceService)
 		pb.RegisterRegionProvinceServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.RegionCityService{}).(*services.RegionCityService)
+		var instance = this.serviceInstance(&services.RegionCityService{}).(*services.RegionCityService)
 		pb.RegisterRegionCityServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.RegionProviderService{}).(*services.RegionProviderService)
+		var instance = this.serviceInstance(&services.RegionProviderService{}).(*services.RegionProviderService)
 		pb.RegisterRegionProviderServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.IPListService{}).(*services.IPListService)
+		var instance = this.serviceInstance(&services.IPListService{}).(*services.IPListService)
 		pb.RegisterIPListServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.IPItemService{}).(*services.IPItemService)
+		var instance = this.serviceInstance(&services.IPItemService{}).(*services.IPItemService)
 		pb.RegisterIPItemServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.LogService{}).(*services.LogService)
+		var instance = this.serviceInstance(&services.LogService{}).(*services.LogService)
 		pb.RegisterLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DNSProviderService{}).(*services.DNSProviderService)
+		var instance = this.serviceInstance(&services.DNSProviderService{}).(*services.DNSProviderService)
 		pb.RegisterDNSProviderServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DNSDomainService{}).(*services.DNSDomainService)
+		var instance = this.serviceInstance(&services.DNSDomainService{}).(*services.DNSDomainService)
 		pb.RegisterDNSDomainServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DNSService{}).(*services.DNSService)
+		var instance = this.serviceInstance(&services.DNSService{}).(*services.DNSService)
 		pb.RegisterDNSServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ACMEUserService{}).(*services.ACMEUserService)
+		var instance = this.serviceInstance(&services.ACMEUserService{}).(*services.ACMEUserService)
 		pb.RegisterACMEUserServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ACMETaskService{}).(*services.ACMETaskService)
+		var instance = this.serviceInstance(&services.ACMETaskService{}).(*services.ACMETaskService)
 		pb.RegisterACMETaskServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ACMEAuthenticationService{}).(*services.ACMEAuthenticationService)
+		var instance = this.serviceInstance(&services.ACMEAuthenticationService{}).(*services.ACMEAuthenticationService)
 		pb.RegisterACMEAuthenticationServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ACMEProviderService{}).(*services.ACMEProviderService)
+		var instance = this.serviceInstance(&services.ACMEProviderService{}).(*services.ACMEProviderService)
 		pb.RegisterACMEProviderServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ACMEProviderAccountService{}).(*services.ACMEProviderAccountService)
+		var instance = this.serviceInstance(&services.ACMEProviderAccountService{}).(*services.ACMEProviderAccountService)
 		pb.RegisterACMEProviderAccountServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.UserService{}).(*services.UserService)
+		var instance = this.serviceInstance(&services.UserService{}).(*services.UserService)
 		pb.RegisterUserServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerDailyStatService{}).(*services.ServerDailyStatService)
+		var instance = this.serviceInstance(&services.ServerDailyStatService{}).(*services.ServerDailyStatService)
 		pb.RegisterServerDailyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.UserBillService{}).(*services.UserBillService)
+		var instance = this.serviceInstance(&services.UserBillService{}).(*services.UserBillService)
 		pb.RegisterUserBillServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerBillService{}).(*services.ServerBillService)
+		var instance = this.serviceInstance(&services.ServerBillService{}).(*services.ServerBillService)
 		pb.RegisterServerBillServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.UserNodeService{}).(*services.UserNodeService)
+		var instance = this.serviceInstance(&services.UserNodeService{}).(*services.UserNodeService)
 		pb.RegisterUserNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.LoginService{}).(*services.LoginService)
+		var instance = this.serviceInstance(&services.LoginService{}).(*services.LoginService)
 		pb.RegisterLoginServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.UserAccessKeyService{}).(*services.UserAccessKeyService)
+		var instance = this.serviceInstance(&services.UserAccessKeyService{}).(*services.UserAccessKeyService)
 		pb.RegisterUserAccessKeyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.SysLockerService{}).(*services.SysLockerService)
+		var instance = this.serviceInstance(&services.SysLockerService{}).(*services.SysLockerService)
 		pb.RegisterSysLockerServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeTaskService{}).(*services.NodeTaskService)
+		var instance = this.serviceInstance(&services.NodeTaskService{}).(*services.NodeTaskService)
 		pb.RegisterNodeTaskServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeValueService{}).(*services.NodeValueService)
+		var instance = this.serviceInstance(&services.NodeValueService{}).(*services.NodeValueService)
 		pb.RegisterNodeValueServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DBService{}).(*services.DBService)
+		var instance = this.serviceInstance(&services.DBService{}).(*services.DBService)
 		pb.RegisterDBServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerRegionCityMonthlyStatService{}).(*services.ServerRegionCityMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerRegionCityMonthlyStatService{}).(*services.ServerRegionCityMonthlyStatService)
 		pb.RegisterServerRegionCityMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerRegionCountryMonthlyStatService{}).(*services.ServerRegionCountryMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerRegionCountryMonthlyStatService{}).(*services.ServerRegionCountryMonthlyStatService)
 		pb.RegisterServerRegionCountryMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerRegionProvinceMonthlyStatService{}).(*services.ServerRegionProvinceMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerRegionProvinceMonthlyStatService{}).(*services.ServerRegionProvinceMonthlyStatService)
 		pb.RegisterServerRegionProvinceMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerRegionProviderMonthlyStatService{}).(*services.ServerRegionProviderMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerRegionProviderMonthlyStatService{}).(*services.ServerRegionProviderMonthlyStatService)
 		pb.RegisterServerRegionProviderMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerClientSystemMonthlyStatService{}).(*services.ServerClientSystemMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerClientSystemMonthlyStatService{}).(*services.ServerClientSystemMonthlyStatService)
 		pb.RegisterServerClientSystemMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerClientBrowserMonthlyStatService{}).(*services.ServerClientBrowserMonthlyStatService)
+		var instance = this.serviceInstance(&services.ServerClientBrowserMonthlyStatService{}).(*services.ServerClientBrowserMonthlyStatService)
 		pb.RegisterServerClientBrowserMonthlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.ServerHTTPFirewallDailyStatService{}).(*services.ServerHTTPFirewallDailyStatService)
+		var instance = this.serviceInstance(&services.ServerHTTPFirewallDailyStatService{}).(*services.ServerHTTPFirewallDailyStatService)
 		pb.RegisterServerHTTPFirewallDailyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.DNSTaskService{}).(*services.DNSTaskService)
+		var instance = this.serviceInstance(&services.DNSTaskService{}).(*services.DNSTaskService)
 		pb.RegisterDNSTaskServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeClusterFirewallActionService{}).(*services.NodeClusterFirewallActionService)
+		var instance = this.serviceInstance(&services.NodeClusterFirewallActionService{}).(*services.NodeClusterFirewallActionService)
 		pb.RegisterNodeClusterFirewallActionServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MonitorNodeService{}).(*services.MonitorNodeService)
+		var instance = this.serviceInstance(&services.MonitorNodeService{}).(*services.MonitorNodeService)
 		pb.RegisterMonitorNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.AuthorityNodeService{}).(*services.AuthorityNodeService)
+		var instance = this.serviceInstance(&services.AuthorityNodeService{}).(*services.AuthorityNodeService)
 		pb.RegisterAuthorityNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.LatestItemService{}).(*services.LatestItemService)
+		var instance = this.serviceInstance(&services.LatestItemService{}).(*services.LatestItemService)
 		pb.RegisterLatestItemServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeThresholdService{}).(*services.NodeThresholdService)
+		var instance = this.serviceInstance(&services.NodeThresholdService{}).(*services.NodeThresholdService)
 		pb.RegisterNodeThresholdServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPFastcgiService{}).(*services.HTTPFastcgiService)
+		var instance = this.serviceInstance(&services.HTTPFastcgiService{}).(*services.HTTPFastcgiService)
 		pb.RegisterHTTPFastcgiServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSClusterService{}).(*nameservers.NSClusterService)
+		var instance = this.serviceInstance(&nameservers.NSClusterService{}).(*nameservers.NSClusterService)
 		pb.RegisterNSClusterServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSNodeService{}).(*nameservers.NSNodeService)
+		var instance = this.serviceInstance(&nameservers.NSNodeService{}).(*nameservers.NSNodeService)
 		pb.RegisterNSNodeServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSDomainService{}).(*nameservers.NSDomainService)
+		var instance = this.serviceInstance(&nameservers.NSDomainService{}).(*nameservers.NSDomainService)
 		pb.RegisterNSDomainServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSRecordService{}).(*nameservers.NSRecordService)
+		var instance = this.serviceInstance(&nameservers.NSRecordService{}).(*nameservers.NSRecordService)
 		pb.RegisterNSRecordServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSRouteService{}).(*nameservers.NSRouteService)
+		var instance = this.serviceInstance(&nameservers.NSRouteService{}).(*nameservers.NSRouteService)
 		pb.RegisterNSRouteServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSKeyService{}).(*nameservers.NSKeyService)
+		var instance = this.serviceInstance(&nameservers.NSKeyService{}).(*nameservers.NSKeyService)
 		pb.RegisterNSKeyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSAccessLogService{}).(*nameservers.NSAccessLogService)
+		var instance = this.serviceInstance(&nameservers.NSAccessLogService{}).(*nameservers.NSAccessLogService)
 		pb.RegisterNSAccessLogServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSRecordHourlyStatService{}).(*nameservers.NSRecordHourlyStatService)
+		var instance = this.serviceInstance(&nameservers.NSRecordHourlyStatService{}).(*nameservers.NSRecordHourlyStatService)
 		pb.RegisterNSRecordHourlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSQuestionOptionService{}).(*nameservers.NSQuestionOptionService)
+		var instance = this.serviceInstance(&nameservers.NSQuestionOptionService{}).(*nameservers.NSQuestionOptionService)
 		pb.RegisterNSQuestionOptionServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&nameservers.NSService{}).(*nameservers.NSService)
+		var instance = this.serviceInstance(&nameservers.NSService{}).(*nameservers.NSService)
 		pb.RegisterNSServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.HTTPAuthPolicyService{}).(*services.HTTPAuthPolicyService)
+		var instance = this.serviceInstance(&services.HTTPAuthPolicyService{}).(*services.HTTPAuthPolicyService)
 		pb.RegisterHTTPAuthPolicyServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MetricItemService{}).(*services.MetricItemService)
+		var instance = this.serviceInstance(&services.MetricItemService{}).(*services.MetricItemService)
 		pb.RegisterMetricItemServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.NodeClusterMetricItemService{}).(*services.NodeClusterMetricItemService)
+		var instance = this.serviceInstance(&services.NodeClusterMetricItemService{}).(*services.NodeClusterMetricItemService)
 		pb.RegisterNodeClusterMetricItemServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MetricStatService{}).(*services.MetricStatService)
+		var instance = this.serviceInstance(&services.MetricStatService{}).(*services.MetricStatService)
 		pb.RegisterMetricStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 	{
-		instance := this.serviceInstance(&services.MetricChartService{}).(*services.MetricChartService)
+		var instance = this.serviceInstance(&services.MetricChartService{}).(*services.MetricChartService)
 		pb.RegisterMetricChartServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.ServerStatBoardService{}).(*services.ServerStatBoardService)
+		var instance = this.serviceInstance(&services.ServerStatBoardService{}).(*services.ServerStatBoardService)
 		pb.RegisterServerStatBoardServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.ServerStatBoardChartService{}).(*services.ServerStatBoardChartService)
+		var instance = this.serviceInstance(&services.ServerStatBoardChartService{}).(*services.ServerStatBoardChartService)
 		pb.RegisterServerStatBoardChartServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.PlanService{}).(*services.PlanService)
+		var instance = this.serviceInstance(&services.PlanService{}).(*services.PlanService)
 		pb.RegisterPlanServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.UserPlanService{}).(*services.UserPlanService)
+		var instance = this.serviceInstance(&services.UserPlanService{}).(*services.UserPlanService)
 		pb.RegisterUserPlanServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.ServerDomainHourlyStatService{}).(*services.ServerDomainHourlyStatService)
+		var instance = this.serviceInstance(&services.ServerDomainHourlyStatService{}).(*services.ServerDomainHourlyStatService)
 		pb.RegisterServerDomainHourlyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.TrafficDailyStatService{}).(*services.TrafficDailyStatService)
+		var instance = this.serviceInstance(&services.TrafficDailyStatService{}).(*services.TrafficDailyStatService)
 		pb.RegisterTrafficDailyStatServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.HTTPCacheTaskKeyService{}).(*services.HTTPCacheTaskKeyService)
+		var instance = this.serviceInstance(&services.HTTPCacheTaskKeyService{}).(*services.HTTPCacheTaskKeyService)
 		pb.RegisterHTTPCacheTaskKeyServiceServer(server, instance)
 		this.rest(instance)
 	}
 
 	{
-		instance := this.serviceInstance(&services.HTTPCacheTaskService{}).(*services.HTTPCacheTaskService)
+		var instance = this.serviceInstance(&services.HTTPCacheTaskService{}).(*services.HTTPCacheTaskService)
 		pb.RegisterHTTPCacheTaskServiceServer(server, instance)
 		this.rest(instance)
 	}
 
+	{
+		var instance = this.serviceInstance(&services.ServerBandwidthStatService{}).(*services.ServerBandwidthStatService)
+		pb.RegisterServerBandwidthStatServiceServer(server, instance)
+		this.rest(instance)
+	}
+
 	APINodeServicesRegister(this, server)
 
 	// TODO check service names
 	for serviceName := range server.GetServiceInfo() {
-		index := strings.LastIndex(serviceName, ".")
+		var index = strings.LastIndex(serviceName, ".")
 		if index >= 0 {
 			serviceName = serviceName[index+1:]
 		}
@@ -602,7 +608,7 @@ func (this *APINode) rest(instance interface{}) {
 	defer this.serviceInstanceLocker.Unlock()
 
 	var name = reflect.TypeOf(instance).String()
-	index := strings.LastIndex(name, ".")
+	var index = strings.LastIndex(name, ".")
 	if index >= 0 {
 		name = name[index+1:]
 	}
@@ -618,7 +624,7 @@ func (this *APINode) serviceInstance(instance interface{}) interface{} {
 	this.serviceInstanceLocker.Lock()
 	defer this.serviceInstanceLocker.Unlock()
 
-	typeName := reflect.TypeOf(instance).String()
+	var typeName = reflect.TypeOf(instance).String()
 	result, ok := this.serviceInstanceMap[typeName]
 	if ok {
 		return result

internal/rpc/services/service_server_bandwidth_stat.go

@@ -0,0 +1,80 @@
+// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package services
+
+import (
+	"context"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
+	"github.com/TeaOSLab/EdgeAPI/internal/goman"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/types"
+	"sync"
+	"time"
+)
+
+var serverBandwidthStatsMap = map[string]*pb.ServerBandwidthStat{} // key => bandwidth
+var serverBandwidthStatsLocker = &sync.Mutex{}
+
+func init() {
+	var ticker = time.NewTicker(5 * time.Minute)
+	if Tea.IsTesting() {
+		ticker = time.NewTicker(1 * time.Minute)
+	}
+
+	dbs.OnReadyDone(func() {
+		goman.New(func() {
+			for range ticker.C {
+				func() {
+					var tx *dbs.Tx
+
+					serverBandwidthStatsLocker.Lock()
+					var m = serverBandwidthStatsMap
+					serverBandwidthStatsMap = map[string]*pb.ServerBandwidthStat{}
+					serverBandwidthStatsLocker.Unlock()
+
+					for _, stat := range m {
+						err := stats.SharedServerBandwidthStatDAO.UpdateServerBandwidth(tx, stat.ServerId, stat.Day, stat.TimeAt, stat.Bytes)
+						if err != nil {
+							remotelogs.Error("ServerBandwidthStatService", "dump bandwidth stats failed: "+err.Error())
+						}
+					}
+				}()
+			}
+		})
+	})
+}
+
+type ServerBandwidthStatService struct {
+	BaseService
+}
+
+// UploadServerBandwidthStats 上传带宽统计
+func (this *ServerBandwidthStatService) UploadServerBandwidthStats(ctx context.Context, req *pb.UploadServerBandwidthStatsRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateNode(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, stat := range req.ServerBandwidthStats {
+		var key = types.String(stat.ServerId) + "@" + stat.Day + "@" + stat.TimeAt
+		serverBandwidthStatsLocker.Lock()
+		oldStat, ok := serverBandwidthStatsMap[key]
+		if ok {
+			oldStat.Bytes += stat.Bytes
+		} else {
+			serverBandwidthStatsMap[key] = &pb.ServerBandwidthStat{
+				Id:       0,
+				ServerId: stat.ServerId,
+				Day:      stat.Day,
+				TimeAt:   stat.TimeAt,
+				Bytes:    stat.Bytes,
+			}
+		}
+		serverBandwidthStatsLocker.Unlock()
+	}
+
+	return this.Success()
+}

internal/rpc/services/service_server_stat_board.go

@@ -9,6 +9,7 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
 	"github.com/TeaOSLab/EdgeAPI/internal/rpc/tasks"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
@@ -84,7 +85,7 @@ func (this *ServerStatBoardService) ComposeServerStatNodeClusterBoard(ctx contex
 	result.CountServers = countServers
 
 	// 按日流量统计
-	dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
+	var dayFrom = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
 	dailyTrafficStats, err := stats.SharedNodeClusterTrafficDailyStatDAO.FindDailyStats(tx, req.NodeClusterId, dayFrom, timeutil.Format("Ymd"))
 	if err != nil {
 		return nil, err
@@ -102,8 +103,8 @@ func (this *ServerStatBoardService) ComposeServerStatNodeClusterBoard(ctx contex
 	}
 
 	// 小时流量统计
-	hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
-	hourTo := timeutil.Format("YmdH")
+	var hourFrom = timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
+	var hourTo = timeutil.Format("YmdH")
 	hourlyTrafficStats, err := stats.SharedNodeTrafficHourlyStatDAO.FindHourlyStatsWithClusterId(tx, req.NodeClusterId, hourFrom, hourTo)
 	if err != nil {
 		return nil, err
@@ -304,7 +305,7 @@ func (this *ServerStatBoardService) ComposeServerStatNodeBoard(ctx context.Conte
 	}
 
 	// 按日流量统计
-	dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
+	var dayFrom = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
 	dailyTrafficStats, err := stats.SharedNodeTrafficDailyStatDAO.FindDailyStats(tx, "node", req.NodeId, dayFrom, timeutil.Format("Ymd"))
 	if err != nil {
 		return nil, err
@@ -322,8 +323,8 @@ func (this *ServerStatBoardService) ComposeServerStatNodeBoard(ctx context.Conte
 	}
 
 	// 小时流量统计
-	hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
-	hourTo := timeutil.Format("YmdH")
+	var hourFrom = timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
+	var hourTo = timeutil.Format("YmdH")
 	hourlyTrafficStats, err := stats.SharedNodeTrafficHourlyStatDAO.FindHourlyStatsWithNodeId(tx, "node", req.NodeId, hourFrom, hourTo)
 	if err != nil {
 		return nil, err
@@ -407,8 +408,45 @@ func (this *ServerStatBoardService) ComposeServerStatBoard(ctx context.Context,
 	var result = &pb.ComposeServerStatBoardResponse{}
 	var tx = this.NullTx()
 
+	// 带宽统计
+	{
+		var bandwidthMinutes = utils.RangeMinutes(time.Now(), 12, 5)
+		var bandwidthStatMap = map[string]*pb.ServerBandwidthStat{}
+		for _, r := range utils.GroupMinuteRanges(bandwidthMinutes) {
+			bandwidthStats, err := stats.SharedServerBandwidthStatDAO.FindServerStats(tx, req.ServerId, r.Day, r.MinuteFrom, r.MinuteTo)
+			if err != nil {
+				return nil, err
+			}
+			for _, stat := range bandwidthStats {
+				bandwidthStatMap[stat.Day+"@"+stat.TimeAt] = &pb.ServerBandwidthStat{
+					Id:       int64(stat.Id),
+					ServerId: int64(stat.ServerId),
+					Day:      stat.Day,
+					TimeAt:   stat.TimeAt,
+					Bytes:    int64(stat.Bytes),
+				}
+			}
+		}
+		var pbBandwidthStats = []*pb.ServerBandwidthStat{}
+		for _, minute := range bandwidthMinutes {
+			stat, ok := bandwidthStatMap[minute.Day+"@"+minute.Minute]
+			if ok {
+				pbBandwidthStats = append(pbBandwidthStats, stat)
+			} else {
+				pbBandwidthStats = append(pbBandwidthStats, &pb.ServerBandwidthStat{
+					Id:       0,
+					ServerId: req.ServerId,
+					Day:      minute.Day,
+					TimeAt:   minute.Minute,
+					Bytes:    0,
+				})
+			}
+		}
+		result.ServerBandwidthStats = pbBandwidthStats
+	}
+
 	// 按日流量统计
-	dayFrom := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
+	var dayFrom = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -14))
 	dailyTrafficStats, err := models.SharedServerDailyStatDAO.FindDailyStats(tx, req.ServerId, dayFrom, timeutil.Format("Ymd"))
 	if err != nil {
 		return nil, err
@@ -426,8 +464,8 @@ func (this *ServerStatBoardService) ComposeServerStatBoard(ctx context.Context,
 	}
 
 	// 小时流量统计
-	hourFrom := timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
-	hourTo := timeutil.Format("YmdH")
+	var hourFrom = timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
+	var hourTo = timeutil.Format("YmdH")
 	hourlyTrafficStats, err := models.SharedServerDailyStatDAO.FindHourlyStats(tx, req.ServerId, hourFrom, hourTo)
 	if err != nil {
 		return nil, err

internal/utils/time.go

@@ -9,6 +9,19 @@ import (
 	"time"
 )
 
+// 分钟时间点
+type timeMinute struct {
+	Day    string
+	Minute string
+}
+
+// 分钟时间范围
+type timeMinuteRange struct {
+	Day        string
+	MinuteFrom string
+	MinuteTo   string
+}
+
 // RangeDays 计算日期之间的所有日期，格式为YYYYMMDD
 func RangeDays(dayFrom string, dayTo string) ([]string, error) {
 	ok, err := regexp.MatchString(`^\d{8}$`, dayFrom)
@@ -79,15 +92,15 @@ func RangeMonths(dayFrom string, dayTo string) ([]string, error) {
 		dayFrom, dayTo = dayTo, dayFrom
 	}
 
-	result := []string{dayFrom[:6]}
+	var result = []string{dayFrom[:6]}
 
-	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)
+	var year = types.Int(dayFrom[:4])
+	var month = types.Int(dayFrom[4:6])
+	var day = types.Int(dayFrom[6:])
+	var t = time.Date(year, time.Month(month), day, 0, 0, 0, 0, time.Local)
 	for {
 		t = t.AddDate(0, 0, 20)
-		newDay := timeutil.Format("Ymd", t)
+		var newDay = timeutil.Format("Ymd", t)
 		if newDay <= dayTo {
 			var monthString = newDay[:6]
 			if !lists.ContainsString(result, monthString) {
@@ -129,17 +142,17 @@ func RangeHours(hourFrom string, hourTo string) ([]string, error) {
 	}
 
 	// 不能超过N天
-	maxHours := 100 - 1 // -1 是去掉默认加入的dayFrom
-	result := []string{hourFrom}
+	var maxHours = 100 - 1 // -1 是去掉默认加入的dayFrom
+	var 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)
+	var year = types.Int(hourFrom[:4])
+	var month = types.Int(hourFrom[4:6])
+	var day = types.Int(hourFrom[6:8])
+	var hour = types.Int(hourFrom[8:])
+	var 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)
+		var newHour = timeutil.Format("YmdH", t)
 		if newHour <= hourTo {
 			result = append(result, newHour)
 		} else {
@@ -154,3 +167,56 @@ func RangeHours(hourFrom string, hourTo string) ([]string, error) {
 
 	return result, nil
 }
+
+// RangeMinutes 计算若干个时间点，返回结果为 [ [day1, minute1], [day2, minute2] ... ]
+func RangeMinutes(toTime time.Time, count int, everyMinutes int64) []timeMinute {
+	var everySeconds = everyMinutes * 60
+	if everySeconds <= 0 {
+		everySeconds = 300
+	}
+	var result = []timeMinute{}
+	var fromTime = time.Unix(toTime.Unix()-everySeconds*int64(count-1), 0)
+	for {
+		var timestamp = fromTime.Unix() / everySeconds * everySeconds
+		result = append(result, timeMinute{
+			Day:    timeutil.FormatTime("Ymd", timestamp),
+			Minute: timeutil.FormatTime("Hi", timestamp),
+		})
+		fromTime = time.Unix(fromTime.Unix()+everySeconds, 0)
+
+		count--
+		if count <= 0 {
+			break
+		}
+	}
+
+	return result
+}
+
+// GroupMinuteRanges 将时间点分组
+func GroupMinuteRanges(minutes []timeMinute) []timeMinuteRange {
+	var result = []*timeMinuteRange{}
+	var lastDay = ""
+	var lastRange *timeMinuteRange
+	for _, minute := range minutes {
+		if minute.Day != lastDay {
+			lastDay = minute.Day
+			lastRange = &timeMinuteRange{
+				Day:        minute.Day,
+				MinuteFrom: minute.Minute,
+				MinuteTo:   minute.Minute,
+			}
+			result = append(result, lastRange)
+		} else {
+			if lastRange != nil {
+				lastRange.MinuteTo = minute.Minute
+			}
+		}
+	}
+
+	var finalResult = []timeMinuteRange{}
+	for _, minutePtr := range result {
+		finalResult = append(finalResult, *minutePtr)
+	}
+	return finalResult
+}

internal/utils/time_test.go

@@ -1,9 +1,13 @@
-package utils
+package utils_test
 
-import "testing"
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"testing"
+	"time"
+)
 
 func TestRangeDays(t *testing.T) {
-	days, err := RangeDays("20210101", "20210115")
+	days, err := utils.RangeDays("20210101", "20210115")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -11,7 +15,7 @@ func TestRangeDays(t *testing.T) {
 }
 
 func TestRangeMonth(t *testing.T) {
-	days, err := RangeMonths("20200101", "20210115")
+	days, err := utils.RangeMonths("20200101", "20210115")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -20,7 +24,7 @@ func TestRangeMonth(t *testing.T) {
 
 func TestRangeHours(t *testing.T) {
 	{
-		hours, err := RangeHours("2021010100", "2021010123")
+		hours, err := utils.RangeHours("2021010100", "2021010123")
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -28,10 +32,49 @@ func TestRangeHours(t *testing.T) {
 	}
 
 	{
-		hours, err := RangeHours("2021010105", "2021010112")
+		hours, err := utils.RangeHours("2021010105", "2021010112")
 		if err != nil {
 			t.Fatal(err)
 		}
 		t.Log(hours)
 	}
 }
+
+func TestRangeMinutes(t *testing.T) {
+	{
+		var minutes = utils.RangeMinutes(time.Now(), 5, 5)
+		t.Log(minutes)
+	}
+
+	{
+		var minutes = utils.RangeMinutes(time.Now(), 5, 3)
+		t.Log(minutes)
+	}
+
+	{
+		var now = time.Now()
+		var hour = now.Hour()
+		var minute = now.Minute()
+		now = now.Add(-time.Duration(hour) * time.Hour)
+		now = now.Add(-time.Duration(minute-7) * time.Minute) // 后一天的 00:07 开始往前计算
+		var minutes = utils.RangeMinutes(now, 5, 5)
+		t.Log(minutes)
+	}
+}
+
+func TestGroupMinuteRanges(t *testing.T) {
+	{
+		var minutes = utils.GroupMinuteRanges(utils.RangeMinutes(time.Now(), 5, 5))
+		t.Log(minutes)
+	}
+
+	{
+		var now = time.Now()
+		var hour = now.Hour()
+		var minute = now.Minute()
+		now = now.Add(-time.Duration(hour) * time.Hour)
+		now = now.Add(-time.Duration(minute-7) * time.Minute) // 后一天的 00:07 开始往前计算
+		var minutes = utils.GroupMinuteRanges(utils.RangeMinutes(now, 5, 5))
+		t.Log(minutes)
+	}
+}