Briq
/
multiparty-meeting
7
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

satisfy ESLint

auto_join_3.3
christian2 2020-05-04 14:22:31 +02:00
parent 84f77f3813
commit f6c76f372a
2 changed files with 236 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

352
server/lib/promExporter.js
View File

@ -4,43 +4,119 @@ const mediasoup = require('mediasoup');
const prom = require('prom-client'); const prom = require('prom-client');
const Logger = require('./Logger'); const Logger = require('./Logger');
const Peer = require('./Peer');
const Room = require('./Room');
const logger = new Logger('prom'); const logger = new Logger('prom');
const resolver = new Resolver(); const resolver = new Resolver();
const workers = new Map(); const workers = new Map();
const label_names = [ const labelNames = [
'pid', 'room_id', 'peer_id', 'display_name', 'user_agent', 'transport_id', 'pid', 'room_id', 'peer_id', 'display_name', 'user_agent', 'transport_id',
'proto', 'local_addr', 'remote_addr', 'id', 'kind', 'codec', 'type' 'proto', 'local_addr', 'remote_addr', 'id', 'kind', 'codec', 'type'
]; ];
const metadata = { const metadata = {
'byteCount': { metric_type: prom.Counter, unit: 'bytes' }, 'byteCount' : { metricType: prom.Counter, unit: 'bytes' },
'score': { metric_type: prom.Gauge } 'score' : { metricType: prom.Gauge }
};
module.exports = async function(rooms, peers, config)
{
const collect = async function(registry)
{
const newMetrics = function(subsystem)
{
const namespace = 'mediasoup';
const metrics = new Map();
for (const key in metadata)
{
if (Object.prototype.hasOwnProperty.call(metadata, key))
{
const value = metadata[key];
const name = key.split(/(?=[A-Z])/).join('_')
.toLowerCase();
const unit = value.unit;
const metricType = value.metricType;
let s = `${namespace}_${subsystem}_${name}`;
if (unit)
{
s += `_${unit}`;
}
const m = new metricType({
name : s, help : `${subsystem}.${key}`, labelNames : labelNames, registers : [ registry ] });
metrics.set(key, m);
}
} }
common_labels = function(both, fn) { return metrics;
for (let [room_id, room] of rooms) { };
for (let [peer_id, peer] of peers) {
if (fn(peer)) {
let display_name = peer._displayName;
let user_agent = peer._socket.client.request.headers['user-agent']; const commonLabels = function(both, fn)
let kind = both.kind; {
let codec = both.rtpParameters.codecs[0].mimeType.split('/')[1]; for (const roomId of rooms.keys())
return { room_id, peer_id, display_name, user_agent, kind, codec }; {
for (const [ peerId, peer ] of peers)
{
if (fn(peer))
{
const displayName = peer._displayName;
const userAgent = peer._socket.client.request.headers['user-agent'];
const kind = both.kind;
const codec = both.rtpParameters.codecs[0].mimeType.split('/')[1];
return { roomId, peerId, displayName, userAgent, kind, codec };
} }
} }
} }
throw new Error('cannot find common labels'); throw new Error('cannot find common labels');
};
const addr = async function(ip, port)
{
if (config.deidentify)
{
const a = ip.split('.');
for (let i = 0; i < a.length - 2; i++)
{
a[i] = 'xx';
} }
set_value = function(key, m, labels, v) { return `${a.join('.')}:${port}`;
logger.debug(`set_value key=${key} v=${v}`); }
switch (metadata[key].metric_type) { else if (config.numeric)
{
return `${ip}:${port}`;
}
else
{
try
{
const a = await resolver.reverse(ip);
ip = a[0];
}
catch (err)
{
logger.error(`reverse DNS query failed: ${ip} ${err.code}`);
}
return `${ip}:${port}`;
}
};
const quiet = function(s)
{
return config.quiet ? '' : s;
};
const setValue = function(key, m, labels, v)
{
logger.debug(`setValue key=${key} v=${v}`);
switch (metadata[key].metricType)
{
case prom.Counter: case prom.Counter:
m.inc(labels, v); m.inc(labels, v);
break; break;
@ -50,179 +126,159 @@ set_value = function(key, m, labels, v) {
default: default:
throw new Error(`unexpected metric: ${m}`); throw new Error(`unexpected metric: ${m}`);
} }
} };
collect = async function(registry, rooms, peers) {
metrics = function(subsystem) {
let namespace = 'mediasoup';
let metrics = new Map();
for (let key in metadata) {
value = metadata[key];
let name = key.split(/(?=[A-Z])/).join('_').toLowerCase();
let unit = value.unit;
let metric_type = value.metric_type;
let s = `${namespace}_${subsystem}_${name}`;
if (unit) {
s += `_${unit}`;
}
m = new metric_type({name: s, help: `${subsystem}.${key}`,
labelNames: label_names, registers: [registry]});
metrics.set(key, m);
}
return metrics;
}
logger.debug('collect'); logger.debug('collect');
const m_rooms = new prom.Gauge({name: 'edumeet_rooms', help: '#rooms', const mRooms = new prom.Gauge({ name: 'edumeet_rooms', help: '#rooms', registers: [ registry ] });
registers: [registry]});
m_rooms.set(rooms.size); mRooms.set(rooms.size);
const m_peers = new prom.Gauge({name: 'edumeet_peers', help: '#peers', const mPeers = new prom.Gauge({ name: 'edumeet_peers', help: '#peers', labelNames: [ 'room_id' ], registers: [ registry ] });
labelNames: ['room_id'], registers: [registry]});
for (let [room_id, room] of rooms) { for (const [ roomId, room ] of rooms)
m_peers.labels(room_id).set(Object.keys(room._peers).length); {
mPeers.labels(roomId).set(Object.keys(room._peers).length);
} }
const m_consumer = metrics('consumer'); const mConsumer = newMetrics('consumer');
const m_producer = metrics('producer'); const mProducer = newMetrics('producer');
for (let [pid, worker] of workers) {
for (const [ pid, worker ] of workers)
{
logger.debug(`visiting worker ${pid}`); logger.debug(`visiting worker ${pid}`);
for (let router of worker._routers) { for (const router of worker._routers)
{
logger.debug(`visiting router ${router.id}`); logger.debug(`visiting router ${router.id}`);
for (let [transport_id, transport] of router._transports) { for (const [ transportId, transport ] of router._transports)
logger.debug(`visiting transport ${transport_id}`); {
let transport_j = await transport.dump(); logger.debug(`visiting transport ${transportId}`);
if (transport_j.iceState != 'completed') { const transportJson = await transport.dump();
logger.debug(`skipping transport ${transport_id}}: ${transport_j.iceState}`);
if (transportJson.iceState != 'completed')
{
logger.debug(`skipping transport ${transportId}}: ${transportJson.iceState}`);
continue; continue;
} }
let ice_selected_tuple = transport_j.iceSelectedTuple; const iceSelectedTuple = transportJson.iceSelectedTuple;
let proto = ice_selected_tuple.protocol const proto = iceSelectedTuple.protocol;
let local_addr = await addr(ice_selected_tuple.localIp, const localAddr = await addr(iceSelectedTuple.localIp,
ice_selected_tuple.localPort); iceSelectedTuple.localPort);
let remote_addr = await addr(ice_selected_tuple.remoteIp, const remoteAddr = await addr(iceSelectedTuple.remoteIp,
ice_selected_tuple.remotePort); iceSelectedTuple.remotePort);
for (let [producer_id, producer] of transport._producers) {
logger.debug(`visiting producer ${producer_id}`); for (const [ producerId, producer ] of transport._producers)
let { room_id, peer_id, display_name, user_agent, kind, codec } = {
common_labels(producer, peer => peer._producers.has(producer_id)); logger.debug(`visiting producer ${producerId}`);
let a = await producer.getStats(); const { roomId, peerId, displayName, userAgent, kind, codec } =
for (let x of a) { commonLabels(producer, (peer) => peer._producers.has(producerId));
let type = x.type; const a = await producer.getStats();
let labels = {
for (const x of a)
{
const type = x.type;
const labels = {
'pid' : pid, 'pid' : pid,
'room_id': room_id, 'room_id' : roomId,
'peer_id': peer_id, 'peer_id' : peerId,
'display_name': display_name, 'display_name' : displayName,
'user_agent': user_agent, 'user_agent' : userAgent,
'transport_id': quiet(transport_id), 'transport_id' : quiet(transportId),
'proto' : proto, 'proto' : proto,
'local_addr': local_addr, 'local_addr' : localAddr,
'remote_addr': remote_addr, 'remote_addr' : remoteAddr,
'id': quiet(producer_id), 'id' : quiet(producerId),
'kind' : kind, 'kind' : kind,
'codec' : codec, 'codec' : codec,
'type' : type 'type' : type
} };
for (let [key, m] of m_producer) {
set_value(key, m, labels, x[key]); for (const [ key, m ] of mProducer)
{
setValue(key, m, labels, x[key]);
} }
} }
} }
for (let [consumer_id, consumer] of transport._consumers) { for (const [ consumerId, consumer ] of transport._consumers)
logger.debug(`visiting consumer ${consumer_id}`); {
let { room_id, peer_id, display_name, user_agent, kind, codec } = logger.debug(`visiting consumer ${consumerId}`);
common_labels(consumer, peer => peer._consumers.has(consumer_id)); const { roomId, peerId, displayName, userAgent, kind, codec } =
let a = await consumer.getStats(); commonLabels(consumer, (peer) => peer._consumers.has(consumerId));
for (let x of a) { const a = await consumer.getStats();
if (x.type == 'inbound-rtp') {
for (const x of a)
{
if (x.type == 'inbound-rtp')
{
continue; continue;
} }
let type = x.type; const type = x.type;
let labels = { const labels =
{
'pid' : pid, 'pid' : pid,
'room_id': room_id, 'room_id' : roomId,
'peer_id': peer_id, 'peer_id' : peerId,
'display_name': display_name, 'display_name' : displayName,
'user_agent': user_agent, 'user_agent' : userAgent,
'transport_id': quiet(transport_id), 'transport_id' : quiet(transportId),
'proto' : proto, 'proto' : proto,
'local_addr': local_addr, 'local_addr' : localAddr,
'remote_addr': remote_addr, 'remote_addr' : remoteAddr,
'id': quiet(consumer_id), 'id' : quiet(consumerId),
'kind' : kind, 'kind' : kind,
'codec' : codec, 'codec' : codec,
'type' : type 'type' : type
} };
for (let [key, m] of m_consumer) {
set_value(key, m, labels, x[key]);
}
}
}
}
}
}
}
module.exports = async function(rooms, peers, config) { for (const [ key, m ] of mConsumer)
{
setValue(key, m, labels, x[key]);
}
}
}
}
}
}
};
addr = async function(ip, port) { try
if (config.deidentify) { {
let a = ip.split('.')
for (let i = 0; i < a.length - 2; i++) {
a[i] = 'xx';
}
return `${a.join('.')}:${port}`;
}
else if (config.numeric) {
return `${ip}:${port}`;
}
else {
try {
let a = await resolver.reverse(ip);
ip = a[0];
}
catch (err) {
logger.error(`reverse DNS query failed: ${ip} ${err.code}`);
}
return `${ip}:${port}`;
}
}
quiet = function(s) {
return config.quiet ? '' : s;
}
try {
logger.debug(`config.deidentify=${config.deidentify}`); logger.debug(`config.deidentify=${config.deidentify}`);
logger.debug(`config.numeric=${config.numeric}`); logger.debug(`config.numeric=${config.numeric}`);
logger.debug(`config.port=${config.port}`); logger.debug(`config.port=${config.port}`);
logger.debug(`config.quiet=${config.quiet}`); logger.debug(`config.quiet=${config.quiet}`);
mediasoup.observer.on('newworker', worker => { mediasoup.observer.on('newworker', (worker) =>
{
logger.debug(`observing newworker ${worker.pid} #${workers.size}`); logger.debug(`observing newworker ${worker.pid} #${workers.size}`);
workers.set(worker.pid, worker); workers.set(worker.pid, worker);
worker.observer.on('close', () => { worker.observer.on('close', () =>
{
logger.debug(`observing close worker ${worker.pid} #${workers.size - 1}`); logger.debug(`observing close worker ${worker.pid} #${workers.size - 1}`);
workers.delete(worker.pid); workers.delete(worker.pid);
}); });
}); });
let app = express(); const app = express();
app.get('/', async (req, res) => {
app.get('/', async (req, res) =>
{
logger.debug(`GET ${req.originalUrl}`); logger.debug(`GET ${req.originalUrl}`);
let registry = new prom.Registry(); const registry = new prom.Registry();
await collect(registry, rooms, peers);
await collect(registry);
res.set('Content-Type', registry.contentType); res.set('Content-Type', registry.contentType);
let data = registry.metrics(); const data = registry.metrics();
res.end(data); res.end(data);
}); });
let server = app.listen(config.port || 8889, () => { const server = app.listen(config.port || 8889, () =>
address = server.address(); {
const address = server.address();
logger.info(`listening ${address.address}:${address.port}`); logger.info(`listening ${address.address}:${address.port}`);
}); });
} }
catch (err) { catch (err)
{
logger.error(err); logger.error(err);
} }
} };

3
server/server.js
View File

@ -134,7 +134,8 @@ async function run()
await interactiveServer(rooms, peers); await interactiveServer(rooms, peers);
// start Prometheus exporter // start Prometheus exporter
if (config.prometheus) { if (config.prometheus)
{
await promExporter(rooms, peers, config.prometheus); await promExporter(rooms, peers, config.prometheus);
} }