OMC/nbi_alarm
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: OMC/nbi_alarm:03bb5ea13196de403a3f9dff4328ecfe822821f2
OMC/nbi_alarm:main
..
compare: OMC/nbi_alarm:70139bd3fd463181b2e38291802778401771708a
OMC/nbi_alarm:main
1 Commits
03bb5ea131 .. 70139bd3fd
Author SHA1 Message Date
admin 70139bd3fd Initial commit 2026-10-29 06:43:41 +00:00
41 changed files with 3 additions and 2490 deletions
Expand all files Collapse all files

No files matched your search

.gitignore
-47
View File
@@ -1,47 +0,0 @@
# ---> VisualStudioCode
.vscode/*
!.vscode/settings.json
!.vscode/tasks.json
!.vscode/launch.json
!.vscode/extensions.json
!.vscode/*.code-snippets
# Local History for Visual Studio Code
.history/
# Run temp file and dir
crontask/log/
crontask/ftp/
crontask/database/
crontask/export/
crontask/temp
crontask/crontask
restagent/backup/
restagent/log/
restagent/upload/
restagent/software/
restagent/database/
restagent/restagent
sshsvc/sshsvc
sshsvc/mmllog/
sshsvc/mmlhome/
sshsvc/log/
tools/loadmconf/loadmconf
vendor
# Built Visual Studio Code Extensions
*.vsix
*.log
*.log-*
*.bak
.idea
omc_log
data/ftp
lib/FTP
assets
__debug_*
.vscode/launch.json
Vendored
-17
View File
@@ -1,17 +0,0 @@
{
// 使用 IntelliSense 了解相关属性。
// 悬停以查看现有属性的描述。
// 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "调试",
"type": "go",
"request": "launch",
"mode": "debug",
"program": "./nbi_alarm.go",
"args": ["-c", "conf/nbi_alarm.json"],
"console": "integratedTerminal"
}
]
}
README.md
+3
View File
@@ -0,0 +1,3 @@
# nbi_alarm
北向-移动集采测试
conf/global.go
-77
View File
@@ -1,77 +0,0 @@
package conf
import (
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"github.com/aceld/zinx/zconf"
"github.com/aceld/zinx/zlog"
)
type Channel struct {
TCPPort int `json:"tcp_port"` //当前通道的TCP监听端口
BindFlag string `json:"bind_flag"` //当前通道bind 的网元信息
Province string `json:"province"` //网元所在省份
DeviceCode string `json:"device_code"` //主机编码 四位，每1位可用0-9、A-Z编码
}
type Config struct {
// 网元通道
Channel []Channel `json:"channel"`
// 数据库连接
Mysql string `json:"mysql"`
//以下是zinx 的配置
Name string `json:"name"`
MaxConn int `json:"max_conn"`
WorkerPoolSize int `json:"worker_pool_size"`
HeartbeatMax int `json:"heartbeat_max"`
HbcheckInterval int `json:"hbcheck_interval"`
LogDir string `json:"log_dir"`
LogFile string `json:"log_file"`
// FTP文件服务
FTPRoot string `json:"ftp_root"`
}
var OmcConf Config
func Reload(path string) error {
confFileExists, _ := zconf.PathExists(path)
if !confFileExists {
zlog.Ins().ErrorF("Config File %s is not exist!!", path)
return errors.New("config file is not exist")
}
data, err := os.ReadFile(path)
if err != nil {
return err
}
err = json.Unmarshal(data, &OmcConf)
if err != nil {
return err
}
return nil
}
func Init(path string) {
if path == "" || strings.HasSuffix(path, "zinx.json") {
pwd, err := os.Getwd()
if err != nil {
pwd = "."
}
path = pwd + "/conf/nbi_alarm.json"
fmt.Println("path", path)
}
if err := Reload(path); err != nil {
zlog.Ins().ErrorF("Config File %s error, ", err)
panic(err)
}
}
conf/nbi_alarm.json
-23
View File
@@ -1,23 +0,0 @@
{
"channel": [
{
"tcp_port": 31232,
"bind_flag": "SMF#SZ_01",
"province": "BJ",
"device_code": "0001"
}
],
"mysql": "root:1000omc@kp!@tcp(192.168.0.229:33066)/omc_db?charset=utf8mb4&parseTime=True&loc=Local",
"mame":"nbi north alarm agent",
"max_conn": 3000,
"worker_pool_size": 100,
"heartbeat_max": 180,
"hbcheck_interval": 60,
"log_dir": "./assets/nbi_alarm",
"log_file":"nbi_alarm.log",
"ftp_root": "/data/ftp"
}
core/consts/consts.go
-7
View File
@@ -1,7 +0,0 @@
package consts
//定义user type
const (
MSG = "msg"
FILE = "ftp"
)
core/db/mysql.go
-37
View File
@@ -1,37 +0,0 @@
package db
import (
"omc/conf"
"github.com/aceld/zinx/zlog"
"gorm.io/driver/mysql"
"gorm.io/gorm"
)
var Client *gorm.DB
func Init() error {
// 日志输出sql观察
// newLogger := logger.New(
// log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer
// logger.Config{
// SlowThreshold: time.Second, // Slow SQL threshold
// LogLevel: logger.Info, // Log level
// IgnoreRecordNotFoundError: true, // Ignore ErrRecordNotFound error for logger
// ParameterizedQueries: false, // Don't include params in the SQL log
// Colorful: false, // Disable color
// },
// )
// db, err := gorm.Open(mysql.Open(conf.OmcConf.Mysql), &gorm.Config{Logger: newLogger})
db, err := gorm.Open(mysql.Open(conf.OmcConf.Mysql), &gorm.Config{})
if err != nil {
zlog.Ins().ErrorF("open mysql %s error, ", conf.OmcConf.Mysql, err)
panic(err)
}
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(20)
sqlDB.SetMaxIdleConns(10)
Client = db
return nil
}
core/decoder/decode_omc.go
-88
View File
@@ -1,88 +0,0 @@
package decoder
import (
"bytes"
"encoding/binary"
"encoding/hex"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"math"
)
//ffff 01 00999999 000a 68656c6c6f2074657374
// +---------------+---------------+--------------------+----------------+-------------------
// | 开始标志 | type | 秒时间戳 | 长度 | 消息体
// | 0xffff(2byte) | uint8(1byte) | uint32(4byte) | uint16(2byte) | bytes(N byte)
// +---------------+---------------+--------------------+----------------+-------------------
const OmcMsgHeaderSize = 9 //表示OMC空包长度
type OmcDecoder struct {
StartSign uint16
MsgType uint8
TimeStamp uint32
LenOfBody uint16
Value []byte
}
func NewOmcDecoder() ziface.IDecoder {
return &OmcDecoder{}
}
func (omc *OmcDecoder) GetLengthField() *ziface.LengthField {
return &ziface.LengthField{
MaxFrameLength: math.MaxUint16 + 9,
LengthFieldOffset: 7,
LengthFieldLength: 2,
LengthAdjustment: 0,
InitialBytesToStrip: 0,
//注意现在默认是大端，使用小端需要指定编码方式
Order: binary.BigEndian,
}
}
func (omc *OmcDecoder) decode(data []byte) *OmcDecoder {
ltvData := OmcDecoder{}
ltvData.StartSign = binary.BigEndian.Uint16(data[0:2])
ltvData.MsgType = data[2]
ltvData.TimeStamp = binary.BigEndian.Uint32(data[3:7])
ltvData.LenOfBody = binary.BigEndian.Uint16(data[7:9])
//Determine the length of V. (确定V的长度)
ltvData.Value = make([]byte, ltvData.LenOfBody)
//5. Get V
binary.Read(bytes.NewBuffer(data[OmcMsgHeaderSize:OmcMsgHeaderSize+ltvData.LenOfBody]), binary.BigEndian, ltvData.Value)
return &ltvData
}
func (omc *OmcDecoder) Intercept(chain ziface.IChain) ziface.IcResp {
//1. Get the IMessage of zinx
iMessage := chain.GetIMessage()
if iMessage == nil {
// Go to the next layer in the chain of responsibility
return chain.ProceedWithIMessage(iMessage, nil)
}
//2. Get Data
data := iMessage.GetData()
zlog.Ins().DebugF("omc-RawData size:%d data:%s\n", len(data), hex.EncodeToString(data))
// (读取的数据不超过包头，直接进入下一层)
if len(data) < OmcMsgHeaderSize {
return chain.ProceedWithIMessage(iMessage, nil)
}
//4. Decode
ltvData := omc.decode(data)
zlog.Ins().DebugF("omc-decode %v", ltvData)
// (将解码后的数据重新设置到IMessage中, Zinx的Router需要MsgID来寻址)
iMessage.SetDataLen(uint32(ltvData.LenOfBody))
iMessage.SetMsgID(uint32(ltvData.MsgType))
iMessage.SetData(ltvData.Value)
//6. Pass the decoded data to the next layer.
// (将解码后的数据进入下一层)
return chain.ProceedWithIMessage(iMessage, *ltvData)
}
core/dpack/datapack_omc.go
-117
View File
@@ -1,117 +0,0 @@
package dpack
import (
"bytes"
"encoding/binary"
"errors"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/zpack"
"time"
"github.com/aceld/zinx/zconf"
"github.com/aceld/zinx/ziface"
)
// +---------------+---------------+--------------------+----------------+-------------------
// | 开始标志 | type | 秒时间戳 | 长度 | 消息体
// | 0xffff(2byte) | uint8(1byte) | uint32(4byte) | uint16(2byte) | bytes(N byte)
// +---------------+---------------+--------------------+----------------+-------------------
var defaultHeaderLen uint32 = 9
type DataPack struct{}
// NewDataPack initializes a packing and unpacking instance
// (封包拆包实例初始化方法)
func NewDataPack() ziface.IDataPack {
return &DataPack{}
}
// GetHeadLen returns the length of the message header
// (获取包头长度方法)
func (dp *DataPack) GetHeadLen() uint32 {
//ID uint32(4 bytes) + DataLen uint32(4 bytes)
return defaultHeaderLen
}
// Pack packs the message (compresses the data)
// (封包方法,压缩数据)
func (dp *DataPack) Pack(msg ziface.IMessage) ([]byte, error) {
zlog.Ins().InfoF("my pack: %v", msg)
// Create a buffer to store the bytes
// (创建一个存放bytes字节的缓冲)
dataBuff := bytes.NewBuffer([]byte{})
// Write the oxffff
if err := binary.Write(dataBuff, binary.BigEndian, uint16(0xffff)); err != nil {
return nil, err
}
//Write the type
if err := binary.Write(dataBuff, binary.BigEndian, uint8(msg.GetMsgID())); err != nil {
return nil, err
}
//Write the timestamp
if err := binary.Write(dataBuff, binary.BigEndian, uint32(time.Now().Unix())); err != nil {
return nil, err
}
//Write the length
if err := binary.Write(dataBuff, binary.BigEndian, uint16(msg.GetDataLen())); err != nil {
return nil, err
}
// Write the data
if err := binary.Write(dataBuff, binary.BigEndian, msg.GetData()); err != nil {
return nil, err
}
return dataBuff.Bytes(), nil
}
// Unpack unpacks the message (decompresses the data)
// (拆包方法,解压数据)
func (dp *DataPack) Unpack(binaryData []byte) (ziface.IMessage, error) {
// Create an ioReader for the input binary data
dataBuff := bytes.NewReader(binaryData)
// Only unpack the header information to obtain the data length and message ID
// (只解压head的信息，得到dataLen和msgID)
msg := &zpack.Message{}
// Read the startSign
var startSign uint16
if err := binary.Read(dataBuff, binary.BigEndian, &startSign); err != nil {
return nil, err
}
// Read the msgID
var msgType uint
if err := binary.Read(dataBuff, binary.BigEndian, &msgType); err != nil {
return nil, err
}
msg.ID = uint32(msgType)
// read timeStamp
var timeStamp uint32
if err := binary.Read(dataBuff, binary.BigEndian, &timeStamp); err != nil {
return nil, err
}
// Read the data length
var length uint16
if err := binary.Read(dataBuff, binary.BigEndian, &length); err != nil {
return nil, err
}
msg.DataLen = uint32(length)
// Check whether the data length exceeds the maximum allowed packet size
// (判断dataLen的长度是否超出我们允许的最大包长度)
if zconf.GlobalObject.MaxPacketSize > 0 && msg.GetDataLen() > zconf.GlobalObject.MaxPacketSize {
return nil, errors.New("too large msg data received")
}
// Only the header data needs to be unpacked, and then another data read is performed from the connection based on the header length
// (这里只需要把head的数据拆包出来就可以了，然后再通过head的长度，再从conn读取一次数据)
return msg, nil
}
core/emun/orig_severity.go
-17
View File
@@ -1,17 +0,0 @@
package emun
// 原始级别 中文转数值
func OrigSeverity(os string) int64 {
switch os {
case "Critical":
return 1
case "Major":
return 2
case "Minor":
return 3
case "Warning":
return 4
default:
return 0
}
}
core/heart_beat/heart_beat.go
-25
View File
@@ -1,25 +0,0 @@
package heartbeat
import (
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
)
var HeadBeatMsgID uint32 = 255
// MyHeartBeatMsg 用户自定义的心跳检测消息处理方法
func MyHeartBeatMsg(conn ziface.IConnection) []byte {
return []byte("heartbeat, I am server, I am alive")
}
// MyHeartBeat 用户自定义心跳发送函数
func MyHeartBeat(conn ziface.IConnection) error {
return nil
}
// MyOnRemoteNotAlive 用户自定义的远程连接不存活时的处理方法
func MyOnRemoteNotAlive(conn ziface.IConnection) {
zlog.Ins().ErrorF("myOnRemoteNotAlive is Called, connID=%v", conn.GetConnID(), "remoteAddr =%v ", conn.RemoteAddr())
//关闭链接
conn.Stop()
}
core/interceptor/global.go
-42
View File
@@ -1,42 +0,0 @@
package interceptor
import (
"encoding/hex"
"fmt"
"omc/core/parse"
"time"
"github.com/aceld/zinx/ziface"
)
type GlobalInterceptor struct{}
func (m *GlobalInterceptor) Intercept(chain ziface.IChain) ziface.IcResp {
request := chain.Request() //从责任链中获取当前拦截器的输入数据
// 这一层是自定义拦截器处理逻辑，这里只是简单打印输入
iRequest := request.(ziface.IRequest) //注意：由于Zinx的Request类型，这里需要做一下断言转换
fmt.Println("\n\n=========自定义拦截器=====")
body, err := parse.RequestBodyDecode(iRequest, nil)
fmt.Printf("消息ID: %v \n", iRequest.GetMsgID())
fmt.Printf("原始数据: %v \n", body.RawData)
fmt.Printf("原始字符: %v \n", hex.EncodeToString(body.RawData))
fmt.Printf("原始字符: %v \n", string(body.RawData))
fmt.Printf("用户ID: %v \n", body.UID)
fmt.Printf("收到消息: %v %v \n", body.Name, body.Data)
fmt.Printf("错误：%v \n", err)
// return chain.Proceed(chain.Request()) //进入并执行下一个拦截器
iMessage := chain.GetIMessage()
resp := chain.ProceedWithIMessage(iMessage, iRequest)
fmt.Printf("目标消息ID: %v \n", iMessage.GetMsgID())
fmt.Printf("收到消息长度: %v \n", iMessage.GetDataLen())
fmt.Printf("信息时间：%v \n", time.Now().String())
fmt.Print("=========自定义拦截器=====\n\n\n")
return resp
}
core/manage/manage.go
-51
View File
@@ -1,51 +0,0 @@
package manage
import (
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
)
// OnConnectionAdd 当客户端建立连接的时候的hook函数
func OnConnectionAdd(conn ziface.IConnection) {
//创建一个user
user := NewUser(conn, conn.RemoteAddrString())
//将当前新上线玩家添加到ChannelManager中
m := GetManager(conn.GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error in GetManager")
conn.Stop()
return
}
m.AddUser(user)
//将该连接绑定属性PID
conn.SetProperty("UID", user.UID)
zlog.Ins().InfoF("====> User uID = %s", user.UID, " arrived ====", "")
}
// OnConnectionLost 当客户端断开连接的时候的hook函数
func OnConnectionLost(conn ziface.IConnection) {
//获取当前连接的PID属性
uID, _ := conn.GetProperty("UID")
var userID string
if uID != nil {
userID = uID.(string)
}
//根据pID获取对应usr
m := GetManager(conn.GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error in GetManager")
return
}
user := m.GetUserByPID(userID)
//触发玩家下线业务
if user != nil {
user.LostConnection(m)
}
zlog.Ins().InfoF("====> User %s-%s", user.UID, user.UserName, " left =====")
}
core/manage/user.go
-50
View File
@@ -1,50 +0,0 @@
package manage
import (
"fmt"
"github.com/aceld/zinx/ziface"
"github.com/google/uuid"
)
// User User 对象
type User struct {
UID string //ID
Conn ziface.IConnection //当前User的连接
LoginState bool
LoginCount int
AlarmType string
UserName string
SeqNo string // 登录随机码
RemoteIp string
}
// NewUser 创建一个user对象
func NewUser(conn ziface.IConnection, addr string) *User {
p := &User{
UID: uuid.NewString(),
Conn: conn,
RemoteIp: addr,
}
return p
}
// LostConnection User下线
func (p *User) LostConnection(m *ChannelManager) {
m.RemoveUserByPID(p.UID)
}
// SendMsg /*
func (p *User) SendMsg(msgID uint32, msg []byte) {
if p.Conn == nil {
fmt.Println("connection in player is nil")
return
}
//调用SendMsg发包
if err := p.Conn.SendMsg(msgID, msg); err != nil {
fmt.Println("Player SendMsg error !")
return
}
}
core/manage/user_manager.go
-237
View File
@@ -1,237 +0,0 @@
package manage
import (
"encoding/json"
"errors"
"omc/core/consts"
"omc/core/db"
"omc/core/parse"
"omc/handle/model"
"omc/handle/service"
"strings"
"sync"
"time"
"github.com/aceld/zinx/zlog"
)
/*
ChannelManager 通道管理管理模块
*/
type ChannelManager struct {
Name string
User map[string]*User //当前在线的User集合
BindFlag string //bind 的网元标识，ne_type#ne_id 格式
Province string //网元所在省份
DeviceCode string //主机编码 四位，每1位可用0-9、A-Z编码
AlarmSeq int64 //当前告警的序列号
pLock sync.RWMutex //保护User的互斥读写机制
}
// MgrObj 多个通道管理模块集合
var MgrObj []*ChannelManager
// NewManager New 提供ChannelManager 初始化 方法
func NewManager(name, bind, province, deviceCode string) *ChannelManager {
newManager := ChannelManager{
Name: name,
BindFlag: bind,
Province: province,
DeviceCode: deviceCode,
User: make(map[string]*User),
}
MgrObj = append(MgrObj, &newManager)
return &newManager
}
func GetManager(name string) *ChannelManager {
for i := range MgrObj {
if MgrObj[i].Name == name {
return MgrObj[i]
}
}
return nil
}
// AddUser 提供添加一个user的功能
func (wm *ChannelManager) AddUser(User *User) {
//将User添加到 世界管理器中
wm.pLock.Lock()
wm.User[User.UID] = User
wm.pLock.Unlock()
}
// Talk User广播消息
func (wm *ChannelManager) Talk(msgID uint32, msg []byte) {
//2. 得到所有的在线user
users := wm.GetAllUser()
//3. 向所有的user发送消息
for _, user := range users {
if user.LoginState && user.AlarmType == consts.MSG {
user.SendMsg(msgID, msg)
}
}
}
// LoginSuccess 登录成功管理
func (wm *ChannelManager) LoginSuccess(UID, name, tp string) error {
wm.pLock.Lock()
defer wm.pLock.Unlock()
//判断是否重复登录
for _, v := range wm.User {
if v.UserName == name && v.AlarmType == tp && v.LoginState {
v.Conn.Stop()
return errors.New("repeat login for the account")
}
}
user, ok := wm.User[UID]
if !ok {
return errors.New("server internal error")
}
user.UserName = name
user.AlarmType = tp
user.LoginState = true
wm.User[UID] = user
return nil
}
// SetSeqNo 设置登录随机码
func (wm *ChannelManager) SetSeqNo(UID, seqNo string) error {
wm.pLock.Lock()
defer wm.pLock.Unlock()
//判断是否重复登录
user, ok := wm.User[UID]
if !ok {
return errors.New("server internal error")
}
user.SeqNo = seqNo
return nil
}
// LoginFail 登录失败管理
func (wm *ChannelManager) LoginFail(UID string) (bool, error) {
wm.pLock.Lock()
defer wm.pLock.Unlock()
//判断是否重复登录
user, ok := wm.User[UID]
if !ok {
return true, errors.New("server internal error")
}
user.LoginCount++
wm.User[UID] = user
if user.LoginCount >= 3 {
return true, errors.New("too many attempts, close connect")
}
return false, nil
}
// RemoveUserByPID 从信息表中移除一个user
func (wm *ChannelManager) RemoveUserByPID(uid string) {
wm.pLock.Lock()
delete(wm.User, uid)
wm.pLock.Unlock()
}
// GetUserByPID 通过user ID 获取对应user信息
func (wm *ChannelManager) GetUserByPID(uID string) *User {
wm.pLock.RLock()
defer wm.pLock.RUnlock()
return wm.User[uID]
}
// GetAllUser 获取所有user的信息
func (wm *ChannelManager) GetAllUser() []*User {
wm.pLock.RLock()
defer wm.pLock.RUnlock()
//创建返回的User集合切片
User := make([]*User, 0)
//添加切片
for _, v := range wm.User {
if v.LoginState && v.AlarmType == consts.MSG {
User = append(User, v)
}
}
//返回
return User
}
func (wm *ChannelManager) RealTimeAlarm() {
for {
wm.pLock.RLock()
//查询
var alarms []model.OmcAlarm
neBind, _ := parse.ConvertBindFlag(wm.BindFlag)
var lastAlarmSeq = wm.AlarmSeq
if wm.AlarmSeq == 0 {
lastAlarmSeq = service.LastAlarmSeq(neBind.NeType, neBind.NeId)
} else {
list, err := service.List(neBind.NeType, neBind.NeId, wm.AlarmSeq)
if err != nil {
zlog.Ins().ErrorF("db error %s", err)
}
alarms = service.ConvertOMCAlarm(list)
}
//上报实时告警信息
if len(alarms) > 0 {
go wm.SendAlarm(alarms)
lastAlarmSeq = service.MaxAlarmSeq(lastAlarmSeq, alarms)
}
var users []string
for _, user := range wm.User {
if user.LoginState && user.AlarmType == consts.MSG {
userInfo := strings.Join([]string{user.UserName, user.RemoteIp}, ";")
users = append(users, userInfo)
}
}
//更新AlarmSeq
wm.AlarmSeq = lastAlarmSeq + 1
wm.pLock.RUnlock()
//记录日志到alarm 日志表
for _, v := range alarms {
for _, u := range users {
var alarmLog model.NbiAlarmLog
ui := strings.Split(u, ";")
if len(ui) == 2 {
alarmLog.OpUser = ui[0]
alarmLog.SrcIp = ui[1]
}
alarmLog.NeType = v.NeType
alarmLog.NeId = v.NeUID
alarmLog.AlarmSeq = int64(v.AlarmSeq)
alarmLog.AlarmId = v.AlarmId
alarmLog.EventTime = v.EventTime
alarmLog.LogTime = time.Now()
db.Client.Create(&alarmLog)
}
}
time.Sleep(3 * time.Second)
}
}
func (wm *ChannelManager) SendAlarm(alarms []model.OmcAlarm) error {
for _, v := range alarms {
//生产告警内容
data, err := json.Marshal(v)
if err != nil {
zlog.Ins().ErrorF("SendAlarm json Marshalerror %v", err)
} else {
//发送告警内容
wm.Talk(0, data)
}
}
return nil
}
func (wm *ChannelManager) UpdateAlarmSeq(alarmSeq int64) {
wm.pLock.Lock()
wm.AlarmSeq = alarmSeq
wm.pLock.Unlock()
}
core/model/body.go
-9
View File
@@ -1,9 +0,0 @@
package model
// 请求数据
type Body struct {
UID string // 连接实例ID
RawData []byte // 原始数据
Name string // 请求名
Data map[string]string // 数据Key
}
core/parse/parse.go
-126
View File
@@ -1,126 +0,0 @@
package parse
import (
"errors"
"fmt"
"omc/core/model"
"strings"
"github.com/aceld/zinx/ziface"
)
// 网元类型#网元标记
type NeBind struct {
NeType string
NeId string
}
// 转换解析服务端绑定的网元
func ConvertBindFlag(bindFlag string) (NeBind, error) {
var neBind NeBind
nb := strings.Split(bindFlag, "#")
if len(nb) != 2 {
return neBind, errors.New("ne bind flag invalid")
}
neBind.NeType = nb[0]
neBind.NeId = nb[1]
return neBind, nil
}
// RequestBodyDecode 请求消息解析
// checker 检查参数必传
func RequestBodyDecode(request ziface.IRequest, checker []string) (model.Body, error) {
// 消息处理
body := model.Body{}
err := Decode(request.GetData(), &body)
if err != nil {
return body, errors.New("inlaid message body")
}
// 检查key
if len(checker) > 0 {
for _, v := range checker {
if _, ok := body.Data[v]; !ok {
return body, errors.New("missing parameter body " + v)
}
}
}
// 当前连接实例ID
uID, err := request.GetConnection().GetProperty("UID")
if err != nil {
request.GetConnection().Stop()
return body, errors.New("server internal error")
}
body.UID = uID.(string)
return body, nil
}
// Decode 数据解析
// reqLoginAlarm;user=yiy;key=qw#$@;type=msg
func Decode(data []byte, body *model.Body) error {
body.RawData = data
multi := strings.Split(string(data), ";")
if len(multi) < 1 {
return errors.New("invalid msg body")
}
// 获取函数名
if multi[0] != "" {
name := multi[0]
idx := strings.LastIndex(name, "\x14")
if idx == -1 {
idx = strings.LastIndex(name, "\x00")
}
if idx == -1 {
idx = strings.LastIndex(name, "\xe2")
}
if idx > 0 {
name = name[idx+1:]
name = strings.Replace(name, "\"", "", 1)
name = strings.Replace(name, "'", "", 1)
name = strings.Replace(name, "#", "", 1)
}
body.Name = name
}
// 解析data KEY
body.Data = make(map[string]string)
for i := 1; i < len(multi); i++ {
str := multi[i]
idx := strings.Index(str, "=")
if idx > 0 {
key := str[:idx]
value := str[idx+1:]
body.Data[key] = value
}
}
return nil
}
// Pack 数据压缩
func Pack(name string, data map[string]string) []byte {
var multi []string
multi = append(multi, name)
// 允许的拓展字段 "seqNo", "fileName"
// 固定值顺序 "result", "reqId", "resDesc"
keys := []string{"seqNo", "reqId", "result", "fileName", "resDesc"}
for _, key := range keys {
if v, ok := data[key]; ok {
// 长度小于32个字符，不允许带分号“;”
if key == "resDesc" && len(v) > 32 {
v = v[0:32]
}
if key == "resDesc" && strings.Contains(v, ";") {
v = strings.ReplaceAll(v, ";", "")
}
multi = append(multi, fmt.Sprintf("%s=%s", key, v))
}
}
raw := strings.Join(multi, ";")
return []byte(raw)
}
core/result.go
-39
View File
@@ -1,39 +0,0 @@
package core
import (
"omc/core/parse"
)
// Result
// 配合 request.GetConnection().SendMsg()
// resDesc 原因长度小于32字符
func Result(name string, data map[string]string) []byte {
return parse.Pack(name, data)
}
// ResultError ackLoginAlarm;result=fail;resDesc=username-error
// request.GetConnection().SendMsg(omc.AckSyncAlarmMsg, core.ResultError("ackSyncAlarmMsg", err.Error(), ""))
func ResultError(name string, desc, reqID string) []byte {
data := map[string]string{
"result": "fail",
"resDesc": desc,
}
if reqID != "" {
data["reqId"] = reqID
}
return Result(name, data)
}
// request.GetConnection().SendMsg(omc.AckSyncAlarmMsg, core.ResultSuccess("ackSyncAlarmMsg", err.Error(), ""))
func ResultSuccess(name string, desc, reqID string) []byte {
data := map[string]string{
"result": "succ",
"resDesc": desc,
}
if reqID != "" {
data["reqId"] = reqID
}
return Result(name, data)
}
core/utils/bcrypt.go
-14
View File
@@ -1,14 +0,0 @@
package utils
import "golang.org/x/crypto/bcrypt"
// Encrypt 加密明文密码
func Encrypt(password string) (string, error) {
hashedBytes, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
return string(hashedBytes), err
}
// Compare 密文校验
func Compare(hashedPassword, password string) error {
return bcrypt.CompareHashAndPassword([]byte(hashedPassword), []byte(password))
}
core/utils/cakey.go
-41
View File
@@ -1,41 +0,0 @@
package utils
import (
"crypto"
"crypto/rsa"
"crypto/sha256"
"crypto/x509"
"encoding/pem"
"fmt"
"strings"
)
// 验证私钥的签名， publicKey从证书里获取公钥 certificate.PublicKey
func VerifySignature(publicKey any, data, signature []byte) bool {
rsaPublicKey := publicKey.(*rsa.PublicKey)
hashed := sha256.Sum256(data)
err := rsa.VerifyPKCS1v15(rsaPublicKey, crypto.SHA256, hashed[:], signature)
return err == nil
}
// 解析证书信息
func ParseCert(carCertificate string) (*x509.Certificate, error) {
// 解析头尾需要换行符
result := strings.Split(carCertificate, "-----")
result[2] = "\n" + result[2] + "\n"
carCertificate = strings.Join(result, "-----")
// 解码证书
block, _ := pem.Decode([]byte(carCertificate))
if block == nil {
return nil, fmt.Errorf("无法解码证书")
}
// 解析证书
certificate, err := x509.ParseCertificate(block.Bytes)
if err != nil {
return nil, err
}
return certificate, nil
}
core/utils/file.go
-145
View File
@@ -1,145 +0,0 @@
package utils
import (
"archive/zip"
"encoding/json"
"fmt"
"io"
"omc/handle/model"
"os"
"path"
"path/filepath"
"strings"
"time"
"github.com/aceld/zinx/zlog"
)
//BJ/HX/RJ/OMC/FM/告警文件生成时间
// /FTP根目录/省份简称/专业简称/厂家编码/OMC名称/数据类别/日期或时间/
//<省份简称>-<数据类别>-<网元类型>[-网元子类]-<主机编号>-<数据版本>-<数据时间>[-登录用户名][-同步请求标识][-Ri][-统计周期] [-序列号].<后缀>
// BJ-FM-OMC-主机编码-v0-告警文件生成时间-001.txt
type FileNameMeta struct {
FTPRoot string `json:"ftp_root"` // FTP根目录
Province string `json:"province"` // 网元所在省份
DeviceCode string `json:"device_code"` // 主机编码 四位，每1位可用0-9、A-Z编码
ReqId string `json:"req_id"` // 同步请求标识
Suffix string `json:"suffix"` // 文件后缀
}
// 生成告警文件文件名
// GD-FM-OMC-100A-V1.1.0-20150611011603-001.txt.zip
func FileName(meta *FileNameMeta) string {
// 路径分层
timeDayStr := time.Now().Format("20060102")
filePath := path.Join(meta.FTPRoot, meta.Province, "HX", "RJ", "OMC", "FM", timeDayStr)
// 文件名组合
reqIdStr := fmt.Sprintf("%03s", meta.ReqId)
timeStr := time.Now().Format("20060102150405")
fileNameGroup := []string{meta.Province, "FM", "OMC", meta.DeviceCode, "V0", timeStr, reqIdStr}
fileName := strings.Join(fileNameGroup, "-")
fileName = strings.ToUpper(fileName) + meta.Suffix
return path.Join(filePath, fileName)
}
func CreateFile(filePath string, data []model.OmcAlarm) error {
// 获取文件所在的目录路径
dirPath := filepath.Dir(filePath)
// 确保文件夹路径存在
err := os.MkdirAll(dirPath, os.ModePerm)
if err != nil {
zlog.Ins().ErrorF("创建文件夹失败 CreateFile %v", err)
}
// 创建或打开文件
file, err := os.Create(filePath)
if err != nil {
return err
}
defer file.Close()
// 遍历切片并将每个元素序列化为 JSON 字符串并写入文件
for _, alarm := range data {
jsonData, err := json.Marshal(alarm)
if err != nil {
return err
}
// 写入 JSON 字符串到文件，并换行
_, err = file.WriteString(string(jsonData) + "\r\n")
if err != nil {
return err
}
}
return nil
}
// 将文件添加到 ZIP 压缩文件
func FileToZip(zipFileName, filePath string) error {
// 创建一个新的 ZIP 文件
newZipFile, err := os.Create(zipFileName)
if err != nil {
return fmt.Errorf("创建 ZIP 文件失败: %w", err)
}
defer newZipFile.Close()
// 创建 ZIP 写入器
zipWriter := zip.NewWriter(newZipFile)
defer zipWriter.Close()
fileToCompress, err := os.Open(filePath)
if err != nil {
return fmt.Errorf("打开文件失败: %w", err)
}
defer fileToCompress.Close()
// 获取文件信息
fileInfo, err := fileToCompress.Stat()
if err != nil {
return fmt.Errorf("获取文件信息失败: %w", err)
}
// 创建文件头
header, err := zip.FileInfoHeader(fileInfo)
if err != nil {
return fmt.Errorf("创建文件头失败: %w", err)
}
// 设置文件头中的名称
header.Name = fileInfo.Name()
// 创建文件在 ZIP 中的写入器
writer, err := zipWriter.CreateHeader(header)
if err != nil {
return fmt.Errorf("创建文件写入器失败: %w", err)
}
// 将文件内容复制到 ZIP 文件中
_, err = io.Copy(writer, fileToCompress)
if err != nil {
return fmt.Errorf("将文件内容复制到 ZIP 失败: %w", err)
}
return nil
}
// 文件同级进行zip压缩
func FileToCompress(filePath string, data []model.OmcAlarm, delOrgFile bool) (string, error) {
defer os.Remove(filePath)
// 导出文件
err := CreateFile(filePath, data)
if err != nil {
return "", err
}
// 压缩文件
zipFfileNmae := filePath + ".zip"
err = FileToZip(zipFfileNmae, filePath)
if err != nil {
return "", err
}
return zipFfileNmae, nil
}
core/utils/strrand.go
-25
View File
@@ -1,25 +0,0 @@
package utils
import (
"math/rand"
"time"
)
// 生成编号 SeqNo 0-9A-Z
func SeqNo(length int) string {
items := []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "A", "B", "C", "D", "E", "F"}
// 创建基于时间的随机生成器
source := rand.NewSource(time.Now().UnixNano())
rng := rand.New(source)
// 生成32位长度的字符串值
result := ""
for i := 0; i < length; i++ {
randomIndex := rng.Intn(len(items))
randomItem := items[randomIndex]
result += randomItem
}
return result
}
go.mod
-18
View File
@@ -1,18 +0,0 @@
module omc
go 1.18
require (
github.com/aceld/zinx v1.1.21
github.com/google/uuid v1.3.0
golang.org/x/crypto v0.11.0
gorm.io/driver/mysql v1.5.1
gorm.io/gorm v1.25.1
)
require (
github.com/go-sql-driver/mysql v1.7.0 // indirect
github.com/gorilla/websocket v1.5.0 // indirect
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
)
go.sum
-40
View File
@@ -1,40 +0,0 @@
github.com/aceld/zinx v1.1.21 h1:8zoZ+hcEAd7gDsl8xOKPaWPEs9vZDRQOvhjG3vuvAnQ=
github.com/aceld/zinx v1.1.21/go.mod h1:nITkdASGtkLSwNKZ5yj88IpcCHTCFCP6cL12JWms1Fo=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=
github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gorm.io/driver/mysql v1.5.1 h1:WUEH5VF9obL/lTtzjmML/5e6VfFR/788coz2uaVCAZw=
gorm.io/driver/mysql v1.5.1/go.mod h1:Jo3Xu7mMhCyj8dlrb3WoCaRd1FhsVh+yMXb1jUInf5o=
gorm.io/gorm v1.25.1 h1:nsSALe5Pr+cM3V1qwwQ7rOkw+6UeLrX5O4v3llhHa64=
gorm.io/gorm v1.25.1/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
handle/api/close_conn_alarm.go
-22
View File
@@ -1,22 +0,0 @@
package api
import (
"omc/core"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/znet"
)
var CloseConnAlarmMsgID uint32 = 10
var CloseConnAlarmMsgType uint32 = 10
var CloseConnAlarmMsgName string = "closeConnAlarm"
// closeConnAlarm 关闭连接
type CloseConnAlarm struct {
znet.BaseRouter
}
func (s *CloseConnAlarm) Handle(request ziface.IRequest) {
request.GetConnection().Stop()
request.GetConnection().SendMsg(CloseConnAlarmMsgType, core.ResultSuccess(CloseConnAlarmMsgName, "", ""))
}
handle/api/req_cmca_login_alarm.go
-80
View File
@@ -1,80 +0,0 @@
package api
import (
"omc/core"
"omc/core/manage"
"omc/core/parse"
"omc/handle/service"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqCMCALoginAlarmMsgID uint32 = 11
var ReqCMCALoginAlarmMsgType uint32 = 11
var ReqCMCALoginAlarmMsgName string = "reqCMCALoginAlarm"
// reqCMCALoginAlarm CMCA认证方式登录
type ReqCMCALoginAlarm struct {
znet.BaseRouter
}
func (*ReqCMCALoginAlarm) Handle(request ziface.IRequest) {
// 登录消息处理
body, err := parse.RequestBodyDecode(request, []string{"user", "key", "cert", "type"})
username := body.Data["user"]
key := body.Data["key"]
cert := body.Data["cert"]
tp := body.Data["type"]
if err != nil || username == "" || cert == "" || key == "" || tp == "" {
zlog.Ins().ErrorF("inlaid message body %s", err.Error())
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
return
}
// 获取当前请求的通道
m := manage.GetManager(request.GetConnection().GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error")
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "server internal error", ""))
return
}
uid, err := request.GetConnection().GetProperty("UID")
if err != nil {
zlog.Ins().ErrorF("GetProperty UID error %s", err)
request.GetConnection().Stop()
return
}
// 得到连接实例的随机码
seqNo := m.GetUserByPID(uid.(string)).SeqNo
content := username + ":" + seqNo
// 检查判断是否对应随机码
if err := service.CMCALogin(username, cert, content, key); err != nil {
zlog.Ins().ErrorF("LoginFail %s", err)
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
// 已登录的，登录错误超过3次,断开连接
if uid != nil || uid != "" {
isClose, _ := m.LoginFail(uid.(string))
if isClose {
request.GetConnection().Stop()
return
}
}
return
}
// manager 用户登录更新
err = m.LoginSuccess(uid.(string), username, tp)
if err != nil {
zlog.Ins().ErrorF("manager:%s", err)
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
return
}
zlog.Ins().InfoF("user login loginSuccess,username:%s, type:%s, channel:%s", username, tp, request.GetConnection().GetName())
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultSuccess(ReqLoginAlarmMsgName, "null", ""))
}
handle/api/req_cmca_login_seq.go
-47
View File
@@ -1,47 +0,0 @@
package api
import (
"omc/core"
"omc/core/manage"
"omc/core/utils"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqCMCALoginSeqMsgID uint32 = 12
var ReqCMCALoginSeqMsgType uint32 = 13
var ReqCMCALoginSeqMsgName string = "ackCMCALoginSeq"
// reqCMCALoginSeq CMCA认证方式登录随机码
type ReqCMCALoginSeq struct {
znet.BaseRouter
}
// reqCMCALoginAlarm;user=omc;key=base64Key;cert=cer;type=msg"
func (s *ReqCMCALoginSeq) Handle(request ziface.IRequest) {
// 获取当前请求的通道
m := manage.GetManager(request.GetConnection().GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error")
request.GetConnection().SendMsg(ReqCMCALoginSeqMsgType, core.ResultError(ReqCMCALoginSeqMsgName, "server internal error", ""))
return
}
uid, err := request.GetConnection().GetProperty("UID")
if err != nil {
zlog.Ins().ErrorF("GetProperty UID error %s", err)
request.GetConnection().Stop()
return
}
// seqNo 32位长度 0-9A-Z
seqNo := utils.SeqNo(32)
m.SetSeqNo(uid.(string), seqNo)
// 发送随机码去sha256加密
msgData := map[string]string{
"seqNo": seqNo,
}
request.GetConnection().SendMsg(ReqCMCALoginSeqMsgType, core.Result(ReqCMCALoginSeqMsgName, msgData))
}
handle/api/req_heart_beat.go
-40
View File
@@ -1,40 +0,0 @@
package api
import (
"omc/core"
"omc/core/parse"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqHeartBeatMsgID uint32 = 8
var ReqHeartBeatMsgType uint32 = 9
var ReqHeartBeatName string = "ackHeartBeat"
// ReqHeartBeat 心跳
type ReqHeartBeat struct {
znet.BaseRouter
}
// reqHeartBeat;reqId=12
func (s *ReqHeartBeat) Handle(request ziface.IRequest) {
// 解包
body, err := parse.RequestBodyDecode(request, nil)
if err != nil {
zlog.Ins().ErrorF("inlaid message body %s", err.Error())
request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultError(ReqHeartBeatName, "inlaid message body", ""))
return
}
reqId, ok := body.Data["reqId"]
if !ok {
zlog.Ins().ErrorF("missing parameter of message body")
request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultError(ReqHeartBeatName, "missing reqId", ""))
return
}
//ack
request.GetConnection().SendMsg(ReqHeartBeatMsgType, core.ResultSuccess(ReqHeartBeatName, "", reqId))
}
handle/api/req_login_alarm.go
-76
View File
@@ -1,76 +0,0 @@
package api
import (
"omc/core"
"omc/core/manage"
"omc/core/parse"
"omc/handle/service"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqLoginAlarmMsgID uint32 = 1
var ReqLoginAlarmMsgType uint32 = 2
var ReqLoginAlarmMsgName string = "ackLoginAlarm"
// reqLoginAlarm 登录
type ReqLoginAlarm struct {
znet.BaseRouter
}
// reqLoginAlarm;user=omc;key=omc;type=msg
func (s *ReqLoginAlarm) Handle(request ziface.IRequest) {
// 登录消息处理
body, err := parse.RequestBodyDecode(request, []string{"user", "key", "type"})
// 账户密码
username := body.Data["user"]
key := body.Data["key"]
tp := body.Data["type"]
if err != nil || username == "" || key == "" {
zlog.Ins().ErrorF("inlaid message body %s", err.Error())
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "inlaid message body", ""))
return
}
// 获取当前请求的通道
m := manage.GetManager(request.GetConnection().GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error")
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, "server internal error", ""))
return
}
uid, err := request.GetConnection().GetProperty("UID")
if err != nil {
zlog.Ins().ErrorF("GetProperty UID error %s", err)
request.GetConnection().Stop()
return
}
// 登录信息
err = service.UserLogin(username, key)
if err != nil {
zlog.Ins().ErrorF("LoginFail %s", err)
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
// 已登录的，登录错误超过3次,断开连接
if uid != nil || uid != "" {
isClose, _ := m.LoginFail(uid.(string))
if isClose {
request.GetConnection().Stop()
return
}
}
return
}
// manager 用户登录更新
err = m.LoginSuccess(uid.(string), username, tp)
if err != nil {
zlog.Ins().ErrorF("manager:%s", err)
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultError(ReqLoginAlarmMsgName, err.Error(), ""))
return
}
zlog.Ins().InfoF("user login loginSuccess,username:%s, type:%s, channel:%s", username, tp, request.GetConnection().GetName())
request.GetConnection().SendMsg(ReqLoginAlarmMsgType, core.ResultSuccess(ReqLoginAlarmMsgName, "null", ""))
}
handle/api/req_sync_alarm_file.go
-113
View File
@@ -1,113 +0,0 @@
package api
import (
"fmt"
"omc/conf"
"omc/core"
"omc/core/manage"
"omc/core/parse"
"omc/core/utils"
"omc/handle/service"
"strconv"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqSyncAlarmFileID uint32 = 5
var ReqSyncAlarmFileType uint32 = 6
var ReqSyncAlarmFileName string = "ackSyncAlarmFile"
var ReqSyncAlarmFileResultType uint32 = 7
var ReqSyncAlarmFileResultName string = "ackSyncAlarmFileResult"
// reqSyncAlarmFile 文件方式同步告警请求
type ReqSyncAlarmFile struct {
znet.BaseRouter
}
func (s *ReqSyncAlarmFile) Handle(request ziface.IRequest) {
// 消息处理
checker := []string{"reqId"}
body, err := parse.RequestBodyDecode(request, checker)
reqId := body.Data["reqId"]
if err != nil || reqId == "" {
zlog.Ins().ErrorF("inlaid message body %s", err.Error())
request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, err.Error(), ""))
return
}
// 参数预设初始值 alarmSeq不一定有
alarmSeq, alarmSeqOk := body.Data["alarmSeq"]
if !alarmSeqOk || alarmSeq == "" {
fmt.Println("== ", alarmSeq)
alarmSeq = "1"
}
//管理模块
m := manage.GetManager(request.GetConnection().GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error")
request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "server internal error", reqId))
return
}
// 检查用户是否登录
u := m.GetUserByPID(body.UID)
if !u.LoginState {
zlog.Ins().ErrorF("no permissions ")
request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "no permissions", reqId))
return
}
// 同步的消息序号
seq, err := strconv.Atoi(alarmSeq)
if err != nil {
seq = 0
}
// check alarmSeq 是否存在
neBind, _ := parse.ConvertBindFlag(m.BindFlag)
alarm, err := service.ListForSaveFile(neBind.NeType, neBind.NeId, int64(seq), body.Data["startTime"], body.Data["endTime"], body.Data["syncSource"])
if err != nil {
zlog.Ins().ErrorF("no permissions ")
request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultError(ReqSyncAlarmFileName, "no permissions", reqId))
return
}
if len(alarm) > 0 {
data := service.ConvertOMCAlarm(alarm)
// 生成文件名
fileNmae := utils.FileName(&utils.FileNameMeta{
FTPRoot: conf.OmcConf.FTPRoot,
Province: m.Province,
DeviceCode: m.DeviceCode,
ReqId: reqId,
Suffix: ".txt",
})
// 打压缩
zipFfileNmae, err := utils.FileToCompress(fileNmae, data, true)
if err != nil {
zlog.Ins().ErrorF("ReqSyncAlarmFileResult FileToCompress %v", err)
request.GetConnection().SendMsg(ReqSyncAlarmFileResultType, core.Result(ReqSyncAlarmFileResultName, map[string]string{
"reqId": reqId,
"result": "fail",
"resDesc": err.Error(),
"fileName": "null",
}))
return
}
//发送文件同步信息
request.GetConnection().SendMsg(ReqSyncAlarmFileResultType, core.Result(ReqSyncAlarmFileResultName, map[string]string{
"reqId": reqId,
"result": "succ",
"resDesc": "null",
"fileName": zipFfileNmae,
}))
return
}
request.GetConnection().SendMsg(ReqSyncAlarmFileType, core.ResultSuccess(ReqSyncAlarmFileName, "null", reqId))
}
handle/api/req_sync_alarm_msg.go
-75
View File
@@ -1,75 +0,0 @@
package api
import (
"omc/core"
"omc/core/manage"
"omc/core/parse"
"omc/handle/service"
"strconv"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/znet"
)
var ReqSyncAlarmMsgID uint32 = 3
var ReqSyncAlarmMsgType uint32 = 4
var ReqSyncAlarmMsgName string = "ackSyncAlarmMsg"
// reqSyncAlarmMsg 消息同步
type ReqSyncAlarmMsg struct {
znet.BaseRouter
}
// reqSyncAlarmMsg
func (s *ReqSyncAlarmMsg) Handle(request ziface.IRequest) {
// 消息处理
checker := []string{"reqId", "alarmSeq"}
body, err := parse.RequestBodyDecode(request, checker)
reqId := body.Data["reqId"]
if err != nil || reqId == "" {
zlog.Ins().ErrorF("inlaid message body %s", err.Error())
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, err.Error(), ""))
return
}
//管理模块
m := manage.GetManager(request.GetConnection().GetName())
if m == nil {
zlog.Ins().ErrorF("server internal error")
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "server internal error", reqId))
return
}
// 检查用户是否登录
u := m.GetUserByPID(body.UID)
if !u.LoginState {
zlog.Ins().ErrorF("no permissions ")
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "no permissions", reqId))
return
}
alarmSeq, err := strconv.Atoi(body.Data["alarmSeq"])
if err != nil || alarmSeq < 1 {
zlog.Ins().ErrorF("invalid alarmSeq of message body")
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "invalid alarmSeq", reqId))
return
}
//check alarmSeq 是否存在
neBind, _ := parse.ConvertBindFlag(m.BindFlag)
alarmSeqCount := service.AlarmSeqCount(neBind.NeType, neBind.NeId, int64(alarmSeq))
if alarmSeqCount > 0 {
if alarmSeqCount > 1000 {
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "over 1000 rows", reqId))
} else {
// 发送更新上报的alarm seq
m.UpdateAlarmSeq(int64(alarmSeq))
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultSuccess(ReqSyncAlarmMsgName, "", reqId))
}
return
}
zlog.Ins().ErrorF("not has of alarmSeq %d", alarmSeq)
request.GetConnection().SendMsg(ReqSyncAlarmMsgType, core.ResultError(ReqSyncAlarmMsgName, "alarmSeq does not exist", reqId))
}
handle/model/alarm.go
-43
View File
@@ -1,43 +0,0 @@
package model
import "time"
// Alarm 实时告警上报信息
type Alarm struct {
ID int64 `json:"id" gorm:"id"`
AlarmSeq int64 `json:"alarm_seq" gorm:"alarm_seq"`
AlarmId string `json:"alarm_id" gorm:"alarm_id"`
AlarmTitle string `json:"alarm_title" gorm:"alarm_title"`
NeType string `json:"ne_type" gorm:"ne_type"`
NeId string `json:"ne_id" gorm:"ne_id"`
AlarmCode int64 `json:"alarm_code" gorm:"alarm_code"`
EventTime time.Time `json:"event_time" gorm:"event_time"`
AlarmType string `json:"alarm_type" gorm:"alarm_type"`
OrigSeverity string `json:"orig_severity" gorm:"orig_severity"` // 1: Critical, 2: Major, 3: Minor, 4: Warning, 5: Event(Only VNF)
PerceivedSeverity string `json:"perceived_severity" gorm:"perceived_severity"` // 1: Critical, 2: Major, 3: Minor, 4: Warning, 5: Event(Only VNF)
PvFlag string `json:"pv_flag" gorm:"pv_flag"`
NeName string `json:"ne_name" gorm:"ne_name"`
ObjectUid string `json:"object_uid" gorm:"object_uid"`
ObjectName string `json:"object_name" gorm:"object_name"`
ObjectType string `json:"object_type" gorm:"object_type"`
LocationInfo string `json:"location_info" gorm:"location_info"`
Province string `json:"province" gorm:"province"`
AlarmStatus int64 `json:"alarm_status" gorm:"alarm_status"`
SpecificProblem string `json:"specific_problem" gorm:"specific_problem"`
SpecificProblemId string `json:"specific_problem_id" gorm:"specific_problem_id"`
AddInfo string `json:"add_info" gorm:"add_info"`
Counter int64 `json:"counter" gorm:"counter"`
LatestEventTime time.Time `json:"latest_event_time" gorm:"latest_event_time"`
AckState int8 `json:"ack_state" gorm:"ack_state"` // 0: Unacked, 1: Acked
AckTime time.Time `json:"ack_time" gorm:"ack_time"`
AckUser string `json:"ack_user" gorm:"ack_user"`
ClearType int8 `json:"clear_type" gorm:"clear_type"` // 0: Unclear, 1: AutoClear, 2: ManualClear
ClearTime time.Time `json:"clear_time" gorm:"clear_time"`
ClearUser string `json:"clear_user" gorm:"clear_user"`
Timestamp time.Time `json:"timestamp" gorm:"timestamp"`
}
// TableName 表名称
func (*Alarm) TableName() string {
return "alarm"
}
handle/model/nbi_alarm_log.go
-24
View File
@@ -1,24 +0,0 @@
package model
import "time"
// NbiAlarmLog undefined
type NbiAlarmLog struct {
ID int64 `json:"id" gorm:"id"`
OpUser string `json:"op_user" gorm:"op_user"`
SrcIp string `json:"src_ip" gorm:"src_ip"`
NeType string `json:"ne_type" gorm:"ne_type"`
NeId string `json:"ne_id" gorm:"ne_id"`
AlarmSeq int64 `json:"alarm_seq" gorm:"alarm_seq"`
AlarmId string `json:"alarm_id" gorm:"alarm_id"`
AlarmCode int64 `json:"alarm_code" gorm:"alarm_code"`
AlarmStatus int64 `json:"alarm_status" gorm:"alarm_status"` // 0/1: 清除告警/活动告警
EventTime string `json:"event_time" gorm:"event_time"`
LogTime time.Time `json:"log_time" gorm:"log_time"`
AId int64 `json:"a_id" gorm:"a_id"`
}
// TableName 表名称
func (*NbiAlarmLog) TableName() string {
return "nbi_alarm_log"
}
handle/model/omc_alarm.go
-25
View File
@@ -1,25 +0,0 @@
package model
// OmcAlarm 告警上报的结构体信息
type OmcAlarm struct {
AlarmSeq int64 `json:"alarmSeq"` //告警序列号
AlarmTitle string `json:"alarmTitle"` //告警事件标题
AlarmStatus int64 `json:"alarmStatus"` //告警状态
AlarmType string `json:"alarmType"` //告警类型
OrigSeverity int64 `json:"origSeverity"` //原始级别
EventTime string `json:"eventTime"` //事件发生时间
AlarmId string `json:"alarmId"` //告警事件唯一标识
SpecificProblemID string `json:"specificProblemID"` //告警问题原因ID
SpecificProblem string `json:"specificProblem"` //告警问题原因
NeUID string `json:"neUID"` //告警网元UID
NeName string `json:"neName"` //告警网元名称
NeType string `json:"neType"` //告警网元设备类型
ObjectUID string `json:"objectUID"` //告警定位对象UID
ObjectName string `json:"objectName"` //告警定位对象名称
ObjectType string `json:"objectType"` //告警定位对象资源类型
LocationInfo string `json:"locationInfo"` //告警定位信息
AddInfo string `json:"addInfo"` //告警辅助信息[条件必选]
PVFlag string `json:"PVFlag"` //网元虚实性[条件必选]
Province string `json:"province"` //网元服务省份
Unit string `json:"unit"` //网元服务省份
}
handle/model/user.go
-41
View File
@@ -1,41 +0,0 @@
package model
import "time"
// User 用户表实体信息
type User struct {
ID int64 `json:"id" gorm:"id"`
AccountId string `json:"account_id" gorm:"account_id"`
Name string `json:"name" gorm:"name"`
RealName string `json:"real_name" gorm:"real_name"`
Sn string `json:"sn" gorm:"sn"`
Gender string `json:"gender" gorm:"gender"`
Email string `json:"email" gorm:"email"`
IdCardNumber string `json:"id_card_number" gorm:"id_card_number"`
Description string `json:"description" gorm:"description"`
TelephoneNumber string `json:"telephone_number" gorm:"telephone_number"`
Phone string `json:"phone" gorm:"phone"`
Mobile string `json:"mobile" gorm:"mobile"`
EmployeeNumber string `json:"employee_number" gorm:"employee_number"`
EmployeeType string `json:"employee_type" gorm:"employee_type"`
Organize string `json:"organize" gorm:"organize"`
SupporterCorpName string `json:"supporter_corp_name" gorm:"supporter_corp_name"`
StartTime string `json:"start_time" gorm:"start_time"`
EndTime string `json:"end_time" gorm:"end_time"`
Password string `json:"password" gorm:"password"`
PasswordSha512 string `json:"password_sha512" gorm:"password_sha512"`
ChangePasswordFlag int8 `json:"change_password_flag" gorm:"change_password_flag"`
PasswordExpiration time.Time `json:"password_expiration" gorm:"password_expiration"`
Status string `json:"status" gorm:"status"`
UserExpiration time.Time `json:"user_expiration" gorm:"user_expiration"`
GroupName string `json:"group_name" gorm:"group_name"`
Profile string `json:"profile" gorm:"profile"`
CreateTime time.Time `json:"create_time" gorm:"create_time"`
UpdateTime time.Time `json:"update_time" gorm:"update_time"`
Unit string `json:"unit" gorm:"unit"`
}
// TableName 表名称
func (*User) TableName() string {
return "user"
}
handle/service/alarm.go
-191
View File
@@ -1,191 +0,0 @@
package service
import (
"errors"
"omc/core/db"
"omc/core/emun"
"omc/handle/model"
"time"
"github.com/aceld/zinx/zlog"
)
// AlarmSeqCount alarm_seq大于等于的记录数
func AlarmSeqCount(neType, neId string, alarmSeq int64) int64 {
var num int64
query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId)
if alarmSeq > 0 {
query = query.Where("alarm_seq >= ?", alarmSeq)
}
if err := query.Order("alarm_seq desc").Count(&num).Error; err != nil {
zlog.Ins().ErrorF("HasAlarmSeq db error %s", err)
return num
}
return num
}
// List 获取告警信息 alarm_seq大于等于的
func List(neType, neId string, alarmSeq int64) ([]model.Alarm, error) {
// 从数据库读取告警记录
var alarms []model.Alarm
if err := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ? and alarm_seq >= ?", neType, neId, alarmSeq).Order("alarm_seq asc").Find(&alarms).Error; err != nil {
zlog.Ins().ErrorF("AlarmList db error %s", err)
return nil, err
}
return alarms, nil
}
// ConvertOMCAlarm 转换上报的告警信息
func ConvertOMCAlarm(alarms []model.Alarm) []model.OmcAlarm {
result := make([]model.OmcAlarm, 0)
for _, v := range alarms {
var item model.OmcAlarm
item.AlarmSeq = v.AlarmSeq
item.AlarmTitle = v.AlarmTitle
item.AlarmStatus = v.AlarmStatus
item.AlarmType = v.AlarmType
item.OrigSeverity = emun.OrigSeverity(v.OrigSeverity)
item.EventTime = v.EventTime.Format("2006-01-02 15:04:05")
item.AlarmId = v.AlarmId
item.SpecificProblemID = v.SpecificProblemId
item.SpecificProblem = v.SpecificProblem
item.NeUID = v.NeId
item.NeName = v.NeName
item.NeType = v.NeType
item.ObjectUID = v.ObjectUid
item.ObjectName = v.NeName
item.ObjectType = v.ObjectType
item.LocationInfo = v.LocationInfo
item.AddInfo = v.AddInfo
item.PVFlag = v.PvFlag
item.Province = v.Province
result = append(result, item)
}
return result
}
// 检查列表里最大的AlarmSeq
func MaxAlarmSeq(current int64, alarms []model.OmcAlarm) int64 {
var maxAlarmSeq = current
for _, v := range alarms {
if v.AlarmSeq > maxAlarmSeq {
maxAlarmSeq = v.AlarmSeq
}
}
return maxAlarmSeq
}
// LastAlarmSeq 获取最新的alarm_seq 序号
func LastAlarmSeq(neType, neId string) int64 {
var alarm model.Alarm
if err := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId).Order("alarm_seq desc").First(&alarm).Error; err != nil {
zlog.Ins().ErrorF("LastAlarmSeq db error %s", err)
return 0
}
return alarm.AlarmSeq
}
// List 获取告警信息为文件保存
func ListForSaveFile(neType, neId string, alarmSeq int64, startTime, endTime, syncSource string) ([]model.Alarm, error) {
// 基于OMC数据库的存量活动告警数据文件
if syncSource == "0" {
// 取告警发生的时间，即告警本身的event_time
return alarmByEventTime(neType, neId, startTime, endTime)
}
// 基于告警消息日志的告警消息流水数据文件
if syncSource == "1" {
if alarmSeq > 0 {
// 只适用于syncSource=1的情况 同步告警的起始告警消息序号
return alarmByAlarmSeq(neType, neId, alarmSeq)
} else {
// 取北向接口记录日志流水的时间
return alarmByLog(neType, neId, startTime, endTime)
}
}
return []model.Alarm{}, nil
}
// 取告警发生的时间，即告警本身的event_time
func alarmByEventTime(neType, neId, startTime, endTime string) ([]model.Alarm, error) {
var alarms []model.Alarm
if startTime == "" && endTime == "" {
return alarms, nil
}
query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ?", neType, neId)
// 从当前告警中最早的开始
if startTime != "" {
startT, err := time.Parse("2006-01-02 15:04:05", startTime)
if err != nil {
return nil, errors.New("startTime invalid")
}
query = query.Where("event_time >= ?", startT)
}
// 取到当前最新的告警截止
if endTime != "" {
endT, err := time.Parse("2006-01-02 15:04:05", endTime)
if err != nil {
return nil, errors.New("endTime invalid")
}
query = query.Where("event_time <= ?", endT)
}
// 查询
if err := query.Order("alarm_seq asc").Find(&alarms).Error; err != nil {
return nil, err
}
return alarms, nil
}
// 同步告警的起始告警消息序号
func alarmByAlarmSeq(neType, neId string, alarmSeq int64) ([]model.Alarm, error) {
var alarms []model.Alarm
// 查询
query := db.Client.Model(&model.Alarm{}).Where("ne_type = ? and ne_id = ? and alarm_seq >= ?", neType, neId, alarmSeq)
if err := query.Order("alarm_seq asc").Find(&alarms).Error; err != nil {
return nil, err
}
return alarms, nil
}
// 取北向接口记录日志流水的时间
func alarmByLog(neType, neId, startTime, endTime string) ([]model.Alarm, error) {
var alarms []model.Alarm
if startTime == "" && endTime == "" {
return alarms, nil
}
var aIDs []int64
query := db.Client.Model(&model.NbiAlarmLog{}).Select("distinct a_id").Where("ne_type = ? and ne_id = ?", neType, neId)
// 从当前告警中最早的开始
if startTime != "" {
startT, err := time.Parse("2006-01-02 15:04:05", startTime)
if err != nil {
return nil, errors.New("startTime invalid")
}
query = query.Where("log_time >= ?", startT)
}
// 取到当前最新的告警截止
if endTime != "" {
endT, err := time.Parse("2006-01-02 15:04:05", endTime)
if err != nil {
return nil, errors.New("endTime invalid")
}
query = query.Where("log_time <= ?", endT)
}
// 查询记录日志流水的id组
if err := query.Order("alarm_seq asc").Find(&aIDs).Error; err != nil {
return nil, err
}
// 查询告警本身的记录
if err := db.Client.Model(&model.Alarm{}).Where("id in (?)", aIDs).Find(&alarms).Error; err != nil {
return nil, err
}
return alarms, nil
}
handle/service/login.go
-80
View File
@@ -1,80 +0,0 @@
package service
import (
"encoding/base64"
"errors"
"fmt"
"omc/core/db"
"omc/core/utils"
"omc/handle/model"
"time"
"github.com/aceld/zinx/zlog"
)
// UserLogin 用户登录
func UserLogin(name, passwd string) error {
// 用户名密码校验
var user model.User
err := db.Client.Model(&model.User{}).Where("account_id=?", name).First(&user).Error
if err != nil {
zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
return errors.New("username-error")
}
if err := utils.Compare(user.Password, passwd); err != nil {
zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
return errors.New("incorrect username and password")
}
return nil
}
// CMCALogin 验证随机码
// content "user:seqNo" 组合的字符
// signatureBase64 签名数据（base64编码）
func CMCALogin(name, carCertificate, content, signatureBase64 string) error {
// 用户名密码校验
var user model.User
err := db.Client.Model(&model.User{}).Where("account_id=?", name).First(&user).Error
if err != nil {
zlog.Ins().ErrorF("Password Login[%s]:%s", name, err)
return errors.New("username-error")
}
if user.AccountId != name {
return fmt.Errorf("username-error")
}
// 解析证书信息
certificate, err := utils.ParseCert(carCertificate)
if err != nil {
zlog.Ins().ErrorF("ParseCert Login[%s]:%s", name, err)
return fmt.Errorf("certificate-error")
}
// 判断证书到期时间
if time.Now().After(certificate.NotAfter) {
zlog.Ins().ErrorF("certificate Not After Now time Login[%s]:%s", name, err)
return errors.New("the certificate has expired")
}
// 判断证书持有人
if certificate.Subject.CommonName != name {
zlog.Ins().ErrorF("certificate Subject CommonName Login[%s]:%s", name, err)
return errors.New("the certificate cn-error")
}
// 解码base64得到签名字节串
signature, err := base64.StdEncoding.DecodeString(signatureBase64)
if err != nil {
zlog.Ins().ErrorF("signatureBase64 Login[%s]:%s", name, err)
return fmt.Errorf("the certificate signature-error")
}
// 验证签名
valid := utils.VerifySignature(certificate.PublicKey, []byte(content), signature)
if valid {
return nil
}
return errors.New("seqNo-error")
}
nbi_alarm.go
-45
View File
@@ -1,45 +0,0 @@
package main
import (
"fmt"
"github.com/aceld/zinx/zutils/commandline/args"
"omc/conf"
"omc/core/db"
"omc/core/manage"
"omc/router"
"os"
"os/signal"
"github.com/aceld/zinx/zlog"
)
// go run nbi_alarm.go -c conf\nbi_alarm.json
func main() {
// 配置初始化
conf.Init(args.Args.ConfigFile)
// 初始数据库
db.Init()
//创建服务器句柄
for _, cg := range conf.OmcConf.Channel {
serverName := fmt.Sprintf("%s:%d", cg.BindFlag, cg.TCPPort)
//注册用户管理模块
m := manage.NewManager(serverName, cg.BindFlag, cg.Province, cg.DeviceCode)
//启动服务
s := router.LoadServer(cg)
go s.Serve()
//启动实时告警
go m.RealTimeAlarm()
}
// close
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt, os.Kill)
sig := <-c
zlog.Ins().InfoF("===exit=== %s", sig)
}
nbi_alarm_test.go
-188
View File
@@ -1,188 +0,0 @@
package main
import (
"bytes"
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
"io"
"net"
"testing"
"time"
"github.com/aceld/zinx/zconf"
)
type Message struct {
StartSign uint16
MsgType uint8
TimeStamp uint32
LenOfBody uint16
Value []byte
}
type TcpClient struct {
conn net.Conn
PID int32
isOnline chan bool
}
func (s *TcpClient) Unpack(headdata []byte) (head *Message, err error) {
fmt.Println("unpack:", hex.EncodeToString(headdata))
headBuf := bytes.NewReader(headdata)
head = &Message{}
// Read the startSign
if err := binary.Read(headBuf, binary.BigEndian, &head.StartSign); err != nil {
return nil, err
}
// Read the msgID
if err := binary.Read(headBuf, binary.BigEndian, &head.MsgType); err != nil {
return nil, err
}
// read timeStamp
if err := binary.Read(headBuf, binary.BigEndian, &head.TimeStamp); err != nil {
return nil, err
}
// Read the data length
if err := binary.Read(headBuf, binary.BigEndian, &head.LenOfBody); err != nil {
return nil, err
}
// Check whether the data length exceeds the maximum allowed packet size
// (判断dataLen的长度是否超出我们允许的最大包长度)
if zconf.GlobalObject.MaxPacketSize > 0 && uint32(head.LenOfBody) > zconf.GlobalObject.MaxPacketSize {
return nil, errors.New("too large msg data received")
}
return head, nil
}
func (s *TcpClient) Pack(msgID uint32, dataBytes []byte) (out []byte, err error) {
outbuff := bytes.NewBuffer([]byte{})
// Write the oxffff
if err := binary.Write(outbuff, binary.BigEndian, uint16(0xffff)); err != nil {
return nil, err
}
//Write the type
if err := binary.Write(outbuff, binary.BigEndian, uint8(msgID)); err != nil {
return nil, err
}
//Write the timestamp
if err := binary.Write(outbuff, binary.BigEndian, uint32(0x999999)); err != nil {
return nil, err
}
//Write the length
if err := binary.Write(outbuff, binary.BigEndian, uint16(len(dataBytes))); err != nil {
return nil, err
}
// Write the data
if err := binary.Write(outbuff, binary.BigEndian, dataBytes); err != nil {
return nil, err
}
out = outbuff.Bytes()
return
}
func (s *TcpClient) SendMsg(msgID uint32, data []byte) {
sendData, err := s.Pack(msgID, data)
if err == nil {
_, err := s.conn.Write(sendData)
fmt.Println("send msg:", hex.EncodeToString(sendData), " err:", err)
} else {
fmt.Println(err)
}
return
}
func (s *TcpClient) Receive() {
for {
//读取服务端发来的数据 ==》 SyncPID
//1.读取8字节
////第一次读取，读取数据头
headData := make([]byte, 9)
if _, err := io.ReadFull(s.conn, headData); err != nil {
fmt.Println(err)
return
}
pkgHead, err := s.Unpack(headData)
if err != nil {
fmt.Println("Unpack", err)
return
}
//data
if pkgHead.LenOfBody > 0 {
pkgHead.Value = make([]byte, pkgHead.LenOfBody)
if _, err := io.ReadFull(s.conn, pkgHead.Value); err != nil {
return
}
}
//处理服务器回执业务
fmt.Println("=================================")
fmt.Println("StartSign:", pkgHead.StartSign)
fmt.Println("MsgType:", pkgHead.MsgType)
fmt.Println("TimeStamp:", pkgHead.TimeStamp)
fmt.Println("LenOfBody:", pkgHead.LenOfBody)
fmt.Println("Value:", string(pkgHead.Value))
fmt.Println("=================================")
time.Sleep(10 * time.Microsecond)
}
}
func (s *TcpClient) Start() {
//登录
// data := "reqLoginAlarm;user=audit;key=omc@password;type=ftp"
// s.SendMsg(1, []byte(data))
//发送同步告警信息
//data = "reqSyncAlarmMsg;reqId=33;alarmSeq=1"
//this.SendMsg(omc.ReqSyncAlarmMsg, []byte(data))
//发送文件同步告警
//data = "reqSyncAlarmFile;reqId=35;alarmSeq=2000;syncSource=1"
//data = "reqSyncAlarmFile;reqId=33;startTime=2023-01-08 00:00:00;syncSource=0"
// data = "reqSyncAlarmFile;reqId=34;startTime=2023-01-08 16:07:00;endTime=2023-07-19 23:59:59;syncSource=1"
// 心跳
data := "reqHeartBeat;reqId=33"
s.SendMsg(8, []byte(data))
go s.Receive()
}
func NewTcpClient(ip string, port int) *TcpClient {
addrStr := fmt.Sprintf("%s:%d", ip, port)
conn, err := net.Dial("tcp", addrStr)
if err != nil {
fmt.Println("net.Dial err: ", err)
panic(err)
}
client := &TcpClient{
conn: conn,
PID: 0,
isOnline: make(chan bool),
}
fmt.Printf("conn: %+v. Connected to server...", conn)
return client
}
func TestFile(t *testing.T) {
client := NewTcpClient("127.0.0.1", 31232)
client.Start()
//防止进程退出，等待中断信号
select {}
}
readme.md
-43
View File
@@ -1,43 +0,0 @@
项目交付文件：
fpt://192.168.0.229/home/guodeng/bin
服务启动：
./nb_alarm_agent -c /home/guodeng/omc/conf/nb_alarm_agent.json
配置文件详情：
{
/* 通道配置项*/
"channel": [
{
"tcp_port": 31232, //通道TCP监听端口
"bind_flag": "SMF#SZ_01", //通道Bind网元（格式为 ne_type#ne_id）
"province": "BJ", //网元所在省份
"device_code": "0001" //网元主机编码
},
{
"tcp_port": 31233, //通道TCP监听端口
"bind_flag": "UDM#SZ_03", //通道Bind网元（格式为 ne_type#ne_id）
"province": "BJ", //网元所在省份
"device_code": "0002" //网元主机编码
}
],
// 数据库配置
"mysql": "root:1000omc@kp!@tcp(192.168.0.229:33066)/omc_db?charset=utf8mb4&parseTime=True&loc=Local",
// FTP服务器根目录
"ftp_root": "ftp:192.168.0.229/data/ftp",
//服务名称
"mame":"north agent",
// 心跳保活时间， 如果服务器检查client 在超过心跳时间没有数据（心跳数据或者业务数据）发送，则断开连接
"heartbeat_max": 180,
// 日志存放目录
"log_dir": "./omc_log",
// 日志文件名称
"log_file":"omc.log"
}
router/router.go
-65
View File
@@ -1,65 +0,0 @@
package router
import (
"fmt"
"omc/conf"
"omc/core/decoder"
"omc/core/dpack"
heartbeat "omc/core/heart_beat"
"omc/core/manage"
"omc/handle/api"
"time"
"github.com/aceld/zinx/zconf"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/znet"
)
func LoadServer(cg conf.Channel) ziface.IServer {
serverName := fmt.Sprintf("%s:%d", cg.BindFlag, cg.TCPPort)
//new 一个TCP服务
s := znet.NewUserConfServer(&zconf.Config{
TCPPort: cg.TCPPort,
Name: serverName,
Host: "0.0.0.0",
MaxConn: conf.OmcConf.MaxConn,
WorkerPoolSize: uint32(conf.OmcConf.WorkerPoolSize),
HeartbeatMax: conf.OmcConf.HeartbeatMax,
LogDir: conf.OmcConf.LogDir,
LogFile: conf.OmcConf.LogFile,
})
//注册客户端连接建立和丢失函数
s.SetOnConnStart(manage.OnConnectionAdd)
s.SetOnConnStop(manage.OnConnectionLost)
//添加LTV数据格式Decoder
s.SetDecoder(decoder.NewOmcDecoder())
//添加LTV数据格式的Pack封包Encoder
s.SetPacket(dpack.NewDataPack())
// (启动心跳检测)
s.StartHeartBeatWithOption(60*time.Second, &ziface.HeartBeatOption{
MakeMsg: heartbeat.MyHeartBeatMsg,
OnRemoteNotAlive: heartbeat.MyOnRemoteNotAlive,
Router: &api.ReqHeartBeat{},
HeadBeatMsgID: api.ReqHeartBeatMsgID,
})
// 设置默认的心跳发送函数
heart := s.GetHeartBeat()
heart.SetHeartbeatFunc(heartbeat.MyHeartBeat)
// 添加自定义拦截器 debug调试输出信息用
// s.AddInterceptor(&interceptor.GlobalInterceptor{})
//注册路由 ================
s.AddRouter(api.ReqLoginAlarmMsgID, &api.ReqLoginAlarm{})
s.AddRouter(api.ReqCMCALoginSeqMsgID, &api.ReqCMCALoginSeq{})
s.AddRouter(api.ReqCMCALoginAlarmMsgID, &api.ReqCMCALoginAlarm{})
s.AddRouter(api.ReqSyncAlarmMsgID, &api.ReqSyncAlarmMsg{})
s.AddRouter(api.ReqSyncAlarmFileID, &api.ReqSyncAlarmFile{})
s.AddRouter(api.CloseConnAlarmMsgID, &api.CloseConnAlarm{})
return s
}