feat: send probe logs to the API (#314)

Author: PavelKopecky

package-lock.json

@@ -26,7 +26,8 @@
 		"socket.io-client": "^4.8.1",
 		"throng": "^5.0.0",
 		"tldts": "^7.0.8",
-		"winston": "^3.17.0"
+		"winston": "^3.17.0",
+		"winston-transport": "^4.9.0"
 	},
 	"devDependencies": {
 		"@commitlint/cli": "^19.8.1",

package.json

@@ -0,0 +1,122 @@
+import Transport from 'winston-transport';
+import { Socket } from 'socket.io-client';
+import { Logger } from 'winston';
+
+export type ApiTransportSettings = {
+	isActive?: boolean;
+	sendInterval?: number;
+	maxBufferSize?: number;
+};
+
+export type ApiTransportOptions = Transport.TransportStreamOptions & ApiTransportSettings & { socket?: Socket };
+
+type Info = {
+	message: string;
+	timestamp: string;
+	level: string;
+	scope: string;
+};
+
+class ApiLogsTransport extends Transport {
+	private logger: Logger | undefined;
+	private socket: Socket | undefined;
+	private isActive: boolean;
+	private sendInterval: number;
+	private maxBufferSize: number;
+	private logBuffer: Info[] = [];
+	private droppedLogs: number = 0;
+	private timer: NodeJS.Timeout | undefined = undefined;
+
+	constructor (opts?: ApiTransportOptions) {
+		super(opts);
+		this.isActive = opts?.isActive ?? false;
+		this.sendInterval = opts?.sendInterval ?? 10000;
+		this.maxBufferSize = opts?.maxBufferSize ?? 100;
+		this.socket = opts?.socket;
+		this.scheduleSend();
+	}
+
+	override log (info: Info, callback?: () => void) {
+		setImmediate(() => this.emit('logged', info));
+
+		this.logBuffer.push(info);
+		const bufferLength = this.logBuffer.length;
+		const bufferOverflow = bufferLength - this.maxBufferSize;
+
+		if (bufferOverflow > 0) {
+			this.logBuffer.splice(0, bufferOverflow);
+			this.droppedLogs += bufferOverflow;
+		}
+
+		callback && callback();
+	}
+
+	setSocket (socket: Socket) {
+		this.socket = socket;
+	}
+
+	setLogger (logger: Logger) {
+		this.logger = logger;
+	}
+
+	getCurrentSettings () {
+		return {
+			isActive: this.isActive,
+			sendInterval: this.sendInterval,
+			maxBufferSize: this.maxBufferSize,
+		};
+	}
+
+	updateSettings (settings: ApiTransportSettings) {
+		this.isActive = settings.isActive ?? this.isActive;
+		this.sendInterval = settings.sendInterval ?? this.sendInterval;
+		this.maxBufferSize = settings.maxBufferSize ?? this.maxBufferSize;
+		this.scheduleSend();
+	}
+
+	private scheduleSend () {
+		clearTimeout(this.timer);
+
+		if (this.isActive) {
+			this.timer = setTimeout(() => {
+				void this.sendLogs();
+			}, this.sendInterval);
+		}
+	}
+
+	private async sendLogs () {
+		if (!this.isActive || !this.socket?.connected || !this.logBuffer.length) {
+			return this.scheduleSend();
+		}
+
+		const payload = {
+			logs: this.logBuffer.slice(),
+			skipped: this.droppedLogs,
+		};
+
+		const droppedInPayload = payload.skipped;
+		const presentInPayload = payload.logs.length;
+
+		try {
+			const response: unknown = await this.socket.emitWithAck('probe:logs', payload);
+
+			if (response === 'success') {
+				const droppedWhileAwaiting = this.droppedLogs - droppedInPayload;
+				const oldLogsRemaining = presentInPayload - droppedWhileAwaiting;
+
+				if (oldLogsRemaining >= 0) {
+					this.logBuffer.splice(0, oldLogsRemaining);
+					this.droppedLogs = 0;
+				} else {
+					this.droppedLogs = -oldLogsRemaining; // === droppedWhileAwaiting - presentInPayload
+				}
+			}
+		} catch (e) {
+			this.logger?.error('Failed to send logs to the API.', e);
+		} finally {
+			this.scheduleSend();
+		}
+	}
+}
+
+export default ApiLogsTransport;

src/lib/api-logs-transport.ts

@@ -2,6 +2,9 @@
 import process from 'node:process';
 import { inspect } from 'node:util';
 import * as winston from 'winston';
+import ApiLogsTransport from './api-logs-transport.js';
+
+export const apiLogsTransport = new ApiLogsTransport();
 
 const objectFormatter = (object: Record<string, any>) => {
 	const entries = Object.entries(object).map(([ key, value ]) => {
@@ -40,7 +43,10 @@ const logger = winston.createLogger({
 	),
 	transports: [
 		new winston.transports.Console(),
+		apiLogsTransport,
 	],
 });
 
 export const scopedLogger = (scope: string): winston.Logger => logger.child({ scope });
+
+apiLogsTransport.setLogger(scopedLogger('api-logs-transport'));

src/lib/logger.ts

@@ -10,7 +10,8 @@ import physicalCpuCount from 'physical-cpu-count';
 import { getFakeIp } from './lib/fake-ip.js';
 import type { CommandInterface, MeasurementRequest } from './types.js';
 import { loadAll as loadAllDeps } from './lib/dependencies.js';
-import { scopedLogger } from './lib/logger.js';
+import { apiLogsTransport, scopedLogger } from './lib/logger.js';
+import { ApiTransportSettings } from './lib/api-logs-transport.js';
 import { initErrorHandler } from './helper/api-error-handler.js';
 import { handleTestError } from './helper/test-error-handler.js';
 import { apiConnectLocationHandler } from './helper/api-connect-handler.js';
@@ -103,6 +104,8 @@ function connect (workerId?: number) {
 	});
 
 	runStatsAgent(socket, worker);
+	apiLogsTransport.setSocket(socket);
+
 	const statusManager = initStatusManager(socket, pingCmd);
 	const errorHandler = initErrorHandler(socket);
 
@@ -150,7 +153,8 @@ function connect (workerId?: number) {
 				}
 			});
 		})
-		.on('probe:adoption:code', logAdoptionCode);
+		.on('probe:adoption:code', logAdoptionCode)
+		.on('api:logs-transport:set', (data: ApiTransportSettings) => apiLogsTransport.updateSettings(data));
 
 	process.on('SIGTERM', () => {
 		logger.debug('SIGTERM received.');