update

app/schedule/creeperxts.js

@@ -7,7 +7,7 @@ class Creeper extends Subscription {
   // 更改执行时间
   static get schedule() {
     return {
-      cron: '0 6 10 * * ?', // 每天晚上23点执行任务
+      cron: '0 15 10 * * ?', // 每天晚上23点执行任务
       // // cron: '0 20 12 * * ?', // 每天晚上23点执行任务
       // interval: '60s', // 1分钟执行一次
       type: 'worker', // 指定所有的 worker 都需要执行
@@ -16,8 +16,8 @@ class Creeper extends Subscription {
 
   // subscribe 是真正定时任务执行时被运行的函数
   async subscribe() {
-    console.log('选调生');
-    await this.ctx.service.creeperxts.creeper();// 选调生
+    console.log('服务器用户监控');
+    await this.ctx.service.creeperxts.creeper(); // 服务器用户监控
   }
 }
 module.exports = Creeper;

app/schedule/creeperxtsqnode.js

@@ -9,14 +9,14 @@ class Creeper extends Subscription {
     return {
       // cron: '0 6 30 * * ?', // 每天晚上23点执行任务
       // // cron: '0 20 12 * * ?', // 每天晚上23点执行任务
-      interval: '60s', // 1分钟执行一次
+      interval: '5m', // 1分钟执行一次
       type: 'worker', // 指定所有的 worker 都需要执行
     };
   }
 
   // subscribe 是真正定时任务执行时被运行的函数
   async subscribe() {
-    console.log('节点查询');
+    console.log('服务器集群节点查询');
     await this.ctx.service.creeperxtsqnode.creeper();// qnodes
   }
 }

app/schedule/creeperxtsread.js

@@ -7,7 +7,7 @@ class Creeper extends Subscription {
   // 更改执行时间
   static get schedule() {
     return {
-      cron: '0 7 10 * * ?', // 每天晚上23点执行任务
+      cron: '0 20 10 * * ?', // 每天晚上23点执行任务
       // // cron: '0 20 12 * * ?', // 每天晚上23点执行任务
       // interval: '60s', // 1分钟执行一次
       type: 'worker', // 指定所有的 worker 都需要执行
@@ -17,7 +17,7 @@ class Creeper extends Subscription {
   // subscribe 是真正定时任务执行时被运行的函数
   async subscribe() {
     console.log('读取文件');
-    await this.ctx.service.creeperxtsread.creeper();// 读取top文件
+    await this.ctx.service.creeperxtsread.creeper();// 读取文件
   }
 }
 module.exports = Creeper;

app/schedule/noderead.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const Subscription = require('egg').Subscription;
+
+class Creeper extends Subscription {
+  // 通过 schedule 属性来设置定时任务的执行间隔等配置
+  // 更改执行时间
+  static get schedule() {
+    return {
+      // cron: '0 6 30 * * ?', // 每天晚上23点执行任务
+      // // cron: '0 20 12 * * ?', // 每天晚上23点执行任务
+      interval: '8m', // 1分钟执行一次
+      type: 'worker', // 指定所有的 worker 都需要执行
+    };
+  }
+
+  // subscribe 是真正定时任务执行时被运行的函数
+  async subscribe() {
+    console.log('服务器集群节点查询');
+    await this.ctx.service.creeperxtsread.qnodes(); // qnodes
+  }
+}
+module.exports = Creeper;

app/service/creeperxts.js

@@ -15,16 +15,19 @@ class CreeperxtsService extends CrudService {
   }
 
   async sshcon() {
-    const nowtime = moment().locale('zh-cn').format('YYYY-MM-DD');
+    const nowdate = moment().locale('zh-cn').format('YYYY-MM-DD');
     const conn = new Client();
+    const pathDir = this.app.config.dataDir;
+    if (!fs.existsSync(pathDir)) {
+      fs.mkdirSync(pathDir);
+    }
     conn.on('ready', function() {
       conn.exec('top -bcn 1', function(err, stream) {
         if (err) throw err;
         stream.on('close', function(code, signal) {
           conn.end();
         }).on('data', function(data) {
-          console.log('STDOUT: ' + data);
-          const path_ = 'app/public/data/top' + nowtime + '.txt';
+          const path_ = `${pathDir}/top${nowdate}.txt`;
           fs.access(path_, err => {
             if (err) {
               fs.writeFile(path_, data.toString(), function(err) {

app/service/creeperxtsqnode.js

@@ -1,9 +1,8 @@
 /* eslint-disable strict */
 const _ = require('lodash');
 // const fs = require('fs');
-const Client = require('ssh2').Client;
+const SSH2Shell = require('ssh2shell');
 const fs = require('fs');
-const moment = require('moment');
 const { CrudService } = require('naf-framework-mongoose/lib/service');
 class CreeperxtsService extends CrudService {
 
@@ -15,38 +14,45 @@ class CreeperxtsService extends CrudService {
   }
 
   async sshcon() {
-    const nowtime = moment().locale('zh-cn').format('YYYY-MM-DD');
-    const conn = new Client();
-    conn.on('ready', function() {
-      conn.exec('cd /home; qnodes', function(err, stream) {
-        if (err) throw err;
-        stream.on('close', function(code, signal) {
-          conn.end();
-        }).on('data', function(data) {
-          console.log('STDOUT: ' + data);
-          const path_ = 'app/public/data/qnodes' + nowtime + '.txt';
-          fs.access(path_, err => {
-            if (err) {
-              fs.writeFile(path_, data.toString(), function(err) {
-                if (err) throw err;
-              });
-            } else {
-              fs.appendFile(path_, data.toString(), function(err) {
-                if (err) throw err;
-              });
-            }
-          });
-        }).stderr.on('data', function(data) {
-          console.log('STDERR: ' + data);
-        });
-      });
-    }).connect({
+    const config = {
       host: '192.168.10.102',
       port: 42973,
-      username: 'free_ly',
+      userName: 'free_ly',
       password: '1qaz2wsx',
-      // privateKey: require('fs').readFileSync('/home/admin/.ssh/id_dsa')
-    });
+    };
+    const pathDir = this.app.config.dataDir;
+    if (!fs.existsSync(pathDir)) {
+      fs.mkdirSync(pathDir);
+    }
+    const host = {
+      server: config,
+      commands: [ 'cd /home', 'qnodes' ],
+      onCommandComplete: (command, response, sshObj) => {
+        // console.group('complete');
+        // console.log(`command=>${command}`);
+        // console.log(`response=>${response}`);
+        // console.log(`sshObj=>${sshObj}`);
+        // console.groupEnd();
+      },
+      onEnd: (sessionText, sshObj) => {
+        // console.log(`sessionText=>${sessionText}`);
+        // console.log(`sshObj=>${JSON.stringify(sshObj)}`);
+        const path_ = `${pathDir}/qnodes.txt`;
+        fs.access(path_, err => {
+          if (err) {
+            fs.writeFile(path_, sessionText, function(err) {
+              if (err) throw err;
+            });
+          } else {
+            fs.appendFile(path_, sessionText, function(err) {
+              if (err) throw err;
+            });
+          }
+        });
+      },
+    };
+    const SSH = new SSH2Shell(host);
+    SSH.connect();
   }
 
 }

app/service/creeperxtsread.js

@@ -4,6 +4,7 @@ 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() {
@@ -15,72 +16,89 @@ class CreeperxtsreadService extends CrudService {
   }
 
   async top() {
-    const nowtime = moment().locale('zh-cn').format('YYYY-MM-DD');
-    const pathread = '../public/data/top' + nowtime + '.txt';
+    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) {
-      let index = 0;
       for (const rl of res.lineList) {
-        if (index > 6) {
-          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];
+        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);
         }
-        index = index + 1;
       }
     }
   }
 
   async qnodes() {
-    const nowtime = moment().locale('zh-cn').format('YYYY-MM-DD');
-    const pathread = '../public/data/qnodes' + nowtime + '.txt';
-    const res = await this.readFileToArr(pathread);
-    if (res) {
-      console.log(res.lineList);
-      // let index = 0;
-      // for (const rl of res.lineList) {
-      //   if (index > 6) {
-      //     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];
-      //     await this.ctx.model.Top.create(newdata);
-      //   }
-      //   index = index + 1;
-      // }
+    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: 50, reverse: false };
+    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();
-    console.log(lineObj);
     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;

config/config.default.js

@@ -3,6 +3,7 @@
 'use strict';
 
 const { jwt } = require('./config.secret');
+const { sep } = require('path');
 
 /**
  * @param {Egg.EggAppInfo} appInfo app info
@@ -12,7 +13,7 @@ module.exports = appInfo => {
    * built-in config
    * @type {Egg.EggAppConfig}
    **/
-  const config = exports = {};
+  const config = (exports = {});
 
   // use for cookie sign key, should change to your own and keep security
   config.keys = appInfo.name + '_1590486343297_6167';
@@ -31,7 +32,19 @@ module.exports = appInfo => {
       port: 3001,
     },
   };
+  // mq配置
+  config.amqp = {
+    client: {
+      hostname: '127.0.0.1',
+      username: 'visit',
+      password: 'visit',
+      vhost: 'train',
+    },
+    app: true,
+    agent: true,
+  };
 
+  config.dataDir = `${appInfo.baseDir}${sep}data`;
   // mongoose config
   config.mongoose = {
     url: 'mongodb://localhost:27017/count',

config/plugin.js

@@ -1,9 +1,5 @@
 'use strict';
-
-/** @type Egg.EggPlugin */
-module.exports = {
-  // had enabled by egg
-  // static: {
-  //   enable: true,
-  // }
+exports.amqp = {
+  enable: true,
+  package: 'egg-naf-amqp',
 };

package.json

@@ -12,9 +12,11 @@
     "jsonwebtoken": "^8.5.1",
     "lodash": "^4.17.15",
     "naf-framework-mongoose": "^0.6.12",
-    "ssh2": "^0.8.9",
     "query-lines-reader": "^0.2.3",
-    "readline": "^1.3.0"
+    "readline": "^1.3.0",
+    "ssh2": "^0.8.9",
+    "ssh2shell": "^1.9.4",
+    "egg-naf-amqp": "0.0.13"
   },
   "devDependencies": {
     "autod": "^3.0.1",