OMC/nms_cxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

merge: 合并OMC分支

Browse Source
This commit is contained in:
TsMask
2024-07-10 14:18:48 +08:00
parent 17c0011c6b
commit 625ed57a50
260 changed files with 9167 additions and 14857 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
features/pm/performance.go
View File

@@ -8,6 +8,7 @@ import (
"math"
"net/http"
"strconv"
"strings"
"time"
"nms_cxy/lib/dborm"
@@ -65,6 +66,26 @@ type GoldKpi struct {
Timestamp string `json:"timestamp"`
}
type KpiData struct {
ID int `json:"id" xorm:"pk 'id' '<-' autoincr"`
NEType string `json:"neType" xorm:"ne_type"`
NEName string `json:"neName" xorm:"ne_name"`
RmUid string `json:"rmUid" xorm:"rm_uid"`
Date string `json:"date" xorm:"date"`
StartTime string `json:"startTime" xorm:"start_time"`
EndTime string `json:"endTime" xorm:"end_time"`
Index int `json:"index" xorm:"index"`
Granularity int8 `json:"granularity" xorm:"granularity"`
KPIValues []KPIVal `json:"kpiValues" xorm:"json 'kpi_values'"`
//CreatedAt int64 `json:"createdAt" xorm:"created 'created_at'"`
CreatedAt int64 `json:"createdAt" xorm:"'created_at'"`
}
type KPIVal struct {
KPIID string `json:"kpi_id" xorm:"kpi_id"`
Value int64 `json:"value" xorm:"value"`
Err string `json:"err" xorm:"err"`
}
var (
// performance management
PerformanceUri = config.DefaultUriPrefix + "/performanceManagement/{apiVersion}/elementType/{elementTypeValue}/objectType/kpiReport/{index}"
@@ -98,9 +119,9 @@ type DatabaseClient struct {
var DbClient DatabaseClient
func InitDbClient(dbType, dbUser, dbPassword, dbHost, dbPort, dbName string) error {
DbClient.dbUrl = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&collation=utf8mb4_general_ci&parseTime=true&loc=Local",
dbUser, dbPassword, dbHost, dbPort, dbName)
func InitDbClient(dbType, dbUser, dbPassword, dbHost, dbPort, dbName, dbParam string) error {
DbClient.dbUrl = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?%s",
dbUser, dbPassword, dbHost, dbPort, dbName, dbParam)
DbClient.dbType = dbType
DbClient.dbConnMaxLifetime = 0
DbClient.dbMaxIdleConns = 0
@@ -108,8 +129,7 @@ func InitDbClient(dbType, dbUser, dbPassword, dbHost, dbPort, dbName string) err
if log.GetLevel() == log.LOG_TRACE {
DbClient.IsShowSQL = true
}
log.Debugf("dbType:%s dbUrl:%s:******@tcp(%s:%s)/%s??charset=utf8mb4&collation=utf8mb4_general_ci&parseTime=true&loc=Local",
dbType, dbUser, dbHost, dbPort, dbName)
log.Debugf("dbType:%s dbUrl:%s:", dbType, DbClient.dbUrl)
var err error
DbClient.XEngine, err = xorm.NewEngine(DbClient.dbType, DbClient.dbUrl)
@@ -120,11 +140,28 @@ func InitDbClient(dbType, dbUser, dbPassword, dbHost, dbPort, dbName string) err
DbClient.XEngine.SetConnMaxLifetime(DbClient.dbConnMaxLifetime)
DbClient.XEngine.SetMaxIdleConns(DbClient.dbMaxIdleConns)
DbClient.XEngine.SetMaxOpenConns(DbClient.dbMaxOpenConns)
DbClient.XEngine.DatabaseTZ = time.Local // 必须
DbClient.XEngine.TZLocation = time.Local // 必须
if DbClient.IsShowSQL {
DbClient.XEngine.ShowSQL(true)
}
xEngine = DbClient.XEngine
// exist, err := xEngine.IsTableExist("kpi_report")
// if err != nil {
// log.Error("Failed to IsTableExist:", err)
// return err
// }
// if exist {
// // 复制表结构到新表
// sql := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` AS SELECT * FROM kpi_report WHERE 1=0", "kpi_report_amf")
// _, err := xEngine.Exec(sql)
// if err != nil {
// log.Error("Failed to Exec:", err)
// return err
// }
// }
return nil
}
@@ -155,7 +192,7 @@ func GetDateTimeFromTimeString(fmtString string, timeString string) string {
return t.Format(global.DateTime)
}
// process alarm post message from NFs
// process KPI report post message from NFs
func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
log.Debug("PostKPIReportFromNF processing... ")
@@ -167,6 +204,91 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
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)
layout := time.RFC3339Nano
//kpiDate := GetDateFromTimeString(layout, kpiReport.Task.Period.StartTime)
kpiIndex, _ := strconv.Atoi(vars["index"])
startTime := global.GetFmtTimeString(layout, kpiReport.Task.Period.StartTime, time.DateTime)
endTime := global.GetFmtTimeString(layout, kpiReport.Task.Period.EndTime, time.DateTime)
// get time granularity from startTime and endTime
seconds, _ := global.GetSecondDuration(startTime, endTime)
var granularity int8 = 60
if seconds != 0 && seconds <= math.MaxInt8 && seconds >= math.MinInt8 {
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
kpiData.Index = kpiIndex
//stime, _ := time.ParseInLocation(time.RFC3339Nano, kpiReport.Task.Period.StartTime, time.Local)
//etime, _ := time.ParseInLocation(time.RFC3339Nano, kpiReport.Task.Period.EndTime, time.Local)
kpiData.StartTime = startTime
kpiData.EndTime = endTime
kpiData.Granularity = granularity
kpiData.NEName = kpiReport.Task.NE.NEName
kpiData.NEType = kpiReport.Task.NE.NeType
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 {
@@ -205,12 +327,32 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
"startIndex": goldKpi.Index,
"timeGroup": goldKpi.StartTime,
}
// 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 _, 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
log.Trace("goldKpi:", goldKpi)
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()
@@ -244,13 +386,22 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
}
}
// 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 goldKpi.NEType == "UPF" {
wsService.NewWSSendImpl.ByGroupID(wsService.GROUP_KPI_UPF, kpiEvent)
}
services.ResponseStatusOK200Null(w)
services.ResponseStatusOK204NoContent(w)
}
type MeasureTask struct {
@@ -903,9 +1054,9 @@ func PostMeasurementFromNF(w http.ResponseWriter, r *http.Request) {
layout := time.RFC3339
measurement.Date = GetDateFromTimeString(layout, measurement.StartTime)
measurement.StartTime = GetDateTimeFromTimeString(layout, measurement.StartTime)
affected, err := session.Table("northbound_pm").Insert(measurement)
affected, err := session.Table("nbi_pm").Insert(measurement)
if err != nil && affected <= 0 {
log.Error("Failed to insert northbound_pm:", err)
log.Error("Failed to insert nbi_pm:", err)
services.ResponseInternalServerError500DatabaseOperationFailed(w)
return
}
@@ -986,9 +1137,9 @@ func GetMeasurementFromNF(w http.ResponseWriter, r *http.Request) {
layout := time.RFC3339
measurement.Date = GetDateFromTimeString(layout, measurement.StartTime)
measurement.StartTime = GetDateTimeFromTimeString(layout, measurement.StartTime)
affected, err := session.Table("northbound_pm").Insert(measurement)
affected, err := session.Table("nbi_pm").Insert(measurement)
if err != nil && affected <= 0 {
log.Error("Failed to insert northbound_pm:", err)
log.Error("Failed to insert nbi_pm:", err)
services.ResponseInternalServerError500DatabaseOperationFailed(w)
return
}