Skip to content

Commit c4d2191

Browse files
iwindiwind
committed
增加服务流量统计
1 parent f6159c6 commit c4d2191

7 files changed

Lines changed: 134 additions & 42 deletions

‎internal/db/models/db_node_initializer_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,21 +12,21 @@ func TestDBNodeInitializer_loop(t *testing.T) {
1212
if err != nil {
1313
t.Fatal(err)
1414
}
15-
t.Log(len(accessLogDBMapping), len(accessLogDAOMapping))
15+
t.Log(len(accessLogDBMapping), len(httpAccessLogDAOMapping))
1616
}
1717

1818
func TestFindAccessLogTable(t *testing.T) {
1919
before := time.Now()
2020
db := SharedHTTPAccessLogDAO.Instance
21-
tableName, err := findAccessLogTable(db, "20201010", false)
21+
tableName, err := findHTTPAccessLogTable(db, "20201010", false)
2222
if err != nil {
2323
t.Fatal(err)
2424
}
2525
t.Log(tableName)
2626
t.Log(time.Since(before).Seconds()*1000, "ms")
2727

2828
before = time.Now()
29-
tableName, err = findAccessLogTable(db, "20201010", false)
29+
tableName, err = findHTTPAccessLogTable(db, "20201010", false)
3030

3131
if err != nil {
3232
t.Fatal(err)
@@ -40,6 +40,6 @@ func BenchmarkFindAccessLogTable(b *testing.B) {
4040

4141
runtime.GOMAXPROCS(1)
4242
for i := 0; i < b.N; i++ {
43-
_, _ = findAccessLogTable(db, "20201010", false)
43+
_, _ = findHTTPAccessLogTable(db, "20201010", false)
4444
}
4545
}

‎internal/db/models/http_access_log_dao_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ func TestCreateHTTPAccessLogs(t *testing.T) {
2424
Status: 200,
2525
Timestamp: time.Now().Unix(),
2626
}
27-
dao := randomAccessLogDAO()
27+
dao := randomHTTPAccessLogDAO()
2828
t.Log("dao:", dao)
2929
err = SharedHTTPAccessLogDAO.CreateHTTPAccessLogsWithDAO(tx, dao, []*pb.HTTPAccessLog{accessLog})
3030
if err != nil {
@@ -41,7 +41,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs(t *testing.T) {
4141
t.Fatal(err)
4242
}
4343

44-
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0)
44+
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "")
4545
if err != nil {
4646
t.Fatal(err)
4747
}
@@ -68,7 +68,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page(t *testing.T) {
6868
times := 0 // 防止循环次数太多
6969
for {
7070
before := time.Now()
71-
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0)
71+
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "")
7272
cost := time.Since(before).Seconds()
7373
if err != nil {
7474
t.Fatal(err)
@@ -99,7 +99,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Reverse(t *testing.T) {
9999
}
100100

101101
before := time.Now()
102-
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0, false, 0)
102+
accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0, false, 0, "")
103103
cost := time.Since(before).Seconds()
104104
if err != nil {
105105
t.Fatal(err)
@@ -124,7 +124,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page_NotExists(t *testing.T) {
124124
times := 0 // 防止循环次数太多
125125
for {
126126
before := time.Now()
127-
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)
127+
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, "")
128128
cost := time.Since(before).Seconds()
129129
if err != nil {
130130
t.Fatal(err)

‎internal/db/models/node_value_dao_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package models
22

33
import (
4+
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
45
_ "github.com/go-sql-driver/mysql"
56
_ "github.com/iwind/TeaGo/bootstrap"
67
"github.com/iwind/TeaGo/maps"
@@ -13,7 +14,7 @@ func TestNodeValueDAO_CreateValue(t *testing.T) {
1314
m := maps.Map{
1415
"hello": "world12344",
1516
}
16-
err := dao.CreateValue(nil, NodeRoleNode, 1, "test", m.AsJSON(), time.Now().Unix())
17+
err := dao.CreateValue(nil, nodeconfigs.NodeRoleNode, 1, "test", m.AsJSON(), time.Now().Unix())
1718
if err != nil {
1819
t.Fatal(err)
1920
}

‎internal/db/models/server_daily_stat_dao.go‎

Lines changed: 54 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"github.com/iwind/TeaGo/dbs"
88
"github.com/iwind/TeaGo/maps"
99
timeutil "github.com/iwind/TeaGo/utils/time"
10+
"regexp"
1011
)
1112

1213
type ServerDailyStatDAO dbs.DAO
@@ -30,24 +31,33 @@ func init() {
3031
})
3132
}
3233

33-
// 提交数据
34+
// SaveStats 提交数据
3435
func (this *ServerDailyStatDAO) SaveStats(tx *dbs.Tx, stats []*pb.ServerDailyStat) error {
3536
for _, stat := range stats {
3637
day := timeutil.FormatTime("Ymd", stat.CreatedAt)
3738
timeFrom := timeutil.FormatTime("His", stat.CreatedAt)
38-
timeTo := timeutil.FormatTime("His", stat.CreatedAt+5*60) // 5分钟
39+
timeTo := timeutil.FormatTime("His", stat.CreatedAt+5*60-1) // 5分钟
3940

4041
_, _, err := this.Query(tx).
4142
Param("bytes", stat.Bytes).
43+
Param("cachedBytes", stat.CachedBytes).
44+
Param("countRequests", stat.CountRequests).
45+
Param("countCachedRequests", stat.CountCachedRequests).
4246
InsertOrUpdate(maps.Map{
43-
"serverId": stat.ServerId,
44-
"regionId": stat.RegionId,
45-
"bytes": dbs.SQL("bytes+:bytes"),
46-
"day": day,
47-
"timeFrom": timeFrom,
48-
"timeTo": timeTo,
47+
"serverId": stat.ServerId,
48+
"regionId": stat.RegionId,
49+
"bytes": dbs.SQL("bytes+:bytes"),
50+
"cachedBytes": dbs.SQL("cachedBytes+:cachedBytes"),
51+
"countRequests": dbs.SQL("countRequests+:countRequests"),
52+
"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
53+
"day": day,
54+
"timeFrom": timeFrom,
55+
"timeTo": timeTo,
4956
}, maps.Map{
50-
"bytes": dbs.SQL("bytes+:bytes"),
57+
"bytes": dbs.SQL("bytes+:bytes"),
58+
"cachedBytes": dbs.SQL("cachedBytes+:cachedBytes"),
59+
"countRequests": dbs.SQL("countRequests+:countRequests"),
60+
"countCachedRequests": dbs.SQL("countCachedRequests+:countCachedRequests"),
5161
})
5262
if err != nil {
5363
return err
@@ -56,7 +66,7 @@ func (this *ServerDailyStatDAO) SaveStats(tx *dbs.Tx, stats []*pb.ServerDailySta
5666
return nil
5767
}
5868

59-
// 根据用户计算某月合计
69+
// SumUserMonthly 根据用户计算某月合计
6070
// month 格式为YYYYMM
6171
func (this *ServerDailyStatDAO) SumUserMonthly(tx *dbs.Tx, userId int64, regionId int64, month string) (int64, error) {
6272
query := this.Query(tx)
@@ -69,7 +79,7 @@ func (this *ServerDailyStatDAO) SumUserMonthly(tx *dbs.Tx, userId int64, regionI
6979
SumInt64("bytes", 0)
7080
}
7181

72-
// 获取某月带宽峰值
82+
// SumUserMonthlyPeek 获取某月带宽峰值
7383
// month 格式为YYYYMM
7484
func (this *ServerDailyStatDAO) SumUserMonthlyPeek(tx *dbs.Tx, userId int64, regionId int64, month string) (int64, error) {
7585
query := this.Query(tx)
@@ -86,7 +96,7 @@ func (this *ServerDailyStatDAO) SumUserMonthlyPeek(tx *dbs.Tx, userId int64, reg
8696
return int64(max), nil
8797
}
8898

89-
// 获取某天流量总和
99+
// SumUserDaily 获取某天流量总和
90100
// day 格式为YYYYMMDD
91101
func (this *ServerDailyStatDAO) SumUserDaily(tx *dbs.Tx, userId int64, regionId int64, day string) (int64, error) {
92102
query := this.Query(tx)
@@ -100,7 +110,7 @@ func (this *ServerDailyStatDAO) SumUserDaily(tx *dbs.Tx, userId int64, regionId
100110
SumInt64("bytes", 0)
101111
}
102112

103-
// 获取某天带宽峰值
113+
// SumUserDailyPeek 获取某天带宽峰值
104114
// day 格式为YYYYMMDD
105115
func (this *ServerDailyStatDAO) SumUserDailyPeek(tx *dbs.Tx, userId int64, regionId int64, day string) (int64, error) {
106116
query := this.Query(tx)
@@ -117,3 +127,34 @@ func (this *ServerDailyStatDAO) SumUserDailyPeek(tx *dbs.Tx, userId int64, regio
117127
}
118128
return int64(max), nil
119129
}
130+
131+
// SumHourlyStat 获取 N 小时内的流量
132+
// hour 格式为YYYYMMDDHH
133+
func (this *ServerDailyStatDAO) SumHourlyStat(tx *dbs.Tx, serverId int64, hour string) (stat *pb.ServerDailyStat, err error) {
134+
stat = &pb.ServerDailyStat{}
135+
136+
if !regexp.MustCompile(`^\d{10}$`).MatchString(hour) {
137+
return
138+
}
139+
140+
one, _, err := this.Query(tx).
141+
Result("SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
142+
Attr("serverId", serverId).
143+
Attr("day", hour[:8]).
144+
Gte("timeFrom", hour[8:]+"0000").
145+
Lte("timeTo", hour[8:]+"5959").
146+
FindOne()
147+
if err != nil {
148+
return nil, err
149+
}
150+
151+
if one == nil {
152+
return
153+
}
154+
155+
stat.Bytes = one.GetInt64("bytes")
156+
stat.CachedBytes = one.GetInt64("cachedBytes")
157+
stat.CountRequests = one.GetInt64("countRequests")
158+
stat.CountCachedRequests = one.GetInt64("countCachedRequests")
159+
return
160+
}

‎internal/db/models/server_daily_stat_dao_test.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
55
_ "github.com/go-sql-driver/mysql"
66
"github.com/iwind/TeaGo/dbs"
7+
"github.com/iwind/TeaGo/logs"
78
timeutil "github.com/iwind/TeaGo/utils/time"
89
"testing"
910
)
@@ -51,3 +52,14 @@ func TestServerDailyStatDAO_SumUserMonthly(t *testing.T) {
5152
}
5253
t.Log("bytes:", bytes)
5354
}
55+
56+
func TestServerDailyStatDAO_SumHourlyRequests(t *testing.T) {
57+
dbs.NotifyReady()
58+
var tx *dbs.Tx
59+
60+
stats, err := NewServerDailyStatDAO().SumHourlyStats(tx, 23, timeutil.Format("YmdH"))
61+
if err != nil {
62+
t.Fatal(err)
63+
}
64+
logs.PrintAsJSON(stats, t)
65+
}

‎internal/db/models/server_daily_stat_model.go‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,32 @@
11
package models
22

3-
// 计费流量统计
3+
// ServerDailyStat 计费流量统计
44
type ServerDailyStat struct {
5-
Id uint64 `field:"id"` // ID
6-
ServerId uint32 `field:"serverId"` // 服务ID
7-
RegionId uint32 `field:"regionId"` // 区域ID
8-
Bytes uint64 `field:"bytes"` // 流量
9-
Day string `field:"day"` // 日期YYYYMMDD
10-
TimeFrom string `field:"timeFrom"` // 开始时间HHMMSS
11-
TimeTo string `field:"timeTo"` // 结束时间
12-
IsCharged uint8 `field:"isCharged"` // 是否已计算费用
5+
Id uint64 `field:"id"` // ID
6+
ServerId uint32 `field:"serverId"` // 服务ID
7+
RegionId uint32 `field:"regionId"` // 区域ID
8+
Bytes uint64 `field:"bytes"` // 流量
9+
CachedBytes uint64 `field:"cachedBytes"` // 缓存的流量
10+
CountRequests uint64 `field:"countRequests"` // 请求数
11+
CountCachedRequests uint64 `field:"countCachedRequests"` // 缓存的请求数
12+
Day string `field:"day"` // 日期YYYYMMDD
13+
TimeFrom string `field:"timeFrom"` // 开始时间HHMMSS
14+
TimeTo string `field:"timeTo"` // 结束时间
15+
IsCharged uint8 `field:"isCharged"` // 是否已计算费用
1316
}
1417

1518
type ServerDailyStatOperator struct {
16-
Id interface{} // ID
17-
ServerId interface{} // 服务ID
18-
RegionId interface{} // 区域ID
19-
Bytes interface{} // 流量
20-
Day interface{} // 日期YYYYMMDD
21-
TimeFrom interface{} // 开始时间HHMMSS
22-
TimeTo interface{} // 结束时间
23-
IsCharged interface{} // 是否已计算费用
19+
Id interface{} // ID
20+
ServerId interface{} // 服务ID
21+
RegionId interface{} // 区域ID
22+
Bytes interface{} // 流量
23+
CachedBytes interface{} // 缓存的流量
24+
CountRequests interface{} // 请求数
25+
CountCachedRequests interface{} // 缓存的请求数
26+
Day interface{} // 日期YYYYMMDD
27+
TimeFrom interface{} // 开始时间HHMMSS
28+
TimeTo interface{} // 结束时间
29+
IsCharged interface{} // 是否已计算费用
2430
}
2531

2632
func NewServerDailyStatOperator() *ServerDailyStatOperator {

‎internal/rpc/services/service_server_daily_stat.go‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,15 @@ import (
66
"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
77
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
88
timeutil "github.com/iwind/TeaGo/utils/time"
9+
"time"
910
)
1011

11-
// 服务统计相关服务
12+
// ServerDailyStatService 服务统计相关服务
1213
type ServerDailyStatService struct {
1314
BaseService
1415
}
1516

16-
// 上传统计
17+
// UploadServerDailyStats 上传统计
1718
func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context, req *pb.UploadServerDailyStatsRequest) (*pb.RPCSuccess, error) {
1819
nodeId, err := this.ValidateNode(ctx)
1920
if err != nil {
@@ -65,3 +66,34 @@ func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context,
6566

6667
return this.Success()
6768
}
69+
70+
// FindServerHourlyStats 按小时读取统计数据
71+
func (this *ServerDailyStatService) FindServerHourlyStats(ctx context.Context, req *pb.FindServerHourlyStatsRequest) (*pb.FindServerHourlyStatsResponse, error) {
72+
_, err := this.ValidateAdmin(ctx, 0)
73+
if err != nil {
74+
return nil, err
75+
}
76+
77+
tx := this.NullTx()
78+
79+
result := []*pb.FindServerHourlyStatsResponse_HourlyStat{}
80+
if req.Hours > 0 {
81+
for i := int32(0); i < req.Hours; i++ {
82+
hourString := timeutil.Format("YmdH", time.Now().Add(-time.Duration(i)*time.Hour))
83+
stat, err := models.SharedServerDailyStatDAO.SumHourlyStat(tx, req.ServerId, hourString)
84+
if err != nil {
85+
return nil, err
86+
}
87+
if stat != nil {
88+
result = append(result, &pb.FindServerHourlyStatsResponse_HourlyStat{
89+
Hour: hourString,
90+
Bytes: stat.Bytes,
91+
CachedBytes: stat.CachedBytes,
92+
CountRequests: stat.CountRequests,
93+
CountCachedRequests: stat.CountCachedRequests,
94+
})
95+
}
96+
}
97+
}
98+
return &pb.FindServerHourlyStatsResponse{Stats: result}, nil
99+
}

0 commit comments

Comments
 (0)