Rewrite statistics export

Add automatic export of metrics from slaves to the master server.  And
rewrite the statistics export plugin to use this new system.
This commit is contained in:
Hornwitser committed 2019-12-20 21:26:29 +01:00
1 parent 8229d1955c
commit 26eefb0476
19 files changed
+334 -226

No files matched your search

-1
View File
@@ -51,7 +51,6 @@ sharedPlugins/*
!sharedPlugins/UPSdisplay/
!sharedPlugins/researchSync/
!sharedPlugins/clusterioMod/
!sharedPlugins/statisticsExporter/
# File used for marking plugins as disabled
sharedPlugins/*/DISABLED
+16
View File
@@ -23,6 +23,10 @@ Version 2.0.0
a base class and use the same WebSocket connection Clusterio uses to
communicate.
### Features
- Added export of pollution statitics.
### Changes
- Fixed sslCert and sslPrivKey entries being ignored in the config.
@@ -111,6 +115,18 @@ Version 2.0.0
- Removed the onLoadFinish hook from master plugins.
- Moved plugins from the sharedPlugins directory to plugins directory.
- Implemented a new plugin system that replaces the old. Breaks all plugins.
- Removed express metric http_request_duration_milliseconds metric.
- Removed socket.io metrics socket_io_connected, socket_io_connect_total,
socket_io_disconnect_total, socket_io_events_received_total,
socket_io_events_sent_total, socket_io_recieve_bytes, and
socket_io_transmit_bytes,
- Removed clusterio_connected_instaces_gauge and added
clusterio_master_connected_clients_count in its place.
- Renamed clusterio_player_count_gauge to clusterio_instance_player_count
- Removed clusterio_UPS_gauge and added clusterio_instance_game_ticks_total in
its place.
- Renamed clusterio_endpoint_hit_gauge to clusterio_http_enpoint_hits_total
- Renamed clusterio_statistics_gauge to clusterio_instance_force_flows
Version 1.2.4
+51
View File
@@ -10,6 +10,7 @@ const factorio = require("lib/factorio");
const link = require("lib/link");
const plugin = require("lib/plugin");
const errors = require("lib/errors");
const prometheus = require('lib/prometheus');
/**
@@ -221,6 +222,30 @@ class Instance extends link.Link{
await this.server.stop();
}
// Clear metrics this instance is exporting
for (let collector of prometheus.defaultRegistry.collectors) {
if (
collector instanceof prometheus.ValueCollector
&& collector.metric.labels.includes('instance_id')
) {
collector.removeAll({ instance_id: String(this.config.id) });
}
}
}
async getMetricsRequestHandler() {
let results = []
for (let pluginInstance of this.plugins.values()) {
let pluginResults = await pluginInstance.onMetrics();
if (pluginResults !== undefined) {
for await (let result of pluginResults) {
results.push(prometheus.serializeResult(result))
}
}
}
return { results };
}
async startInstanceRequestHandler() {
@@ -500,6 +525,32 @@ class Slave extends link.Link {
return instanceConnection;
}
async getMetricsRequestHandler() {
let requests = [];
for (let instanceConnection of this.instanceConnections.values()) {
requests.push(link.messages.getMetrics.send(instanceConnection));
}
let results = [];
for (let response of await Promise.all(requests)) {
results.push(...response.results);
}
for await (let result of prometheus.defaultRegistry.collect()) {
if (result.metric.name.startsWith('process_')) {
results.push(prometheus.serializeResult(result, {
addLabels: { 'slave_id': String(this.config.id) },
metricName: result.metric.name.replace('process_', 'clusterio_slave_'),
}));
} else {
results.push(prometheus.serializeResult(result));
}
}
return { results };
}
async startInstanceRequestHandler(message, request) {
let instanceConnection = await this._connectInstance(message.data.instance_id);
return await request.send(instanceConnection, message.data);
+2
View File
@@ -170,6 +170,8 @@ indents. Should you be modifiying such code that is mixing spaces and
tabs as indending you should update the indenting to use tabs _on the
code you change_ while leaving the rest of the code untouched.
Lua code should be indented with 4 spaces.
### Line Length
+49
View File
@@ -301,3 +301,52 @@ plugin's `info.js` module. In other words:
For the Request class the send method is async and returns the response
data received from the target it was sent to, or throws an error if the
request failed.
Collecting Statistics
---------------------
Clusterio comes with it's own Prometheus client implementation, one part
due to Not Invented Here and another part due to collectors in
prom-client being difficult to get to work nicely with collecting data
from plugins optionally loaded at runtime on different computers.
In its simplest form collecting data from a plugins consists of defining
the metric and updating it somewhere in the plugin code. For example:
const { Counter } = require('lib/prometheus');
const fooMetric = new Counter(
'clusterio_foo_metric', "Measures the level of foo",
);
// Somewhere in the master plugin code
fooMetric.inc();
This works for master plugins, and the metric will be automatically
available through the /metric HTTP endpoint. For metrics that are
per-instance you must define an `instance_id` label and set it
accordingly, for example:
const { Counter } = require('lib/prometheus');
const barMetric = new Gauge(
'clusterio_bar_metric', "Bar instance level",
{ labels: ['instance_id'] }
);
// Somewhere in the instance plugin code
barMetric.labels(String(this.config.id)).set(someValue);
Metrics are automatically registered to the default registry, and this
default registry is automatically polled by the master server on slaves.
This means that it's important that you place the definition of the
metric at module level so that it's not created more than once over the
lifetime of a slave. Since the metrics remember their values and would
continue to be exported after an instance is shutdown there's code at
instance shutdown that removes all the values where the `instance_id`
label matches the id of the instance shut down.
For statistics you need to update on collection there's an `onMetrics`
hook on both master and instance plugins that is runned before the
metrics in the default registry is collected.
+8
View File
@@ -292,6 +292,14 @@ messages.sendRcon = new Request({
}
});
messages.getMetrics = new Request({
type: 'statistics_exporter:get_metrics',
links: ['master-slave', 'slave-instance'],
responseProperties: {
"results": { type: "array" },
},
});
/**
* Represents an event sent over the link
+34
View File
@@ -19,6 +19,23 @@ class BaseInstancePlugin {
*/
async init() { }
/**
* Called before collecting Prometheus metrics
*
* Invoked before the default metrics of prometheus is collected. Note
* that since this is done while replying to /metrics this should not
* take a long time, otherwise the requst will time out.
*
* Although you can return the results of collecting your prometheus
* collectors directly, all gauges in the default registry is already
* automatically collected. It's therefore recommended to only update
* the Gauges/Counters/etc in onMetric and let Clusterio deal will
* collecting the values.
*
* @returns an async iterator of prometheus metric results or undefined.
*/
async onMetrics() { }
/**
* Called after the Factorio server is started
*/
@@ -66,6 +83,23 @@ class BaseMasterPlugin {
*/
async init() { }
/**
* Called before collecting Prometheus metrics
*
* Invoked before the default metrics of prometheus is collected. Note
* that since this is done while replying to /metrics this should not
* take a long time, otherwise the requst will time out.
*
* Although you can return the results of collecting your prometheus
* collectors directly, all gauges in the default registry is already
* automatically collected. It's therefore recommended to only update
* the Gauges/Counters/etc in onMetric and let Clusterio deal will
* collecting the values.
*
* @returns an async iterator of prometheus metric results or undefined.
*/
async onMetrics() { }
/**
* Called when the master server is shutting down
*/
+59 -71
View File
@@ -39,6 +39,7 @@ const schema = require("lib/schema");
const link = require("lib/link");
const errors = require("lib/errors");
const plugin = require("lib/plugin");
const prometheus = require("lib/prometheus");
// homemade express middleware for token auth
const authenticate = require("lib/authenticate");
@@ -83,79 +84,69 @@ require("./routes/api/getPictures")(app);
// Set folder to serve static content from (the website)
app.use(express.static('static'));
// set up logging software
const prometheusPrefix = "clusterio_";
const Prometheus = require('prom-client');
const expressPrometheus = require('express-prometheus-request-metrics');
Prometheus.collectDefaultMetrics({ timeout: 10000 }); // collects RAM usage etc every 10 s
const endpointHitCounter = new prometheus.Gauge(
'clusterio_master_http_endpoint_hits_total', "How many requests a particular endpoint has gotten",
{ labels: ['route'] }
);
// collect express request durations ms
app.use(expressPrometheus(Prometheus));
const endpointHitCounter = new Prometheus.Gauge({
name: prometheusPrefix+'endpoint_hit_gauge',
help: "How many requests a particular endpoint has gotten",
labelNames: ['route'],
});
const prometheusConnectedInstancesCounter = new Prometheus.Gauge({
name: prometheusPrefix+'connected_instaces_gauge',
help: "How many instances are currently connected to this master server",
});
const prometheusWsUsageCounter = new Prometheus.Counter({
name: prometheusPrefix+'websocket_usage_counter',
help: 'Websocket traffic',
labelNames: ["connectionType", "instanceID"],
});
const prometheusPlayerCountGauge = new Prometheus.Gauge({
name: prometheusPrefix+'player_count_gauge',
help: 'Amount of players connected to this cluster',
labelNames: ["instanceID", "instanceName"],
});
const prometheusUPSGauge = new Prometheus.Gauge({
name: prometheusPrefix+'UPS_gauge',
help: 'UPS of the current server',
labelNames: ["instanceID", "instanceName"],
});
/**
GET Prometheus metrics endpoint. Returns performance and usage metrics in a prometheus readable format.
@memberof clusterioMaster
@instance
@alias /metrics
*/
app.get('/metrics', (req, res) => {
// Prometheus polling endpoint
async function getMetrics(req, res, next) {
endpointHitCounter.labels(req.route.path).inc();
res.set('Content-Type', Prometheus.register.contentType);
/// gather some static metrics
registerMoreMetrics();
res.end(Prometheus.register.metrics());
});
function registerMoreMetrics(){
for (let [instanceID, slave] of db.slaves) {
// playercount
try{
prometheusPlayerCountGauge.labels(instanceID, slave.instanceName).set(Number(slave.playerCount) || 0);
}catch(e){}
// UPS
try{
prometheusUPSGauge.labels(instanceID, slave.instanceName).set(Number(slave.meta.UPS) || 60);
if(slave.meta.tick && typeof slave.meta.tick === "number") prometheusUPSGauge.labels(instanceID, slave.instanceName).set(Number(slave.meta.tick) || 0);
}catch(e){}
}
// plugins
let results = []
for (let masterPlugin of masterPlugins.values()) {
masterPlugin.onMetrics();
let pluginResults = await masterPlugin.onMetrics();
if (pluginResults !== undefined) {
for await (let result of pluginResults) {
results.push(result)
}
}
}
// Slave count
let numberOfActiveSlaves = 0;
for(let slave of db.slaves.values()){
if(Date.now() - Number(slave.time) < 1000 * 30) numberOfActiveSlaves++;
let requests = [];
for (let slaveConnection of slaveConnections.values()) {
// XXX this needs a timeout
requests.push(link.messages.getMetrics.send(slaveConnection));
}
prometheusConnectedInstancesCounter.set(numberOfActiveSlaves);
for await (let result of await prometheus.defaultRegistry.collect()) {
results.push(result);
}
let resultMap = new Map();
for (let response of await Promise.all(requests)) {
for (let result of response.results) {
if (!resultMap.has(result.metric.name)) {
resultMap.set(result.metric.name, result);
} else {
// Merge metrics received by multiple slaves
let stored = resultMap.get(result.metric.name);
stored.samples.push(...result.samples);
}
}
}
for (let result of resultMap.values()) {
results.push(prometheus.deserializeResult(result));
}
let text = await prometheus.exposition(results);
res.set('Content-Type', prometheus.exposition.contentType);
res.send(text);
}
app.get('/metrics', (req, res, next) => getMetrics(req, res, next).catch(next));
const masterConnectedClientsCount = new prometheus.Gauge(
'clusterio_master_connected_clients_count', "How many clients are currently connected to this master server",
{
labels: ['type'], callback: async function(gauge) {
gauge.labels("slave").set(slaveConnections.size);
gauge.labels("control").set(controlConnections.length);
},
},
);
// set up database
const db = {};
@@ -269,9 +260,6 @@ app.get("/api/getFactorioLocale", function(req,res){
// socket.io connection for slaves
var io = require("socket.io")({});
const ioMetrics = require("socket.io-prometheus");
ioMetrics(io);
class BaseConnection extends link.Link {
constructor(target, connector) {
@@ -408,7 +396,7 @@ class SlaveConnection extends BaseConnection {
});
this.connector.on('message', () => {
prometheusWsUsageCounter.labels('message', "other").inc();
// XXX prometheusWsUsageCounter.labels('message', "other").inc();
});
this.connector.on('disconnect', () => {
@@ -525,7 +513,7 @@ io.on('connection', function (socket) {
// Handle socket handshake
socket.once('message', (payload) => {
prometheusWsUsageCounter.labels('message', "other").inc();
// XXX prometheusWsUsageCounter.labels('message', "other").inc();
if (!schema.clientHandshake(payload)) {
console.log(`SOCKET | closing ${socket.handshake.address} after receiving invalid handshake`, payload);
connector.send('close', { reason: "Invalid handshake" });
@@ -588,7 +576,7 @@ async function loadPlugins() {
main:new masterPlugin({
TODO?
config, socketio: io, express: app,
db, Prometheus, prometheusPrefix, endpointHitCounter,
db,
}),
pluginConfig,
});*/
-3
View File
@@ -42,7 +42,6 @@
"ejs": "^2.5.7",
"express": "^4.16.2",
"express-fileupload": "1.1.5",
"express-prometheus-request-metrics": "github:burnash/express-prometheus-request-metrics",
"follow-redirects": "^1.5.0",
"fs-extra": "^8.1.0",
"ini": "^1.3.5",
@@ -55,7 +54,6 @@
"needle": "^2.4.0",
"node-factorio-api": "^0.3.8",
"node-forge": "^0.7.1",
"prom-client": "^10.2.3",
"rcon-client": "^4.2.0",
"request": "^2.83.0",
"sanitizer": "^0.1.3",
@@ -64,7 +62,6 @@
"single-line-log": "^1.1.2",
"socket.io": "^2.2.0",
"socket.io-client": "^2.2.0",
"socket.io-prometheus": "^0.2.1",
"sync-request": "^4.0.1",
"typescript-collections": "^1.3.2",
"yargs": "^14.2.0"
+9
View File
@@ -0,0 +1,9 @@
let link = require("lib/link");
module.exports = {
name: "statistics_exporter",
title: "Prometheus Statistics Export",
description: "Exports statistics to Prometheus",
version: "2.0.0-alpha",
instanceEntrypoint: "instance",
}
+62
View File
@@ -0,0 +1,62 @@
const plugin = require('lib/plugin');
const { Gauge } = require('lib/prometheus');
const instancePlayerCount = new Gauge(
'clusterio_instance_player_count', "Amount of players connected to this cluster",
{ labels: ['instance_id'] }
);
const instanceGameTicksTotal = new Gauge(
'clusterio_instance_game_ticks_total', "Game tick an instance has progressed to",
{ labels: ['instance_id'] }
);
const instanceForceFlowStatistics = new Gauge(
'clusterio_instance_force_flow_statistics', "Items/fluids/enemies/buildings produced/built/killed by a force",
{ labels: ['instance_id', 'force', 'statistic', 'direction', 'name'] },
);
const instanceGameFlowStatistics = new Gauge(
'clusterio_instance_game_flow_statistics', "Pollution produced/consumed in the game",
{ labels: ['instance_id', 'statistic', 'direction', 'name'] },
);
class InstancePlugin extends plugin.BaseInstancePlugin {
async onMetrics() {
let string = await this.instance.server.sendRcon('/sc remote.call("statistics_exporter", "export")');
let stats = JSON.parse(string);
let instanceId = this.instance.config.id;
instanceGameTicksTotal.labels(String(instanceId)).set(stats.game_tick);
instancePlayerCount.labels(String(instanceId)).set(stats.player_count);
for (let [forceName, flowStatistics] of Object.entries(stats.force_flow_statistics)) {
for (let [statisticName, statistic] of Object.entries(flowStatistics)) {
for (let [direction, counts] of Object.entries(statistic)) {
for (let [item, value] of Object.entries(counts)) {
instanceForceFlowStatistics.labels(
String(instanceId), forceName, statisticName, direction, item
).set(value);
}
}
}
}
for (let [direction, counts] of Object.entries(stats.game_flow_statistics.pollution_statistics)) {
for (let [item, value] of Object.entries(counts)) {
instanceGameFlowStatistics.labels(
String(instanceId), "pollution_statistics", direction, item
).set(value);
}
}
}
}
module.exports = {
InstancePlugin,
// For testing only
_instancePlayerCount: instancePlayerCount,
_instanceGameTicksTotal: instanceGameTicksTotal,
_instanceForceFlowStatistics: instanceForceFlowStatistics,
_instanceGameFlowStatistics: instanceGameFlowStatistics,
}
@@ -0,0 +1,39 @@
local statistics = {
"item_production_statistics",
"fluid_production_statistics",
"kill_count_statistics",
"entity_build_count_statistics",
}
local function export()
local stats = {
game_tick = game.tick,
player_count = #game.connected_players,
game_flow_statistics = {
pollution_statistics = {
input = game.pollution_statistics.input_counts,
output = game.pollution_statistics.output_counts,
},
},
force_flow_statistics = {}
}
for _, force in pairs(game.forces) do
local flow_statistics = {}
for _, statName in pairs(statistics) do
flow_statistics[statName] = {
input = force[statName].input_counts,
output = force[statName].output_counts,
}
end
stats.force_flow_statistics[force.name] = flow_statistics
end
rcon.print(game.table_to_json(stats))
end
return {
add_remote_interface = function()
remote.add_interface("statistics_exporter", { export = export })
end
}
Whitespace-only changes.
@@ -1,9 +0,0 @@
/*
Clusterio plugin for gathering ingame statistics
*/
module.exports = {
name: "statisticsExporter",
version: "1.0.0",
description: "Gathers ingame stats such as production, consumption, kills and buildings placed",
masterPlugin: "masterPlugin.js"
}
-40
View File
@@ -1,40 +0,0 @@
const needle = require("needle");
module.exports = class remoteCommands {
constructor(mergedConfig, messageInterface, extras){
this.messageInterface = messageInterface;
this.config = mergedConfig;
const needleOptionsWithTokenAuthHeader = {
headers: {
'x-access-token': this.config.masterAuthToken
},
};
// Gather and submit statistics every 15 seconds
setInterval(async () => {
let stats = await this.getStats();
stats.forEach(force => {
// I have no idea why I need to do JSON.stringify before passing it to needle.post,
// but if you do forceData: force.data, it doesn't work...
let request = {
instanceID: this.config.unique,
force: force.forceName,
forceData: JSON.stringify(force.data),
};
needle.post(`${mergedConfig.masterURL}/api/logStats`,
request,
needleOptionsWithTokenAuthHeader, (err, resp) => {
if(!err && resp.body && resp.body.ok == false) console.log("Error calling /api/logStats: " + resp.body);
}
);
});
}, 15000);
}
async getStats() {
let string = await this.messageInterface('/sc remote.call("statisticsExporter", "export")');
return JSON.parse(string);
}
}
@@ -1,30 +0,0 @@
local statistics = {
"item_production_statistics",
"fluid_production_statistics",
"kill_count_statistics",
"entity_build_count_statistics",
}
local function export()
local stats = {}
for _, force in pairs(game.forces) do
local data = {}
for _, statName in pairs(statistics) do
data[statName] = {
input = force[statName].input_counts,
output = force[statName].output_counts,
}
end
table.insert(stats, {
forceName = force.name,
data = data,
})
end
rcon.print(game.table_to_json(stats))
end
return {
add_remote_interface = function()
remote.add_interface("statisticsExporter", { export = export })
end
}
@@ -1,67 +0,0 @@
const fs = require("fs-extra");
const path = require("path");
class masterPlugin {
constructor({config, socketio, express, Prometheus}){
const authenticate = require("lib/authenticate");
this.config = config;
this.io = socketio;
const app = express;
const prometheusStatisticsGauge = new Prometheus.Gauge({
name: 'clusterio_statistics_gauge',
help: 'Items produced/built/killed by a force',
labelNames: ["instanceID", "force", "statistic", "direction", "itemName"],
});
/** --- TODO: Replace this with a simpler Prometheus exporter. Requires changes to the mod as well.
POST endpoint to log production graph statistics. Should contain an instanceID (also reffered to as "unique")
and of course the timeSeries data organized by category.
@memberof clusterioMaster
@instance
@alias api/logStats
@param {object} JSON {instanceID: "string", force: "string" data: {"item_production_statistics": {"input":{"iron-plate":150}, "output":{}}, ...}}
@returns {string} failure
*/
app.post("/api/logStats", authenticate.middleware, function(req,res) {
let data = req.body.forceData;
if(typeof req.body.forceData == "string") {
data = JSON.parse(req.body.forceData);
}
// endpointHitCounter.labels(req.route.path).inc();
if(typeof req.body == "object"
&& req.body.instanceID
&& typeof req.body.instanceID == "string"
&& req.body.instanceID.length < 100
&& data
&& typeof data == "object"
&& !Array.isArray(data)){
try{
Object.keys(data).forEach(statisticName => {
Object.keys(data[statisticName]).forEach(direction => {
let statHash = data[statisticName][direction];
Object.keys(statHash).forEach(item => {
let value = Number(statHash[item]);
if(value) prometheusStatisticsGauge.labels(req.body.instanceID, req.body.force, statisticName, direction, item).set(value);
});
});
});
res.send({ok:true, msg:`Statistics successfully reported at ${Date.now()}`});
}catch(e){
res.send({ok:false, msg:"Statistics reporter crashed, see sharedPlugins/statisticsExporter/masterPlugin.js"});
console.log(e)
};
} else {
console.log(`/api/logStats: Invalid request. instanceID: ${req.body.instanceID}, dataIsArray: `
+ `${Array.isArray(data)}, ${typeof req.body.instanceID}, ${typeof data}`);
res.send({
ok:false,
msg:"Invalid request parameters"
});
}
});
}
}
module.exports = masterPlugin;
@@ -1,5 +0,0 @@
// These are libraries that does not have any units test, they are
// required here in order to track coverage for them.
require('../index');
require('../masterPlugin');
+5
View File
@@ -57,10 +57,12 @@ class MockConnector extends events.EventEmitter {
class MockServer {
constructor() {
this.rconCommands = [];
this.rconCommandResults = new Map();
}
async sendRcon(command) {
this.rconCommands.push(command);
return this.rconCommandResults.get(command) || '';
}
}
@@ -69,6 +71,9 @@ class MockInstance extends link.Link {
super('instance', 'slave', new MockConnector());
this.server = new MockServer();
this.name = "test";
this.config = {
id: 7357,
}
}
}