Added prometheus monitoring metrics for k8s
ADD
src/monitoring/ComposedMonitoringMetric.ts
+import client, { Gauge, LabelValues } from 'prom-client';
+
+class ComposedMonitoringMetric {
+ private gaugeMetricAvg: client.Gauge;
+ private gaugeMetricCount: client.Gauge;
+ private metricCount = 0;
+ private metricAvg = 0;
+
+ // Normal signature with defaults
+ public constructor(prefix: string, metricname: string, suffix: number, metrichelp: string, labelNames: string[]) {
+ this.gaugeMetricAvg = new client.Gauge({
+ name: prefix + '_' + metricname + '_avg_' + suffix,
+ help: metrichelp,
+ labelNames: labelNames,
+ });
+ this.gaugeMetricCount = new client.Gauge({
+ name: prefix + '_' + metricname + '_count_' + suffix,
+ help: metrichelp,
+ labelNames: labelNames,
+ });
+ }
+
+ public register(registry: client.Registry): void {
+ registry.registerMetric(this.gaugeMetricAvg);
+ registry.registerMetric(this.gaugeMetricCount);
+ }
+
+ public setValue(labels: LabelValues<string>, value: number) {
+ if (this.metricCount === 0) {
+ this.metricAvg = 0;
+ }
+ this.metricAvg = (this.metricAvg * this.metricCount + value) / (this.metricCount + 1);
+ this.metricCount++;
+ this.gaugeMetricCount.labels(labels).set(this.metricCount);
+ this.gaugeMetricAvg.labels(labels).set(this.metricAvg);
+ }
+
+ public clear() {
+ this.metricAvg = 0;
+ this.metricCount = 0;
+ this.gaugeMetricCount.reset();
+ this.gaugeMetricAvg.reset();
+ }
+}
+export { ComposedMonitoringMetric };

MODIFY
src/monitoring/MonitoringServer.ts
src/monitoring/MonitoringServer.ts
+import client from 'prom-client';
+import { ComposedMonitoringMetric } from './ComposedMonitoringMetric';
+
export default abstract class MonitoringServer {
public abstract start(): void;
+
+ public abstract getGauge(name: string): client.Gauge | undefined;
+
+ public abstract getComposedMetric(
+ prefix: string,
+ metricname: string,
+ suffix: number,
+ metrichelp: string,
+ labelNames: string[]
+ ): ComposedMonitoringMetric;
}

MODIFY
src/monitoring/prometheus/PrometheusMonitoringServer.ts
src/monitoring/prometheus/PrometheusMonitoringServer.ts
import { Application, NextFunction, Request, Response } from 'express';
import { ServerAction, ServerType } from '../../types/Server';
+import client, { Gauge } from 'prom-client';
+import Constants from '../../utils/Constants';
import ExpressUtils from '../../server/ExpressUtils';
import Logging from '../../utils/Logging';
import MonitoringConfiguration from '../../types/configuration/MonitoringConfiguration';
+import { ComposedMonitoringMetric } from '../ComposedMonitoringMetric';
import MonitoringServer from '../MonitoringServer';
import { ServerUtils } from '../../server/ServerUtils';
-import client from 'prom-client';
+import global from '../../types/GlobalType';
const MODULE_NAME = 'PrometheusMonitoringServer';
export default class PrometheusMonitoringServer extends MonitoringServer {
private monitoringConfig: MonitoringConfiguration;
private expressApplication: Application;
+ private mapGauge = new Map<string, Gauge>();
+ private mapComposedMetric = new Map<string, ComposedMonitoringMetric>();
+ private clientRegistry = new client.Registry();
public constructor(monitoringConfig: MonitoringConfiguration) {
super();
// Keep params
this.monitoringConfig = monitoringConfig;
- // Create a Registry which registers the metrics
- const register = new client.Registry();
+ client.collectDefaultMetrics({ register: this.clientRegistry });
// Add a default label which is added to all metrics
- register.setDefaultLabels({
+ this.clientRegistry.setDefaultLabels({
app: 'e-Mobility'
});
- // Enable the collection of default metrics
- client.collectDefaultMetrics({
- register,
- });
+ if (process.env.K8S) {
+ this.createGaugeMetric(Constants.WEB_SOCKET_OCPP_CONNECTIONS_COUNT, 'The number of ocpp web sockets');
+ this.createGaugeMetric(Constants.WEB_SOCKET_REST_CONNECTIONS_COUNT, 'The number of rest web sockets');
+ this.createGaugeMetric(Constants.WEB_SOCKET_QUEUED_REQUEST, 'The number of web sockets that are queued');
+ this.createGaugeMetric(Constants.WEB_SOCKET_RUNNING_REQUEST, 'The number of web sockets that are running');
+ this.createGaugeMetric(Constants.WEB_SOCKET_RUNNING_REQUEST_RESPONSE, 'The number of web sockets request + response that are running');
+ this.createGaugeMetric(Constants.WEB_SOCKET_CURRRENT_REQUEST, 'JSON WS Requests in cache');
+ this.createGaugeMetric(Constants.MONGODB_CONNECTION_READY, 'The number of connection that are ready');
+ this.createGaugeMetric(Constants.MONGODB_CONNECTION_CREATED, 'The number of connection created');
+ this.createGaugeMetric(Constants.MONGODB_CONNECTION_CLOSED, 'The number of connection closed');
+ }
// Create HTTP Server
this.expressApplication = ExpressUtils.initApplication();
// Handle requests
@@ -37,8 +49,11 @@ export default class PrometheusMonitoringServer extends MonitoringServer {
// Trace Request
await Logging.traceExpressRequest(req, res, next, ServerAction.MONITORING);
// Process
- res.setHeader('Content-Type', register.contentType);
- res.end(await register.metrics());
+ res.setHeader('Content-Type', this.clientRegistry.contentType);
+ res.end(await this.clientRegistry.metrics());
+ for (const val of this.mapComposedMetric.values()) {
+ val.clear();
+ }
next();
// Trace Response
Logging.traceExpressResponse(req, res, next, ServerAction.MONITORING);
@@ -48,8 +63,44 @@ export default class PrometheusMonitoringServer extends MonitoringServer {
ExpressUtils.postInitApplication(this.expressApplication);
}
+ public getGauge(name: string): client.Gauge | undefined {
+ return this.mapGauge.get(name) ;
+ }
+
public start(): void {
+ global.monitoringServer = this;
ServerUtils.startHttpServer(this.monitoringConfig,
ServerUtils.createHttpServer(this.monitoringConfig, this.expressApplication), MODULE_NAME, ServerType.MONITORING_SERVER);
}
+
+ private createGaugeMetric(metricname : string, metrichelp : string, labelNames? : string[]) : Gauge {
+ let gaugeMetric : client.Gauge;
+ if (Array.isArray(labelNames)) {
+ gaugeMetric = new client.Gauge({
+ name: metricname,
+ help: metrichelp,
+ labelNames: labelNames
+ });
+ } else {
+ gaugeMetric = new client.Gauge({
+ name: metricname,
+ help: metrichelp
+ });
+ }
+ this.mapGauge.set(metricname, gaugeMetric);
+ this.clientRegistry.registerMetric(gaugeMetric);
+ return gaugeMetric;
+ }
+
+ public getComposedMetric(prefix : string, metricname: string, suffix: number, metrichelp: string, labelNames: string[]) : ComposedMonitoringMetric {
+ const key = prefix + '_' + metricname + '_' + suffix;
+ let composedMetric : ComposedMonitoringMetric = this.mapComposedMetric.get(key);
+ if (composedMetric) {
+ return composedMetric;
+ }
+ composedMetric = new ComposedMonitoringMetric(prefix, metricname,suffix,metrichelp,labelNames);
+ composedMetric.register(this.clientRegistry);
+ this.mapComposedMetric.set(key, composedMetric);
+ return composedMetric;
+ }
}

MODIFY
src/server/ocpp/json/JsonOCPPServer.ts
src/server/ocpp/json/JsonOCPPServer.ts
@@ -706,6 +706,14 @@ export default class JsonOCPPServer extends OCPPServer {
`${sizeOfCurrentRequestsBytes / 1000} kB used in JSON WS cache`
]
}).catch(() => { /* Intentional */ });
+ if ((global.monitoringServer) && (process.env.K8S)) {
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_RUNNING_REQUEST_RESPONSE).set(this.runningWSMessages);
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_OCPP_CONNECTIONS_COUNT).set(this.jsonWSConnections.size);
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_REST_CONNECTIONS_COUNT).set(this.jsonRestWSConnections.size);
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_CURRRENT_REQUEST).set(numberOfCurrentRequests);
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_RUNNING_REQUEST).set(Object.keys(this.runningWSRequestsMessages).length);
+ global.monitoringServer.getGauge(Constants.WEB_SOCKET_QUEUED_REQUEST).set(this.waitingWSMessages);
+ }
if (this.isDebug()) {
Logging.logConsoleDebug('=====================================');
Logging.logConsoleDebug(`** ${this.jsonWSConnections.size} JSON Connection(s)`);

MODIFY
src/utils/Logging.ts
src/utils/Logging.ts
@@ -97,6 +97,17 @@ export default class Logging {
Logging.logConsoleWarning('====================================');
}
}
+ if ((global.monitoringServer) && (process.env.K8S)) {
+ const labels = { tenant: tenant.subdomain, module: module, method: method };
+ const values = Object.values(labels).toString();
+ const hashCode = Utils.positiveHashcode(values);
+ const durationMetric = global.monitoringServer.getComposedMetric('mongodb', 'Duration', hashCode, 'db duration', Object.keys(labels));
+ durationMetric.setValue(labels, executionDurationMillis);
+ const requestSizeMetric = global.monitoringServer.getComposedMetric('mongodb', 'RequestSize', hashCode, 'db duration', Object.keys(labels));
+ requestSizeMetric.setValue(labels, sizeOfRequestDataKB);
+ const responseSizeMetric = global.monitoringServer.getComposedMetric('mongodb', 'ResponseSize', hashCode, 'db duration', Object.keys(labels));
+ responseSizeMetric.setValue(labels, sizeOfResponseDataKB);
+ }
await PerformanceStorage.savePerformanceRecord(
Utils.buildPerformanceRecord({
tenantSubdomain: tenant.subdomain,
@@ -710,6 +721,13 @@ export default class Logging {
})
);
const message = `${direction} OCPP Request '${action}~${Utils.last5Chars(performanceID)}' on '${chargingStationID}' has been ${direction === '>>' ? 'received' : 'sent'} - Req ${sizeOfRequestDataKB} KB`;
+ if ((global.monitoringServer) && (process.env.K8S)) {
+ const labels = { ocppComand : action, direction: ((direction === '<<') ? 'in' : 'out'), tenant: tenant.subdomain, siteId: chargingStationDetails.siteID, siteAreaID: chargingStationDetails.siteAreaID, companyID: chargingStationDetails.companyID };
+ const values = Object.values(labels).toString();
+ const hashCode = Utils.positiveHashcode(values);
+ const durationMetric = global.monitoringServer.getComposedMetric('ocpp', 'requestSize', hashCode, 'ocpp response time ', Object.keys(labels));
+ durationMetric.setValue(labels,sizeOfRequestDataKB);
+ }
Utils.isDevelopmentEnv() && Logging.logConsoleInfo(message);
await Logging.logDebug({
tenantID: tenant.id,
@@ -754,6 +772,13 @@ export default class Logging {
Logging.logConsoleWarning('====================================');
}
}
+ if ((global.monitoringServer) && (process.env.K8S)) {
+ const labels = { ocppComand : action, direction: ((direction === '<<') ? 'in' : 'out'), tenant: tenant.subdomain, siteId: chargingStationDetails.siteID, siteAreaID: chargingStationDetails.siteAreaID, companyID: chargingStationDetails.companyID };
+ const values = Object.values(labels).toString();
+ const hashCode = Utils.positiveHashcode(values);
+ const durationMetric = global.monitoringServer.getComposedMetric('ocpp', 'responsetime', hashCode, 'ocpp response time ', Object.keys(labels));
+ durationMetric.setValue(labels,executionDurationMillis);
+ }
if (response && response['status'] === OCPPStatus.REJECTED) {
await Logging.logError({
tenantID: tenant?.id,

Added prometheus monitoring metrics for k8s