kpi search optimize

src/modules/network_data/repository/perf_kpi.go

@@ -7,9 +7,15 @@ type IPerfKPI interface {
 	// SelectGoldKPI 通过网元指标数据信息
 	SelectGoldKPI(query model.GoldKPIQuery, kpiIds []string) []map[string]any
 
+	// select from new kpi report table, exp. kpi_report_upf
+	SelectKpiReport(query model.GoldKPIQuery, kpiIds []string) []map[string]any
+
 	// SelectGoldKPITitle 网元对应的指标名称
 	SelectGoldKPITitle(neType string) []model.GoldKPITitle
 
 	// SelectUPFTotalFlow 查询UPF总流量 N3上行 N6下行
 	SelectUPFTotalFlow(neType, rmUID, startDate, endDate string) map[string]any
+
+	// select upf throughput from new kpi_report
+	SelectUPFThroughput(neType, rmUID, startDate, endDate string) map[string]any
 }

src/modules/network_data/repository/perf_kpi.impl.go

@@ -86,6 +86,79 @@ func (r *PerfKPIImpl) SelectGoldKPI(query model.GoldKPIQuery, kpiIds []string) [
 	return results
 }
 
+func (r *PerfKPIImpl) SelectKpiReport(query model.GoldKPIQuery, kpiIds []string) []map[string]any {
+	// 查询条件拼接
+	var conditions []string
+	var params []any
+	var tableName string = "kpi_report_"
+	if query.RmUID != "" {
+		conditions = append(conditions, "gk.rm_uid = ?")
+		params = append(params, query.RmUID)
+	}
+	if query.NeType != "" {
+		conditions = append(conditions, "gk.ne_type = ?")
+		params = append(params, query.NeType)
+		tableName += strings.ToLower(query.NeType)
+	}
+	var dateTimeStr string = "CONCAT(gk.`date`, \" \", gk.start_time)"
+	if query.StartTime != "" {
+		conditions = append(conditions, dateTimeStr+" >= ?")
+		params = append(params, query.StartTime)
+	}
+	if query.EndTime != "" {
+		conditions = append(conditions, dateTimeStr+" <= ?")
+		params = append(params, query.EndTime)
+	}
+	// 构建查询条件语句
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += " where " + strings.Join(conditions, " and ")
+	}
+
+	// 查询字段列
+	timeFormat := "DATE_FORMAT(" + dateTimeStr + ", '%Y-%m-%d %H:%i:')"
+	secondGroup := fmt.Sprintf("LPAD(FLOOR(SECOND(gk.start_time) / %d) * %d, 2, '0')", query.Interval, query.Interval)
+	groupByField := fmt.Sprintf("CONCAT( %s, %s ) AS timeGroup", timeFormat, secondGroup)
+	if query.Interval > 60 {
+		minute := query.Interval / 60
+		timeFormat = "DATE_FORMAT(" + dateTimeStr + ", '%Y-%m-%d %H:')"
+		minuteGroup := fmt.Sprintf("LPAD(FLOOR(MINUTE(gk.start_time) / %d) * %d, 2, '0')", minute, minute)
+		groupByField = fmt.Sprintf("CONCAT( %s, %s ) AS timeGroup", timeFormat, minuteGroup)
+	}
+	var fields = []string{
+		groupByField,
+		"min(CASE WHEN gk.index != '' THEN gk.index ELSE 0 END) AS startIndex",
+		"min(CASE WHEN gk.ne_type != '' THEN gk.ne_type ELSE 0 END) AS neType",
+		"min(CASE WHEN gk.ne_name != '' THEN gk.ne_name ELSE 0 END) AS neName",
+	}
+	for i, kid := range kpiIds {
+		// 特殊字段，只取最后一次收到的非0值
+		if kid == "AMF.01" || kid == "UDM.01" || kid == "UDM.02" || kid == "UDM.03" || kid == "SMF.01" {
+			str := fmt.Sprintf("IFNULL(SUBSTRING_INDEX(GROUP_CONCAT( CASE WHEN JSON_EXTRACT(gk.kpi_values, '$[%d].kpi_id') = '%s' THEN JSON_EXTRACT(gk.kpi_values, '$[%d].value') END  ), ',', 1), 0) AS '%s'", i, kid, i, kid)
+			fields = append(fields, str)
+		} else {
+			str := fmt.Sprintf("sum(CASE WHEN JSON_EXTRACT(gk.kpi_values, '$[%d].kpi_id') = '%s' THEN JSON_EXTRACT(gk.kpi_values, '$[%d].value') ELSE 0 END) AS '%s'", i, kid, i, kid)
+			fields = append(fields, str)
+		}
+	}
+	fieldsSql := strings.Join(fields, ",")
+
+	// 查询数据
+	if query.SortField == "" {
+		query.SortField = "timeGroup"
+	}
+	if query.SortOrder == "" {
+		query.SortOrder = "desc"
+	}
+	orderSql := fmt.Sprintf(" order by %s %s", query.SortField, query.SortOrder)
+	querySql := fmt.Sprintf("SELECT %s FROM %s gk %s GROUP BY timeGroup %s", fieldsSql, tableName, whereSql, orderSql)
+	results, err := datasource.RawDB("", querySql, params)
+	if err != nil {
+		logger.Errorf("query err => %v", err)
+	}
+	return results
+}
+
 // SelectGoldKPITitle 网元对应的指标名称
 func (r *PerfKPIImpl) SelectGoldKPITitle(neType string) []model.GoldKPITitle {
 	result := []model.GoldKPITitle{}
@@ -131,3 +204,39 @@ func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID, startDate, endDate strin
 	}
 	return results[0]
 }
+
+// SelectUPFTotalFlow 查询UPF总流量 N3上行 N6下行
+func (r *PerfKPIImpl) SelectUPFThroughput(neType, rmUID, startDate, endDate string) map[string]any {
+	// 查询条件拼接
+	var conditions []string
+	var params []any
+	if neType != "" {
+		conditions = append(conditions, "gk.ne_type = ?")
+		params = append(params, neType)
+	}
+	if rmUID != "" {
+		conditions = append(conditions, "gk.rm_uid = ?")
+		params = append(params, rmUID)
+	}
+	if startDate != "" {
+		conditions = append(conditions, "gk.date >= ?")
+		params = append(params, startDate)
+	}
+	if endDate != "" {
+		conditions = append(conditions, "gk.date <= ?")
+		params = append(params, endDate)
+	}
+	// 构建查询条件语句
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += " where " + strings.Join(conditions, " and ")
+	}
+
+	// 查询数据
+	querySql := fmt.Sprintf("SELECT sum( CASE WHEN  JSON_EXTRACT(gk.kpi_values, '$[2].kpi_id') = 'UPF.03' THEN JSON_EXTRACT(gk.kpi_values, '$[2].value') ELSE 0 END ) AS 'up', sum( CASE WHEN  JSON_EXTRACT(gk.kpi_values, '$[5].kpi_id') = 'UPF.06' THEN JSON_EXTRACT(gk.kpi_values, '$[5].value') ELSE 0 END ) AS 'down' FROM kpi_report_upf gk %s", whereSql)
+	results, err := datasource.RawDB("", querySql, params)
+	if err != nil {
+		logger.Errorf("query err => %v", err)
+	}
+	return results[0]
+}

src/modules/network_data/service/perf_kpi.impl.go

@@ -31,7 +31,8 @@ func (r *PerfKPIImpl) SelectGoldKPI(query model.GoldKPIQuery) []map[string]any {
 		kpiIds = append(kpiIds, kpiId.KPIID)
 	}
 
-	data := r.perfKPIRepository.SelectGoldKPI(query, kpiIds)
+	//data := r.perfKPIRepository.SelectGoldKPI(query, kpiIds)
+	data := r.perfKPIRepository.SelectKpiReport(query, kpiIds)
 	if data == nil {
 		return []map[string]any{}
 	}
@@ -66,7 +67,8 @@ func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID string, day int) map[stri
 		}
 	}
 
-	info = r.perfKPIRepository.SelectUPFTotalFlow(neType, rmUID, startDate, endDate)
+	//info = r.perfKPIRepository.SelectUPFTotalFlow(neType, rmUID, startDate, endDate)
+	info = r.perfKPIRepository.SelectUPFThroughput(neType, rmUID, startDate, endDate)
 
 	// 保存到缓存
 	infoJSON, _ := json.Marshal(info)