OMC/be.ems
1
0
Fork 0
Code Issues Releases 3 Activity

Merge remote-tracking branch 'origin/main' into practical-training

Browse Source
This commit is contained in:
TsMask
2024-07-27 14:18:54 +08:00
parent c7d3b0885f 6e5530ad5e
commit c5cd851e7a
64 changed files with 1795 additions and 2156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/framework/config/config/config.default.yaml
View File

@@ -1,7 +1,7 @@
# 项目信息
framework:
name: "CN EMS"
version: "2.2407.3"
version: "2.2407.4"
# 应用服务配置
server:

49
src/framework/utils/file/zip.go
View File

@@ -8,6 +8,55 @@ import (
"path/filepath"
)
// UnZip 解 ZIP 压缩文件输出到目录下
func UnZip(zipFilePath, dirPath string) error {
// 打开ZIP文件进行读取
r, err := zip.OpenReader(zipFilePath)
if err != nil {
return err
}
defer r.Close()
// 创建本地输出目录
if err := os.MkdirAll(dirPath, 0775); err != nil {
return err
}
// 遍历ZIP文件中的每个文件并解压缩到输出目录
for _, f := range r.File {
// 打开ZIP文件中的文件
rc, err := f.Open()
if err != nil {
return err
}
defer rc.Close()
// 创建解压后的文件
path := filepath.ToSlash(filepath.Join(dirPath, f.Name))
if f.FileInfo().IsDir() {
// 如果是目录，创建目录
if err := os.MkdirAll(path, 0775); err != nil {
return err
}
} else {
if err = os.MkdirAll(filepath.Dir(path), 0775); err != nil {
return err
}
out, err := os.Create(path)
if err != nil {
return err
}
defer out.Close()
if _, err = io.Copy(out, rc); err != nil {
return err
}
}
}
return nil
}
// CompressZipByFile 将单文件添加到 ZIP 压缩文件
func CompressZipByFile(zipFilePath, filePath string) error {
// 创建一个新的 ZIP 文件

65
src/framework/utils/ssh/sftp.go
View File

@@ -23,6 +23,13 @@ func (s *SSHClientSFTP) Close() {
// CopyDirRemoteToLocal 复制目录-远程到本地
func (s *SSHClientSFTP) CopyDirRemoteToLocal(remoteDir, localDir string) error {
// 创建本地目录
err := os.MkdirAll(localDir, 0775)
if err != nil {
logger.Errorf("CopyDirRemoteToLocal failed to creating local directory %s: => %s", localDir, err.Error())
return err
}
// 列出远程目录中的文件和子目录
remoteFiles, err := s.Client.ReadDir(remoteDir)
if err != nil {
@@ -30,13 +37,6 @@ func (s *SSHClientSFTP) CopyDirRemoteToLocal(remoteDir, localDir string) error {
return err
}
// 创建本地目录
err = os.MkdirAll(localDir, 0775)
if err != nil {
logger.Errorf("CopyDirRemoteToLocal failed to creating local directory %s: => %s", localDir, err.Error())
return err
}
// 遍历远程文件和子目录并复制到本地
for _, remoteFile := range remoteFiles {
remotePath := filepath.ToSlash(filepath.Join(remoteDir, remoteFile.Name()))
@@ -62,56 +62,33 @@ func (s *SSHClientSFTP) CopyDirRemoteToLocal(remoteDir, localDir string) error {
// CopyDirRemoteToLocal 复制目录-本地到远程
func (s *SSHClientSFTP) CopyDirLocalToRemote(localDir, remoteDir string) error {
// 创建远程目录
err := s.Client.MkdirAll(remoteDir)
if err != nil {
logger.Errorf("CopyDirLocalToRemote failed to creating remote directory %s: => %s", remoteDir, err.Error())
return err
}
// 遍历本地目录中的文件和子目录并复制到远程
err = filepath.Walk(localDir, func(localPath string, info os.FileInfo, err error) error {
err := filepath.Walk(localDir, func(localPath string, info os.FileInfo, err error) error {
if err != nil {
return err
}
// 生成远程路径
remotePath := filepath.Join(remoteDir, localPath[len(localDir):])
remotePath := filepath.ToSlash(filepath.Join(remoteDir, localPath[len(localDir):]))
if info.IsDir() {
// 如果是子目录，则创建远程目录
err := s.Client.MkdirAll(remotePath)
if err != nil {
if err := s.Client.MkdirAll(remotePath); err != nil {
logger.Errorf("CopyDirLocalToRemote failed to creating remote directory %s: => %s", remotePath, err.Error())
return nil
return err
}
} else {
// 如果是文件，则复制文件内容
localFile, err := os.Open(localPath)
if err != nil {
logger.Errorf("CopyDirLocalToRemote failed to opening local file %s: => %s", localPath, err.Error())
return nil
}
defer localFile.Close()
remoteFile, err := s.Client.Create(remotePath)
if err != nil {
logger.Errorf("CopyDirLocalToRemote failed to creating remote file %s: => %s", remotePath, err.Error())
return nil
}
defer remoteFile.Close()
_, err = io.Copy(remoteFile, localFile)
if err != nil {
logger.Errorf("CopyDirLocalToRemote failed to copying file contents from %s to %s: => %s", localPath, remotePath, err.Error())
return nil
if err := s.CopyFileLocalToRemote(localPath, remotePath); err != nil {
logger.Errorf("CopyDirLocalToRemote failed to copying remote file %s: => %s", localPath, err.Error())
return err
}
}
return nil
})
if err != nil {
logger.Errorf("CopyDirLocalToRemote failed to walking local directory: => %s", err.Error())
logger.Errorf("CopyDirLocalToRemote failed to walking remote directory: => %s", err.Error())
return err
}
return nil
@@ -140,8 +117,7 @@ func (s *SSHClientSFTP) CopyFileRemoteToLocal(remotePath, localPath string) erro
defer localFile.Close()
// 复制文件内容
_, err = io.Copy(localFile, remoteFile)
if err != nil {
if _, err = io.Copy(localFile, remoteFile); err != nil {
logger.Errorf("CopyFileRemoteToLocal failed to copying contents: => %s", err.Error())
return err
}
@@ -158,6 +134,12 @@ func (s *SSHClientSFTP) CopyFileLocalToRemote(localPath, remotePath string) erro
}
defer localFile.Close()
// 创建远程目录
// if err := s.Client.MkdirAll(filepath.Dir(remotePath)); err != nil {
// logger.Errorf("CopyFileLocalToRemote failed to creating remote directory %s: => %s", remotePath, err.Error())
// return err
// }
// 创建远程文件
remoteFile, err := s.Client.Create(remotePath)
if err != nil {
@@ -167,8 +149,7 @@ func (s *SSHClientSFTP) CopyFileLocalToRemote(localPath, remotePath string) erro
defer remoteFile.Close()
// 复制文件内容
_, err = io.Copy(remoteFile, localFile)
if err != nil {
if _, err = io.Copy(remoteFile, localFile); err != nil {
logger.Errorf("CopyFileLocalToRemote failed to copying contents: => %s", err.Error())
return err
}

1
src/framework/utils/ssh/ssh.go
View File

@@ -100,6 +100,7 @@ func (c *ConnSSH) RunCMD(cmd string) (string, error) {
defer session.Close()
buf, err := session.CombinedOutput(cmd)
if err != nil {
logger.Infof("RunCMD failed run command: => %s", cmd)
logger.Errorf("RunCMD failed run command: => %s", err.Error())
}
c.LastResult = string(buf)

2
src/modules/crontask/processor/monitor_sys_resource/monitor_sys_resource.go
View File

@@ -1,4 +1,4 @@
package monitorsysresource
package monitor_sys_resource
import (
"encoding/json"

61
src/modules/crontask/processor/ne_config_backup/ne_config_backup.go Normal file
View File

@@ -0,0 +1,61 @@
package ne_config_backup
import (
"fmt"
"path/filepath"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
neModel "be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
)
var NewProcessor = &NeConfigBackupProcessor{
neConfigBackupService: neService.NewNeConfigBackupImpl,
neInfoService: neService.NewNeInfoImpl,
count: 0,
}
// NeConfigBackupProcessor 网元配置文件定期备份
type NeConfigBackupProcessor struct {
// 网元配置文件备份记录服务
neConfigBackupService neService.INeConfigBackup
// 网元信息服务
neInfoService neService.INeInfo
// 执行次数
count int
}
func (s *NeConfigBackupProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复 %v 任务ID %s", options.Repeat, sysJob.JobID)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
neList := s.neInfoService.SelectList(neModel.NeInfo{}, false, false)
for _, neInfo := range neList {
neTypeAndId := fmt.Sprintf("%s_%s", neInfo.NeType, neInfo.NeId)
// 将网元文件备份到本地
zipFilePath, err := s.neConfigBackupService.NeConfigNeToLocal(neInfo)
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
// 新增备份记录
item := neModel.NeConfigBackup{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
Name: filepath.Base(zipFilePath),
Path: zipFilePath,
CreateBy: "system",
}
s.neConfigBackupService.Insert(item)
result[neTypeAndId] = "ok"
}
return result, nil
}

7
src/modules/crontask/processor/processor.go
View File

@@ -7,13 +7,16 @@ import (
"be.ems/src/modules/crontask/processor/deleteExpiredRecord"
"be.ems/src/modules/crontask/processor/genNeStateAlarm"
"be.ems/src/modules/crontask/processor/getStateFromNE"
monitorsysresource "be.ems/src/modules/crontask/processor/monitor_sys_resource"
processorMonitorSysResource "be.ems/src/modules/crontask/processor/monitor_sys_resource"
processorNeConfigBackup "be.ems/src/modules/crontask/processor/ne_config_backup"
)
// InitCronQueue 初始定时任务队列
func InitCronQueue() {
// 监控-系统资源
cron.CreateQueue("monitor_sys_resource", monitorsysresource.NewProcessor)
cron.CreateQueue("monitor_sys_resource", processorMonitorSysResource.NewProcessor)
// 网元-网元配置文件定期备份
cron.CreateQueue("ne_config_backup", processorNeConfigBackup.NewProcessor)
// delete expired NE backup file
cron.CreateQueue("delExpiredNeBackup", delExpiredNeBackup.NewProcessor)
cron.CreateQueue("deleteExpiredRecord", deleteExpiredRecord.NewProcessor)

13
src/modules/network_data/controller/all_kpi.go
View File

@@ -3,7 +3,6 @@ package controller
import (
"be.ems/src/framework/i18n"
"be.ems/src/framework/utils/ctx"
"be.ems/src/framework/utils/date"
"be.ems/src/framework/vo/result"
"be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
@@ -37,18 +36,6 @@ func (s *PerfKPIController) GoldKPI(c *gin.Context) {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
// 时间格式校验
startTime := date.ParseStrToDate(querys.StartTime, date.YYYY_MM_DD_HH_MM_SS)
if startTime.IsZero() {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
endTime := date.ParseStrToDate(querys.EndTime, date.YYYY_MM_DD_HH_MM_SS)
if endTime.IsZero() {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
if querys.Interval < 5 || querys.Interval > 3600 {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return

43
src/modules/network_data/controller/udm_auth.go
View File

@@ -15,6 +15,7 @@ import (
"be.ems/src/framework/vo/result"
"be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
neFetchlink "be.ems/src/modules/network_element/fetch_link"
neService "be.ems/src/modules/network_element/service"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
@@ -411,7 +412,11 @@ func (s *UDMAuthController) Export(c *gin.Context) {
data := [][]string{}
data = append(data, []string{"imsi", "ki", "algo", "amf", "opc"})
for _, v := range list {
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, v.Opc})
opc := v.Opc
if opc == "-" {
opc = ""
}
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, opc})
}
// 输出到文件
err := file.WriterFileCSV(data, filePath)
@@ -425,7 +430,11 @@ func (s *UDMAuthController) Export(c *gin.Context) {
// 转换数据
data := [][]string{}
for _, v := range list {
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, v.Opc})
opc := v.Opc
if opc == "-" {
opc = ""
}
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, opc})
}
// 输出到文件
err = file.WriterFileTXT(data, ",", filePath)
@@ -446,6 +455,8 @@ func (s *UDMAuthController) Import(c *gin.Context) {
var body struct {
NeId string `json:"neId" binding:"required"`
UploadPath string `json:"uploadPath" binding:"required"`
TypeVal string `json:"typeVal" binding:"required,oneof=default k4"`
TypeData any `json:"typeData"`
}
if err := c.ShouldBindBodyWith(&body, binding.JSON); err != nil {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
@@ -497,16 +508,30 @@ func (s *UDMAuthController) Import(c *gin.Context) {
}
defer telnetClient.Close()
// 发送MML
cmd := fmt.Sprintf("import authdat:path=%s", neFilePath)
data, err := telnet.ConvertToStr(telnetClient, cmd)
if err != nil {
c.JSON(200, result.ErrMsg(err.Error()))
// 结果信息
var resultMsg string
var resultErr error
// 默认的情况 发送MML
if body.TypeVal == "default" {
cmd := fmt.Sprintf("import authdat:path=%s", neFilePath)
resultMsg, resultErr = telnet.ConvertToStr(telnetClient, cmd)
}
// K4类型发特定请求
if body.TypeVal == "k4" {
resultMsg, resultErr = neFetchlink.UDMImportAuth(neInfo.IP, map[string]any{
"path": neFilePath, "k4": body.TypeData,
})
}
if resultErr != nil {
c.JSON(200, result.ErrMsg(resultErr.Error()))
return
}
// 命令ok时
if strings.Contains(data, "ok") {
if strings.Contains(resultMsg, "ok") {
if strings.HasSuffix(body.UploadPath, ".csv") {
data := file.ReadFileCSV(localFilePath)
neId := ""
@@ -518,5 +543,5 @@ func (s *UDMAuthController) Import(c *gin.Context) {
go s.udmAuthService.InsertData(neId, "txt", data)
}
}
c.JSON(200, result.OkMsg(data))
c.JSON(200, result.OkMsg(resultMsg))
}

9
src/modules/network_data/model/cdr_event_smf.go
View File

@@ -13,15 +13,6 @@ type CDREventSMF struct {
CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;default:CURRENT_TIMESTAMP"`
// ====== 非数据库字段属性 ======
// RecordType string `json:"recordType" gorm:"column:record_type"`
// ChargingID string `json:"chargingID" gorm:"column:charging_id"`
// SubscriberID string `json:"subscriberID" gorm:"column:subscriber_id"`
// Duration string `json:"duration" gorm:"column:duration"`
// DataVolumeUplink string `json:"dataVolumeUplink" gorm:"column:data_volume_uplink"`
// DataVolumeDownlink string `json:"dataVolumeDownlink" gorm:"column:data_volume_downlink"`
// DataTotalVolume string `json:"dataTotalVolume" gorm:"column:data_total_volume"`
// PDUAddress string `json:"pduAddress" gorm:"column:pdu_address"`
}
// CDREventSMFQuery CDR会话对象SMF查询参数结构体

2
src/modules/network_data/model/perf_kpi.go
View File

@@ -16,7 +16,7 @@ type GoldKPIQuery struct {
NeID string `form:"neId" binding:"required"`
StartTime string `form:"startTime" binding:"required"`
EndTime string `form:"endTime" binding:"required"`
Interval int64 `form:"interval" binding:"required"`
Interval int64 `form:"interval" binding:"required,oneof=5 60 300 900 1800 3600"`
RmUID string `form:"rmUID"`
SortField string `form:"sortField" binding:"omitempty,oneof=timeGroup"`
SortOrder string `form:"sortOrder" binding:"omitempty,oneof=asc desc"`

15
src/modules/network_data/repository/cdr_event_ims.impl.go
View File

@@ -84,13 +84,24 @@ func (r *CDREventIMSImpl) SelectPage(querys model.CDREventIMSQuery) map[string]a
conditions = append(conditions, "JSON_EXTRACT(cdr_json, '$.calledParty') = ?")
params = append(params, querys.CalledParty)
}
// MySQL8支持的
// if querys.RecordType != "" {
// recordTypes := strings.Split(querys.RecordType, ",")
// placeholder := repo.KeyPlaceholderByQuery(len(recordTypes))
// conditions = append(conditions, fmt.Sprintf("JSON_EXTRACT(cdr_json, '$.recordType') in (%s)", placeholder))
// for _, recordType := range recordTypes {
// params = append(params, recordType)
// }
// }
// Mariadb不支持json in查询改or
if querys.RecordType != "" {
recordTypes := strings.Split(querys.RecordType, ",")
placeholder := repo.KeyPlaceholderByQuery(len(recordTypes))
conditions = append(conditions, fmt.Sprintf("JSON_EXTRACT(cdr_json, '$.recordType') in (%s)", placeholder))
var queryStrArr []string
for _, recordType := range recordTypes {
queryStrArr = append(queryStrArr, "JSON_EXTRACT(cdr_json, '$.recordType') = ?")
params = append(params, recordType)
}
conditions = append(conditions, fmt.Sprintf("( %s )", strings.Join(queryStrArr, " OR ")))
}
// 构建查询条件语句

15
src/modules/network_data/repository/cdr_event_smf.impl.go
View File

@@ -14,7 +14,6 @@ import (
// 实例化数据层 CDREventSMFImpl 结构体
var NewCDREventSMFImpl = &CDREventSMFImpl{
selectSql: `select id, ne_type, ne_name, rm_uid, timestamp, cdr_json, created_at from cdr_event_smf`,
// selectSql: `select id, ne_type, ne_name, rm_uid, timestamp, JSON_EXTRACT(cdr_json, '$.recordType') AS record_type, JSON_EXTRACT(cdr_json, '$.chargingID') AS charging_id, JSON_EXTRACT(cdr_json, '$.subscriberIdentifier.subscriptionIDData') AS subscriber_id, JSON_EXTRACT(cdr_json, '$.duration') AS duration, JSON_EXTRACT(cdr_json, '$.listOfMultipleUnitUsage[*].usedUnitContainer[*].dataVolumeUplink') AS data_volume_uplink, JSON_EXTRACT(cdr_json, '$.listOfMultipleUnitUsage[*].usedUnitContainer[*].dataVolumeDownlink') AS data_volume_downlink, JSON_EXTRACT(cdr_json, '$.listOfMultipleUnitUsage[*].usedUnitContainer[*].dataTotalVolume') AS data_total_volume, JSON_EXTRACT(cdr_json, '$.pDUSessionChargingInformation.pDUAddress') AS pdu_address, created_at from cdr_event_smf`,
resultMap: map[string]string{
"id": "ID",
@@ -24,20 +23,6 @@ var NewCDREventSMFImpl = &CDREventSMFImpl{
"timestamp": "Timestamp",
"cdr_json": "CDRJSONStr",
"created_at": "CreatedAt",
// "id": "ID",
// "ne_type": "NeType",
// "ne_name": "NeName",
// "rm_uid": "RmUID",
// "timestamp": "Timestamp",
// "record_type": "RecordType",
// "charging_id": "ChargingID",
// "subscriber_id": "SubscriberID",
// "duration": "Duration",
// "data_volume_uplink": "DataVolumeUplink",
// "data_volume_downlink": "DataVolumeDownlink",
// "data_total_volume": "DataTotalVolume",
// "pdu_address": "PDUAddress",
// "created_at": "CreatedAt",
},
}

6
src/modules/network_data/repository/perf_kpi.go
View File

@@ -7,15 +7,9 @@ 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
}

170
src/modules/network_data/repository/perf_kpi.impl.go
View File

@@ -17,76 +17,6 @@ type PerfKPIImpl struct{}
// SelectGoldKPI 通过网元指标数据信息
func (r *PerfKPIImpl) SelectGoldKPI(query model.GoldKPIQuery, kpiIds []string) []map[string]any {
// 查询条件拼接
var conditions []string
var params []any
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)
}
if query.StartTime != "" {
conditions = append(conditions, "gk.start_time >= ?")
params = append(params, query.StartTime)
}
if query.EndTime != "" {
conditions = append(conditions, "gk.start_time <= ?")
params = append(params, query.EndTime)
}
// 构建查询条件语句
whereSql := ""
if len(conditions) > 0 {
whereSql += " where " + strings.Join(conditions, " and ")
}
// 查询字段列
timeFormat := "DATE_FORMAT(gk.start_time, '%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(gk.start_time, '%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 _, 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 gk.kpi_id = '%s' and gk.VALUE != 0 THEN gk.VALUE END ), ',', 1), 0) AS '%s'", kid, kid)
fields = append(fields, str)
} else {
str := fmt.Sprintf("sum(CASE WHEN gk.kpi_id = '%s' THEN gk.value ELSE 0 END) AS '%s'", kid, 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 gold_kpi gk %s GROUP BY timeGroup %s", fieldsSql, whereSql, orderSql)
results, err := datasource.RawDB("", querySql, params)
if err != nil {
logger.Errorf("query err => %v", err)
}
return results
}
func (r *PerfKPIImpl) SelectKpiReport(query model.GoldKPIQuery, kpiIds []string) []map[string]any {
// 查询条件拼接
var conditions []string
var params []any
@@ -100,43 +30,15 @@ func (r *PerfKPIImpl) SelectKpiReport(query model.GoldKPIQuery, kpiIds []string)
// params = append(params, query.NeType)
tableName += strings.ToLower(query.NeType)
}
var dateStr1, dateStr2, timeStr1, timeStr2 string
if query.StartTime != "" {
dateStr1 = query.StartTime[:10]
timeStr1 = query.StartTime[11:]
conditions = append(conditions, "gk.created_at >= ?")
params = append(params, query.StartTime)
}
if query.EndTime != "" {
dateStr2 = query.EndTime[:10]
timeStr2 = query.EndTime[11:]
}
if dateStr1 == dateStr2 && dateStr1 != "" {
conditions = append(conditions, "gk.`date` = ?")
params = append(params, dateStr1)
conditions = append(conditions, "gk.`start_time` >= ?")
params = append(params, timeStr1)
conditions = append(conditions, "gk.`start_time` <= ?")
params = append(params, timeStr2)
} else {
if dateStr1 != "" {
conditions = append(conditions, "(gk.`date` > ? OR (gk.`date` = ? AND gk.`start_time` >= ?))")
params = append(params, dateStr1, dateStr1, timeStr1)
}
if dateStr2 != "" {
conditions = append(conditions, "(gk.`date` < ? OR (gk.`date` = ? AND gk.`start_time` <= ?))")
params = append(params, dateStr2, dateStr2, timeStr2)
}
conditions = append(conditions, "gk.created_at <= ?")
params = append(params, query.EndTime)
}
// 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 {
@@ -144,18 +46,9 @@ func (r *PerfKPIImpl) SelectKpiReport(query model.GoldKPIQuery, kpiIds []string)
}
// 查询字段列
var dateTimeStr string = "CONCAT(gk.`date`, \" \", gk.start_time)"
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,
// fmt.Sprintf("FROM_UNIXTIME(FLOOR(gk.created_at / (%d * 1000)) * %d) AS timeGroup", query.Interval, query.Interval),
fmt.Sprintf("CONCAT(FLOOR(gk.created_at / (%d * 1000)) * (%d * 1000)) AS timeGroup", query.Interval, query.Interval), // 时间戳毫秒
"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",
@@ -204,19 +97,19 @@ func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID, startDate, endDate strin
var conditions []string
var params []any
if neType != "" {
conditions = append(conditions, "gk.ne_type = ?")
conditions = append(conditions, "kupf.ne_type = ?")
params = append(params, neType)
}
if rmUID != "" {
conditions = append(conditions, "gk.rm_uid = ?")
conditions = append(conditions, "kupf.rm_uid = ?")
params = append(params, rmUID)
}
if startDate != "" {
conditions = append(conditions, "gk.date >= ?")
conditions = append(conditions, "kupf.created_at >= ?")
params = append(params, startDate)
}
if endDate != "" {
conditions = append(conditions, "gk.date <= ?")
conditions = append(conditions, "kupf.created_at <= ?")
params = append(params, endDate)
}
// 构建查询条件语句
@@ -226,44 +119,11 @@ func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID, startDate, endDate strin
}
// 查询数据
querySql := fmt.Sprintf("SELECT sum( CASE WHEN gk.kpi_id = 'UPF.03' THEN gk.VALUE ELSE 0 END ) AS 'up', sum( CASE WHEN gk.kpi_id = 'UPF.06' THEN gk.VALUE ELSE 0 END ) AS 'down' FROM gold_kpi gk %s", whereSql)
results, err := datasource.RawDB("", querySql, params)
if err != nil {
logger.Errorf("query err => %v", err)
}
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)
querySql := `SELECT
sum( CASE WHEN JSON_EXTRACT(kupf.kpi_values, '$[2].kpi_id') = 'UPF.03' THEN JSON_EXTRACT(kupf.kpi_values, '$[2].value') ELSE 0 END ) AS 'up',
sum( CASE WHEN JSON_EXTRACT(kupf.kpi_values, '$[5].kpi_id') = 'UPF.06' THEN JSON_EXTRACT(kupf.kpi_values, '$[5].value') ELSE 0 END ) AS 'down'
FROM kpi_report_upf kupf`
results, err := datasource.RawDB("", querySql+whereSql, params)
if err != nil {
logger.Errorf("query err => %v", err)
}

22
src/modules/network_data/service/perf_kpi.impl.go
View File

@@ -31,8 +31,7 @@ func (r *PerfKPIImpl) SelectGoldKPI(query model.GoldKPIQuery) []map[string]any {
kpiIds = append(kpiIds, kpiId.KPIID)
}
//data := r.perfKPIRepository.SelectGoldKPI(query, kpiIds)
data := r.perfKPIRepository.SelectKpiReport(query, kpiIds)
data := r.perfKPIRepository.SelectGoldKPI(query, kpiIds)
if data == nil {
return []map[string]any{}
}
@@ -46,12 +45,11 @@ func (r *PerfKPIImpl) SelectGoldKPITitle(neType string) []model.GoldKPITitle {
// SelectUPFTotalFlow 查询UPF总流量 N3上行 N6下行
func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID string, day int) map[string]any {
// 获取当前日期
now := time.Now()
endDate := now.Format("2006-01-02")
// 获取当前日期
endDate := fmt.Sprint(now.UnixMilli())
// 将当前日期前几天数
afterDays := now.AddDate(0, 0, -day)
startDate := afterDays.Format("2006-01-02")
startDate := fmt.Sprint(now.AddDate(0, 0, -day).Truncate(24 * time.Hour).UnixMilli())
var info map[string]any
@@ -61,14 +59,18 @@ func (r *PerfKPIImpl) SelectUPFTotalFlow(neType, rmUID string, day int) map[stri
if infoStr != "" {
json.Unmarshal([]byte(infoStr), &info)
expireSecond, _ := redis.GetExpire("", key)
expireMinute := (time.Duration(int64(expireSecond)) * time.Second)
if expireMinute > 2*time.Minute {
if expireSecond > 120 {
return info
}
}
//info = r.perfKPIRepository.SelectUPFTotalFlow(neType, rmUID, startDate, endDate)
info = r.perfKPIRepository.SelectUPFThroughput(neType, rmUID, startDate, endDate)
info = r.perfKPIRepository.SelectUPFTotalFlow(neType, rmUID, startDate, endDate)
if v, ok := info["up"]; ok && v == nil {
info["up"] = 0
}
if v, ok := info["down"]; ok && v == nil {
info["down"] = 0
}
// 保存到缓存
infoJSON, _ := json.Marshal(info)

54
src/modules/network_data/udm_k4_test.go Normal file
View File

@@ -0,0 +1,54 @@
package networkdata
import (
"crypto/des"
"errors"
"testing"
)
// 加密
func encrypt(origData, key []byte) ([]byte, error) {
if len(origData) < 1 || len(key) < 1 {
return nil, errors.New("wrong data or key")
}
block, err := des.NewCipher(key)
if err != nil {
return nil, err
}
bs := block.BlockSize()
if len(origData)%bs != 0 {
return nil, errors.New("wrong padding")
}
out := make([]byte, len(origData))
dst := out
for len(origData) > 0 {
block.Encrypt(dst, origData[:bs])
origData = origData[bs:]
dst = dst[bs:]
}
return out, nil
}
func TestEncrypt(t *testing.T) {
// key := []byte{0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef}
// 0123456789abcdef
// ki := []byte{0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef}
// 0123456789abcdef0123456789abcdef
// 密码
key := []byte{0x12, 0x34, 0x12, 0x34, 0x12, 0x34, 0x12, 0x34}
// 1234123412341234
// 要加密的ki
ki := []byte{0x80, 0x5D, 0xAD, 0xC6, 0xE8, 0xA5, 0x4A, 0x0D, 0x59, 0xD6, 0x22, 0xC7, 0xA0, 0x4D, 0x08, 0xE0}
// 805DADC6E8A54A0D59D622C7A04D08E0
kis, err := encrypt(ki, key)
// 加密后的，放导入导入文件里ki
t.Errorf("kis: %x\n", kis)
// 3e479135bb16f45dc874a18831b54d71
t.Errorf("err: %v\n", err)
}

204
src/modules/network_element/controller/ne_config_backup.go Normal file
View File

@@ -0,0 +1,204 @@
package controller
import (
"os"
"path/filepath"
"strings"
"be.ems/src/framework/i18n"
"be.ems/src/framework/utils/ctx"
"be.ems/src/framework/utils/file"
"be.ems/src/framework/utils/parse"
"be.ems/src/framework/vo/result"
"be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
)
// NewNeConfigBackup 实例化控制层 NeConfigBackupController 结构体
var NewNeConfigBackup = &NeConfigBackupController{
neConfigBackupService: neService.NewNeConfigBackupImpl,
neInfoService: neService.NewNeInfoImpl,
}
// 网元配置文件备份记录
//
// PATH /config/backup
type NeConfigBackupController struct {
// 网元配置文件备份记录服务
neConfigBackupService neService.INeConfigBackup
// 网元信息服务
neInfoService neService.INeInfo
}
// 网元配置文件备份记录列表
//
// GET /list
func (s *NeConfigBackupController) List(c *gin.Context) {
querys := ctx.QueryMap(c)
data := s.neConfigBackupService.SelectPage(querys)
c.JSON(200, result.Ok(data))
}
// 网元配置文件备份记录信息
//
// GET /download?id=xx
func (s *NeConfigBackupController) Download(c *gin.Context) {
language := ctx.AcceptLanguage(c)
id, ok := c.GetQuery("id")
if !ok || id == "" {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
item := s.neConfigBackupService.SelectById(id)
if item.ID != id {
// 没有可访问主机命令数据！
c.JSON(200, result.ErrMsg(i18n.TKey(language, "neConfigBackup.noData")))
return
}
if _, err := os.Stat(item.Path); err != nil {
c.JSON(200, result.ErrMsg(i18n.TKey(language, "neConfigBackup.notFoundFile")))
return
}
c.FileAttachment(item.Path, item.Name)
}
// 网元配置文件备份记录修改
//
// PUT /
func (s *NeConfigBackupController) Edit(c *gin.Context) {
language := ctx.AcceptLanguage(c)
var body struct {
ID string `json:"id" binding:"required"` // 记录ID
Name string `json:"name" binding:"required"` // 名称
Remark string `json:"remark" binding:"required"` // 备注
}
err := c.ShouldBindBodyWith(&body, binding.JSON)
if err != nil || body.ID == "" {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
// 检查是否存在
data := s.neConfigBackupService.SelectById(body.ID)
if data.ID != body.ID {
// 没有可访问主机命令数据！
c.JSON(200, result.ErrMsg(i18n.TKey(language, "neConfig.noData")))
return
}
data.Name = body.Name
data.Remark = body.Remark
data.UpdateBy = ctx.LoginUserToUserName(c)
rows := s.neConfigBackupService.Update(data)
if rows > 0 {
c.JSON(200, result.Ok(nil))
return
}
c.JSON(200, result.Err(nil))
}
// 网元配置文件备份记录删除
//
// DELETE /?id=xx
func (s *NeConfigBackupController) Remove(c *gin.Context) {
language := ctx.AcceptLanguage(c)
id, ok := c.GetQuery("id")
if !ok || id == "" {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
// 处理字符转id数组后去重
ids := strings.Split(id, ",")
uniqueIDs := parse.RemoveDuplicates(ids)
if len(uniqueIDs) <= 0 {
c.JSON(200, result.Err(nil))
return
}
rows, err := s.neConfigBackupService.DeleteByIds(uniqueIDs)
if err != nil {
c.JSON(200, result.ErrMsg(i18n.TKey(language, err.Error())))
return
}
msg := i18n.TTemplate(language, "app.common.deleteSuccess", map[string]any{"num": rows})
c.JSON(200, result.OkMsg(msg))
}
// 网元配置文件备份导入
//
// POST /import
func (s *NeConfigBackupController) Import(c *gin.Context) {
language := ctx.AcceptLanguage(c)
var body struct {
NeType string `json:"neType" binding:"required"`
NeId string `json:"neId" binding:"required"`
Type string `json:"type" binding:"required,oneof=backup upload"` // 导入类型
Path string `json:"path" binding:"required"` // 文件路径
}
if err := c.ShouldBindBodyWith(&body, binding.JSON); err != nil {
c.JSON(200, result.ErrMsg(i18n.TKey(language, "app.common.noNEInfo")))
return
}
if !strings.HasSuffix(body.Path, ".zip") {
c.JSON(200, result.ErrMsg("Only supports decompression of zip files"))
return
}
// 查网元
neInfo := s.neInfoService.SelectNeInfoByNeTypeAndNeID(body.NeType, body.NeId)
if neInfo.NeId != body.NeId || neInfo.IP == "" {
c.JSON(200, result.ErrMsg(i18n.TKey(language, "app.common.noNEInfo")))
return
}
// 将zip文件解压到本地后复制到网元端
localFilePath := body.Path
if body.Type == "upload" {
localFilePath = file.ParseUploadFilePath(body.Path)
}
if err := s.neConfigBackupService.NeConfigLocalToNe(neInfo, localFilePath); err != nil {
c.JSON(200, result.ErrMsg(err.Error()))
return
}
c.JSON(200, result.Ok(nil))
}
// 网元配置文件备份导出
//
// POST /export
func (s *NeConfigBackupController) Export(c *gin.Context) {
language := ctx.AcceptLanguage(c)
var body struct {
NeType string `json:"neType" binding:"required"`
NeId string `json:"neId" binding:"required"`
}
if err := c.ShouldBindBodyWith(&body, binding.JSON); err != nil {
c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
return
}
// 查网元
neInfo := s.neInfoService.SelectNeInfoByNeTypeAndNeID(body.NeType, body.NeId)
if neInfo.NeId != body.NeId || neInfo.IP == "" {
c.JSON(200, result.ErrMsg(i18n.TKey(language, "app.common.noNEInfo")))
return
}
// 将网元文件备份到本地
zipFilePath, err := s.neConfigBackupService.NeConfigNeToLocal(neInfo)
if err != nil {
c.JSON(200, result.ErrMsg(err.Error()))
return
}
// 新增备份记录
item := model.NeConfigBackup{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
Name: filepath.Base(zipFilePath),
Path: zipFilePath,
CreateBy: ctx.LoginUserToUserName(c),
}
s.neConfigBackupService.Insert(item)
c.FileAttachment(item.Path, item.Name)
}

3
src/modules/network_element/fetch_link/udm.go
View File

@@ -3,6 +3,7 @@ package fetchlink
import (
"encoding/json"
"fmt"
"strings"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/fetch"
@@ -30,7 +31,7 @@ func UDMImportAuth(udmIP string, data map[string]any) (string, error) {
return "", err
}
if v, ok := resData["code"]; ok && v == "00000" {
return "ok", nil
return strings.TrimSpace(strings.ToLower(resData["message"])), nil
}
return "", fmt.Errorf(resData["message"])
}

23
src/modules/network_element/model/ne_config_backup.go Normal file
View File

@@ -0,0 +1,23 @@
package model
// NeConfigBackup 网元配置文件备份记录 ne_config_backup
type NeConfigBackup struct {
ID string `json:"id" gorm:"column:id;primaryKey;autoIncrement"`
NeType string `json:"neType" gorm:"ne_type"` // 网元类型
NeId string `json:"neId" gorm:"ne_id"` // 网元ID
Name string `json:"name" gorm:"name"` // 压缩包名称
Path string `json:"path" gorm:"path"` // 压缩包位置
Remark string `json:"remark" gorm:"remark"` // 备注
CreateBy string `json:"createBy" gorm:"create_by"` // 创建者
CreateTime int64 `json:"createTime" gorm:"create_time"` // 创建时间
UpdateBy string `json:"updateBy" gorm:"update_by"` // 更新者
UpdateTime int64 `json:"updateTime" gorm:"update_time"` // 更新时间
// ====== 非数据库字段属性 ======
}
// TableName 表名称
func (*NeConfigBackup) TableName() string {
return "ne_config_backup"
}

33
src/modules/network_element/network_element.go
View File

@@ -309,6 +309,39 @@ func Setup(router *gin.Engine) {
controller.NewNeConfig.DataRemove,
)
}
// 网元配置文件备份记录
neConfigBackupGroup := neGroup.Group("/config/backup")
{
neConfigBackupGroup.GET("/list",
middleware.PreAuthorize(nil),
controller.NewNeConfigBackup.List,
)
neConfigBackupGroup.GET("/download",
middleware.PreAuthorize(nil),
controller.NewNeConfigBackup.Download,
)
neConfigBackupGroup.PUT("",
middleware.PreAuthorize(nil),
collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.neConfigBackup", collectlogs.BUSINESS_TYPE_UPDATE)),
controller.NewNeConfigBackup.Edit,
)
neConfigBackupGroup.DELETE("",
middleware.PreAuthorize(nil),
collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.neConfigBackup", collectlogs.BUSINESS_TYPE_DELETE)),
controller.NewNeConfigBackup.Remove,
)
neConfigBackupGroup.POST("/import",
middleware.PreAuthorize(nil),
collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.neConfigBackup", collectlogs.BUSINESS_TYPE_IMPORT)),
controller.NewNeConfigBackup.Import,
)
neConfigBackupGroup.POST("/export",
middleware.PreAuthorize(nil),
collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.neConfigBackup", collectlogs.BUSINESS_TYPE_EXPORT)),
controller.NewNeConfigBackup.Export,
)
}
}
// InitLoad 初始参数

24
src/modules/network_element/repository/ne_config_backup.go Normal file
View File

@@ -0,0 +1,24 @@
package repository
import "be.ems/src/modules/network_element/model"
// INeConfigBackup 网元配置文件备份记录 数据层接口
type INeConfigBackup interface {
// SelectPage 根据条件分页查询字典类型
SelectPage(query map[string]any) map[string]any
// SelectList 根据实体查询
SelectList(item model.NeConfigBackup) []model.NeConfigBackup
// SelectByIds 通过ID查询
SelectByIds(ids []string) []model.NeConfigBackup
// Insert 新增信息
Insert(item model.NeConfigBackup) string
// Update 修改信息
Update(item model.NeConfigBackup) int64
// DeleteByIds 批量删除信息
DeleteByIds(ids []string) int64
}

262
src/modules/network_element/repository/ne_config_backup.impl.go Normal file
View File

@@ -0,0 +1,262 @@
package repository
import (
"strings"
"time"
"be.ems/src/framework/datasource"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/parse"
"be.ems/src/framework/utils/repo"
"be.ems/src/modules/network_element/model"
)
// 实例化数据层 NewNeConfigBackupImpl 结构体
var NewNeConfigBackupImpl = &NeConfigBackupImpl{
selectSql: `select
id, ne_type, ne_id, name, path, remark, create_by, create_time, update_by, update_time
from ne_config_backup`,
resultMap: map[string]string{
"id": "ID",
"ne_type": "NeType",
"ne_id": "NeId",
"name": "Name",
"path": "Path",
"remark": "Remark",
"create_by": "CreateBy",
"create_time": "CreateTime",
"update_by": "UpdateBy",
"update_time": "UpdateTime",
},
}
// NeConfigBackupImpl 网元配置文件备份记录 数据层处理
type NeConfigBackupImpl struct {
// 查询视图对象SQL
selectSql string
// 结果字段与实体映射
resultMap map[string]string
}
// convertResultRows 将结果记录转实体结果组
func (r *NeConfigBackupImpl) convertResultRows(rows []map[string]any) []model.NeConfigBackup {
arr := make([]model.NeConfigBackup, 0)
for _, row := range rows {
item := model.NeConfigBackup{}
for key, value := range row {
if keyMapper, ok := r.resultMap[key]; ok {
repo.SetFieldValue(&item, keyMapper, value)
}
}
arr = append(arr, item)
}
return arr
}
// SelectPage 根据条件分页查询字典类型
func (r *NeConfigBackupImpl) SelectPage(query map[string]any) map[string]any {
// 查询条件拼接
var conditions []string
var params []any
if v, ok := query["neType"]; ok && v != "" {
conditions = append(conditions, "ne_type = ?")
params = append(params, strings.Trim(v.(string), " "))
}
if v, ok := query["neId"]; ok && v != "" {
conditions = append(conditions, "ne_id = ?")
params = append(params, strings.Trim(v.(string), " "))
}
if v, ok := query["name"]; ok && v != "" {
conditions = append(conditions, "name like concat(concat('%', ?), '%')")
params = append(params, strings.Trim(v.(string), " "))
}
// 构建查询条件语句
whereSql := ""
if len(conditions) > 0 {
whereSql += " where " + strings.Join(conditions, " and ")
}
result := map[string]any{
"total": 0,
"rows": []model.NeHost{},
}
// 查询数量 长度为0直接返回
totalSql := "select count(1) as 'total' from ne_config_backup"
totalRows, err := datasource.RawDB("", totalSql+whereSql, params)
if err != nil {
logger.Errorf("total err => %v", err)
return result
}
total := parse.Number(totalRows[0]["total"])
if total == 0 {
return result
} else {
result["total"] = total
}
// 分页
pageNum, pageSize := repo.PageNumSize(query["pageNum"], query["pageSize"])
pageSql := " order by id desc limit ?,? "
params = append(params, pageNum*pageSize)
params = append(params, pageSize)
// 查询数据
querySql := r.selectSql + whereSql + pageSql
results, err := datasource.RawDB("", querySql, params)
if err != nil {
logger.Errorf("query err => %v", err)
return result
}
// 转换实体
result["rows"] = r.convertResultRows(results)
return result
}
// SelectList 根据实体查询
func (r *NeConfigBackupImpl) SelectList(item model.NeConfigBackup) []model.NeConfigBackup {
// 查询条件拼接
var conditions []string
var params []any
if item.NeType != "" {
conditions = append(conditions, "ne_type = ?")
params = append(params, item.NeType)
}
if item.NeId != "" {
conditions = append(conditions, "ne_id = ?")
params = append(params, item.NeId)
}
// 构建查询条件语句
whereSql := ""
if len(conditions) > 0 {
whereSql += " where " + strings.Join(conditions, " and ")
}
// 查询数据
querySql := r.selectSql + whereSql + " order by id desc "
results, err := datasource.RawDB("", querySql, params)
if err != nil {
logger.Errorf("query err => %v", err)
}
// 转换实体
return r.convertResultRows(results)
}
// SelectByIds 通过ID查询
func (r *NeConfigBackupImpl) SelectByIds(cmdIds []string) []model.NeConfigBackup {
placeholder := repo.KeyPlaceholderByQuery(len(cmdIds))
querySql := r.selectSql + " where id in (" + placeholder + ")"
parameters := repo.ConvertIdsSlice(cmdIds)
results, err := datasource.RawDB("", querySql, parameters)
if err != nil {
logger.Errorf("query err => %v", err)
return []model.NeConfigBackup{}
}
// 转换实体
return r.convertResultRows(results)
}
// Insert 新增信息
func (r *NeConfigBackupImpl) Insert(item model.NeConfigBackup) string {
// 参数拼接
params := make(map[string]any)
if item.NeType != "" {
params["ne_type"] = item.NeType
}
if item.NeId != "" {
params["ne_id"] = item.NeId
}
if item.Name != "" {
params["name"] = item.Name
}
if item.Path != "" {
params["path"] = item.Path
}
if item.Remark != "" {
params["remark"] = item.Remark
}
if item.CreateBy != "" {
params["create_by"] = item.CreateBy
params["create_time"] = time.Now().UnixMilli()
}
// 构建执行语句
keys, placeholder, values := repo.KeyPlaceholderValueByInsert(params)
sql := "insert into ne_config_backup (" + strings.Join(keys, ",") + ")values(" + placeholder + ")"
db := datasource.DefaultDB()
// 开启事务
tx := db.Begin()
// 执行插入
err := tx.Exec(sql, values...).Error
if err != nil {
logger.Errorf("insert row : %v", err.Error())
tx.Rollback()
return ""
}
// 获取生成的自增 ID
var insertedID string
err = tx.Raw("select last_insert_id()").Row().Scan(&insertedID)
if err != nil {
logger.Errorf("insert last id : %v", err.Error())
tx.Rollback()
return ""
}
// 提交事务
tx.Commit()
return insertedID
}
// Update 修改信息
func (r *NeConfigBackupImpl) Update(item model.NeConfigBackup) int64 {
// 参数拼接
params := make(map[string]any)
if item.NeType != "" {
params["ne_type"] = item.NeType
}
if item.NeId != "" {
params["ne_id"] = item.NeId
}
if item.Name != "" {
params["name"] = item.Name
}
if item.Path != "" {
params["path"] = item.Path
}
params["remark"] = item.Remark
if item.UpdateBy != "" {
params["update_by"] = item.UpdateBy
params["update_time"] = time.Now().UnixMilli()
}
// 构建执行语句
keys, values := repo.KeyValueByUpdate(params)
sql := "update ne_config_backup set " + strings.Join(keys, ",") + " where id = ?"
// 执行更新
values = append(values, item.ID)
rows, err := datasource.ExecDB("", sql, values)
if err != nil {
logger.Errorf("update row : %v", err.Error())
return 0
}
return rows
}
// DeleteByIds 批量删除信息
func (r *NeConfigBackupImpl) DeleteByIds(ids []string) int64 {
placeholder := repo.KeyPlaceholderByQuery(len(ids))
sql := "delete from ne_config_backup where id in (" + placeholder + ")"
parameters := repo.ConvertIdsSlice(ids)
results, err := datasource.ExecDB("", sql, parameters)
if err != nil {
logger.Errorf("delete err => %v", err)
return 0
}
return results
}

30
src/modules/network_element/service/ne_config_backup.go Normal file
View File

@@ -0,0 +1,30 @@
package service
import "be.ems/src/modules/network_element/model"
// INeConfigBackup 网元配置文件备份记录 服务层接口
type INeConfigBackup interface {
// SelectNeHostPage 分页查询列表数据
SelectPage(query map[string]any) map[string]any
// SelectList 根据实体查询
SelectList(item model.NeConfigBackup) []model.NeConfigBackup
// SelectByIds 通过ID查询
SelectById(id string) model.NeConfigBackup
// Insert 新增信息
Insert(item model.NeConfigBackup) string
// Update 修改信息
Update(item model.NeConfigBackup) int64
// DeleteByIds 批量删除信息
DeleteByIds(ids []string) (int64, error)
// NeConfigLocalToNe 网元配置文件复制到网元端覆盖
NeConfigLocalToNe(neInfo model.NeInfo, localFile string) error
// NeConfigNeToLocal 网元备份文件网元端复制到本地
NeConfigNeToLocal(neInfo model.NeInfo) (string, error)
}

198
src/modules/network_element/service/ne_config_backup.impl.go Normal file
View File

@@ -0,0 +1,198 @@
package service
import (
"fmt"
"os"
"runtime"
"strings"
"time"
"be.ems/src/framework/utils/date"
"be.ems/src/framework/utils/file"
"be.ems/src/modules/network_element/model"
"be.ems/src/modules/network_element/repository"
)
// NewNeConfigBackupImpl 网元配置文件备份记录 实例化服务层
var NewNeConfigBackupImpl = &NeConfigBackupImpl{
neConfigBackupRepository: repository.NewNeConfigBackupImpl,
}
// NeConfigBackupImpl 网元配置文件备份记录 服务层处理
type NeConfigBackupImpl struct {
// 网元配置文件备份记录
neConfigBackupRepository repository.INeConfigBackup
}
// SelectNeHostPage 分页查询列表数据
func (r *NeConfigBackupImpl) SelectPage(query map[string]any) map[string]any {
return r.neConfigBackupRepository.SelectPage(query)
}
// SelectConfigList 查询列表
func (r *NeConfigBackupImpl) SelectList(item model.NeConfigBackup) []model.NeConfigBackup {
return r.neConfigBackupRepository.SelectList(item)
}
// SelectByIds 通过ID查询
func (r *NeConfigBackupImpl) SelectById(id string) model.NeConfigBackup {
if id == "" {
return model.NeConfigBackup{}
}
arr := r.neConfigBackupRepository.SelectByIds([]string{id})
if len(arr) > 0 {
return arr[0]
}
return model.NeConfigBackup{}
}
// Insert 新增信息
func (r *NeConfigBackupImpl) Insert(item model.NeConfigBackup) string {
return r.neConfigBackupRepository.Insert(item)
}
// Update 修改信息
func (r *NeConfigBackupImpl) Update(item model.NeConfigBackup) int64 {
return r.neConfigBackupRepository.Update(item)
}
// DeleteByIds 批量删除信息
func (r *NeConfigBackupImpl) DeleteByIds(ids []string) (int64, error) {
// 检查是否存在
data := r.neConfigBackupRepository.SelectByIds(ids)
if len(data) <= 0 {
return 0, fmt.Errorf("neConfigBackup.noData")
}
if len(data) == len(ids) {
rows := r.neConfigBackupRepository.DeleteByIds(ids)
return rows, nil
}
// 删除信息失败！
return 0, fmt.Errorf("delete fail")
}
// NeConfigLocalToNe 网元配置文件复制到网元端覆盖
func (r *NeConfigBackupImpl) NeConfigLocalToNe(neInfo model.NeInfo, localFile string) error {
neTypeLower := strings.ToLower(neInfo.NeType)
// 网管本地路径
omcPath := "/usr/local/etc/omc/ne_config"
if runtime.GOOS == "windows" {
omcPath = fmt.Sprintf("C:%s", omcPath)
}
localDirPath := fmt.Sprintf("%s/%s/%s/backup/tmp_import", omcPath, neTypeLower, neInfo.NeId)
if err := file.UnZip(localFile, localDirPath); err != nil {
return fmt.Errorf("unzip err")
}
// 网元主机的SSH客户端
sshClient, err := NewNeInfoImpl.NeRunSSHClient(neInfo.NeType, neInfo.NeId)
if err != nil {
return fmt.Errorf("ne info ssh client err")
}
defer sshClient.Close()
// 网元主机的SSH客户端进行文件传输
sftpClient, err := sshClient.NewClientSFTP()
if err != nil {
return fmt.Errorf("ne info sftp client err")
}
defer sftpClient.Close()
// 网元配置端上的临时目录
neDirTemp := fmt.Sprintf("/tmp/omc/ne_config/%s/%s", neTypeLower, neInfo.NeId)
sshClient.RunCMD(fmt.Sprintf("mkdir -p /tmp/omc && sudo chmod 777 -R /tmp/omc && sudo rm -rf %s", neDirTemp))
// 复制到网元端
if err = sftpClient.CopyDirLocalToRemote(localDirPath, neDirTemp); err != nil {
return fmt.Errorf("copy config to ne err")
}
// 配置复制到网元内
if neTypeLower == "ims" {
// ims目录
imsDirArr := [...]string{"bgcf", "icscf", "ismc", "mmtel", "mrf", "oam_manager.yaml", "pcscf", "scscf", "vars.cfg", "zlog"}
for _, v := range imsDirArr {
sshClient.RunCMD(fmt.Sprintf("sudo mkdir -p /usr/local/etc/ims && sudo cp -rf %s/ims/%s /usr/local/etc/ims/%v && sudo chmod 755 -R /usr/local/etc/ims/%s", neDirTemp, v, v, v))
}
// mf目录
sshClient.RunCMD(fmt.Sprintf("sudo mkdir -p /usr/local/etc/mf && sudo cp -rf %s/mf/* /usr/local/etc/mf && sudo chmod 755 -R /usr/local/etc/mf", neDirTemp))
// rtproxy目录
sshClient.RunCMD(fmt.Sprintf("sudo mkdir -p /usr/local/etc/rtproxy && sudo cp -rf %s/rtproxy/* /usr/local/etc/rtproxy && sudo chmod 755 /usr/local/etc/rtproxy/rtproxy.conf", neDirTemp))
// iwf目录
sshClient.RunCMD(fmt.Sprintf("sudo mkdir -p /usr/local/etc/iwf && sudo cp -rf %s/iwf/* /usr/local/etc/iwf && sudo chmod 755 /usr/local/etc/iwf/*.yaml", neDirTemp))
} else {
neEtcPath := fmt.Sprintf("/usr/local/etc/%s", neTypeLower)
chmodFile := fmt.Sprintf("sudo chmod 755 %s/*.yaml", neEtcPath)
if neTypeLower == "mme" {
chmodFile = fmt.Sprintf("sudo chmod 755 %s/*.{yaml,conf}", neEtcPath)
}
sshClient.RunCMD(fmt.Sprintf("sudo cp -rf %s/* %s && %s", neDirTemp, neEtcPath, chmodFile))
}
_ = os.RemoveAll(localDirPath) // 删除本地临时目录
sshClient.RunCMD(fmt.Sprintf("sudo rm -rf %s", neDirTemp)) // 删除临时目录
return nil
}
// NeConfigNeToLocal 网元备份文件网元端复制到本地
func (r *NeConfigBackupImpl) NeConfigNeToLocal(neInfo model.NeInfo) (string, error) {
// 网元主机的SSH客户端
sshClient, err := NewNeInfoImpl.NeRunSSHClient(neInfo.NeType, neInfo.NeId)
if err != nil {
return "", fmt.Errorf("ne info ssh client err")
}
defer sshClient.Close()
// 网元主机的SSH客户端进行文件传输
sftpClient, err := sshClient.NewClientSFTP()
if err != nil {
return "", fmt.Errorf("ne info sftp client err")
}
defer sftpClient.Close()
neTypeLower := strings.ToLower(neInfo.NeType)
// 网管本地路径
omcPath := "/usr/local/etc/omc/ne_config"
if runtime.GOOS == "windows" {
omcPath = fmt.Sprintf("C:%s", omcPath)
}
localDirPath := fmt.Sprintf("%s/%s/%s/backup/tmp_export", omcPath, neTypeLower, neInfo.NeId)
// 网元配置文件先复制到临时目录
sshClient.RunCMD("mkdir -p /tmp/omc && sudo chmod 777 -R /tmp/omc")
neDirTemp := fmt.Sprintf("/tmp/omc/ne_config/%s/%s", neTypeLower, neInfo.NeId)
if neTypeLower == "ims" {
// ims目录
sshClient.RunCMD(fmt.Sprintf("mkdir -p %s/ims", neDirTemp))
imsDirArr := [...]string{"bgcf", "icscf", "ismc", "mmtel", "mrf", "oam_manager.yaml", "pcscf", "scscf", "vars.cfg", "zlog"}
for _, v := range imsDirArr {
sshClient.RunCMD(fmt.Sprintf("sudo cp -rf /usr/local/etc/ims/%s %s/ims", v, neDirTemp))
}
// mf目录
sshClient.RunCMD(fmt.Sprintf("mkdir -p %s/mf && sudo cp -rf /usr/local/etc/mf %s", neDirTemp, neDirTemp))
// rtproxy目录
sshClient.RunCMD(fmt.Sprintf("mkdir -p %s/rtproxy && sudo cp -rf /usr/local/etc/rtproxy/rtproxy.conf %s/rtproxy", neDirTemp, neDirTemp))
// iwf目录
sshClient.RunCMD(fmt.Sprintf("mkdir -p %s/iwf && sudo cp -rf /usr/local/etc/iwf/*.yaml %s/iwf", neDirTemp, neDirTemp))
} else {
nePath := fmt.Sprintf("/usr/local/etc/%s/*.yaml", neTypeLower)
if neTypeLower == "mme" {
nePath = fmt.Sprintf("/usr/local/etc/%s/*.{yaml,conf}", neTypeLower)
}
sshClient.RunCMD(fmt.Sprintf("mkdir -p %s && sudo cp -rf %s %s", neDirTemp, nePath, neDirTemp))
}
// 网元端复制到本地
if err = sftpClient.CopyDirRemoteToLocal(neDirTemp, localDirPath); err != nil {
return "", fmt.Errorf("copy config err")
}
// 压缩zip文件名
zipFileName := fmt.Sprintf("%s-%s-etc-%s.zip", neTypeLower, neInfo.NeId, date.ParseDateToStr(time.Now(), date.YYYYMMDDHHMMSS))
zipFilePath := fmt.Sprintf("%s/%s/%s/backup/%s", omcPath, neTypeLower, neInfo.NeId, zipFileName)
if err := file.CompressZipByDir(zipFilePath, localDirPath); err != nil {
return "", fmt.Errorf("compress zip err")
}
_ = os.RemoveAll(localDirPath) // 删除本地临时目录
sshClient.RunCMD(fmt.Sprintf("sudo rm -rf %s", neDirTemp)) // 删除临时目录
return zipFilePath, nil
}

8
src/modules/ws/service/ws_send.impl.go
View File

@@ -12,10 +12,10 @@ import (
const (
// 组号-其他
GROUP_OTHER = "0"
// 组号-指标
GROUP_KPI = "10"
// 组号-指标UPF
GROUP_KPI_UPF = "12"
// 组号-指标通用 10_neType_neId
GROUP_KPI = "10_"
// 组号-指标UPF 12_neId
GROUP_KPI_UPF = "12_"
// 组号-IMS_CDR会话事件
GROUP_IMS_CDR = "1005"
// 组号-SMF_CDR会话事件