third/APIPark
1
0
Fork 0
mirror of https://github.com/APIParkLab/APIPark.git synced 2026-06-04 10:13:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix:ai token monitor bug

Browse Source
This commit is contained in:
Liujian
2025-04-30 13:59:50 +08:00
parent 1d36f4b821
commit 8927211ea2
10 changed files with 275 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/monitor/driver/influxdb-v2/executor.go
+78 -24
View File
@@ -363,41 +363,95 @@ func (e *executor) CommonStatistics(ctx context.Context, start, end time.Time, g
}
return resultMap, nil
}
func (e *executor) overviewByStatusCode(ctx context.Context, start, end time.Time, table string, wheres []monitor.MonWhereItem, statusCode []string, dataFields []string, fn flux.AggregateFn) ([]time.Time, map[string][]int64, error) {
newStartTime, every, windowOffset, bucket := getTimeIntervalAndBucket(start, end)
var returnDates []time.Time
var returnResult = make(map[string][]int64)
for _, s := range statusCode {
newWheres := make([]monitor.MonWhereItem, 0, len(wheres)+1)
newWheres = append(newWheres, wheres...)
newWheres = append(newWheres, monitor.MonWhereItem{
Key: "status_code",
Operation: "=",
Values: []string{s},
})
dates, result, err := e.fluxQuery.CommonTendency(ctx, e.openApi, newStartTime, end, bucket, table, formatFilter(newWheres), dataFields, every, windowOffset, fn)
if err != nil {
return nil, nil, err
}
if len(dates) > 0 {
returnDates = dates
}
for _, v := range dataFields {
key := fmt.Sprintf("%s_%s", s, v)
if _, ok := returnResult[key]; !ok {
returnResult[key] = make([]int64, 0, len(returnDates))
}
returnResult[key] = append(returnResult[key], result[v]...)
}
}
return returnDates, returnResult, nil
}
func (e *executor) TrafficOverviewByStatusCode(ctx context.Context, start time.Time, end time.Time, wheres []monitor.MonWhereItem) ([]time.Time, *monitor.StatusCodeOverview, []*monitor.StatusCodeOverview, error) {
newStartTime, every, windowOffset, bucket := getTimeIntervalAndBucket(start, end)
filters := formatFilter(wheres)
fieldsConditions := []string{"s2xx_request", "s4xx_request", "s5xx_request", "s2xx_response", "s4xx_response", "s5xx_response"}
dates, groupValues, err := e.fluxQuery.CommonTendency(ctx, e.openApi, newStartTime, end, bucket, "request", filters, fieldsConditions, every, windowOffset, flux.SumFn)
fieldsConditions := []string{"request", "response"}
statusFilters := []string{"2xx", "4xx", "5xx"}
dates, overview, err := e.overviewByStatusCode(ctx, start, end, "request", wheres, statusFilters, fieldsConditions, flux.SumFn)
if err != nil {
return nil, nil, nil, err
}
s2xxRequest := groupValues["s2xx_request"]
s4xxRequest := groupValues["s4xx_request"]
s5xxRequest := groupValues["s5xx_request"]
s2xxRequest := overview["2xx_request"]
s2xxRequestLen := len(s2xxRequest)
s4xxRequest := overview["4xx_request"]
s4xxRequestLen := len(s4xxRequest)
s5xxRequest := overview["5xx_request"]
s5xxRequestLen := len(s5xxRequest)
s2xxResponse := overview["2xx_response"]
s2xxResponseLen := len(s2xxResponse)
s4xxResponse := overview["4xx_response"]
s4xxResponseLen := len(s4xxResponse)
s5xxResponse := overview["5xx_response"]
s5xxResponseLen := len(s5xxResponse)
s2xxResponse := groupValues["s2xx_response"]
s4xxResponse := groupValues["s4xx_response"]
s5xxResponse := groupValues["s5xx_response"]
totalOverview := new(monitor.StatusCodeOverview)
result := make([]*monitor.StatusCodeOverview, 0, len(dates))
for i := range dates {
overview := new(monitor.StatusCodeOverview)
overview.Status2xx = s2xxRequest[i] + s2xxResponse[i]
overview.Status4xx = s4xxRequest[i] + s4xxResponse[i]
overview.Status5xx = s5xxRequest[i] + s5xxResponse[i]
overview.StatusTotal = overview.Status2xx + overview.Status4xx + overview.Status5xx
r := new(monitor.StatusCodeOverview)
if s2xxRequestLen > i {
r.Status2xx = s2xxRequest[i]
totalOverview.Status2xx += r.Status2xx
}
if s4xxRequestLen > i {
r.Status4xx = s4xxRequest[i]
totalOverview.Status4xx += r.Status4xx
}
if s5xxRequestLen > i {
r.Status5xx = s5xxRequest[i]
totalOverview.Status5xx += r.Status5xx
}
if s2xxResponseLen > i {
r.Status2xx += s2xxResponse[i]
totalOverview.Status2xx += r.Status2xx
}
if s4xxResponseLen > i {
r.Status4xx += s4xxResponse[i]
totalOverview.Status4xx += r.Status4xx
}
if s5xxResponseLen > i {
r.Status5xx += s5xxResponse[i]
totalOverview.Status5xx += r.Status5xx
}
r.StatusTotal += r.Status2xx + r.Status4xx + r.Status5xx
totalOverview.StatusTotal += r.StatusTotal
result = append(result, r)
totalOverview.StatusTotal += overview.StatusTotal
totalOverview.Status2xx += overview.Status2xx
totalOverview.Status4xx += overview.Status4xx
totalOverview.Status5xx += overview.Status5xx
result = append(result, overview)
}
return dates, totalOverview, result, nil
@@ -539,7 +593,7 @@ func (e *executor) TopN(ctx context.Context, start time.Time, end time.Time, lim
{
Measurement: "request",
AggregateFn: "sum()",
Fields: []string{"total", "request", "total_token"},
Fields: []string{"total", "request", "response", "total_token"},
},
{
Measurement: "proxy",
@@ -558,7 +612,7 @@ func (e *executor) TopN(ctx context.Context, start time.Time, end time.Time, lim
n.Key = key
n.Request = result.Total
n.Token = result.TotalToken
n.Traffic = result.TotalRequest
n.Traffic = result.TotalRequest + result.TotalResponse
topN = append(topN, n)
}
module/monitor/driver/influxdb-v2/flux/flux.go
+19 -13
View File
@@ -61,24 +61,30 @@ func (f *fluxQuery) CommonStatistics(ctx context.Context, queryApi api.QueryAPI,
totalRequest := common.FmtIntFromInterface(maps["request"])
maxRequest := common.FmtIntFromInterface(maps["request_max"])
minRequest := common.FmtIntFromInterface(maps["request_min"])
totalResponse := common.FmtIntFromInterface(maps["response"])
maxResponse := common.FmtIntFromInterface(maps["response_max"])
minResponse := common.FmtIntFromInterface(maps["response_min"])
totalToken := common.FmtIntFromInterface(maps["total_token"])
maxToken := common.FmtIntFromInterface(maps["total_token_max"])
minToken := common.FmtIntFromInterface(maps["total_token_min"])
resultMap[key] = &FluxStatistics{
Total: total,
Success: success,
ProxyTotal: pTotal,
ProxySuccess: pSuccess,
TotalTiming: totalTiming,
MaxTiming: maxMinTiming,
MinTiming: minTiming,
TotalRequest: totalRequest,
RequestMax: maxRequest,
RequestMin: minRequest,
TotalToken: totalToken,
TokenMax: maxToken,
TokenMin: minToken,
Total: total,
Success: success,
ProxyTotal: pTotal,
ProxySuccess: pSuccess,
TotalTiming: totalTiming,
MaxTiming: maxMinTiming,
MinTiming: minTiming,
TotalRequest: totalRequest,
RequestMax: maxRequest,
RequestMin: minRequest,
TotalResponse: totalResponse,
ResponseMax: maxResponse,
ResponseMin: minResponse,
TotalToken: totalToken,
TokenMax: maxToken,
TokenMin: minToken,
}
}
module/monitor/driver/influxdb-v2/flux/monitor_flux.go
+17 -14
View File
@@ -2,20 +2,23 @@ package flux
// FluxStatistics flux统计通用字段
type FluxStatistics struct {
Total int64 `json:"total"` //总数
Success int64 `json:"success"` //成功数
S2xx int64 `json:"s2xx"` //2xx
ProxyTotal int64 `json:"p_total"` //转发总数
ProxySuccess int64 `json:"p_success"` //转发成功数
TotalTiming int64 `json:"timing"` //平均响应时间
MaxTiming int64 `json:"timing_max"` //最大响应时间
MinTiming int64 `json:"timing_min"` //最小响应时间
TotalRequest int64 `json:"request"` //总请求流量
RequestMax int64 `json:"request_max"` //最大流量
RequestMin int64 `json:"request_min"` //最小流量
TotalToken int64 `json:"total_token"` //总token流量
TokenMax int64 `json:"total_token_max"` //最大token流量
TokenMin int64 `json:"total_token_min"` //最小token流量
Total int64 `json:"total"` //总数
Success int64 `json:"success"` //成功数
S2xx int64 `json:"s2xx"` //2xx
ProxyTotal int64 `json:"p_total"` //转发总数
ProxySuccess int64 `json:"p_success"` //转发成功数
TotalTiming int64 `json:"timing"` //平均响应时间
MaxTiming int64 `json:"timing_max"` //最大响应时间
MinTiming int64 `json:"timing_min"` //最小响应时间
TotalRequest int64 `json:"request"` //总请求流量
RequestMax int64 `json:"request_max"` //最大流量
RequestMin int64 `json:"request_min"` //最小流量
TotalResponse int64 `json:"response"` //总请求流量
ResponseMax int64 `json:"response_max"` //最大流量
ResponseMin int64 `json:"response_min"` //最小流量
TotalToken int64 `json:"total_token"` //总token流量
TokenMax int64 `json:"total_token_max"` //最大token流量
TokenMin int64 `json:"total_token_min"` //最小token流量
}
// FluxWarnStatistics flux统计告警通用字段
module/monitor/driver/influxdb-v2/flux/tasks/day.yaml
+14
View File
@@ -26,6 +26,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -43,6 +44,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/hour")
@@ -66,6 +68,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -83,6 +86,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/hour")
@@ -106,6 +110,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -123,6 +128,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/hour")
@@ -144,6 +150,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -161,6 +168,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -191,6 +199,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -209,6 +218,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
@@ -232,6 +242,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -250,6 +261,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
@@ -273,6 +285,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -291,5 +304,6 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
module/monitor/driver/influxdb-v2/flux/tasks/hour.yaml
+14
View File
@@ -26,6 +26,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -43,6 +44,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/minute")
@@ -66,6 +68,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -83,6 +86,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/minute")
@@ -106,6 +110,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -123,6 +128,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/minute")
@@ -144,6 +150,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -161,6 +168,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -191,6 +199,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -209,6 +218,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
@@ -232,6 +242,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -250,6 +261,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
@@ -273,6 +285,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
"_measurement",
],
@@ -291,5 +304,6 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
module/monitor/driver/influxdb-v2/flux/tasks/minute.yaml
+61 -3
View File
@@ -18,6 +18,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -36,6 +37,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_request
@@ -52,6 +54,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -70,6 +73,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_request
@@ -87,6 +91,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -110,6 +115,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -129,6 +135,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_response
@@ -145,6 +152,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -163,6 +171,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_response
@@ -180,6 +189,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -203,6 +213,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -221,7 +232,8 @@
"node",
"cluster",
"provider",
"api_kind",
"api_kind",
"status_code",
],
)
request_retry
@@ -238,6 +250,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -256,6 +269,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -279,6 +293,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -298,6 +313,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_status
@@ -315,6 +331,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -333,6 +350,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -352,6 +370,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_status
@@ -370,6 +389,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -393,6 +413,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -412,6 +433,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_timing
@@ -428,6 +450,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -446,6 +469,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_timing
@@ -463,6 +487,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -487,6 +512,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -507,6 +533,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_timing
@@ -525,6 +552,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_timing
@@ -543,6 +571,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_timing
@@ -561,6 +590,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -585,6 +615,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -605,6 +636,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_status
@@ -623,6 +655,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -641,6 +674,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -661,6 +695,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_status
@@ -680,6 +715,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -704,6 +740,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -724,6 +761,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_request
@@ -741,6 +779,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -760,6 +799,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -784,6 +824,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -803,6 +844,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
proxy_response
@@ -820,6 +862,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -839,6 +882,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -863,6 +907,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -882,6 +927,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_input_token
@@ -899,6 +945,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -918,6 +965,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_input_token
@@ -936,6 +984,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -960,6 +1009,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -979,6 +1029,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_output_token
@@ -996,6 +1047,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -1015,6 +1067,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_output_token
@@ -1033,6 +1086,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
@@ -1041,7 +1095,6 @@
cron: "* * * * *"
offset: "32s"
flux: |
request_total_token =
from(bucket: "apinto")
|> range(start: -1m)
@@ -1058,6 +1111,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_measurement",
],
)
@@ -1077,6 +1131,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_total_token
@@ -1094,6 +1149,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
timeColumn: "_start",
)
@@ -1113,6 +1169,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
request_total_token
@@ -1131,5 +1188,6 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
)
module/monitor/driver/influxdb-v2/flux/tasks/week.yaml
+14
View File
@@ -26,6 +26,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -43,6 +44,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/day")
@@ -66,6 +68,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -83,6 +86,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/day")
@@ -106,6 +110,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -123,6 +128,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/day")
@@ -144,6 +150,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -161,6 +168,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
-
@@ -191,6 +199,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -209,6 +218,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/day")
@@ -231,6 +241,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -249,6 +260,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
from(bucket: "apinto/day")
@@ -271,6 +283,7 @@
"cluster",
"provider",
"api_kind",
"status_code",
"_field",
],
)
@@ -289,5 +302,6 @@
"cluster",
"provider",
"api_kind",
"status_code",
],
)
module/monitor/driver/influxdb-v2/util.go
+14 -6
View File
@@ -15,9 +15,9 @@ const (
tenDay = 10 * oneDay
oneYear = 365 * oneDay
bucketMinuteRetention = (7 - 1) * oneDay
bucketHourRetention = (90 - 1) * oneDay
bucketDayRetention = (5*365 - 1) * oneDay
bucketMinuteRetention = (7) * oneDay
bucketHourRetention = (90) * oneDay
bucketDayRetention = (5 * 365) * oneDay
bucketMinute = "apinto/minute"
bucketHour = "apinto/hour"
@@ -127,11 +127,11 @@ func getTimeIntervalAndBucket(startTime, endTime time.Time) (time.Time, string,
switch minimumBucket {
case bucketMinute:
offset := ""
offsetTime := startTime.Minute() % 5
offsetTime := startTime.Minute() % 10
if offsetTime != 0 {
offset = fmt.Sprintf("%dm", offsetTime)
}
return startTime, "5m", offset, bucketMinute
return startTime, "10m", offset, bucketMinute
case bucketHour:
newStart := formatStartTimeHour(startTime, location)
@@ -148,7 +148,15 @@ func getTimeIntervalAndBucket(startTime, endTime time.Time) (time.Time, string,
} else if diff <= tenDay {
switch minimumBucket {
case bucketMinute, bucketHour:
case bucketMinute:
offset := ""
offsetTime := startTime.Minute()
if offsetTime != 0 {
offset = fmt.Sprintf("%dm", offsetTime)
}
return startTime, "1h", offset, bucketMinute
case bucketHour:
newStart := formatStartTimeHour(startTime, location)
return newStart, "1h", "", bucketHour
case bucketDay:
module/monitor/format.go
+3 -3
View File
@@ -10,9 +10,9 @@ const (
oneDay = 24 * oneHour
tenDay = 10 * oneDay
oneYear = 365 * oneDay
bucketMinuteRetention = (7 - 1) * oneDay
bucketHourRetention = (90 - 1) * oneDay
bucketDayRetention = (5*365 - 1) * oneDay
bucketMinuteRetention = (7) * oneDay
bucketHourRetention = (90) * oneDay
bucketDayRetention = (5 * 365) * oneDay
)
// getTimeIntervalAndBucket 根据start和end来获取窗口时间间隔，窗口偏移量offset，以及使用的bucket, 查询的startTime也会格式化
module/monitor/iml.go
+41 -8
View File
@@ -110,8 +110,8 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
}
subscriberNum := int64(len(appMap))
var wg sync.WaitGroup
wg.Add(2)
errChan := make(chan error, 2)
wg.Add(3)
errChan := make(chan error, 3)
result := new(monitor_dto.ChartAIOverview)
go func() {
defer wg.Done()
@@ -136,7 +136,19 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
result.AvgRequestPerSubscriber = common.FormatCount(summary.StatusTotal / subscriberNum)
result.RequestTotal = common.FormatCount(summary.StatusTotal)
}()
avgResponseTimes := make([]int64, 0)
go func() {
defer wg.Done()
_, _, items, err := executor.AvgResponseTimeOverview(ctx, formatTimeByMinute(start), formatTimeByMinute(end), wheres)
if err != nil {
errChan <- err
return
}
for _, item := range items {
avgResponseTimes = append(avgResponseTimes, item)
}
}()
totalTokens := make([]int64, 0)
go func() {
defer wg.Done()
startTime := formatTimeByMinute(start)
@@ -146,7 +158,6 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
errChan <- err
return
}
timeInterval := getTimeInterval(startTime, endTime)
result.TokenOverview = make([]*monitor_dto.TokenOverview, 0, len(items))
result.AvgTokenOverview = make([]int64, 0, len(items))
result.AvgTokenPerSubscriberOverview = make([]*monitor_dto.TokenOverview, 0, len(items))
@@ -163,7 +174,7 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
OutputToken: item.OutputToken,
InputToken: item.InputToken,
})
result.AvgTokenOverview = append(result.AvgTokenOverview, item.TotalToken/timeInterval)
totalTokens = append(totalTokens, item.TotalToken)
result.AvgTokenPerSubscriberOverview = append(result.AvgTokenPerSubscriberOverview, &monitor_dto.TokenOverview{
TotalToken: item.TotalToken / subscriberNum,
OutputToken: item.OutputToken / subscriberNum,
@@ -172,9 +183,9 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
}
result.AvgTokenPerSubscriber = common.FormatCount(summary.TotalToken / subscriberNum)
result.MaxToken = fmt.Sprintf("%s/s", common.FormatCount(maxToken/timeInterval))
result.MinToken = fmt.Sprintf("%s/s", common.FormatCount(minToken/timeInterval))
result.AvgToken = fmt.Sprintf("%s/s", common.FormatCount(summary.OutputToken/timeInterval))
//result.MaxToken = fmt.Sprintf("%s/s", common.FormatCount(maxToken/timeInterval))
//result.MinToken = fmt.Sprintf("%s/s", common.FormatCount(minToken/timeInterval))
//result.AvgToken = fmt.Sprintf("%s/s", common.FormatCount(summary.OutputToken/timeInterval))
result.TokenTotal = common.FormatCount(summary.TotalToken)
}()
go func() {
@@ -190,6 +201,27 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service
if len(errs) > 0 {
return nil, fmt.Errorf("errors occurred: %v", errs)
}
var maxTokenPerSecond, minTokenPerSecond, avgTokenPerSecond int64 = 0, 0, 0
for index, token := range totalTokens {
var p int64 = 0
if len(avgResponseTimes) > index && avgResponseTimes[index] > 0 {
// 由于时间单位是ms，因此需要✖️1000
p = int64(float64(token) * 1000 / float64(avgResponseTimes[index]))
}
result.AvgTokenOverview = append(result.AvgTokenOverview, p)
if maxTokenPerSecond < p {
maxTokenPerSecond = p
}
if minTokenPerSecond == 0 || minTokenPerSecond > p {
minTokenPerSecond = p
}
avgTokenPerSecond += p
}
if len(avgResponseTimes) > 0 {
result.AvgToken = fmt.Sprintf("%s/s", common.FormatCount(avgTokenPerSecond/int64(len(avgResponseTimes))))
}
result.MaxToken = fmt.Sprintf("%s/s", common.FormatCount(maxTokenPerSecond))
result.MinToken = fmt.Sprintf("%s/s", common.FormatCount(minTokenPerSecond))
return result, nil
}
@@ -290,6 +322,7 @@ func (i *imlMonitorStatisticModule) RestChartOverview(ctx context.Context, servi
})
result.AvgTrafficPerSubscriberOverview = append(result.AvgTrafficPerSubscriberOverview, item.StatusTotal/subscriberNum)
}
result.TrafficTotal = common.FormatByte(summary.StatusTotal)
result.AvgTrafficPerSubscriber = common.FormatCount(summary.StatusTotal / subscriberNum)
}()
go func() {