simonzhangsz/omc_api
1
0
Fork 0
Code Activity

merge: 合并代码20241018

Browse Source
This commit is contained in:
TsMask
2024-10-18 17:26:59 +08:00
parent 49860c2f28
commit 17f57175c7
289 changed files with 21476 additions and 12863 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
features/pm/performance.go
View File

@@ -2,7 +2,6 @@ package pm
import (
"encoding/json"
"errors"
"fmt"
"io"
"math"
@@ -11,17 +10,21 @@ import (
"strings"
"time"
"be.ems/features/pm/kpi_c_report"
"be.ems/features/pm/kpi_c_title"
"be.ems/lib/dborm"
evaluate "be.ems/lib/eval"
"be.ems/lib/global"
"be.ems/lib/log"
"be.ems/lib/services"
"be.ems/restagent/config"
"xorm.io/xorm"
neService "be.ems/src/modules/network_element/service"
wsService "be.ems/src/modules/ws/service"
"github.com/go-resty/resty/v2"
_ "github.com/go-sql-driver/mysql"
"github.com/gorilla/mux"
"xorm.io/xorm"
)
type Response struct {
@@ -41,7 +44,7 @@ type KpiReport struct {
NeType string `json:"NeType"`
KPIs []struct {
KPIID string `json:"KPIID"`
Value int `json:"Value"`
Value int64 `json:"Value"`
Err string `json:"Err"`
} `json:"KPIs"`
} `json:"NE"`
@@ -59,7 +62,7 @@ type GoldKpi struct {
RmUid string `json:"rmUid" xorm:"rm_uid"`
NEType string `json:"neType" xorm:"ne_type"`
KpiId string `json:"kpiId" xorm:"kpi_id"`
Value int `json:"value"`
Value int64 `json:"value"`
Error string `json:"error"`
Timestamp string `json:"timestamp"`
}
@@ -226,14 +229,6 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
granularity = int8(seconds)
}
// 黄金指标事件对象
kpiEvent := map[string]any{
// kip_id ...
"neType": kpiReport.Task.NE.NeType,
"neName": kpiReport.Task.NE.NEName,
"startIndex": kpiIndex,
"timeGroup": startTime,
}
// insert into new kpi_report_xxx table
kpiData := new(KpiData)
kpiData.Date = startTime
@@ -248,141 +243,29 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
kpiData.RmUid = kpiReport.Task.NE.RmUID
kpiVal := new(KPIVal)
kpiData.CreatedAt = time.Now().UnixMilli()
for _, k := range kpiReport.Task.NE.KPIs {
kpiEvent[k.KPIID] = k.Value // kip_id
kpiVal.KPIID = k.KPIID
kpiVal.Value = int64(k.Value)
kpiVal.Err = k.Err
kpiData.KPIValues = append(kpiData.KPIValues, *kpiVal)
}
// insert kpi_report table, no session
tableName := "kpi_report_" + strings.ToLower(kpiReport.Task.NE.NeType)
affected, err := xEngine.Table(tableName).Insert(kpiData)
if err != nil && affected <= 0 {
log.Errorf("Failed to insert %s:%v", tableName, err)
services.ResponseInternalServerError500ProcessError(w, err)
return
}
// 推送到ws订阅组
wsService.NewWSSendImpl.ByGroupID(wsService.GROUP_KPI, kpiEvent)
if kpiReport.Task.NE.NeType == "UPF" {
wsService.NewWSSendImpl.ByGroupID(wsService.GROUP_KPI_UPF, kpiEvent)
}
services.ResponseStatusOK204NoContent(w)
}
// post kpi report from NEs, insert insto gold_kpi table, discard...
func PostGoldKPIFromNF(w http.ResponseWriter, r *http.Request) {
log.Debug("PostKPIReportFromNF processing... ")
vars := mux.Vars(r)
apiVer := vars["apiVersion"]
if apiVer != global.ApiVersionV1 {
log.Error("Uri api version is invalid. apiVersion:", apiVer)
services.ResponseNotFound404UriNotExist(w, r)
return
}
// body, err := io.ReadAll(io.LimitReader(r.Body, global.RequestBodyMaxLen))
body, err := io.ReadAll(io.LimitReader(r.Body, global.RequestBodyMaxLen))
if err != nil {
log.Error("Faile to io.ReadAll: ", err)
services.ResponseNotFound404UriNotExist(w, r)
return
}
log.Trace("Request body:", string(body))
kpiReport := new(KpiReport)
_ = json.Unmarshal(body, &kpiReport)
log.Trace("kpiReport:", kpiReport)
session := xEngine.NewSession()
defer session.Close()
goldKpi := new(GoldKpi)
layout := time.RFC3339Nano
goldKpi.Date = GetDateFromTimeString(layout, kpiReport.Task.Period.StartTime)
goldKpi.Index, _ = strconv.Atoi(vars["index"])
goldKpi.StartTime = global.GetFmtTimeString(layout, kpiReport.Task.Period.StartTime, time.DateTime)
goldKpi.EndTime = global.GetFmtTimeString(layout, kpiReport.Task.Period.EndTime, time.DateTime)
// get time granularity from startTime and endTime
seconds, _ := global.GetSecondDuration(goldKpi.StartTime, goldKpi.EndTime)
goldKpi.Granularity = 60
if seconds != 0 && seconds <= math.MaxInt8 && seconds >= math.MinInt8 {
goldKpi.Granularity = int8(seconds)
}
goldKpi.NEName = kpiReport.Task.NE.NEName
goldKpi.RmUid = kpiReport.Task.NE.RmUID
goldKpi.NEType = kpiReport.Task.NE.NeType
goldKpi.Timestamp = global.GetFmtTimeString(layout, kpiReport.Timestamp, time.DateTime)
// 黄金指标事件对象
kpiEvent := map[string]any{
// kip_id ...
"neType": goldKpi.NEType,
"neName": goldKpi.NEName,
"startIndex": goldKpi.Index,
"timeGroup": goldKpi.StartTime,
"neType": kpiReport.Task.NE.NeType,
"neName": kpiReport.Task.NE.NEName,
"rmUID": kpiReport.Task.NE.RmUID,
"startIndex": kpiIndex,
"timeGroup": kpiData.CreatedAt,
}
// insert into new kpi_report_xxx table
kpiData := new(KpiData)
kpiData.Date = goldKpi.Date
kpiData.Index = goldKpi.Index
//st, _ := time.ParseInLocation(time.RFC3339Nano, kpiReport.Task.Period.StartTime, time.Local)
//et, _ := time.ParseInLocation(time.RFC3339Nano, kpiReport.Task.Period.EndTime, time.Local)
kpiData.StartTime = goldKpi.StartTime
kpiData.EndTime = goldKpi.EndTime
kpiData.Granularity = goldKpi.Granularity
kpiData.NEName = goldKpi.NEName
kpiData.NEType = goldKpi.NEType
kpiData.RmUid = goldKpi.RmUid
kpiVal := new(KPIVal)
kpiData.CreatedAt = time.Now().UnixMilli()
// for custom kpi
kpiValMap := map[string]any{}
for _, k := range kpiReport.Task.NE.KPIs {
kpiEvent[k.KPIID] = k.Value // kip_id
goldKpi.KpiId = k.KPIID
goldKpi.Value = k.Value
goldKpi.Error = k.Err
kpiVal.KPIID = k.KPIID
kpiVal.Value = int64(k.Value)
kpiVal.Err = k.Err
kpiData.KPIValues = append(kpiData.KPIValues, *kpiVal)
//log.Trace("goldKpi:", goldKpi)
// 启动事务
err := session.Begin()
if err != nil {
log.Error("Failed to Begin gold_kpi:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
}
gkpi := &GoldKpi{}
_, err = session.Where("id = ?", 1).ForUpdate().Get(gkpi)
if err != nil {
// 回滚事务
session.Rollback()
log.Error("Failed to ForUpdate gold_kpi:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
}
affected, err := session.Insert(goldKpi)
if err != nil && affected <= 0 {
session.Rollback()
log.Error("Failed to insert gold_kpi:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
}
// 提交事务
err = session.Commit()
if err != nil {
log.Error("Failed to Commit gold_kpi:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
}
kpiValMap[k.KPIID] = k.Value
}
kpiValMap["granularity"] = kpiData.Granularity
// insert kpi_report table, no session
tableName := "kpi_report_" + strings.ToLower(kpiReport.Task.NE.NeType)
@@ -393,10 +276,60 @@ func PostGoldKPIFromNF(w http.ResponseWriter, r *http.Request) {
return
}
// 推送到ws订阅组
wsService.NewWSSendImpl.ByGroupID(wsService.GROUP_KPI, kpiEvent)
if goldKpi.NEType == "UPF" {
wsService.NewWSSendImpl.ByGroupID(wsService.GROUP_KPI_UPF, kpiEvent)
report := kpi_c_report.KpiCReport{
NeType: &kpiData.NEType,
NeName: &kpiData.NEName,
RmUID: &kpiData.RmUid,
Date: kpiData.Date,
StartTime: &kpiData.StartTime,
EndTime: &kpiData.EndTime,
Index: int16(kpiData.Index),
Granularity: &kpiData.Granularity,
}
// 发送到匹配的网元
neInfo := neService.NewNeInfo.SelectNeInfoByRmuid(kpiData.RmUid)
// custom kpi report to FE
kpiCEvent := map[string]any{
// kip_id ...
"neType": kpiData.NEType,
"neId": neInfo.NeId,
"neName": kpiData.NEName,
"rmUID": kpiData.RmUid,
"startIndex": kpiData.Index,
"timeGroup": kpiData.Date[:10] + " " + kpiData.EndTime,
"createdAt": kpiData.CreatedAt,
"granularity": kpiData.Granularity,
}
kpiCList := kpi_c_title.GetActiveKPICList(kpiData.NEType)
for _, k := range kpiCList {
result, err := evaluate.CalcExpr(*k.Expression, kpiValMap)
kpiCVal := new(kpi_c_report.KpiCVal)
kpiCVal.KPIID = *k.KpiID
if err != nil {
kpiCVal.Value = 0.0
kpiCVal.Err = err.Error()
} else {
kpiCVal.Value = result
}
report.KpiValues = append(report.KpiValues, *kpiCVal)
// set KPIC event kpiid and value
kpiCEvent[kpiCVal.KPIID] = kpiCVal.Value
}
// KPI自定义指标入库
kpi_c_report.InsertKpiCReport(kpiData.NEType, report)
if neInfo.RmUID == kpiData.RmUid {
// 推送到ws订阅组
wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI, neInfo.NeType, neInfo.NeId), kpiEvent)
// 推送自定义KPI到ws订阅组
wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI_C, neInfo.NeType, neInfo.NeId), kpiCEvent)
if neInfo.NeType == "UPF" {
wsService.NewWSSend.ByGroupID(wsService.GROUP_KPI_UPF+neInfo.NeId, kpiEvent)
}
}
services.ResponseStatusOK204NoContent(w)
@@ -639,7 +572,7 @@ func PostMeasureTaskToNF(w http.ResponseWriter, r *http.Request) {
return
}
if neInfo == nil {
err := errors.New(fmt.Sprintf("not found target NE neType=%s, neId=%s", neType, neId))
err := fmt.Errorf("not found target NE neType=%s, neId=%s", neType, neId)
log.Error(err)
services.ResponseInternalServerError500ProcessError(w, err)
return
@@ -692,7 +625,7 @@ func PostMeasureTaskToNF(w http.ResponseWriter, r *http.Request) {
return
}
default:
err = errors.New(fmt.Sprintf("measure task status must be inactive id=%d", id))
err = fmt.Errorf("measure task status must be inactive id=%d", id)
log.Error("Unable to active measure task:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
@@ -719,7 +652,7 @@ func PostMeasureTaskToNF(w http.ResponseWriter, r *http.Request) {
services.TransportResponse(w, response.StatusCode(), response.Body())
return
} else {
err = errors.New(fmt.Sprintf("failed to active measure task, NF return error status=%v", response.Status()))
err = fmt.Errorf("failed to active measure task, NF return error status=%v", response.Status())
log.Error("Unable to active measure task:", err)
services.ResponseInternalServerError500ProcessError(w, err)
return
@@ -932,8 +865,6 @@ func PatchMeasureTaskToNF(w http.ResponseWriter, r *http.Request) {
return
}
if neInfo == nil {
em := errors.New("Not found NE info in database")
log.Error(em)
taskInfo := new(dborm.MeasureTask)
taskInfo.Status = dborm.MeasureTaskStatusInactive
affected, err := dborm.XormUpdateTableById(id, dborm.TableNameMeasureTask, taskInfo)
@@ -986,7 +917,6 @@ func PatchMeasureTaskToNF(w http.ResponseWriter, r *http.Request) {
}
services.ResponseWithJson(w, response.StatusCode(), respMsg)
return
}
type Measurement struct {