add: custom kpi features

.gitignore

@@ -49,3 +49,4 @@ vendor
 *.exe
 __debug_bin*.exe
 
+tools/evaluate/*.go

features/features.go

@@ -0,0 +1,17 @@
+package features
+
+import (
+	"be.ems/features/pm"
+	"be.ems/lib/log"
+	"github.com/gin-gonic/gin"
+)
+
+func InitServiceEngine(r *gin.Engine) {
+	log.Info("======init feature group gin.Engine")
+
+	// featuresGroup := r.Group("/")
+	// 注册 PM 模块的路由
+	pm.InitSubServiceRoute(r)
+
+	// return featuresGroup
+}

features/pm/kpi_c_report/controller.go

@@ -0,0 +1,164 @@
+package kpi_c_report
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"be.ems/src/framework/datasource"
+	"github.com/gin-gonic/gin"
+)
+
+func (k *KpiCReport) Get(c *gin.Context) {
+	var reports []KpiCReport
+	var conditions []string
+	var params []any
+
+	var querys KpiCReportQuery
+	if err := c.ShouldBindQuery(&querys); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	// construct condition to get
+	if querys.NeType != "" {
+		conditions = append(conditions, "ne_type = ?")
+		params = append(params, strings.ToUpper(querys.NeType))
+	} else {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Not found NE type"})
+		return
+	}
+	tableName := TableName() + "_" + strings.ToLower(querys.NeType)
+	dborm := datasource.DefaultDB().Table(tableName)
+
+	if querys.StartTime != "" {
+		conditions = append(conditions, "created_at >= ?")
+		params = append(params, querys.StartTime)
+	}
+	if querys.EndTime != "" {
+		conditions = append(conditions, "created_at <= ?")
+		params = append(params, querys.EndTime)
+	}
+
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += strings.Join(conditions, " and ")
+		dborm = dborm.Where(whereSql, params...)
+	}
+	// page number and size
+	if pageSize := querys.PageSize; pageSize > 0 {
+		dborm = dborm.Limit(pageSize)
+		if pageNum := querys.PageNum; pageNum > 0 {
+			dborm = dborm.Offset((pageNum - 1) * pageSize)
+		}
+	}
+
+	// order by
+	if sortField, sortOrder := querys.SortField, querys.SortOrder; sortField != "" && sortOrder != "" {
+		orderBy := fmt.Sprintf("%s %s", sortField, sortOrder)
+		dborm = dborm.Order(orderBy)
+	}
+
+	//err := datasource.DefaultDB().Table(tableName).Where(whereSql, params...).Find(&reports).Error
+	err := dborm.Find(&reports).Error
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	//c.JSON(http.StatusOK, map[string]any{"data": titles})
+	c.JSON(http.StatusOK, reports)
+}
+
+func (k *KpiCReport) Total(c *gin.Context) {
+	var conditions []string
+	var params []any
+
+	var querys KpiCReportQuery
+	if err := c.ShouldBindQuery(&querys); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	// construct condition to get
+	if querys.NeType != "" {
+		conditions = append(conditions, "ne_type = ?")
+		params = append(params, strings.ToUpper(querys.NeType))
+	} else {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Not found NE type"})
+		return
+	}
+	tableName := TableName() + "_" + strings.ToLower(querys.NeType)
+	dborm := datasource.DefaultDB().Table(tableName)
+
+	if querys.StartTime != "" {
+		conditions = append(conditions, "created_at >= ?")
+		params = append(params, querys.StartTime)
+	}
+	if querys.EndTime != "" {
+		conditions = append(conditions, "created_at <= ?")
+		params = append(params, querys.EndTime)
+	}
+
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += strings.Join(conditions, " and ")
+		dborm = dborm.Where(whereSql, params...)
+	}
+	var total int64 = 0
+	err := dborm.Count(&total).Error
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, map[string]any{"total": total})
+}
+
+func (k *KpiCReport) Post(c *gin.Context) {
+	var report KpiCReport
+
+	if err := c.ShouldBindJSON(&report); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if err := datasource.DefaultDB().Create(&report).Error; err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, report)
+}
+
+func (k *KpiCReport) Put(c *gin.Context) {
+	var report KpiCReport
+	id := c.Param("id")
+
+	if err := datasource.DefaultDB().First(&report, id).Error; err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "KPI report not found"})
+		return
+	}
+
+	if err := c.ShouldBindJSON(&report); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	datasource.DefaultDB().Save(&report)
+	c.JSON(http.StatusOK, report)
+}
+
+func (k *KpiCReport) Delete(c *gin.Context) {
+	id := c.Param("id")
+
+	if err := datasource.DefaultDB().Delete(&KpiCReport{}, id).Error; err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "KPI report not found"})
+		return
+	}
+
+	c.JSON(http.StatusNoContent, nil) // 204 No Content
+}
+
+func InsertKpiCReport(neType string, report KpiCReport) {
+	tableName := TableName() + "_" + strings.ToLower(neType)
+	if err := datasource.DefaultDB().Table(tableName).Create(&report).Error; err != nil {
+		return
+	}
+}

features/pm/kpi_c_report/model.go

@@ -0,0 +1,62 @@
+package kpi_c_report
+
+import (
+	"database/sql/driver"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+type KpiCVal struct {
+	KPIID string  `json:"kpi_id" gorm:"column:kpi_id"`
+	Value float64 `json:"value" gorm:"column:value"`
+	Err   string  `json:"err" gorm:"column:err"`
+}
+
+type KpiCValues []KpiCVal
+
+type KpiCReport struct {
+	ID          int        `gorm:"column:id;primary_key;auto_increment" json:"id"`
+	NeType      *string    `gorm:"column:ne_type;default:NULL" json:"ne_type,omitempty"`
+	NeName      *string    `gorm:"column:ne_name;default:" json:"ne_name,omitempty"`
+	RmUID       *string    `gorm:"column:rm_uid;default:NULL" json:"rm_uid,omitempty"`
+	Date        string     `gorm:"column:date" json:"date"` // time.Time `gorm:"column:date" json:"date"`
+	StartTime   *string    `gorm:"column:start_time;default:NULL" json:"start_time,omitempty"`
+	EndTime     *string    `gorm:"column:end_time;default:NULL" json:"end_time,omitempty"`
+	Index       int16      `gorm:"column:index" json:"index"`
+	Granularity *int8      `gorm:"column:granularity;default:60" json:"granularity,omitempty"` //Time granualarity: 5/10/.../60/300 (second)
+	KpiValues   KpiCValues `gorm:"column:kpi_values;type:json" json:"kpi_values,omitempty"`
+	CreatedAt   *time.Time `gorm:"column:created_at;default:current_timestamp()" json:"created_at,omitempty"`
+	TenantID    *string    `gorm:"column:tenant_id;default:NULL" json:"tenant_id,omitempty"`
+}
+
+type KpiCReportQuery struct {
+	NeType     string `json:"neType" form:"neType" binding:"required"`
+	RmUID      string `json:"rmUID" form:"rmUID"`
+	StartTime  string `json:"startTime" form:"startTime"`
+	EndTime    string `json:"endTime" form:"endTime"`
+	TenantName string `json:"tenantName" form:"tenantName"`
+	UserName   string `json:"userName" form:"userName"`
+	SortField  string `json:"sortField" form:"sortField" binding:"omitempty,oneof=created_at"` // 排序字段，填写结果字段
+	SortOrder  string `json:"sortOrder" form:"sortOrder" binding:"omitempty,oneof=asc desc"`   // 排序升降序，asc desc
+	PageNum    int    `json:"pageNum" form:"pageNum"`
+	PageSize   int    `json:"pageSize" form:"pageSize"`
+}
+
+func TableName() string {
+	return "kpi_c_report"
+}
+
+// 将 KpiCValues 转换为 JSON 字节
+func (k KpiCValues) Value() (driver.Value, error) {
+	return json.Marshal(k)
+}
+
+// 从字节中扫描 KpiCValues
+func (k *KpiCValues) Scan(value interface{}) error {
+	b, ok := value.([]byte)
+	if !ok {
+		return fmt.Errorf("failed to scan value: %v", value)
+	}
+	return json.Unmarshal(b, k)
+}

features/pm/kpi_c_report/route.go

@@ -0,0 +1,35 @@
+package kpi_c_report
+
+import (
+	"be.ems/src/framework/middleware"
+	"github.com/gin-gonic/gin"
+)
+
+// Register Routes for kpi_c_report
+func Register(r *gin.RouterGroup) {
+
+	pmKPIC := r.Group("/kpiC")
+	{
+		var k *KpiCReport
+		pmKPIC.GET("/report",
+			middleware.PreAuthorize(nil),
+			k.Get,
+		)
+		pmKPIC.GET("/report/total",
+			middleware.PreAuthorize(nil),
+			k.Total,
+		)
+		pmKPIC.POST("/report",
+			middleware.PreAuthorize(nil),
+			k.Post,
+		)
+		pmKPIC.PUT("/report/:id",
+			middleware.PreAuthorize(nil),
+			k.Put,
+		)
+		pmKPIC.DELETE("/report/:id",
+			middleware.PreAuthorize(nil),
+			k.Delete,
+		)
+	}
+}

features/pm/kpi_c_title/controller.go

@@ -0,0 +1,113 @@
+package kpi_c_title
+
+import (
+	"net/http"
+	"strings"
+
+	"be.ems/src/framework/datasource"
+	"github.com/gin-gonic/gin"
+)
+
+func (k *KpiCTitle) Get(c *gin.Context) {
+	var titles []KpiCTitle
+	var conditions []string
+	var params []any
+
+	// construct condition to get
+	if neType := c.Query("neType"); neType != "" {
+		conditions = append(conditions, "ne_type = ?")
+		params = append(params, strings.ToUpper(neType))
+	}
+	if status := c.Query("status"); status != "" {
+		conditions = append(conditions, "status = ?")
+		params = append(params, status)
+	}
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += strings.Join(conditions, " and ")
+	}
+	if err := datasource.DefaultDB().Where(whereSql, params...).Find(&titles).Error; err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	//c.JSON(http.StatusOK, map[string]any{"data": titles})
+	c.JSON(http.StatusOK, titles)
+}
+
+func (k *KpiCTitle) Total(c *gin.Context) {
+	var conditions []string
+	var params []any
+
+	// construct condition to get
+	if neType := c.Query("neType"); neType != "" {
+		conditions = append(conditions, "ne_type = ?")
+		params = append(params, strings.ToUpper(neType))
+	}
+	if status := c.Query("status"); status != "" {
+		conditions = append(conditions, "status = ?")
+		params = append(params, status)
+	}
+	whereSql := ""
+	if len(conditions) > 0 {
+		whereSql += strings.Join(conditions, " and ")
+	}
+	var total int64 = 0
+	if err := datasource.DefaultDB().Table(k.TableName()).Where(whereSql, params...).Count(&total).Error; err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, map[string]any{"total": total})
+}
+
+func (k *KpiCTitle) Post(c *gin.Context) {
+	var title KpiCTitle
+
+	if err := c.ShouldBindJSON(&title); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if err := datasource.DefaultDB().Create(&title).Error; err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, title)
+}
+
+func (k *KpiCTitle) Put(c *gin.Context) {
+	var title KpiCTitle
+	id := c.Param("id")
+
+	if err := datasource.DefaultDB().First(&title, id).Error; err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Title not found"})
+		return
+	}
+
+	if err := c.ShouldBindJSON(&title); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	datasource.DefaultDB().Save(&title)
+	c.JSON(http.StatusOK, title)
+}
+
+func (k *KpiCTitle) Delete(c *gin.Context) {
+	id := c.Param("id")
+
+	if err := datasource.DefaultDB().Delete(&KpiCTitle{}, id).Error; err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Title not found"})
+		return
+	}
+
+	c.JSON(http.StatusNoContent, nil) // 204 No Content
+}
+
+func GetActiveKPICList(neType string) []KpiCTitle {
+	k := new([]KpiCTitle)
+
+	err := datasource.DefaultDB().Where("`ne_type` = ? and `status` = 'Active'", neType).Find(&k).Error
+	if err != nil {
+		return nil
+	}
+	return *k
+}

features/pm/kpi_c_title/model.go

@@ -0,0 +1,19 @@
+package kpi_c_title
+
+import "time"
+
+type KpiCTitle struct {
+	ID          int        `gorm:"column:id;primary_key;auto_increment" json:"id"`
+	NeType      *string    `gorm:"column:ne_type;default:NULL," json:"ne_type,omitempty"`
+	KpiID       *string    `gorm:"column:kpi_id;default:NULL," json:"kpi_id,omitempty"`
+	Title       *string    `gorm:"column:title;default:NULL," json:"title,omitempty"`
+	Expression  *string    `gorm:"column:expression;default:NULL," json:"expression,omitempty"`
+	Status      *string    `gorm:"column:status" json:"status,omitempty"`
+	Description *string    `gorm:"column:description;default:NULL," json:"description,omitempty"`
+	CreatedBy   *string    `gorm:"column:created_by;default:NULL," json:"created_by,omitempty"`
+	UpdatedAt   *time.Time `gorm:"column:updated_at;default:current_timestamp()," json:"updated_at,omitempty"`
+}
+
+func (k *KpiCTitle) TableName() string {
+	return "kpi_c_title"
+}

features/pm/kpi_c_title/route.go

@@ -0,0 +1,35 @@
+package kpi_c_title
+
+import (
+	"be.ems/src/framework/middleware"
+	"github.com/gin-gonic/gin"
+)
+
+// Register Routes for kpi_c_title
+func Register(r *gin.RouterGroup) {
+
+	pmKPIC := r.Group("/kpiC")
+	{
+		var k *KpiCTitle
+		pmKPIC.GET("/title",
+			middleware.PreAuthorize(nil),
+			k.Get,
+		)
+		pmKPIC.GET("/title/total",
+			middleware.PreAuthorize(nil),
+			k.Total,
+		)
+		pmKPIC.POST("/title",
+			middleware.PreAuthorize(nil),
+			k.Post,
+		)
+		pmKPIC.PUT("/title/:id",
+			middleware.PreAuthorize(nil),
+			k.Put,
+		)
+		pmKPIC.DELETE("/title/:id",
+			middleware.PreAuthorize(nil),
+			k.Delete,
+		)
+	}
+}

features/pm/performance.go

@@ -10,7 +10,10 @@ 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"
@@ -251,6 +254,9 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
 		"startIndex": kpiIndex,
 		"timeGroup":  kpiData.CreatedAt,
 	}
+
+	// for custom kpi
+	kpiValMap := map[string]any{}
 	for _, k := range kpiReport.Task.NE.KPIs {
 		kpiEvent[k.KPIID] = k.Value // kip_id
 
@@ -258,7 +264,9 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
 		kpiVal.Value = int64(k.Value)
 		kpiVal.Err = k.Err
 		kpiData.KPIValues = append(kpiData.KPIValues, *kpiVal)
+		kpiValMap[k.KPIID] = k.Value
 	}
+	kpiValMap["granularity"] = kpiData.Granularity
 
 	// set tenant_name if exist
 	where := fmt.Sprintf("status='1' and tenancy_type='UPF' and tenancy_key='%s'", kpiData.RmUid)
@@ -272,6 +280,34 @@ func PostKPIReportFromNF(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	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,
+		TenantID:    &kpiData.TenantID,
+	}
+
+	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)
+	}
+
+	kpi_c_report.InsertKpiCReport(kpiData.NEType, report)
 	// 发送到匹配的网元
 	neInfo := neService.NewNeInfoImpl.SelectNeInfoByRmuid(kpiData.RmUid)
 	if neInfo.RmUID == kpiData.RmUid {

features/pm/service.go

@@ -0,0 +1,19 @@
+package pm
+
+import (
+	"be.ems/features/pm/kpi_c_report"
+	"be.ems/features/pm/kpi_c_title"
+	"be.ems/lib/log"
+	"github.com/gin-gonic/gin"
+)
+
+func InitSubServiceRoute(r *gin.Engine) {
+	log.Info("======init PM group gin.Engine")
+
+	pmGroup := r.Group("/pm")
+	// register sub modules routes
+	kpi_c_title.Register(pmGroup)
+	kpi_c_report.Register(pmGroup)
+
+	// return featuresGroup
+}

lib/eval/evaluate.go

@@ -0,0 +1,111 @@
+package evaluate
+
+import (
+	"fmt"
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// Parse and caculate expression
+func CalcExpr(expr string, paramValues map[string]any) (float64, error) {
+	// match parameter with ''
+	re := regexp.MustCompile(`'([^']+)'`)
+	matches := re.FindAllStringSubmatch(expr, -1)
+
+	// replace to value
+	for _, match := range matches {
+		paramName := match[1]
+		value, exists := paramValues[paramName]
+		if !exists {
+			return 0, fmt.Errorf("parameter '%s' not found", paramName)
+		}
+
+		expr = strings.Replace(expr, match[0], fmt.Sprintf("%v", value), 1)
+	}
+
+	// expression to evaluate
+	result, err := evalExpr(expr)
+	return result, err
+}
+
+// eval 解析和计算表达式
+func evalExpr(expr string) (float64, error) {
+	//fset := token.NewFileSet()
+	node, err := parser.ParseExpr(expr)
+	if err != nil {
+		return 0, err
+	}
+	return evalNode(node)
+}
+
+// EvaluateExpr 解析并计算给定的表达式
+func EvalExpr(expr string, values map[string]any) (float64, error) {
+	// 解析表达式
+	node, err := parser.ParseExpr(expr)
+	if err != nil {
+		return 0, err
+	}
+
+	// 遍历 AST 并替换变量
+	ast.Inspect(node, func(n ast.Node) bool {
+		if ident, ok := n.(*ast.Ident); ok {
+			if val, ok := values[ident.Name]; ok {
+				// 替换标识符为对应值
+				ident.Name = fmt.Sprintf("%v", val)
+			}
+		}
+		return true
+	})
+
+	// 计算表达式
+	return evalNode(node)
+}
+
+// eval 递归计算 AST 节点
+func evalNode(node ast.Node) (float64, error) {
+	var result float64
+
+	switch n := node.(type) {
+	case *ast.BinaryExpr:
+		left, err := evalNode(n.X)
+		if err != nil {
+			return 0, err
+		}
+		right, err := evalNode(n.Y)
+		if err != nil {
+			return 0, err
+		}
+		switch n.Op {
+		case token.ADD:
+			result = left + right
+		case token.SUB:
+			result = left - right
+		case token.MUL:
+			result = left * right
+		case token.QUO:
+			result = left / right
+		}
+	case *ast.BasicLit:
+		var err error
+		result, err = strconv.ParseFloat(n.Value, 64)
+		if err != nil {
+			return 0, err
+		}
+	case *ast.Ident:
+		val, err := strconv.ParseFloat(n.Name, 64)
+		if err != nil {
+			return 0, fmt.Errorf("unsupported expression: %s", n.Name)
+		}
+		result = val
+	case *ast.ParenExpr:
+		return evalNode(n.X) // 递归评估括号中的表达式
+	default:
+		return 0, fmt.Errorf("unsupported expression: %T", n)
+	}
+
+	return result, nil
+}

restagent/restagent.go

@@ -11,6 +11,7 @@ import (
 
 	_ "net/http/pprof"
 
+	"be.ems/features"
 	"be.ems/features/dbrest"
 	"be.ems/features/event"
 	"be.ems/features/fm"
@@ -251,6 +252,9 @@ func main() {
 	// AMF上报的UE事件, 无前缀，暂时特殊处理
 	app.POST(event.UriUEEventAMF, event.PostUEEventFromAMF)
 
+	// register feature service gin.Engine
+	features.InitServiceEngine(app)
+
 	var listenLocalhost bool = false
 	for _, rest := range conf.Rest {
 		// ipv4 goroutines