From fef49eb32c49c4a2628a1d4bcdf0fc1477ee8fff Mon Sep 17 00:00:00 2001 From: Liujian <824010343@qq.com> Date: Wed, 30 Apr 2025 18:55:01 +0800 Subject: [PATCH] tmp commit --- common/format.go | 17 +- module/monitor/driver/driver.go | 1 + module/monitor/driver/influxdb-v2/executor.go | 8 + .../monitor/driver/influxdb-v2/flux/flux.go | 42 ++++ module/monitor/dto/output.go | 30 ++- module/monitor/format.go | 52 ---- module/monitor/iml.go | 236 +++++++++++------- 7 files changed, 230 insertions(+), 156 deletions(-) delete mode 100644 module/monitor/format.go diff --git a/common/format.go b/common/format.go index e389aa1a..1b799d21 100644 --- a/common/format.go +++ b/common/format.go @@ -5,7 +5,7 @@ import ( "strconv" ) -func FormatCount(count int64) string { +func FormatCountInt64(count int64) string { switch { case count < 1000: return strconv.FormatInt(count, 10) @@ -20,6 +20,21 @@ func FormatCount(count int64) string { } } +func FormatCountFloat64(count float64) string { + switch { + case count < 1000: + return strconv.FormatFloat(count, 'f', -1, 64) + case count < 1000000: + return fmt.Sprintf("%.1fK", count/1000) + case count < 1000000000: + return fmt.Sprintf("%.1fM", count/1000000) + case count < 1000000000000: + return fmt.Sprintf("%.1fB", count/1000000000) + default: + return fmt.Sprintf("%.1fT", count/1000000000000) + } +} + func FormatTime(t int64) string { if t < 1000 { return strconv.FormatInt(t, 10) + "ms" diff --git a/module/monitor/driver/driver.go b/module/monitor/driver/driver.go index d9d93701..c596e983 100644 --- a/module/monitor/driver/driver.go +++ b/module/monitor/driver/driver.go @@ -32,6 +32,7 @@ type IExecutor interface { type IBasicOverview interface { RequestOverview(ctx context.Context, start time.Time, end time.Time, wheres []monitor.MonWhereItem) ([]time.Time, *monitor.StatusCodeOverview, []*monitor.StatusCodeOverview, error) TopN(ctx context.Context, start time.Time, end time.Time, limit int, groupBy string, wheres []monitor.MonWhereItem) ([]*monitor.TopN, error) + ConsumerOverview(ctx context.Context, start time.Time, end time.Time, wheres []monitor.MonWhereItem) (int64, map[time.Time]int64, error) } type IRestOverview interface { diff --git a/module/monitor/driver/influxdb-v2/executor.go b/module/monitor/driver/influxdb-v2/executor.go index b558c7fc..2fc7ab26 100644 --- a/module/monitor/driver/influxdb-v2/executor.go +++ b/module/monitor/driver/influxdb-v2/executor.go @@ -684,3 +684,11 @@ func (e *executor) TokenOverview(ctx context.Context, start time.Time, end time. } return dates, totalOverview, result, nil } + +func (e *executor) ConsumerOverview(ctx context.Context, start time.Time, end time.Time, wheres []monitor.MonWhereItem) (int64, map[time.Time]int64, error) { + newStartTime, every, _, bucket := getTimeIntervalAndBucket(start, end) + filters := formatFilter(wheres) + + return e.fluxQuery.CommonTendencyTag(ctx, e.openApi, newStartTime, end, bucket, "request", filters, every, "app") + +} diff --git a/module/monitor/driver/influxdb-v2/flux/flux.go b/module/monitor/driver/influxdb-v2/flux/flux.go index bdbbcd87..94fccd8c 100644 --- a/module/monitor/driver/influxdb-v2/flux/flux.go +++ b/module/monitor/driver/influxdb-v2/flux/flux.go @@ -15,6 +15,7 @@ type IFluxQuery interface { CommonStatistics(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, groupBy, filters string, statisticsConf []*StatisticsFilterConf, limit int) (map[string]*FluxStatistics, error) CommonProxyStatistics(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, groupBy, filters string, statisticsConf []*StatisticsFilterConf, limit int) (map[string]*FluxStatistics, error) CommonTendency(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, table, filters string, dataFields []string, every, windowOffset string, fn AggregateFn) ([]time.Time, map[string][]int64, error) + CommonTendencyTag(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, table, filters, every, tag string) (int64, map[time.Time]int64, error) // CommonQueryOnce 查询只返回一条结果 CommonQueryOnce(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, filters string, fieldsConf *StatisticsFilterConf) (map[string]interface{}, error) CommonWarnStatistics(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, groupBy, filters string, statisticsConf *StatisticsFilterConf) (map[string]*FluxWarnStatistics, error) @@ -171,6 +172,35 @@ func (f *fluxQuery) CommonTendency(ctx context.Context, queryApi api.QueryAPI, s return dates, resultMap, nil } +func (f *fluxQuery) CommonTendencyTag(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, table, filters, every, tag string) (int64, map[time.Time]int64, error) { + query := f.assembleTendencyTagFlux(start, end, bucket, table, filters, every, tag) + log.Info("flux sql=", query) + result, err := queryApi.Query(ctx, query) + if err != nil { + log.Error("flux err=", err) + return 0, nil, err + } + dateMap := map[time.Time]map[string]struct{}{} + tagMap := make(map[string]struct{}) + defer result.Close() + for result.Next() { + date := result.Record().Values()["_time"].(time.Time).In(time.Local) + if _, ok := dateMap[date]; !ok { + dateMap[date] = map[string]struct{}{} + } + if vv, ok := result.Record().Values()[tag]; ok { + v := vv.(string) + tagMap[v] = struct{}{} + dateMap[date][v] = struct{}{} + } + } + returnMap := make(map[time.Time]int64) + for k, v := range dateMap { + returnMap[k] = int64(len(v)) + } + return int64(len(tagMap)), returnMap, nil +} + func (f *fluxQuery) CommonQueryOnce(ctx context.Context, queryApi api.QueryAPI, start, end time.Time, bucket, filters string, fieldsConf *StatisticsFilterConf) (map[string]interface{}, error) { query := f.getCircularMapFlux(start, end, bucket, filters, fieldsConf) @@ -180,6 +210,7 @@ func (f *fluxQuery) CommonQueryOnce(ctx context.Context, queryApi api.QueryAPI, log.Error("flux err=", err) return nil, err } + defer result.Close() for result.Next() { return result.Record().Values(), nil @@ -324,6 +355,17 @@ func (f *fluxQuery) assembleTendencyFlux(start, end time.Time, bucket, table, fi } +func (f *fluxQuery) assembleTendencyTagFlux(start, end time.Time, bucket, table, filters string, every, tag string) string { + return fmt.Sprintf(` +from(bucket: "%s") + |> range(start: %d, stop: %d) + |> filter(fn: (r) => r["_measurement"] == "%s") + %s +|> keep(columns: ["_time", "%s"]) + |> window(every: %s) + |> distinct(column: "%s")`, bucket, start.Unix(), end.Unix(), table, filters, tag, every, tag) +} + // assembleTendencyFieldCondition 封装趋势图需要的Field数据 func (f *fluxQuery) assembleTendencyFieldCondition(fieldConditions []string) string { /* diff --git a/module/monitor/dto/output.go b/module/monitor/dto/output.go index bb9c8e40..96104a96 100644 --- a/module/monitor/dto/output.go +++ b/module/monitor/dto/output.go @@ -154,10 +154,16 @@ type TokenOverview struct { InputToken int64 `json:"input_token"` //最小token流量 } +type TokenFloatOverview struct { + TotalToken float64 `json:"total_token"` //总token流量 + OutputToken float64 `json:"output_token"` + InputToken float64 `json:"input_token"` //最小token流量 +} + type ChartAIOverview struct { RequestOverview []*StatusCodeOverview `json:"request_overview"` AvgRequestPerSubscriber string `json:"avg_request_per_subscriber"` //请求概况 - AvgRequestPerSubscriberOverview []int64 `json:"avg_request_per_subscriber_overview"` //平均响应时间概况 + AvgRequestPerSubscriberOverview []float64 `json:"avg_request_per_subscriber_overview"` //平均响应时间概况 RequestTotal string `json:"request_total"` Request2xxTotal string `json:"request_2xx_total"` Request4xxTotal string `json:"request_4xx_total"` @@ -166,8 +172,8 @@ type ChartAIOverview struct { InputTokenTotal string `json:"input_token_total"` OutputTokenTotal string `json:"output_token_total"` //最大token流量 TokenOverview []*TokenOverview `json:"token_overview"` //token概况 - AvgTokenOverview []int64 `json:"avg_token_overview"` - AvgTokenPerSubscriberOverview []*TokenOverview `json:"avg_token_per_subscriber_overview"` + AvgTokenOverview []float64 `json:"avg_token_overview"` + AvgTokenPerSubscriberOverview []*TokenFloatOverview `json:"avg_token_per_subscriber_overview"` AvgToken string `json:"avg_token"` MaxToken string `json:"max_token"` MinToken string `json:"min_token"` @@ -178,7 +184,7 @@ type ChartAIOverview struct { type ChartRestOverview struct { RequestOverview []*StatusCodeOverview `json:"request_overview"` //请求概况 AvgRequestPerSubscriber string `json:"avg_request_per_subscriber"` - AvgRequestPerSubscriberOverview []int64 `json:"avg_request_per_subscriber_overview"` //平均响应时间概况 + AvgRequestPerSubscriberOverview []float64 `json:"avg_request_per_subscriber_overview"` //平均响应时间概况 RequestTotal string `json:"request_total"` Request2xxTotal string `json:"request_2xx_total"` @@ -190,14 +196,14 @@ type ChartRestOverview struct { Traffic4xxTotal string `json:"traffic_4xx_total"` //流量概况 Traffic5xxTotal string `json:"traffic_5xx_total"` //流量概况 - AvgResponseTimeOverview []int64 `json:"avg_response_time_overview"` //平均响应时间概况 - AvgTrafficPerSubscriberOverview []int64 `json:"avg_traffic_per_subscriber_overview"` - TrafficTotal string `json:"traffic_total"` - AvgResponseTime string `json:"avg_response_time"` //平均响应时间 - MaxResponseTime string `json:"max_response_time"` //最大响应时间 - MinResponseTime string `json:"min_response_time"` //最小响应时间 - AvgTrafficPerSubscriber string `json:"avg_traffic_per_subscriber"` - Date []string `json:"date"` + AvgResponseTimeOverview []int64 `json:"avg_response_time_overview"` //平均响应时间概况 + AvgTrafficPerSubscriberOverview []float64 `json:"avg_traffic_per_subscriber_overview"` + TrafficTotal string `json:"traffic_total"` + AvgResponseTime string `json:"avg_response_time"` //平均响应时间 + MaxResponseTime string `json:"max_response_time"` //最大响应时间 + MinResponseTime string `json:"min_response_time"` //最小响应时间 + AvgTrafficPerSubscriber string `json:"avg_traffic_per_subscriber"` + Date []string `json:"date"` } type ServiceChartRestOverview struct { diff --git a/module/monitor/format.go b/module/monitor/format.go deleted file mode 100644 index 55054dbf..00000000 --- a/module/monitor/format.go +++ /dev/null @@ -1,52 +0,0 @@ -package monitor - -import ( - "time" -) - -const ( - oneMinute = 60 - oneHour = 3600 - oneDay = 24 * oneHour - tenDay = 10 * oneDay - oneYear = 365 * oneDay - bucketMinuteRetention = (7) * oneDay - bucketHourRetention = (90) * oneDay - bucketDayRetention = (5 * 365) * oneDay -) - -// getTimeIntervalAndBucket 根据start和end来获取窗口时间间隔，窗口偏移量offset，以及使用的bucket, 查询的startTime也会格式化 -func getTimeInterval(startTime, endTime time.Time) int64 { - startToNow := time.Now().Unix() - startTime.Unix() - - //结合可使用的最小桶，根据end-start时间间隔来得出合适的桶和趋势图时间间隔 - diff := endTime.Unix() - startTime.Unix() - if diff <= oneHour { - return 5 * oneMinute - } else if diff <= oneDay { - - switch { - case startToNow <= bucketHourRetention: - return oneHour - case startToNow <= bucketDayRetention: - return oneDay - default: - return 7 * oneDay - } - - } else if diff <= tenDay { - - switch { - case startToNow <= bucketHourRetention: - return oneHour - case startToNow <= bucketDayRetention: - return oneDay - default: - return 7 * oneDay - } - - } else { - return 7 * oneDay - } - -} diff --git a/module/monitor/iml.go b/module/monitor/iml.go index 2ca4587e..9bc9637a 100644 --- a/module/monitor/iml.go +++ b/module/monitor/iml.go @@ -83,32 +83,36 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service return nil, err } - serviceIds := make([]string, 0) - // 从数据库中获取相关信息 - if serviceId == "" { - // 获取全部服务 - list, err := i.serviceService.ServiceListByKind(ctx, service.AIService) - if err != nil { - - return nil, err - } - serviceIds = utils.SliceToSlice(list, func(t *service.Service) string { - return t.Id - }) - } else { - serviceIds = append(serviceIds, serviceId) + //serviceIds := make([]string, 0) + //// 从数据库中获取相关信息 + //if serviceId == "" { + // // 获取全部服务 + // list, err := i.serviceService.ServiceListByKind(ctx, service.AIService) + // if err != nil { + // + // return nil, err + // } + // serviceIds = utils.SliceToSlice(list, func(t *service.Service) string { + // return t.Id + // }) + //} else { + // serviceIds = append(serviceIds, serviceId) + //} + //appMap := make(map[string]struct{}) + //for _, sId := range serviceIds { + // items, err := i.subscribeService.ListBySubscribeStatus(ctx, sId, subscribe.ApplyStatusSubscribe) + // if err != nil { + // return nil, err + // } + // for _, item := range items { + // appMap[item.Application] = struct{}{} + // } + //} + //subscriberNum := int64(len(appMap)) + totalConsumerCount, consumerMap, err := executor.ConsumerOverview(ctx, formatTimeByMinute(start), formatTimeByMinute(end), wheres) + if err != nil { + return nil, err } - appMap := make(map[string]struct{}) - for _, sId := range serviceIds { - items, err := i.subscribeService.ListBySubscribeStatus(ctx, sId, subscribe.ApplyStatusSubscribe) - if err != nil { - return nil, err - } - for _, item := range items { - appMap[item.Application] = struct{}{} - } - } - subscriberNum := int64(len(appMap)) var wg sync.WaitGroup wg.Add(3) errChan := make(chan error, 3) @@ -123,21 +127,30 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service result.Date = utils.SliceToSlice(date, func(t time.Time) string { return t.Format("2006/01/02 15:04") }) - result.AvgRequestPerSubscriberOverview = make([]int64, 0, len(items)) + result.AvgRequestPerSubscriberOverview = make([]float64, 0, len(items)) result.RequestOverview = make([]*monitor_dto.StatusCodeOverview, 0, len(items)) - for _, item := range items { - result.AvgRequestPerSubscriberOverview = append(result.AvgRequestPerSubscriberOverview, item.StatusTotal/subscriberNum) + for index, item := range items { + consumerNum := consumerMap[date[index]] + avgRequestPerSubscriber := 0.0 + if consumerNum != 0 { + avgRequestPerSubscriber = float64(item.StatusTotal) / float64(consumerNum) + } + result.AvgRequestPerSubscriberOverview = append(result.AvgRequestPerSubscriberOverview, avgRequestPerSubscriber) result.RequestOverview = append(result.RequestOverview, &monitor_dto.StatusCodeOverview{ Status2xx: item.Status2xx, Status4xx: item.Status4xx, Status5xx: item.Status5xx, }) } - result.AvgRequestPerSubscriber = common.FormatCount(summary.StatusTotal / subscriberNum) - result.RequestTotal = common.FormatCount(summary.StatusTotal) - result.Request2xxTotal = common.FormatCount(summary.Status2xx) - result.Request4xxTotal = common.FormatCount(summary.Status4xx) - result.Request5xxTotal = common.FormatCount(summary.Status5xx) + avgRequestPerSubscriber := 0.0 + if totalConsumerCount != 0 { + avgRequestPerSubscriber = float64(summary.StatusTotal) / float64(totalConsumerCount) + } + result.AvgRequestPerSubscriber = common.FormatCountFloat64(avgRequestPerSubscriber) + result.RequestTotal = common.FormatCountInt64(summary.StatusTotal) + result.Request2xxTotal = common.FormatCountInt64(summary.Status2xx) + result.Request4xxTotal = common.FormatCountInt64(summary.Status4xx) + result.Request5xxTotal = common.FormatCountInt64(summary.Status5xx) }() sumResponseTimes := make([]int64, 0) go func() { @@ -156,16 +169,16 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service defer wg.Done() startTime := formatTimeByMinute(start) endTime := formatTimeByMinute(end) - _, summary, items, err := executor.TokenOverview(ctx, startTime, endTime, wheres) + date, summary, items, err := executor.TokenOverview(ctx, startTime, endTime, wheres) if err != nil { errChan <- err return } 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)) + result.AvgTokenOverview = make([]float64, 0, len(items)) + result.AvgTokenPerSubscriberOverview = make([]*monitor_dto.TokenFloatOverview, 0, len(items)) var maxToken, minToken int64 = 0, 0 - for _, item := range items { + for index, item := range items { if maxToken < item.TotalToken { maxToken = item.TotalToken } @@ -178,23 +191,41 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service InputToken: item.InputToken, }) totalTokens = append(totalTokens, item.TotalToken) - result.AvgTokenPerSubscriberOverview = append(result.AvgTokenPerSubscriberOverview, &monitor_dto.TokenOverview{ - TotalToken: item.TotalToken / subscriberNum, - OutputToken: item.OutputToken / subscriberNum, - InputToken: item.InputToken / subscriberNum, + consumerNum := consumerMap[date[index]] + avgTotalPerSubscriber := 0.0 + avgOutputPerSubscriber := 0.0 + avgInputPerSubscriber := 0.0 + if consumerNum != 0 { + avgTotalPerSubscriber = float64(item.TotalToken) / float64(consumerNum) + avgOutputPerSubscriber = float64(item.OutputToken) / float64(consumerNum) + avgInputPerSubscriber = float64(item.InputToken) / float64(consumerNum) + } + + result.AvgTokenPerSubscriberOverview = append(result.AvgTokenPerSubscriberOverview, &monitor_dto.TokenFloatOverview{ + TotalToken: avgTotalPerSubscriber, + OutputToken: avgOutputPerSubscriber, + InputToken: avgInputPerSubscriber, }) } - result.AvgTokenPerSubscriber = common.FormatCount(summary.TotalToken / subscriberNum) - result.TokenTotal = common.FormatCount(summary.TotalToken) - result.InputTokenTotal = common.FormatCount(summary.InputToken) - result.OutputTokenTotal = common.FormatCount(summary.OutputToken) + avgTokenPerSubscriber := 0.0 + if totalConsumerCount != 0 { + avgTokenPerSubscriber = float64(summary.TotalToken) / float64(totalConsumerCount) + } + result.AvgToken = common.FormatCountFloat64(avgTokenPerSubscriber) + result.MaxToken = common.FormatCountInt64(maxToken) + result.MinToken = common.FormatCountInt64(minToken) + result.AvgTokenPerSubscriber = common.FormatCountFloat64(avgTokenPerSubscriber) + result.AvgTokenPerSubscriber = common.FormatCountInt64(summary.TotalToken / totalConsumerCount) + result.TokenTotal = common.FormatCountInt64(summary.TotalToken) + result.InputTokenTotal = common.FormatCountInt64(summary.InputToken) + result.OutputTokenTotal = common.FormatCountInt64(summary.OutputToken) }() go func() { wg.Wait() close(errChan) }() - errs := make([]error, 0, 2) + errs := make([]error, 0, 3) // 收集错误 for err := range errChan { errs = append(errs, err) @@ -203,12 +234,12 @@ 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 + var maxTokenPerSecond, minTokenPerSecond, avgTokenPerSecond float64 = 0, 0, 0 for index, token := range totalTokens { - var p int64 = 0 + var p float64 = 0 if len(sumResponseTimes) > index && sumResponseTimes[index] > 0 { // 由于时间单位是ms，因此需要✖️1000 - p = int64(float64(token) * 1000 / float64(sumResponseTimes[index])) + p = float64(token) * 1000 / float64(sumResponseTimes[index]) } result.AvgTokenOverview = append(result.AvgTokenOverview, p) if maxTokenPerSecond < p { @@ -220,10 +251,10 @@ func (i *imlMonitorStatisticModule) AIChartOverview(ctx context.Context, service avgTokenPerSecond += p } if len(sumResponseTimes) > 0 { - result.AvgToken = fmt.Sprintf("%s Token/s", common.FormatCount(avgTokenPerSecond/int64(len(sumResponseTimes)))) + result.AvgToken = fmt.Sprintf("%s Token/s", common.FormatCountFloat64(avgTokenPerSecond/float64(len(sumResponseTimes)))) } - result.MaxToken = fmt.Sprintf("%s Token/s", common.FormatCount(maxTokenPerSecond)) - result.MinToken = fmt.Sprintf("%s Token/s", common.FormatCount(minTokenPerSecond)) + result.MaxToken = fmt.Sprintf("%s Token/s", common.FormatCountFloat64(maxTokenPerSecond)) + result.MinToken = fmt.Sprintf("%s Token/s", common.FormatCountFloat64(minTokenPerSecond)) return result, nil } @@ -237,31 +268,36 @@ func (i *imlMonitorStatisticModule) RestChartOverview(ctx context.Context, servi return nil, err } - serviceIds := make([]string, 0) - // 从数据库中获取相关信息 - if serviceId == "" { - // 获取全部服务 - list, err := i.serviceService.ServiceListByKind(ctx, service.RestService) - if err != nil { - return nil, err - } - serviceIds = utils.SliceToSlice(list, func(t *service.Service) string { - return t.Id - }) - } else { - serviceIds = append(serviceIds, serviceId) + //serviceIds := make([]string, 0) + //// 从数据库中获取相关信息 + //if serviceId == "" { + // // 获取全部服务 + // list, err := i.serviceService.ServiceListByKind(ctx, service.RestService) + // if err != nil { + // return nil, err + // } + // serviceIds = utils.SliceToSlice(list, func(t *service.Service) string { + // return t.Id + // }) + //} else { + // serviceIds = append(serviceIds, serviceId) + //} + //appMap := make(map[string]struct{}) + //for _, sId := range serviceIds { + // items, err := i.subscribeService.ListBySubscribeStatus(ctx, sId, subscribe.ApplyStatusSubscribe) + // if err != nil { + // return nil, err + // } + // for _, item := range items { + // appMap[item.Id] = struct{}{} + // } + //} + //subscriberNum := int64(len(appMap)) + totalConsumerCount, consumerMap, err := executor.ConsumerOverview(ctx, formatTimeByMinute(start), formatTimeByMinute(end), wheres) + if err != nil { + return nil, err } - appMap := make(map[string]struct{}) - for _, sId := range serviceIds { - items, err := i.subscribeService.ListBySubscribeStatus(ctx, sId, subscribe.ApplyStatusSubscribe) - if err != nil { - return nil, err - } - for _, item := range items { - appMap[item.Id] = struct{}{} - } - } - subscriberNum := int64(len(appMap)) + var wg sync.WaitGroup wg.Add(3) errChan := make(chan error, 2) @@ -276,21 +312,30 @@ func (i *imlMonitorStatisticModule) RestChartOverview(ctx context.Context, servi result.Date = utils.SliceToSlice(date, func(t time.Time) string { return t.Format("2006/01/02 15:04") }) - result.AvgRequestPerSubscriberOverview = make([]int64, 0, len(items)) + result.AvgRequestPerSubscriberOverview = make([]float64, 0, len(items)) result.RequestOverview = make([]*monitor_dto.StatusCodeOverview, 0, len(items)) - for _, item := range items { - result.AvgRequestPerSubscriberOverview = append(result.AvgRequestPerSubscriberOverview, item.StatusTotal/int64(subscriberNum)) + for index, item := range items { + consumerNum := consumerMap[date[index]] + avgRequestPerSubscriber := 0.0 + if consumerNum != 0 { + avgRequestPerSubscriber = float64(summary.StatusTotal) / float64(consumerNum) + } + result.AvgRequestPerSubscriberOverview = append(result.AvgRequestPerSubscriberOverview, avgRequestPerSubscriber) result.RequestOverview = append(result.RequestOverview, &monitor_dto.StatusCodeOverview{ Status2xx: item.Status2xx, Status4xx: item.Status4xx, Status5xx: item.Status5xx, }) } - result.AvgRequestPerSubscriber = common.FormatCount(summary.StatusTotal / subscriberNum) - result.RequestTotal = common.FormatCount(summary.StatusTotal) - result.Request2xxTotal = common.FormatCount(summary.Status2xx) - result.Request4xxTotal = common.FormatCount(summary.Status4xx) - result.Request5xxTotal = common.FormatCount(summary.Status5xx) + avgRequestPerSubscriber := 0.0 + if totalConsumerCount != 0 { + avgRequestPerSubscriber = float64(summary.StatusTotal) / float64(totalConsumerCount) + } + result.AvgRequestPerSubscriber = common.FormatCountFloat64(avgRequestPerSubscriber) + result.RequestTotal = common.FormatCountInt64(summary.StatusTotal) + result.Request2xxTotal = common.FormatCountInt64(summary.Status2xx) + result.Request4xxTotal = common.FormatCountInt64(summary.Status4xx) + result.Request5xxTotal = common.FormatCountInt64(summary.Status5xx) }() go func() { @@ -312,26 +357,35 @@ func (i *imlMonitorStatisticModule) RestChartOverview(ctx context.Context, servi defer wg.Done() startTime := formatTimeByMinute(start) endTime := formatTimeByMinute(end) - _, summary, items, err := executor.TrafficOverviewByStatusCode(ctx, startTime, endTime, wheres) + date, summary, items, err := executor.TrafficOverviewByStatusCode(ctx, startTime, endTime, wheres) if err != nil { errChan <- err return } result.TrafficOverview = make([]*monitor_dto.StatusCodeOverview, 0, len(items)) - result.AvgTrafficPerSubscriberOverview = make([]int64, 0, len(items)) - for _, item := range items { + result.AvgTrafficPerSubscriberOverview = make([]float64, 0, len(items)) + for index, item := range items { result.TrafficOverview = append(result.TrafficOverview, &monitor_dto.StatusCodeOverview{ Status2xx: item.Status2xx, Status4xx: item.Status4xx, Status5xx: item.Status5xx, }) - result.AvgTrafficPerSubscriberOverview = append(result.AvgTrafficPerSubscriberOverview, item.StatusTotal/subscriberNum) + consumerNum := consumerMap[date[index]] + avgTrafficPerSubscriber := 0.0 + if consumerNum != 0 { + avgTrafficPerSubscriber = float64(item.StatusTotal) / float64(consumerNum) + } + result.AvgTrafficPerSubscriberOverview = append(result.AvgTrafficPerSubscriberOverview, avgTrafficPerSubscriber) } result.TrafficTotal = common.FormatByte(summary.StatusTotal) result.Traffic2xxTotal = common.FormatByte(summary.Status2xx) result.Traffic4xxTotal = common.FormatByte(summary.Status4xx) result.Traffic5xxTotal = common.FormatByte(summary.Status5xx) - result.AvgTrafficPerSubscriber = common.FormatCount(summary.StatusTotal / subscriberNum) + avgTrafficPerSubscriber := 0.0 + if totalConsumerCount != 0 { + avgTrafficPerSubscriber = float64(summary.StatusTotal) / float64(totalConsumerCount) + } + result.AvgTrafficPerSubscriber = common.FormatCountFloat64(avgTrafficPerSubscriber) }() go func() { wg.Wait() @@ -353,13 +407,13 @@ func generateTopN(id string, name string, item *monitor.TopN, apiKind string) *m n := &monitor_dto.TopN{ Id: id, Name: name, - Request: common.FormatCount(item.Request), + Request: common.FormatCountInt64(item.Request), } switch apiKind { case "rest": n.Traffic = common.FormatByte(item.Traffic) case "ai": - n.Token = common.FormatCount(item.Token) + n.Token = common.FormatCountInt64(item.Token) } return n }