free-sun
/
OperationalService
15
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

(*)修复启动控制台报错; kafka告警处理加异常类型字段

dev
wuqun 2 years ago
parent
9df3845913
commit
81598a0105
2 changed files with 104 additions and 100 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      api/app/lib/service/kafka.js
  2. 183
      api/app/lib/utils/helper.js

21
api/app/lib/service/kafka.js
View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
const moment = require('moment'); const moment = require('moment');
const Kafka = require('kafka-node'); const Kafka = require('kafka-node');
const { getAxyStructs } = require('../utils/helper'); //const { getAxyStructs } = require('../utils/helper');
module.exports = async function factory(app, opts) { module.exports = async function factory(app, opts) {
@ -11,7 +11,6 @@ module.exports = async function factory(app, opts) {
expireTime: null//10分钟更新一次结构物列表 expireTime: null//10分钟更新一次结构物列表
} }
//const { utils: { getAxyStructs } } = app.fs
const client = new Kafka.KafkaClient({ kafkaHost: opts.kafka.rootURL }); const client = new Kafka.KafkaClient({ kafkaHost: opts.kafka.rootURL });
const producer = new Kafka.HighLevelProducer(client); const producer = new Kafka.HighLevelProducer(client);
producer.on('error', function (err) { producer.on('error', function (err) {
@ -70,10 +69,11 @@ module.exports = async function factory(app, opts) {
}) })
async function getStructsAche(app) { async function getStructsAche() {
const { utils: { getAxyStructs } } = app.fs
try { try {
if (!structsAche.dataList.length || moment() > moment(structsAche.expireTime)) { if (!structsAche.dataList.length || moment() > moment(structsAche.expireTime)) {
let structList = await getAxyStructs(app); let structList = await getAxyStructs();
structsAche.dataList = structList; structsAche.dataList = structList;
structsAche.expireTime = moment().add(10, 'minute').format('YYYY-MM-DD HH:mm:ss'); structsAche.expireTime = moment().add(10, 'minute').format('YYYY-MM-DD HH:mm:ss');
} }
@ -85,22 +85,27 @@ module.exports = async function factory(app, opts) {
//保存告警[发现] //保存告警[发现]
async function savePullAlarm(msg) { async function savePullAlarm(msg) {
const { clickHouse } = app.fs
try { try {
let { messageMode, structureId, sourceName, alarmTypeCode, alarmCode, content, time } = msg; let { messageMode, structureId, sourceName, alarmTypeCode, alarmCode, content, time } = msg;
let structsAche = await getStructsAche(app); let structsAche = await getStructsAche();
if (structsAche) { if (structsAche) {
let structs = structsAche.dataList;//结构物列表 let structs = structsAche.dataList;//结构物列表
const { models } = app.fs.dc const { models } = app.fs.dc
let exist = structs.find(s => s.strucId == structureId); let exist = structs.find(s => s.strucId == structureId);
if (exist) { if (exist) {
let alarmType = await clickHouse.anxinyun.query(
`SELECT name FROM t_alarm_type WHERE code='${alarmTypeCode}'`).toPromise()
let type = alarmType.length ? alarmType[0].name : ''
let projects = exist.pomsProject.filter(d => !d.del).map(p => p.id); let projects = exist.pomsProject.filter(d => !d.del).map(p => p.id);
if (messageMode == 'AlarmGeneration') {//告警产生--------------------------------------------------1 if (messageMode == 'AlarmGeneration') {//告警产生--------------------------------------------------1
let datas = projects.map(d => {//需要 项目,告警源,异常类型,时间 let datas = projects.map(d => {//需要 项目,告警源,异常类型,时间
return { return {
projectCorrelationId: d, projectCorrelationId: d,
alarmInfo: { structureId, sourceName, alarmTypeCode, alarmCode, content }, alarmInfo: { sourceName, alarmTypeCode },
time: time, time: time,
type: 1//异常类型 TODO type//异常类型
} }
}) })
let rslt = await models.AlarmAppearRecord.bulkCreate(datas, { returning: true }); let rslt = await models.AlarmAppearRecord.bulkCreate(datas, { returning: true });
@ -118,7 +123,7 @@ module.exports = async function factory(app, opts) {
return { return {
pepUserId: null, pepUserId: null,
projectCorrelationId: d, projectCorrelationId: d,
alarmInfo: { sourceName, alarmTypeCode, alarmCode },//包含告警id,type,source TODO alarmInfo: { source: sourceName, type },//包含告警id,type,source
confirmTime: time, confirmTime: time,
confirmContent: '自动恢复' confirmContent: '自动恢复'
} }

183
api/app/lib/utils/helper.js
View File

@ -2,29 +2,30 @@
const moment = require('moment') const moment = require('moment')
async function getAxyStructs(app, pepProjectId) { module.exports = function (app, opts) {
const { models } = app.fs.dc const { models } = app.fs.dc
const { clickHouse } = app.fs const { clickHouse } = app.fs
const { database: anxinyun } = clickHouse.anxinyun.opts.config const { database: anxinyun } = clickHouse.anxinyun.opts.config
try { async function getAxyStructs(pepProjectId) {
try { try {
const { pepProjectRes, bindRes } = await pomsWithPepRangeParams(app, pepProjectId) try {
// 获取不重复的 安心云项目 id const { pepProjectRes, bindRes } = await pomsWithPepRangeParams(pepProjectId)
const anxinProjectIds = [ // 获取不重复的 安心云项目 id
...(bindRes).reduce( const anxinProjectIds = [
(arr, b) => { ...(bindRes).reduce(
for (let sid of b.anxinProjectId) { (arr, b) => {
arr.add(sid); for (let sid of b.anxinProjectId) {
} arr.add(sid);
return arr; }
}, return arr;
new Set() },
) new Set()
] )
// 查询安心云项目及结构物信息 ]
const undelStrucRes = anxinProjectIds.length ? // 查询安心云项目及结构物信息
await clickHouse.anxinyun.query( const undelStrucRes = anxinProjectIds.length ?
`SELECT await clickHouse.anxinyun.query(
`SELECT
t_project.id AS projectId, t_project.id AS projectId,
t_structure.id AS strucId, t_structure.id AS strucId,
t_structure.name AS strucName, t_structure.name AS strucName,
@ -56,81 +57,78 @@ async function getAxyStructs(app, pepProjectId) {
AND AND
t_project.id IN (${anxinProjectIds.join(',')})`).toPromise() : [] t_project.id IN (${anxinProjectIds.join(',')})`).toPromise() : []
// 构建安心云结构物和项企项目的关系 // 构建安心云结构物和项企项目的关系
// 并保存信息至数据 // 并保存信息至数据
let undelStruc = [] let undelStruc = []
for (let s of undelStrucRes) { for (let s of undelStrucRes) {
let corStruc = undelStruc.find(us => us.strucId == s.strucId) let corStruc = undelStruc.find(us => us.strucId == s.strucId)
if (corStruc) { if (corStruc) {
if (!corStruc.project.some(cp => cp.id == s.projectId)) { if (!corStruc.project.some(cp => cp.id == s.projectId)) {
corStruc.project.push({ corStruc.project.push({
id: s.projectId id: s.projectId
})
}
} else {
corStruc = {
strucId: s.strucId,
strucName: s.strucName,
projectId: s.projectId,
project: [{
id: s.projectId,
}],
pomsProject: []
}
undelStruc.push(corStruc)
}
for (let { dataValues: br } of bindRes) {
if (br.anxinProjectId.some(braId => braId == s.projectId)) {
let corPepProject = pepProjectRes.find(pp => pp.id == br.pepProjectId)
let corPomsProject = corStruc.pomsProject.find(cp => cp.id == br.id)
if (corPomsProject) {
// poms 的 project 和 pep 的 project 是一对一的关系 所以这个情况不用处理
} else {
corStruc.pomsProject.push({
...br,
pepProject: corPepProject
}) })
} }
} else {
corStruc = {
strucId: s.strucId,
strucName: s.strucName,
projectId: s.projectId,
project: [{
id: s.projectId,
}],
pomsProject: []
}
undelStruc.push(corStruc)
}
for (let { dataValues: br } of bindRes) {
if (br.anxinProjectId.some(braId => braId == s.projectId)) {
let corPepProject = pepProjectRes.find(pp => pp.id == br.pepProjectId)
let corPomsProject = corStruc.pomsProject.find(cp => cp.id == br.id)
if (corPomsProject) {
// poms 的 project 和 pep 的 project 是一对一的关系 所以这个情况不用处理
} else {
corStruc.pomsProject.push({
...br,
pepProject: corPepProject
})
}
}
} }
} }
return undelStruc
} catch (error) {
console.error(error);
} }
return undelStruc
} catch (error) { } catch (error) {
console.error(error); console.log(error)
} }
} catch (error) {
console.log(error)
} }
}
async function pomsWithPepRangeParams(app, pepProjectId) { async function pomsWithPepRangeParams(pepProjectId) {
const { models } = app.fs.dc try {
const { clickHouse } = app.fs let findOption = {
const { database: anxinyun } = clickHouse.anxinyun.opts.config where: {
try { del: false
let findOption = { }
where: {
del: false
} }
} if (pepProjectId) {
if (pepProjectId) { // 有 特定的项目id 就按此查询
// 有 特定的项目id 就按此查询 findOption.where.id = pepProjectId
findOption.where.id = pepProjectId
}
const bindRes = await models.ProjectCorrelation.findAll(findOption);
// 获取不重复的 项企项目id
let pepProjectIds = []
for (let b of bindRes) {
if (b.pepProjectId) {
pepProjectIds.push(b.pepProjectId)
} }
} const bindRes = await models.ProjectCorrelation.findAll(findOption);
// 查询项企项目的信息 // 获取不重复的 项企项目id
const pepProjectRes = pepProjectIds.length ? let pepProjectIds = []
await clickHouse.projectManage.query( for (let b of bindRes) {
`SELECT if (b.pepProjectId) {
pepProjectIds.push(b.pepProjectId)
}
}
// 查询项企项目的信息
const pepProjectRes = pepProjectIds.length ?
await clickHouse.projectManage.query(
`SELECT
t_pim_project.id AS id, t_pim_project.id AS id,
t_pim_project.project_name AS projectName, t_pim_project.project_name AS projectName,
t_pim_project.isdelete AS isdelete, t_pim_project.isdelete AS isdelete,
@ -144,17 +142,18 @@ async function pomsWithPepRangeParams(app, pepProjectId) {
ON t_pim_project_construction.construction_status_id = t_pim_project_state.id ON t_pim_project_construction.construction_status_id = t_pim_project_state.id
WHERE WHERE
id IN (${pepProjectIds.join(',')})` id IN (${pepProjectIds.join(',')})`
).toPromise() : []; ).toPromise() : [];
return { return {
pepProjectRes, bindRes pepProjectRes, bindRes
} }
} catch (error) { } catch (error) {
console.error(error); console.error(error);
}
} }
}
module.exports = { return {
getAxyStructs getAxyStructs
}; }
}
Write Preview
Loading…
Cancel
Save