Authored by htoooth

init

{
"extends": "yoho",
"parserOptions": {
"sourceType": "module",
"ecmaVersion": 2017
}
}
\ No newline at end of file
... ...
node_modules/
package-lock.json
coverage/
.nyc_output/
... ...
# influx-batch-sender
buffer message and send them in a bulk.
## useage
``` javascript
const Sender = require('../index');
const sender = new Sender({
host: 'influxd.yoho.cn',
db: 'web-apm',
measurement: 'api-duration',
duration: 2000, // 多长时间发送一次,默认值是 2000ms
records: 200, // 累积多少条消息发送一次,默认值是100
immediate: true // 是否立刻发送消息,设置为 true 会忽略 records 设置,默认值是 false
path: "/url", // 重新设置发送消息的路径,默认值是 /write
});
setInterval(() => {
sender.addMessage({
tags: {
reqid: 'sdasda'+Math.random(),
route: 'test'
},
fields: {
dasd: "d123asda"
}
});
}, Math.random() * 100)
```
\ No newline at end of file
... ...
/**
* test addmessages
*/
const Sender = require('../index');
const sender = new Sender({
host: 'influxd.yoho.cn',
db: 'web-apm',
measurement: 'api-duration',
duration: 2000,
records: 200
});
let num = 1;
let num2 = 1;
setInterval(() => {
if (num > 10000) {
console.log('1 done!');
return;
}
sender.addMessage({
tags: {
reqid: num + '',
route: 'test'
},
fields: {
dasd: 'd123asda'
}
});
num++;
}, Math.random() * 10);
setInterval(() => {
if (num2 > 10000) {
console.log('2 done!');
return;
}
sender.addMessage({
tags: {
reqid: num2 + '',
route: 'test2'
},
fields: {
dasd: '12131'
}
});
num2++;
}, Math.random() * 10);
\ No newline at end of file
... ...
const debug = require('debug')('apm-agent');
const EventEmitter = require('events');
const request = require('request');
/**
* buffer message and send them in a bulk
*
* const options =
* {
* host:'xxx.xxx.xxx',
* port:80,
* db:'xxxxx',
* measurement:'api-duration'
* duration:2000 //ms per 200 send message,
* records:100 //when message over 100 send them
* immediate: true // send message immediate not wart records count if true
* path: "/url", // set target path default is /write
* }
*/
class Sender extends EventEmitter {
constructor(options) {
super();
this.bulks = [];
this.batchMessages = [];
this.options = options;
this.options.duration = this.options.duration || 2000;
this.options.records = this.options.records || 100;
this.options.path = this.options.path || '/write';
this.options.port = this.options.port || 80;
if (!this.options.db || !this.options.host) {
console.error('config error: db or host undefined!');
return;
}
if (this.options.immediate) {
return;
}
// pre duration send array
setInterval(() => {
this._send();
debug('batchMessages duration sent!');
}, this.options.duration);
}
/**
* add message to batchMessage
*
* @example
* {
* measurement:'test',
* tags:{
* type:'api',
* preqID:'MdHy21313',
* api:'app.brand.newBrandList',
* route:'/sada/dsa/test'
* },
* fields:{
* duration:123
* }
* }
* @param {Object} message
*/
addMessage(message) {
message.time = message.time || new Date().getTime() * 1000000;
debug('add message! %O', message);
this.batchMessages.push(message);
if ((this.batchMessages.length > this.options.records) || this.options.immediate) {
this._send();
debug('batchMessages over records and sent!');
}
}
/**
* use tcp send message
* @private
*/
_send() {
if (!this.options.db || !this.options.host) {
return;
}
const len = this.batchMessages.length;
if (len < 1) {
debug('batchMessages is empty!');
return;
}
const bulk = this.batchMessages.splice(0, len);
debug('send bulkMessages! %O', bulk);
const options = {
headers: {
'content-type': 'application/json'
},
url: `http://${this.options.host}:${this.options.port}${this.options.path}`,
qs: {
db: this.options.db
},
method: 'POST',
body: JSON.stringify(bulk)
};
request(options, (error, res, body) => {
if (error) {
debug('send error: %O', error);
console.error(error);
this.emit('sendError', error);
return;
}
debug('status code: %o', res.statusCode);
if (res.statusCode !== 204) {
console.error(`Send failed! statusCode:${res.statusCode}`);
console.error(body);
this.emit('failed', res.statusCode);
} else {
this.emit('ok', res.statusCode);
}
});
}
}
module.exports = Sender;
\ No newline at end of file
... ...
{
"name": "yoho-apm-agent",
"version": "0.1.12",
"main": "index.js",
"scripts": {
"test": "node_modules/.bin/nyc node_modules/.bin/ava",
"posttest": "node_modules/.bin/nyc report --reporter=html",
"lint": "node node_modules/eslint\bin/eslint ./*.js ./example/*.js"
},
"dependencies": {
"debug": "^3.0.1",
"lodash": "^4.17.4",
"request": "^2.83.0"
},
"devDependencies": {
"ava": "^0.15.2",
"eslint": "^4.4.1",
"eslint-config-yoho": "^1.0.1",
"nyc": "^11.1.0"
}
}
... ...
const test = require('ava');
const Sender = require('../index');
test.before(() => {
const http = require('http');
const server = http.createServer((req, res) => {
res.statusCode = 204;
res.end('');
});
server.listen(3003);
});
test('test normal send message', t => {
const sender = new Sender({
host: 'localhost',
port: 3003,
db: 'test',
measurement: 'test2'
});
sender.addMessage({
tags: {
a: 'a',
b: 'b'
},
fields: {
a: 'a',
b: 'b'
}
});
sender.on('ok', code => {
t.is(code, 204);
});
});
test('test normal send message immediately', t => {
const sender = new Sender({
host: 'localhost',
port: 3003,
db: 'test',
measurement: 'test2',
immediate: true
});
sender.addMessage({
tags: {
a: 'a',
b: 'b'
},
fields: {
a: 'a',
b: 'b'
}
});
sender.on('ok', code => {
t.is(code, 204);
});
});
test('test send json message immediately', t => {
const sender = new Sender({
host: 'localhost',
port: 3003,
db: 'test',
measurement: 'test2',
immediate: true,
path: '/alert'
});
sender.addMessage({
tags: {
a: 'a',
b: 'b'
},
fields: {
a: 'a',
b: 'b'
}
});
sender.on('ok', code => {
t.is(code, 204);
});
});
\ No newline at end of file
... ...
This diff could not be displayed because it is too large.