service-count/app/service/creeperxtsread.js

/* eslint-disable strict */
const _ = require('lodash');
const path = require('path');
const moment = require('moment');
const QueryLinesReader = require('query-lines-reader');
const { CrudService } = require('naf-framework-mongoose/lib/service');
const fs = require('fs');
class CreeperxtsreadService extends CrudService {

  async creeper() {
    // 链接linux服务器
    console.log('进入读取取回的文本文件');
    // await this.sshcon();
    await this.top();
    await this.qnodes();
  }

  async top() {
    const nowdate = moment().locale('zh-cn').format('YYYY-MM-DD');
    const pathDir = this.app.config.dataDir;
    const pathread = `${pathDir}/top${nowdate}.txt`;
    const res = await this.readFileToArr(pathread);
    if (res) {
      for (const rl of res.lineList) {
        const strsplit = rl.split(' ');
        const strs = _.pull(strsplit, '');
        const newdata = {};
        newdata.pid = strs[0];
        newdata.user = strs[1];
        newdata.pr = strs[2];
        newdata.ni = strs[3];
        newdata.virt = strs[4];
        newdata.res = strs[5];
        newdata.shr = strs[6];
        newdata.s = strs[7];
        newdata.cpu = strs[8];
        newdata.mem = strs[9];
        newdata.time = strs[10];
        newdata.command = strs[11];
        if (newdata.pid && newdata.pid !== '0' && newdata.user && newdata.user !== '0') {
          await this.ctx.model.Top.create(newdata);
        }
      }
    }
  }

  async qnodes() {
    const nowdate = moment().locale('zh-cn').format('YYYY-MM-DD HH:mm:ss');
    const pathDir = this.app.config.dataDir;
    const pathread = `${pathDir}/qnodes.txt`;
    let { lineList } = await this.readFileToArr(pathread);
    // res,去掉前两行和最后一行
    lineList = _.drop(lineList, 2);
    lineList = _.dropRight(lineList, 1);
    const arr = [];
    if (_.isArray(lineList)) {
      const match = /^(gpu|cu)(\d+)$/i;
      // 规律,当line符合 cuxx(数字)或gpuxx(数字) 时,之后的行数就是集群属性,直到 line === ''时截止,再扫到条件1的时候又是开始
      let obj = {};
      for (let line of lineList) {
        if (line.match(match)) {
          // 找到开始
          obj.node = _.trim(line);
        } else if (line === '') {
          // 找到结束,推进去,重置
          obj.time = nowdate;
          arr.push(_.cloneDeep(obj));
          obj = {};
        } else {
          line = _.trim(line);
          const arr = line.split(',');
          for (const i of arr) {
            const iarr = i.split('=');
            if (iarr && iarr.length >= 2) {
              obj[_.trim(iarr[0])] = _.trim(iarr[1]);
            }
          }
        }
      }
    }
    this.toMq(arr);
    fs.unlinkSync(`${pathread}`);
  }

  async readFileToArr(path_) {
    const options = { start: 0, end: 1, reverse: false };
    const toGetTotal = new QueryLinesReader(path.resolve(__dirname, path_), options);
    const total = (await toGetTotal.getTotal()) || 0;
    options.end = total;
    const queryLinesReader = new QueryLinesReader(path.resolve(__dirname, path_), options);
    const lineObj = await queryLinesReader.queryLines();
    return lineObj;
  }

  async toMq(data) {
    if (this.ctx.mq) {
      const exchange = 'service-count';
      const routerKey = 'nodes';
      const parm = { durable: true };
      await this.mq.fanout(exchange, routerKey, JSON.stringify(data), parm);
    }
  }
}
module.exports = CreeperxtsreadService;